#!/usr/bin/env python3 """ burry_tracker.py — Michael Burry-style deep value screener (yfinance edition) Screens for fundamentally strong, conservatively financed companies trading at a discount — inspired by Burry's documented approach: -> High and consistent ROE / ROA -> Strong cash flow relative to capital expenditure -> Equity growth that validates the ROE numbers -> Consistent earnings growth -> Conservative debt and low dividend payout ratio -> Cheap valuation (low P/E, P/B) Scoring pillars (weights sum to 100): Quality 35% — ROE, ROA, net margin, operating margin Capital 25% — CF/CapEx ratio, FCF margin, OCF margin Safety 25% — debt/equity, payout ratio, current ratio Value 15% — trailing P/E, price/book, earnings growth Reuses the same yfinance cache as rk_tracker.py — no extra API calls for already-cached tickers. Run: python burry_tracker.py --config config.yaml --universe_csv us_universe.csv --resume python burry_tracker.py --config config.yaml --finalize_only """ from __future__ import annotations import argparse import datetime as dt import threading import time from pathlib import Path from typing import Any, Dict, Optional 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, ) # --------------------------------------------------------------------------- # Per-ticker row computation # --------------------------------------------------------------------------- def compute_row(cache: YFCache, ticker: str) -> 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")) # ── Price / market data ──────────────────────────────────────────────── row["current_price"] = safe_float(info.get("currentPrice")) row["market_cap"] = safe_float(info.get("marketCap")) # ── Quality metrics ──────────────────────────────────────────────────── row["roe"] = safe_float(info.get("returnOnEquity")) row["roa"] = safe_float(info.get("returnOnAssets")) row["profit_margin"] = safe_float(info.get("profitMargins")) row["operating_margin"] = safe_float(info.get("operatingMargins")) row["gross_margin"] = safe_float(info.get("grossMargins")) # ── Capital efficiency ───────────────────────────────────────────────── op_cf = safe_float(info.get("operatingCashflow")) capex = safe_float(info.get("capitalExpenditures")) # typically negative fcf = safe_float(info.get("freeCashflow")) rev = safe_float(info.get("totalRevenue")) # CF/CapEx: operating cash flow divided by absolute capital expenditure if op_cf is not None and capex is not None and capex != 0: row["cf_to_capex"] = op_cf / abs(capex) else: row["cf_to_capex"] = None # FCF margin: free cash flow / total revenue if fcf is not None and rev is not None and rev > 0: row["fcf_margin"] = fcf / rev else: row["fcf_margin"] = None # OCF margin: operating cash flow / total revenue if op_cf is not None and rev is not None and rev > 0: row["ocf_margin"] = op_cf / rev else: row["ocf_margin"] = None # ── Balance sheet / safety ───────────────────────────────────────────── de_raw = safe_float(info.get("debtToEquity")) # yfinance sometimes returns D/E as a percentage (150 = 1.5x) — normalise if de_raw is not None: row["debt_to_equity"] = de_raw / 100.0 if de_raw > 20 else de_raw else: row["debt_to_equity"] = None row["payout_ratio"] = safe_float(info.get("payoutRatio")) row["current_ratio"] = safe_float(info.get("currentRatio")) # ── Value metrics ────────────────────────────────────────────────────── pe = safe_float(info.get("trailingPE")) row["trailing_pe"] = pe if (pe is not None and pe > 0) else None row["price_to_book"] = safe_float(info.get("priceToBook")) row["earnings_growth"] = safe_float( info.get("earningsGrowth") or info.get("earningsQuarterlyGrowth") ) row["revenue_growth"] = safe_float(info.get("revenueGrowth")) return row # --------------------------------------------------------------------------- # Scoring — 4 Burry pillars # --------------------------------------------------------------------------- def build_scores(df: pd.DataFrame, weights: Dict[str, float]) -> pd.DataFrame: out = filter_common_stock_rows(df.copy()) if out.empty: return out # ── Pillar 1: Quality (35%) — higher is always better ───────────────── out["_q_roe"] = pct_rank(out["roe"], ascending=True) * 100 out["_q_roa"] = pct_rank(out["roa"], ascending=True) * 100 out["_q_pm"] = pct_rank(out["profit_margin"], ascending=True) * 100 out["_q_om"] = pct_rank(out["operating_margin"], ascending=True) * 100 out["quality_score"] = nanmean_rows( out["_q_roe"], out["_q_roa"], out["_q_pm"], out["_q_om"] ) # ── Pillar 2: Capital efficiency (25%) ──────────────────────────────── # Cap CF/CapEx at 50× to prevent outliers from dominating the ranking cf_cap = out["cf_to_capex"].clip(upper=50) out["_c_cfcapex"] = pct_rank(cf_cap, ascending=True) * 100 out["_c_fcf"] = pct_rank(out["fcf_margin"], ascending=True) * 100 out["_c_ocf"] = pct_rank(out["ocf_margin"], ascending=True) * 100 out["capital_score"] = nanmean_rows( out["_c_cfcapex"], out["_c_fcf"], out["_c_ocf"] ) # ── Pillar 3: Safety (25%) ───────────────────────────────────────────── # D/E: lower is better out["_s_de"] = (1.0 - pct_rank(out["debt_to_equity"], ascending=True)) * 100 # Payout ratio: lower is better (cap at 1.0 = 100%) pr_capped = out["payout_ratio"].clip(upper=1.0) out["_s_pr"] = (1.0 - pct_rank(pr_capped, ascending=True)) * 100 # Current ratio: higher is better out["_s_cr"] = pct_rank(out["current_ratio"], ascending=True) * 100 out["safety_score"] = nanmean_rows( out["_s_de"], out["_s_pr"], out["_s_cr"] ) # ── Pillar 4: Value (15%) ────────────────────────────────────────────── # P/E and P/B: lower is better out["_v_pe"] = (1.0 - pct_rank(out["trailing_pe"], ascending=True)) * 100 out["_v_pb"] = (1.0 - pct_rank(out["price_to_book"], ascending=True)) * 100 out["_v_eg"] = pct_rank(out["earnings_growth"], ascending=True) * 100 out["burry_value_score"] = nanmean_rows( out["_v_pe"], out["_v_pb"], out["_v_eg"] ) w = weights out = weighted_score_from_available_pillars(out, [ ("quality_score", "Quality", float(w.get("quality", 35))), ("capital_score", "Capital efficiency", float(w.get("capital", 25))), ("safety_score", "Safety", float(w.get("safety", 25))), ("burry_value_score", "Value", float(w.get("value", 15))), ]) out["total_score"] = out["total_score"].clip(lower=0, upper=100) # Drop internal scoring columns out = out.drop(columns=[c for c in out.columns if c.startswith("_")]) return out # --------------------------------------------------------------------------- # Output writing # --------------------------------------------------------------------------- COMPACT_COLS = [ "ticker", "total_score", "score_confidence", "missing_pillars", "unscorable_reason", "quality_score", "capital_score", "safety_score", "burry_value_score", "roe", "roa", "profit_margin", "operating_margin", "gross_margin", "sector", "industry", "quote_type", "type_disp", "country", "cf_to_capex", "fcf_margin", "ocf_margin", "debt_to_equity", "payout_ratio", "current_ratio", "trailing_pe", "price_to_book", "earnings_growth", "revenue_growth", "current_price", "market_cap", ] def write_outputs(scored: pd.DataFrame, outputs_dir: Path, suffix: str = "") -> None: scored.to_csv(outputs_dir / f"burry{suffix}.csv", index=False) cols = [c for c in COMPACT_COLS if c in scored.columns] scored[cols].to_csv(outputs_dir / f"burry_compact{suffix}.csv", index=False) # --------------------------------------------------------------------------- # Pipeline orchestration # --------------------------------------------------------------------------- def _jsonl_path(cfg: Dict, outputs_dir: Path) -> Path: return outputs_dir / cfg.get("burry", {}).get("results_jsonl", "burry_rows.jsonl") 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")) jsonl_path = _jsonl_path(cfg, outputs_dir) if not jsonl_path.exists(): raise FileNotFoundError(f"Missing {jsonl_path}. Run without --finalize_only first.") df = enrich_classification_from_cache(read_jsonl_latest(jsonl_path), cache_dir) weights = cfg.get("burry", {}).get("scoring_weights", { "quality": 35, "capital": 25, "safety": 25, "value": 15 }) scored = build_scores(df, weights=weights) scored = scored.sort_values(["total_score", "ticker"], ascending=[False, True]) write_outputs(scored, outputs_dir) out_path = outputs_dir / "burry_compact.csv" print(f"Wrote {len(scored)} stocks → {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")) weights = cfg.get("burry", {}).get("scoring_weights", { "quality": 35, "capital": 25, "safety": 25, "value": 15 }) 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 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) 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 = processed_count if "error" in row: print(f"[{curr_processed}/{total_todo}] {t} ERROR {dur:.1f}s: {row['error']}") else: print(f"[{curr_processed}/{total_todo}] {t} ok {dur:.1f}s") if checkpoint_every > 0 and curr_processed % 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) sc = sc.sort_values(["total_score", "ticker"], ascending=[False, True]) with file_lock: write_outputs(sc, outputs_dir) write_outputs(sc, outputs_dir, suffix="_checkpoint") print(f" → checkpoint after {curr_processed} 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="Burry-style deep value 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)