#!/usr/bin/env python3 """ Export lightweight but analysis-friendly scan outputs for sharing/committing. Each screener gets a compact CSV containing scored names only, with: - shared context fields (ticker, score, sector, industry, price, market cap) - a few screener-specific metrics useful for quick triage Also writes a combined cross-screener file with one row per scored result. """ from __future__ import annotations import csv from pathlib import Path from typing import Dict, Iterable, List ROOT = Path(__file__).parent OUTPUTS_DIR = ROOT / "data" / "outputs" SHARE_DIR = ROOT / "data" / "share" COMMON_COLUMNS = [ "ticker", "total_score", "score_confidence", "sector", "industry", "country", "current_price", "market_cap", ] SOURCES: Dict[str, Dict[str, List[str] | str]] = { "rk": { "filename": "scores_compact.csv", "extra_columns": [ "value_score", "health_score", "sentiment_crowding_score", "quality_momentum_score", "price_to_book", "price_to_sales", "net_cash_to_market_cap", "drawdown_52w", ], }, "burry": { "filename": "burry_compact.csv", "extra_columns": [ "quality_score", "capital_score", "safety_score", "burry_value_score", "roe", "roa", "fcf_margin", "debt_to_equity", ], }, "kulamagi": { "filename": "kulamagi_compact.csv", "extra_columns": [ "momentum_score", "structure_score", "quality_score", "liquidity_score", "return_6m", "return_12m", "pct_from_52w_high", "dollar_volume", ], }, "buffett": { "filename": "buffett_compact.csv", "extra_columns": [ "quality_score", "cash_score", "moat_score", "balance_score", "valuation_score", "roic_5y_median", "fcf_yield", "net_debt_to_ebitda", ], }, } def is_scored(value: str | None) -> bool: if value is None: return False text = str(value).strip() return text not in {"", "nan", "NaN", "None"} def normalize_value(value: str | None) -> str: if value is None: return "" text = str(value).strip() return "" if text in {"nan", "NaN", "None"} else text def unique_columns(columns: Iterable[str]) -> List[str]: ordered: List[str] = [] seen = set() for column in columns: if column in seen: continue seen.add(column) ordered.append(column) return ordered def read_scored_rows(path: Path, columns: List[str]) -> List[Dict[str, str]]: rows: List[Dict[str, str]] = [] seen = set() with path.open(newline="", encoding="utf-8") as handle: for row in csv.DictReader(handle): ticker = str(row.get("ticker", "")).upper().strip() if not ticker or ticker in seen or not is_scored(row.get("total_score")): continue seen.add(ticker) record = {column: normalize_value(row.get(column)) for column in columns} record["ticker"] = ticker rows.append(record) return rows def read_csv_rows(path: Path) -> List[Dict[str, str]]: with path.open(newline="", encoding="utf-8") as handle: return [{k: normalize_value(v) for k, v in row.items()} for row in csv.DictReader(handle)] def write_rows(path: Path, columns: List[str], rows: List[Dict[str, str]]) -> int: with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=columns) writer.writeheader() writer.writerows(rows) return len(rows) def main() -> None: SHARE_DIR.mkdir(parents=True, exist_ok=True) combined_columns = [ "source", *COMMON_COLUMNS, "metric_1_name", "metric_1_value", "metric_2_name", "metric_2_value", "metric_3_name", "metric_3_value", "metric_4_name", "metric_4_value", ] combined_rows: List[Dict[str, str]] = [] for label, cfg in SOURCES.items(): src = OUTPUTS_DIR / str(cfg["filename"]) if not src.exists(): print(f"Skipping {label}: missing {src}") continue extra_columns = list(cfg["extra_columns"]) output_columns = unique_columns([*COMMON_COLUMNS, *extra_columns]) rows = read_scored_rows(src, output_columns) out_path = SHARE_DIR / f"{label}_scored_tickers.csv" count = write_rows(out_path, output_columns, rows) print(f"Wrote {count} rows -> {out_path}") padded_metrics = extra_columns[:4] + [""] * max(0, 4 - len(extra_columns)) for row in rows: combined_row = {column: row.get(column, "") for column in COMMON_COLUMNS} combined_row["source"] = label for idx, metric_name in enumerate(padded_metrics[:4], start=1): combined_row[f"metric_{idx}_name"] = metric_name combined_row[f"metric_{idx}_value"] = row.get(metric_name, "") if metric_name else "" combined_rows.append(combined_row) combined_path = SHARE_DIR / "all_scored_tickers.csv" total = write_rows(combined_path, combined_columns, combined_rows) print(f"Wrote {total} rows -> {combined_path}") export_buffett_diagnostics() def export_buffett_diagnostics() -> None: src = OUTPUTS_DIR / "buffett_compact.csv" if not src.exists(): return rows = read_csv_rows(src) summary_counts: Dict[str, int] = {} for row in rows: reason = row.get("unscorable_reason") or "scored" summary_counts[reason] = summary_counts.get(reason, 0) + 1 summary_rows = [ {"unscorable_reason": reason, "count": str(count)} for reason, count in sorted(summary_counts.items(), key=lambda item: (-item[1], item[0])) ] summary_path = SHARE_DIR / "buffett_unscorable_summary.csv" write_rows(summary_path, ["unscorable_reason", "count"], summary_rows) print(f"Wrote {len(summary_rows)} rows -> {summary_path}") low_conf_columns = [ "ticker", "score_confidence", "missing_pillars", "unscorable_reason", "sector", "industry", "country", "current_price", "market_cap", "quality_score", "cash_score", "moat_score", "balance_score", "valuation_score", "roic_5y_median", "fcf_yield", "net_debt_to_ebitda", ] low_conf_rows = [] for row in rows: if row.get("unscorable_reason") != "low_confidence": continue low_conf_rows.append({column: row.get(column, "") for column in low_conf_columns}) low_conf_path = SHARE_DIR / "buffett_low_confidence.csv" write_rows(low_conf_path, low_conf_columns, low_conf_rows) print(f"Wrote {len(low_conf_rows)} rows -> {low_conf_path}") if __name__ == "__main__": main()