#!/usr/bin/env python3 """ buffett_tracker.py — Buffett-style quality-compounder screener (SEC edition) Philosophy: high-quality business + durable cash generation + low financial risk + sensible valuation. Cash-flow and return-on-capital based, never price-chart based. Two-stage design: Stage 1 — eligibility gates (ruthless elimination): • ≥ 4 fiscal years of SEC data • FCF positive in ≥ 60% of available years (up to 10) • net debt / EBITDA ≤ 4 • share count CAGR ≤ 5%/yr (no serial diluters) • financials excluded (banks/insurers need different metrics) • market cap ≥ $1B (configurable) Stage 2 — percentile-ranked pillars among survivors (weights sum to 100): Quality 35% — 5Y median ROIC, ROIC ex-goodwill, ROIC stability Cash 20% — FCF conversion, capex burden, FCF consistency Moat 15% — gross & operating margin stability (5-10Y) Balance 15% — net debt/EBITDA, interest coverage Valuation 15% — normalized FCF yield (median 3Y FCF / EV) Percentile ranks (not the raw multiplicative score) are used deliberately: a product of ratios is fragile to negative FCF, tiny denominators and missing data. The raw "ROIC × FCF yield" is still reported per stock as `buffett_simple` for reference. Data: data/sec/{TICKER}.json (run download_sec_fundamentals.py first) + the shared yfinance info cache for market cap / sector / country. Run: python download_sec_fundamentals.py --universe_csv us_universe.csv python buffett_tracker.py --config config.yaml --universe_csv us_universe.csv --resume python buffett_tracker.py --config config.yaml --finalize_only """ from __future__ import annotations import argparse import datetime as dt import json import threading import time from pathlib import Path from typing import Any, Dict, List, Optional import numpy as np import pandas as pd from rk_tracker import ( YFCache, append_jsonl, enrich_classification_from_cache, ensure_dir, fetch_info, filter_common_stock_rows, load_universe_from_csv, nanmean_rows, pct_rank, read_jsonl_latest, read_jsonl_tickers, read_yaml, safe_float, safe_text, weighted_score_from_available_pillars, ) DEFAULT_SEC_DIR = "data/sec" DEFAULT_WEIGHTS = {"quality": 35, "cash": 20, "moat": 15, "balance": 15, "valuation": 15} DEFAULT_GATES = { "min_years": 4, "min_fcf_positive": 0.60, # fraction of available years "max_net_debt_ebitda": 4.0, "max_share_cagr": 0.05, "min_market_cap": 1e9, "exclude_financials": True, "exclude_reits": True, "min_score_confidence": 0.75, # demote names scored on too few pillars } # Tag alternatives, first hit wins per year (later alts only fill gaps) _REVENUE = ["Revenues", "RevenueFromContractWithCustomerExcludingAssessedTax", "SalesRevenueNet"] _GROSS = ["GrossProfit"] _OPINC = ["OperatingIncomeLoss"] _NETINC = ["NetIncomeLoss"] _TAX = ["IncomeTaxExpenseBenefit"] _INT = ["InterestExpense"] _DA = ["DepreciationDepletionAndAmortization", "DepreciationAndAmortization", "Depreciation"] _OCF = ["NetCashProvidedByUsedInOperatingActivities", "NetCashProvidedByUsedInOperatingActivitiesContinuingOperations"] _CAPEX = ["PaymentsToAcquirePropertyPlantAndEquipment", "PaymentsToAcquireProductiveAssets", "PaymentsToAcquireOtherProductiveAssets"] _EQUITY = ["StockholdersEquity", "StockholdersEquityIncludingPortionAttributableToNoncontrollingInterest"] _GOODW = ["Goodwill"] _LT_DEBT = ["LongTermDebtAndCapitalLeaseObligations", "LongTermDebtNoncurrent", "LongTermDebt"] _CU_DEBT = ["LongTermDebtAndCapitalLeaseObligationsCurrent", "LongTermDebtCurrent", "DebtCurrent", "ShortTermBorrowings", "CommercialPaper"] _CASH = ["CashAndCashEquivalentsAtCarryingValue"] _SHARES = ["WeightedAverageNumberOfDilutedSharesOutstanding", "WeightedAverageNumberOfSharesOutstandingBasic", "CommonStockSharesOutstanding", "EntityCommonStockSharesOutstanding"] _FINANCIAL_INDUSTRY_HINTS = ( "bank", "insur", "capital markets", "credit services", "asset management", "financial data", "mortgage", ) # --------------------------------------------------------------------------- # SEC annual series helpers # --------------------------------------------------------------------------- def load_sec(sec_dir: Path, ticker: str) -> Optional[dict]: path = sec_dir / f"{ticker}.json" if not path.exists(): return None try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return None def facts_asof(tags: dict, as_of: str) -> dict: """ Point-in-time view of a ticker's SEC facts: keep only entries whose 10-K was FILED on or before `as_of` ("YYYY-MM-DD"). Entries without a filed date are dropped (conservative — unknowable is unusable). This is true PIT — no earnings-lag approximation needed. """ return { tag: [e for e in rows if e.get("filed") and str(e["filed"]) <= as_of] for tag, rows in tags.items() } def annual_series(tags: dict, alts: List[str]) -> Dict[int, float]: """ Build {fiscal_year_end_year: value}. Iterates alternatives in order; later alternatives only fill years the earlier ones didn't cover. Within a tag, the most recently *filed* value per year wins (restatements supersede originals). """ out: Dict[int, float] = {} for alt in alts: best_filed: Dict[int, str] = {} for e in tags.get(alt, []): end, val = e.get("end"), e.get("val") if end is None or val is None: continue try: year, fval = int(end[:4]), float(val) except (ValueError, TypeError): continue filed = str(e.get("filed") or "") if year in out and year not in best_filed: continue # year already covered by earlier alt if year not in best_filed or filed > best_filed[year]: best_filed[year] = filed out[year] = fval return out def _last(series: Dict[int, float], years: List[int]) -> Optional[float]: for y in years: if y in series: return series[y] return None def _median(vals: List[float]) -> Optional[float]: vals = [v for v in vals if v is not None and np.isfinite(v)] return float(np.median(vals)) if vals else None def _std(vals: List[float]) -> Optional[float]: vals = [v for v in vals if v is not None and np.isfinite(v)] return float(np.std(vals)) if len(vals) >= 3 else None # --------------------------------------------------------------------------- # Per-ticker fundamental computation # --------------------------------------------------------------------------- def compute_sec_metrics(sec: dict) -> Dict[str, Any]: """All Buffett metrics derivable from the SEC annual series.""" tags = sec.get("tags", {}) m: Dict[str, Any] = {} rev = annual_series(tags, _REVENUE) gross = annual_series(tags, _GROSS) opinc = annual_series(tags, _OPINC) netinc = annual_series(tags, _NETINC) tax = annual_series(tags, _TAX) intex = annual_series(tags, _INT) dna = annual_series(tags, _DA) ocf = annual_series(tags, _OCF) capex = annual_series(tags, _CAPEX) equity = annual_series(tags, _EQUITY) goodw = annual_series(tags, _GOODW) ltd = annual_series(tags, _LT_DEBT) cud = annual_series(tags, _CU_DEBT) cash = annual_series(tags, _CASH) years = sorted(set(opinc) | set(netinc) | set(ocf), reverse=True) m["n_years"] = len(years) if not years: return m y10 = years[:10] y5 = years[:5] y0 = years[0] def debt(y: int) -> float: return (ltd.get(y) or 0.0) + (cud.get(y) or 0.0) # ── ROIC per year: NOPAT / invested capital ─────────────────────────── roics, roics_xgw = [], [] for y in y5: oi, eq = opinc.get(y), equity.get(y) if oi is None or eq is None: continue ni, tx = netinc.get(y), tax.get(y) if ni is not None and tx is not None and (ni + tx) > 0: rate = min(max(tx / (ni + tx), 0.0), 0.45) else: rate = 0.21 nopat = oi * (1.0 - rate) # Winsorize yearly ROIC to [-100%, +200%]: near-zero invested capital # (heavy buybacks → tiny equity, e.g. MCK/HD) makes raw ROIC explode. ic = eq + debt(y) - (cash.get(y) or 0.0) if ic > 0: roics.append(min(max(nopat / ic, -1.0), 2.0)) ic_x = ic - (goodw.get(y) or 0.0) if ic_x > 0: roics_xgw.append(min(max(nopat / ic_x, -1.0), 2.0)) m["roic_5y_median"] = _median(roics) m["roic_xgw_5y_median"] = _median(roics_xgw) m["roic_std"] = _std(roics) # ── Free cash flow series ────────────────────────────────────────────── # Require capex data for the year — treating missing capex as zero # silently inflates FCF when a filer switches capex tags (e.g. VZ). fcf = {y: ocf[y] - abs(capex[y]) for y in ocf if y in capex} fcf_hist = [fcf[y] for y in y10 if y in fcf] m["fcf_years_available"] = len(fcf_hist) m["fcf_positive_ratio"] = ( sum(1 for v in fcf_hist if v > 0) / len(fcf_hist) if fcf_hist else None ) m["fcf_norm"] = _median([fcf[y] for y in years[:3] if y in fcf]) ni_5 = [netinc[y] for y in y5 if y in netinc] fcf_5 = [fcf[y] for y in y5 if y in fcf] if fcf_5 and ni_5 and sum(ni_5) > 0: m["fcf_conversion"] = sum(fcf_5) / sum(ni_5) burdens = [abs(capex.get(y) or 0.0) / ocf[y] for y in y5 if ocf.get(y, 0) > 0] m["capex_burden"] = _median(burdens) # ── Margin stability (moat proxy) ────────────────────────────────────── gms = [gross[y] / rev[y] for y in y10 if y in gross and rev.get(y, 0) > 0] oms = [opinc[y] / rev[y] for y in y10 if y in opinc and rev.get(y, 0) > 0] m["gross_margin"] = gms[0] if gms else None m["gm_std"] = _std(gms) m["om_std"] = _std(oms) m["operating_margin"] = oms[0] if oms else None # ── Balance sheet ────────────────────────────────────────────────────── ebitda = None if opinc.get(y0) is not None: ebitda = opinc[y0] + (dna.get(y0) or 0.0) net_debt = debt(y0) - (cash.get(y0) or 0.0) m["net_debt"] = net_debt m["ebitda"] = ebitda m["total_debt"] = debt(y0) m["sec_cash"] = cash.get(y0) if ebitda and ebitda > 0: m["net_debt_to_ebitda"] = net_debt / ebitda if intex.get(y0) and intex[y0] > 0 and opinc.get(y0) is not None: m["interest_coverage"] = opinc[y0] / intex[y0] # ── Share count discipline ───────────────────────────────────────────── # Do NOT mix share tags across years (weighted-diluted vs point-in-time # entity shares are different concepts — mixing fakes jumps). Use the # first tag that alone covers ≥ 3 years. shares = {} for alt in _SHARES: cand = annual_series(tags, [alt]) if len([y for y in cand if y in y10]) >= 3: shares = cand break sh_years = sorted([y for y in shares if y in y10]) if sh_years: m["shares_latest"] = shares[sh_years[-1]] # for historical mcap in backtests if len(sh_years) >= 3: # Split-robust dilution: median of adjacent-year growth ratios. # Historical filings report PRE-split counts, so an endpoint CAGR sees # a 4:1 split (AAPL 2020) as 300% dilution. A split shows up as one # outlier ratio; the median ignores it. Ratios beyond 1.8x / 0.55x # (likely splits/reverse splits) are also dropped outright. ratios = [] for a, b in zip(sh_years, sh_years[1:]): if shares[a] > 0 and (b - a) <= 2: r = (shares[b] / shares[a]) ** (1.0 / (b - a)) if 0.55 < r < 1.8: ratios.append(r) if len(ratios) >= 2: m["share_cagr"] = float(np.median(ratios)) - 1.0 # ── Growth (context, lightly weighted via quality only) ──────────────── rev_years = sorted([y for y in rev if y in y10]) if len(rev_years) >= 3 and rev[rev_years[0]] > 0: span = rev_years[-1] - rev_years[0] if span >= 2: m["revenue_cagr"] = (rev[rev_years[-1]] / rev[rev_years[0]]) ** (1.0 / span) - 1.0 return m def compute_row(cache: YFCache, ticker: str, sec_dir: Path) -> Dict[str, Any]: ticker_u = ticker.upper() row: Dict[str, Any] = {"ticker": ticker_u} info = fetch_info(cache, ticker_u) row["sector"] = safe_text(info.get("sector")) row["industry"] = safe_text(info.get("industry")) row["quote_type"] = safe_text(info.get("quoteType")) row["type_disp"] = safe_text(info.get("typeDisp")) row["long_name"] = safe_text(info.get("longName") or info.get("shortName")) row["country"] = safe_text(info.get("country")) row["current_price"] = safe_float(info.get("currentPrice")) row["market_cap"] = safe_float(info.get("marketCap")) sec = load_sec(sec_dir, ticker_u) if sec is None: row["sec_data"] = False return row row["sec_data"] = True row.update(compute_sec_metrics(sec)) # ── Valuation: normalized FCF yield on enterprise value ──────────────── mcap = row.get("market_cap") if mcap and mcap > 0 and row.get("fcf_norm") is not None: ev = mcap + (row.get("total_debt") or 0.0) - (row.get("sec_cash") or 0.0) if ev > 0: row["ev"] = ev row["fcf_yield"] = row["fcf_norm"] / ev # The compact reference score (reported, not used for ranking) if row.get("roic_5y_median") is not None and row.get("fcf_yield") is not None: row["buffett_simple"] = row["roic_5y_median"] * row["fcf_yield"] return row def load_cached_info(cache_dir: Path, ticker: str) -> Dict[str, Any]: path = cache_dir / f"info_{ticker.upper()}.json" if not path.exists(): return {} try: raw = json.loads(path.read_text(encoding="utf-8")) except Exception: return {} return raw if isinstance(raw, dict) else {} def compute_row_from_local_cache(cache_dir: Path, ticker: str, sec_dir: Path) -> Dict[str, Any]: ticker_u = ticker.upper() row: Dict[str, Any] = {"ticker": ticker_u} info = load_cached_info(cache_dir, ticker_u) row["sector"] = safe_text(info.get("sector")) row["industry"] = safe_text(info.get("industry")) row["quote_type"] = safe_text(info.get("quoteType")) row["type_disp"] = safe_text(info.get("typeDisp")) row["long_name"] = safe_text(info.get("longName") or info.get("shortName")) row["country"] = safe_text(info.get("country")) row["current_price"] = safe_float(info.get("currentPrice")) row["market_cap"] = safe_float(info.get("marketCap")) sec = load_sec(sec_dir, ticker_u) if sec is None: row["sec_data"] = False return row row["sec_data"] = True row.update(compute_sec_metrics(sec)) mcap = row.get("market_cap") if mcap and mcap > 0 and row.get("fcf_norm") is not None: ev = mcap + (row.get("total_debt") or 0.0) - (row.get("sec_cash") or 0.0) if ev > 0: row["ev"] = ev row["fcf_yield"] = row["fcf_norm"] / ev if row.get("roic_5y_median") is not None and row.get("fcf_yield") is not None: row["buffett_simple"] = row["roic_5y_median"] * row["fcf_yield"] return row # --------------------------------------------------------------------------- # Stage 1: eligibility gates # --------------------------------------------------------------------------- def _is_financial(sector: Any, industry: Any) -> bool: s = str(sector or "").lower() i = str(industry or "").lower() if "financial" in s: return True return any(h in i for h in _FINANCIAL_INDUSTRY_HINTS) def apply_gates(df: pd.DataFrame, gates: Dict[str, Any]) -> pd.DataFrame: """Set gate_reason (None = passes all gates).""" out = df.copy() reasons = pd.Series([None] * len(out), index=out.index, dtype=object) def flag(mask: pd.Series, reason: str) -> None: mask = mask.fillna(False) if mask.dtype == object else mask reasons.loc[mask & reasons.isna()] = reason if not {"sec_data"}.issubset(out.columns): out["gate_reason"] = "no_sec_data" return out flag(out["sec_data"] != True, "no_sec_data") flag(out.get("n_years", pd.Series(0, index=out.index)).fillna(0) < float(gates["min_years"]), "insufficient_history") if gates.get("exclude_financials", True): fin = out.apply(lambda r: _is_financial(r.get("sector"), r.get("industry")), axis=1) flag(fin, "financial_sector") if gates.get("exclude_reits", True): sec_s = out.get("sector", pd.Series(None, index=out.index)).astype(str).str.lower() ind_s = out.get("industry", pd.Series(None, index=out.index)).astype(str).str.lower() flag(sec_s.str.contains("real estate") | ind_s.str.contains("reit"), "reit") # FCF/ROIC misleading for REITs (depreciation-heavy; needs FFO) # Unknown market cap → exclude. Exchange-traded notes / baby bonds (HWCPZ, # MGR-series...) map to the parent's SEC filings but have no equity market # cap; without this gate they'd sail through with the parent's fundamentals. mc = out.get("market_cap", pd.Series(np.nan, index=out.index)) flag(mc.isna() & (out["sec_data"] == True), "no_market_cap") flag(mc.notna() & (mc < float(gates["min_market_cap"])), "below_min_market_cap") fpr = out.get("fcf_positive_ratio", pd.Series(np.nan, index=out.index)) flag(fpr.notna() & (fpr < float(gates["min_fcf_positive"])), "inconsistent_fcf") flag(fpr.isna() & (out["sec_data"] == True), "no_fcf_history") nde = out.get("net_debt_to_ebitda", pd.Series(np.nan, index=out.index)) flag(nde.notna() & (nde > float(gates["max_net_debt_ebitda"])), "over_levered") sc = out.get("share_cagr", pd.Series(np.nan, index=out.index)) flag(sc.notna() & (sc > float(gates["max_share_cagr"])), "serial_diluter") out["gate_reason"] = reasons return out # --------------------------------------------------------------------------- # Stage 2: pillar scoring (percentile ranks among survivors) # --------------------------------------------------------------------------- def build_scores(df: pd.DataFrame, weights: Dict[str, float], gates: Optional[Dict[str, Any]] = None) -> pd.DataFrame: gates = {**DEFAULT_GATES, **(gates or {})} out = filter_common_stock_rows(df.copy()) if out.empty: return out out = apply_gates(out, gates) eligible = out[out["gate_reason"].isna()].copy() ineligible = out[out["gate_reason"].notna()].copy() ineligible["unscorable_reason"] = ineligible["gate_reason"] if eligible.empty: ineligible["total_score"] = np.nan return ineligible e = eligible # ── Quality (35) ─────────────────────────────────────────────────────── e["_q_roic"] = pct_rank(e["roic_5y_median"], ascending=True) * 100 e["_q_roicx"] = pct_rank(e.get("roic_xgw_5y_median"), ascending=True) * 100 e["_q_stab"] = (1.0 - pct_rank(e.get("roic_std"), ascending=True)) * 100 e["quality_score"] = nanmean_rows(e["_q_roic"], e["_q_roicx"], e["_q_stab"]) # ── Cash discipline (20) ─────────────────────────────────────────────── conv = e.get("fcf_conversion") conv = conv.clip(lower=-1.0, upper=2.0) if conv is not None else None e["_c_conv"] = pct_rank(conv, ascending=True) * 100 e["_c_burden"] = (1.0 - pct_rank(e.get("capex_burden"), ascending=True)) * 100 e["_c_streak"] = pct_rank(e.get("fcf_positive_ratio"), ascending=True) * 100 e["cash_score"] = nanmean_rows(e["_c_conv"], e["_c_burden"], e["_c_streak"]) # ── Moat proxy: margin stability (15) ────────────────────────────────── e["_m_gm"] = (1.0 - pct_rank(e.get("gm_std"), ascending=True)) * 100 e["_m_om"] = (1.0 - pct_rank(e.get("om_std"), ascending=True)) * 100 e["moat_score"] = nanmean_rows(e["_m_gm"], e["_m_om"]) # ── Balance sheet (15) ───────────────────────────────────────────────── # net debt/EBITDA: lower (or negative = net cash) is better e["_b_nde"] = (1.0 - pct_rank(e.get("net_debt_to_ebitda"), ascending=True)) * 100 cov = e.get("interest_coverage") cov = cov.clip(upper=50.0) if cov is not None else None e["_b_cov"] = pct_rank(cov, ascending=True) * 100 e["balance_score"] = nanmean_rows(e["_b_nde"], e["_b_cov"]) # ── Valuation (15) ───────────────────────────────────────────────────── e["_v_fcfy"] = pct_rank(e.get("fcf_yield"), ascending=True) * 100 e["valuation_score"] = e["_v_fcfy"] w = weights e = weighted_score_from_available_pillars(e, [ ("quality_score", "Quality", float(w.get("quality", 35))), ("cash_score", "Cash", float(w.get("cash", 20))), ("moat_score", "Moat", float(w.get("moat", 15))), ("balance_score", "Balance", float(w.get("balance", 15))), ("valuation_score", "Valuation", float(w.get("valuation", 15))), ]) e["total_score"] = e["total_score"].clip(lower=0, upper=100) e = e.drop(columns=[c for c in e.columns if c.startswith("_")]) # ── Post-scoring demotions ───────────────────────────────────────────── # 1. Confidence floor: weights renormalize over available pillars, so a # name with one strong pillar and no data for the rest can top the # list (e.g. RRC ranked #1 on 20% coverage). Demote, don't rank. min_conf = float(gates.get("min_score_confidence", 0.0)) if min_conf > 0 and "score_confidence" in e.columns: low = e["score_confidence"].fillna(0) < min_conf if low.any(): demoted = e[low].copy() demoted["unscorable_reason"] = "low_confidence" demoted["total_score"] = np.nan ineligible = pd.concat([ineligible, demoted], ignore_index=True) e = e[~low].copy() # 2. Dual-class dedup: keep one line per company (larger market cap wins; # LEN vs LEN-B, BRK-A vs BRK-B...). import re as _re tickers = set(e["ticker"]) drop: List[str] = [] for t in tickers: mobj = _re.match(r"^(.+)-[AB]$", t) if mobj and mobj.group(1) in tickers: pair = e[e["ticker"].isin([mobj.group(1), t])] keep = pair.sort_values("market_cap", ascending=False)["ticker"].iloc[0] drop.extend(x for x in (mobj.group(1), t) if x != keep) if drop: dup = e[e["ticker"].isin(drop)].copy() dup["unscorable_reason"] = "dual_class_duplicate" dup["total_score"] = np.nan ineligible = pd.concat([ineligible, dup], ignore_index=True) e = e[~e["ticker"].isin(drop)].copy() combined = pd.concat([e, ineligible], axis=0, ignore_index=True) return combined # --------------------------------------------------------------------------- # Output writing # --------------------------------------------------------------------------- COMPACT_COLS = [ "ticker", "total_score", "score_confidence", "missing_pillars", "unscorable_reason", "quality_score", "cash_score", "moat_score", "balance_score", "valuation_score", "sector", "industry", "quote_type", "type_disp", "country", "roic_5y_median", "roic_xgw_5y_median", "roic_std", "fcf_yield", "fcf_conversion", "fcf_positive_ratio", "capex_burden", "gross_margin", "gm_std", "operating_margin", "om_std", "net_debt_to_ebitda", "interest_coverage", "share_cagr", "revenue_cagr", "buffett_simple", "n_years", "gate_reason", "current_price", "market_cap", ] def write_outputs(scored: pd.DataFrame, outputs_dir: Path, suffix: str = "") -> None: scored.to_csv(outputs_dir / f"buffett{suffix}.csv", index=False) cols = [c for c in COMPACT_COLS if c in scored.columns] scored[cols].to_csv(outputs_dir / f"buffett_compact{suffix}.csv", index=False) # --------------------------------------------------------------------------- # Pipeline orchestration (mirrors burry_tracker) # --------------------------------------------------------------------------- def _cfg_block(cfg: Dict) -> Dict: return cfg.get("buffett", {}) def _jsonl_path(cfg: Dict, outputs_dir: Path) -> Path: return outputs_dir / _cfg_block(cfg).get("results_jsonl", "buffett_rows.jsonl") def _local_finalize_tickers(cfg: Dict, sec_dir: Path) -> List[str]: universe_csv = Path("us_universe.csv") if universe_csv.exists(): return load_universe_from_csv(universe_csv) universe = [str(t).upper().strip() for t in cfg.get("universe", []) if str(t).strip()] if universe: return sorted(set(universe)) return sorted({p.stem.upper() for p in sec_dir.glob("*.json") if p.is_file()}) def finalize_from_jsonl(config_path: str) -> Path: cfg = read_yaml(config_path) cache_dir = ensure_dir(cfg.get("cache_dir", "data/raw")) outputs_dir = ensure_dir(cfg.get("outputs_dir", "data/outputs")) sec_dir = Path(_cfg_block(cfg).get("sec_dir", DEFAULT_SEC_DIR)) jsonl_path = _jsonl_path(cfg, outputs_dir) if not jsonl_path.exists(): tickers = _local_finalize_tickers(cfg, sec_dir) if not tickers: raise FileNotFoundError(f"Missing {jsonl_path} and no local SEC cache found in {sec_dir}.") print(f"Missing {jsonl_path}; rebuilding Buffett rows from local SEC/info cache for {len(tickers)} tickers.") df = pd.DataFrame(compute_row_from_local_cache(cache_dir, ticker, sec_dir) for ticker in tickers) else: df = read_jsonl_latest(jsonl_path) df = enrich_classification_from_cache(df, cache_dir) block = _cfg_block(cfg) weights = block.get("scoring_weights", DEFAULT_WEIGHTS) gates = block.get("gates", {}) scored = build_scores(df, weights=weights, gates=gates) if "total_score" not in scored.columns: scored["total_score"] = np.nan scored = scored.sort_values(["total_score", "ticker"], ascending=[False, True], na_position="last") write_outputs(scored, outputs_dir) out_path = outputs_dir / "buffett_compact.csv" n_scored = scored["total_score"].notna().sum() if "total_score" in scored.columns else 0 print(f"Wrote {len(scored)} stocks ({n_scored} scored, rest gated out) → {out_path}") return out_path def run(config_path: str, universe_csv: Optional[str] = None, resume: bool = False) -> Path: cfg = read_yaml(config_path) cache_dir = ensure_dir(cfg.get("cache_dir", "data/raw")) outputs_dir = ensure_dir(cfg.get("outputs_dir", "data/outputs")) block = _cfg_block(cfg) weights = block.get("scoring_weights", DEFAULT_WEIGHTS) gates = block.get("gates", {}) sec_dir = Path(block.get("sec_dir", DEFAULT_SEC_DIR)) checkpoint_every = int(cfg.get("checkpoint_every", 25)) min_delay = float(cfg.get("min_delay_seconds", 0.5)) max_workers = int(cfg.get("max_workers", 5)) results_jsonl = _jsonl_path(cfg, outputs_dir) if not sec_dir.exists() or not any(sec_dir.glob("*.json")): raise SystemExit( f"No SEC data in {sec_dir}/ — run:\n" f" python download_sec_fundamentals.py --universe_csv us_universe.csv" ) if universe_csv: universe = load_universe_from_csv(universe_csv) else: universe = [str(t).upper().strip() for t in cfg.get("universe", []) if str(t).strip()] if not universe: raise ValueError("Universe is empty.") if resume: done = read_jsonl_tickers(results_jsonl) else: done = set() if results_jsonl.exists(): results_jsonl.write_text("") todo = [t.upper().strip() for t in universe if t.upper().strip() and t.upper().strip() not in done] if not todo: return finalize_from_jsonl(config_path) cache = YFCache(cache_dir, min_delay_seconds=min_delay) from concurrent.futures import ThreadPoolExecutor, as_completed file_lock = threading.Lock() processed_lock = threading.Lock() processed_count = 0 total_todo = len(todo) print(f" → Screening {total_todo} tickers using {max_workers} threads...") def process_ticker(t: str): nonlocal processed_count t0 = time.time() try: row = compute_row(cache, t, sec_dir) except Exception as e: row = {"ticker": t, "error": str(e)} row["processed_utc"] = dt.datetime.utcnow().isoformat() with file_lock: append_jsonl(results_jsonl, row) dur = time.time() - t0 with processed_lock: processed_count += 1 curr = processed_count if "error" in row: print(f"[{curr}/{total_todo}] {t} ERROR {dur:.1f}s: {row['error']}") else: print(f"[{curr}/{total_todo}] {t} ok {dur:.1f}s") if checkpoint_every > 0 and curr % checkpoint_every == 0: try: with file_lock: tmp = read_jsonl_latest(results_jsonl) tmp = enrich_classification_from_cache(tmp, cache_dir) sc = build_scores(tmp, weights=weights, gates=gates) sc = sc.sort_values(["total_score", "ticker"], ascending=[False, True], na_position="last") with file_lock: write_outputs(sc, outputs_dir) print(f" → checkpoint after {curr} tickers (dashboard updated)") except Exception as e: print(f" → checkpoint failed: {e}") with ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [executor.submit(process_ticker, t) for t in todo] for fut in as_completed(futures): try: fut.result() except Exception as e: print(f" → Thread execution error: {e}") return finalize_from_jsonl(config_path) # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": parser = argparse.ArgumentParser(description="Buffett-style quality compounder screener") parser.add_argument("--config", default="config.yaml") parser.add_argument("--universe_csv", default=None) parser.add_argument("--resume", action="store_true") parser.add_argument("--finalize_only", action="store_true") args = parser.parse_args() if args.finalize_only: finalize_from_jsonl(args.config) else: run(args.config, universe_csv=args.universe_csv, resume=args.resume)