#!/usr/bin/env python3 """ combined_tracker.py — best turnaround setups across all four screeners Takes the top-N names from each screener export (Roaring Kitty, Burry, Kulamägi, Buffett), overlays the price-structure setup scan on each, and ranks the union by a combined score: combined = SETUP_W × setup score (2y weekly chart: deep drawdown, recovery off the low, 20w trend, range tightening, volume interest) + FUND_W × best screener score (each is a 0-100 cross-sectional rank inside its own philosophy) + multi-screener bonus (a name that is simultaneously cheap, high-quality and turning up is rarer and more interesting than any single screen hit) Sources: data/share/{rk,burry,kulamagi,buffett}_scored_tickers.csv (falls back to data/outputs/*_compact.csv when share files absent) Prices: data/weekly/{TICKER}.csv Output: data/share/combined_setups.csv Usage: python combined_tracker.py # top 50 per screener python combined_tracker.py --top_n 100 --min_market_cap 1e8 """ from __future__ import annotations import argparse from pathlib import Path from typing import Dict, List, Optional import numpy as np import pandas as pd ROOT = Path(__file__).parent SHARE_DIR = ROOT / "data" / "share" OUTPUTS_DIR = ROOT / "data" / "outputs" WEEKLY_DIR = ROOT / "data" / "weekly" SCREENERS = { "rk": ("rk_scored_tickers.csv", "scores_compact.csv"), "burry": ("burry_scored_tickers.csv", "burry_compact.csv"), "kulamagi": ("kulamagi_scored_tickers.csv", "kulamagi_compact.csv"), "buffett": ("buffett_scored_tickers.csv", "buffett_compact.csv"), } SETUP_W = 0.50 # price structure FUND_W = 0.50 # best screener score MULTI_PTS = 6.0 # bonus per additional screener that surfaced the name MIN_CONFIDENCE = 0.75 # ── Price-structure setup scan (2y weekly chart) ───────────────────────────── def compute_setup(ticker: str, weekly_dir: Path = WEEKLY_DIR) -> Optional[Dict]: """Turnaround-setup metrics from the last 104 weeks of prices.""" path = weekly_dir / f"{ticker}.csv" if not path.exists(): return None try: w = pd.read_csv(path, parse_dates=["Date"]).dropna(subset=["Close"]) except Exception: return None w = w.sort_values("Date").tail(104) if len(w) < 60: return None c, v = w.Close.values, w.Volume.values last = c[-1] ipk = int(np.argmax(c)) # 2y peak peak = c[ipk] itr = ipk + int(np.argmin(c[ipk:])) # trough after the peak trough = c[itr] if peak <= 0 or trough <= 0: return None rets = pd.Series(c).pct_change().dropna() return { "dd_2y": trough / peak - 1, # drawdown depth "below_peak": last / peak - 1, # still down from peak "recovery_off_low": last / trough - 1, # turn evidence "weeks_since_low": len(c) - 1 - itr, "above_20w_ma": bool(last > pd.Series(c).rolling(20).mean().iloc[-1]), "vol_contraction": float(rets.tail(10).std() / max(rets.iloc[-40:-10].std(), 1e-9)), "volume_ratio": float(np.nanmean(v[-8:]) / max(np.nanmean(v[-52:-8]), 1)), "ret_13w": last / c[-14] - 1 if len(c) >= 14 else np.nan, "px_last_date": str(w.Date.iloc[-1].date()), } def setup_score(d: pd.DataFrame) -> pd.Series: """0-100: deep drawdown + turning up + trend + consolidation + volume.""" def clip01(x): return np.clip(x, 0, 1) return ( 35 * clip01((-d["dd_2y"] - 0.25) / 0.45) # depth 25%..70%+ + 25 * clip01(d["recovery_off_low"] / 0.30) * (d["recovery_off_low"] < 1.0) # turning, not run away + 15 * d["above_20w_ma"].astype(float) # trend confirmation + 10 * clip01((1.1 - d["vol_contraction"]) / 0.6) # tightening range + 15 * clip01((d["volume_ratio"] - 0.9) / 0.8) # volume interest ) # ── Screener loading ────────────────────────────────────────────────────────── def load_screener(name: str, top_n: int, min_mcap: float, min_conf: float = MIN_CONFIDENCE) -> pd.DataFrame: share_name, compact_name = SCREENERS[name] for path in (SHARE_DIR / share_name, OUTPUTS_DIR / compact_name): if path.exists(): df = pd.read_csv(path) break else: print(f" [{name}] no export found — skipping") return pd.DataFrame() df = df[df["total_score"].notna()] if "score_confidence" in df.columns: df = df[df["score_confidence"].fillna(1) >= min_conf] if "market_cap" in df.columns: df = df[df["market_cap"].fillna(0) >= min_mcap] df = df.sort_values("total_score", ascending=False).head(top_n) out = df[["ticker", "total_score"]].copy() for c in ("sector", "industry", "country", "market_cap", "current_price"): if c in df.columns: out[c] = df[c].values out["ticker"] = out["ticker"].astype(str).str.upper().str.strip() return out # ── Main ────────────────────────────────────────────────────────────────────── def run(top_n: int, min_mcap: float, weekly_dir: Path) -> pd.DataFrame: # Union of top names, remembering every screener that surfaced each one entries: Dict[str, Dict] = {} for name in SCREENERS: df = load_screener(name, top_n, min_mcap) print(f" [{name}] {len(df)} names") for _, r in df.iterrows(): e = entries.setdefault(r.ticker, { "ticker": r.ticker, "scores": {}, "meta": {}}) e["scores"][name] = float(r.total_score) for c in ("sector", "industry", "country", "market_cap", "current_price"): if c in r.index and pd.notna(r.get(c)): e["meta"].setdefault(c, r[c]) rows: List[Dict] = [] for t, e in entries.items(): setup = compute_setup(t, weekly_dir) if setup is None: continue row = { "ticker": t, "sources": "+".join(sorted(e["scores"])), "n_sources": len(e["scores"]), "best_screener_score": max(e["scores"].values()), **{f"{k}_score": v for k, v in e["scores"].items()}, **e["meta"], **setup, } rows.append(row) d = pd.DataFrame(rows) if d.empty: return d d["setup_score"] = setup_score(d) d["combined_score"] = ( SETUP_W * d["setup_score"] + FUND_W * d["best_screener_score"] + MULTI_PTS * (d["n_sources"] - 1) ).round(2) d = d.sort_values("combined_score", ascending=False).reset_index(drop=True) # Dashboard compatibility: the UI ranks/filters on total_score d["total_score"] = d["combined_score"] return d def main() -> None: ap = argparse.ArgumentParser(description="Cross-screener turnaround setups") ap.add_argument("--top_n", type=int, default=50, help="Top N names taken from EACH screener (default 50)") ap.add_argument("--min_market_cap", type=float, default=1e8) ap.add_argument("--weekly_dir", default=str(WEEKLY_DIR)) ap.add_argument("--out", default=str(SHARE_DIR / "combined_setups.csv")) ap.add_argument("--min_setup", type=float, default=0.0, help="Require setup_score ≥ this (e.g. 60) so names must be " "BOTH a screener pick AND a real turnaround setup. " "Default 0 = blended ranking only.") args = ap.parse_args() print(f"Combined tracker — top {args.top_n} per screener, " f"mcap ≥ ${args.min_market_cap/1e6:.0f}M") d = run(args.top_n, args.min_market_cap, Path(args.weekly_dir)) if d.empty: raise SystemExit("No names produced — are the share exports / weekly data present?") if args.min_setup > 0: before = len(d) d = d[d["setup_score"] >= args.min_setup].reset_index(drop=True) print(f" --min_setup {args.min_setup:.0f}: {before} → {len(d)} names " f"(screener pick AND turnaround setup)") out = Path(args.out) out.parent.mkdir(parents=True, exist_ok=True) d.to_csv(out, index=False) cols = ["ticker", "combined_score", "setup_score", "best_screener_score", "sources", "dd_2y", "recovery_off_low", "above_20w_ma", "volume_ratio"] print(f"\nTop 20 of {len(d)} (weekly data through {d.px_last_date.max()}):\n") print(d[cols].head(20).round(2).to_string(index=False)) multi = d[d.n_sources >= 2] print(f"\n{len(multi)} names surfaced by ≥2 screeners; " f"{len(d[d.n_sources >= 3])} by ≥3.") print(f"\nFull table → {out}") if __name__ == "__main__": main()