#!/usr/bin/env python3 """ bt_buffett.py — Walk-forward backtest of the Buffett quality-compounder screener True point-in-time: every SEC fact carries the date its 10-K was FILED, so each quarterly rebalance only sees statements that were actually public. No earnings-lag approximation, no static-fundamentals fallback. At each rebalance date (Jan / Apr / Jul / Oct): 1. facts_asof(filed ≤ date) → PIT annual series per ticker 2. compute_sec_metrics() → ROIC history, FCF series, margins, leverage... 3. Historical market cap = price(date) × latest reported share count → PIT enterprise value → PIT FCF yield 4. buffett_tracker.build_scores() → gates + cross-sectional pillar ranks 5. Hold top-N equal-weight for one quarter; costs charged on turnover Shares the transaction-cost model, benchmark and metrics with bt_engine. Caveats (inherited from the data): • Universe = today's listings → survivorship bias (returns overstated) • SEC data covers 10-K filers only (foreign 20-F filers absent) • Sector/industry (for the financials gate) is today's classification Usage: python bt_buffett.py # full available SEC universe python bt_buffett.py --top_n 20 --tc_bps 20 --train_start 2016-01-01 """ from __future__ import annotations import argparse import json import sys import time from pathlib import Path from typing import Any, Dict, List, Optional import numpy as np import pandas as pd from bt_engine import ( BENCHMARK, DEFAULT_TC_BPS, _build_rebalance_dates, compute_metrics, load_prices_panel, portfolio_return, print_metrics, ) from buffett_tracker import ( DEFAULT_GATES, DEFAULT_WEIGHTS, build_scores, compute_sec_metrics, facts_asof, ) # ── Data loading ────────────────────────────────────────────────────────────── def load_sec_cache(sec_dir: Path) -> Dict[str, dict]: """Load every data/sec/{TICKER}.json into memory.""" cache: Dict[str, dict] = {} for path in sorted(sec_dir.glob("*.json")): t = path.stem.upper() if t.startswith("_"): continue try: data = json.loads(path.read_text(encoding="utf-8")) if data.get("tags"): cache[t] = data except Exception: continue return cache def load_classification(outputs_dir: Path) -> pd.DataFrame: """ Static sector/industry/name map for the financials gate and the common-stock filter. Prefers buffett output, falls back to RK scores. (Today's classification — sectors rarely change, but note the caveat.) """ cols = ["ticker", "sector", "industry", "quote_type", "type_disp", "long_name"] for name in ("buffett_compact.csv", "scores_compact.csv"): p = outputs_dir / name if p.exists(): df = pd.read_csv(p) keep = [c for c in cols if c in df.columns] out = df[keep].drop_duplicates("ticker", keep="last") out["ticker"] = out["ticker"].astype(str).str.upper().str.strip() return out.set_index("ticker") return pd.DataFrame(columns=cols[1:]) # ── PIT scoring at one rebalance date ───────────────────────────────────────── def score_at_date( date: pd.Timestamp, sec_cache: Dict[str, dict], close_panel: pd.DataFrame, classification: pd.DataFrame, weights: Dict[str, float], gates: Dict[str, Any], ) -> pd.DataFrame: """Build the PIT feature table for `date` and run gates + pillar scoring.""" as_of = date.strftime("%Y-%m-%d") rows: List[Dict[str, Any]] = [] for t, sec in sec_cache.items(): if t not in close_panel.columns: continue col = close_panel[t].dropna() hist = col[col.index <= date] if hist.empty: continue price = float(hist.iloc[-1]) if price <= 0: continue tags_pit = facts_asof(sec.get("tags", {}), as_of) m = compute_sec_metrics({"tags": tags_pit}) if not m.get("n_years"): continue row: Dict[str, Any] = {"ticker": t, "sec_data": True, **m} # Historical market cap / EV / FCF yield shares = m.get("shares_latest") if shares and shares > 0: mcap = price * shares row["market_cap"] = mcap ev = mcap + (m.get("total_debt") or 0.0) - (m.get("sec_cash") or 0.0) if ev > 0 and m.get("fcf_norm") is not None: row["ev"] = ev row["fcf_yield"] = m["fcf_norm"] / ev if m.get("roic_5y_median") is not None: row["buffett_simple"] = m["roic_5y_median"] * row["fcf_yield"] if t in classification.index: for c in ("sector", "industry", "quote_type", "type_disp", "long_name"): if c in classification.columns: row[c] = classification.loc[t, c] rows.append(row) if not rows: return pd.DataFrame() df = pd.DataFrame(rows) scored = build_scores(df, weights=weights, gates=gates) if "ticker" in scored.columns: scored = scored.set_index("ticker") return scored # ── Backtest loop ───────────────────────────────────────────────────────────── def run_backtest( sec_cache: Dict[str, dict], close_panel: pd.DataFrame, classification: pd.DataFrame, start_date: str, end_date: str, top_n: int = 20, weights: Optional[Dict[str, float]] = None, gates: Optional[Dict[str, Any]] = None, tc_bps: float = DEFAULT_TC_BPS, label: str = "", ) -> pd.DataFrame: weights = weights or DEFAULT_WEIGHTS gates = {**DEFAULT_GATES, **(gates or {})} dates = _build_rebalance_dates(start_date, end_date) if len(dates) < 2: raise ValueError(f"Need ≥ 2 rebalance dates in [{start_date}, {end_date}]") tag = f"[{label}] " if label else "" print(f"\n{tag}Buffett backtest {start_date} → {end_date} " f"({len(dates)-1} quarters, top-{top_n}) [SEC filed-date PIT] [tc {tc_bps:.0f}bps]") rows = [] tc_rate = tc_bps / 1e4 prev_portfolio: Optional[set] = None for i, rb_date in enumerate(dates[:-1]): next_date = dates[i + 1] t0 = time.time() scored = score_at_date(rb_date, sec_cache, close_panel, classification, weights, gates) if scored.empty or "total_score" not in scored.columns: print(f" {rb_date.date()} | no scorable universe — skipping") continue eligible = scored[scored["total_score"].notna()] if len(eligible) < top_n: print(f" {rb_date.date()} | only {len(eligible)} eligible — skipping") continue portfolio = eligible.nlargest(top_n, "total_score").index.tolist() port_set = set(portfolio) if prev_portfolio is None: turnover = 1.0 else: turnover = 2.0 * len(port_set - prev_portfolio) / top_n tc_cost = tc_rate * turnover prev_portfolio = port_set p_gross, ticker_rets = portfolio_return(portfolio, close_panel, rb_date, next_date) if np.isnan(p_gross): p_ret = np.nan elif tc_cost == 0.0: p_ret = p_gross else: p_ret = (1.0 + p_gross) * (1.0 - tc_cost) - 1.0 if BENCHMARK in close_panel.columns: b_ret, _ = portfolio_return([BENCHMARK], close_panel, rb_date, next_date) else: b_ret = np.nan alpha = (p_ret - b_ret) if not (np.isnan(p_ret) or np.isnan(b_ret)) else np.nan n_valid = sum(1 for v in ticker_rets.values() if not np.isnan(v)) sign = "+" if (not np.isnan(alpha) and alpha >= 0) else "" print(f" {rb_date.date()} → {next_date.date()} | " f"net {p_ret:>+7.1%} (gross {p_gross:>+7.1%}, tc {tc_cost:.2%}) " f"SPY {b_ret:>+7.1%} alpha {sign}{alpha:.1%} " f"({n_valid}/{top_n}, univ {len(eligible)}, {time.time()-t0:.1f}s)") rows.append({ "rebalance_date": rb_date, "next_date": next_date, "portfolio_return": p_ret, "portfolio_return_gross": p_gross, "turnover": turnover, "tc_cost": tc_cost, "spy_return": b_ret, "outperformance": alpha, "portfolio_tickers": ",".join(portfolio), "n_valid": n_valid, "n_eligible": len(eligible), }) return pd.DataFrame(rows) # ── CLI ─────────────────────────────────────────────────────────────────────── def main() -> None: ap = argparse.ArgumentParser(description="Walk-forward Buffett screener backtest") ap.add_argument("--sec_dir", default="data/sec") ap.add_argument("--weekly_dir", default="data/weekly") ap.add_argument("--outputs_dir", default="data/outputs") ap.add_argument("--out_dir", default="data/backtest") ap.add_argument("--train_start", default="2015-01-01") ap.add_argument("--train_end", default="2021-12-31") ap.add_argument("--eval_start", default="2022-01-01") ap.add_argument("--eval_end", default=None) ap.add_argument("--top_n", type=int, default=20) ap.add_argument("--tc_bps", type=float, default=DEFAULT_TC_BPS) ap.add_argument("--min_market_cap", type=float, default=None, help="Override the gate (default from buffett_tracker: $1B)") args = ap.parse_args() sec_dir = Path(args.sec_dir) weekly_dir = Path(args.weekly_dir) out_dir = Path(args.out_dir) out_dir.mkdir(parents=True, exist_ok=True) eval_end = args.eval_end or pd.Timestamp.today().strftime("%Y-%m-%d") if not sec_dir.exists(): sys.exit(f"No SEC dir ({sec_dir}/). Run download_sec_fundamentals.py first.") if not weekly_dir.exists(): sys.exit(f"No weekly price dir ({weekly_dir}/). Run download_weekly_history.py first.") print(f"Loading SEC cache ({sec_dir}/)...") t0 = time.time() sec_cache = load_sec_cache(sec_dir) print(f" {len(sec_cache):,} tickers with SEC data ({time.time()-t0:.1f}s)") if len(sec_cache) < 50: print(" WARNING: small SEC universe — results are a smoke test, not a benchmark.") print(f"Loading weekly prices ({weekly_dir}/)...") t0 = time.time() tickers = list(sec_cache.keys()) + [BENCHMARK] close_panel = load_prices_panel(weekly_dir, tickers) print(f" {close_panel.shape[1]:,} tickers {close_panel.shape[0]:,} weeks " f"({time.time()-t0:.1f}s)") if BENCHMARK not in close_panel.columns: sys.exit(f"{BENCHMARK}.csv missing from {weekly_dir}/ — needed as benchmark.") classification = load_classification(Path(args.outputs_dir)) gates = {} if args.min_market_cap is not None: gates["min_market_cap"] = args.min_market_cap results = {} for label, s, e in (("IN-SAMPLE", args.train_start, args.train_end), ("OUT-OF-SAMPLE", args.eval_start, eval_end)): res = run_backtest( sec_cache=sec_cache, close_panel=close_panel, classification=classification, start_date=s, end_date=e, top_n=args.top_n, tc_bps=args.tc_bps, gates=gates, label=label, ) results[label] = res m = compute_metrics(res, label=f"{label} ({s} – {e})") print_metrics(m) all_res = pd.concat( [results["IN-SAMPLE"].assign(split="train"), results["OUT-OF-SAMPLE"].assign(split="test")], ignore_index=True, ) all_res.to_csv(out_dir / "buffett_results_quarterly.csv", index=False) print(f"\n Results → {out_dir}/buffett_results_quarterly.csv") print(f" Compare with the RK screener: python bt_engine.py --strict_pit\n") if __name__ == "__main__": main()