#!/usr/bin/env python3 """ kulamagi_tracker.py — Kulamägi-style breakout screener (yfinance edition) Screens for the strongest U.S. stocks with: - High liquidity (dollar volume filter) - Strong multi-timeframe momentum (prior big move) - Tight structure near 52-week highs (orderly consolidation) - Fundamental quality (revenue / earnings growth) Reuses the same yfinance cache as rk_tracker.py — no extra API calls for already-cached tickers. Scoring pillars: Momentum 40% — 1m, 3m, 6m, 12m returns Structure 35% — proximity to 52W high, above MAs, vol contraction Quality 15% — revenue & earnings growth Liquidity 10% — dollar volume rank Run: # Score from existing cache (fast — reuses rk_tracker data) python kulamagi_tracker.py --config config.yaml --universe_csv us_universe.csv --resume # Rebuild scores from cached JSONL only python kulamagi_tracker.py --config config.yaml --finalize_only """ from __future__ import annotations import argparse import datetime as dt import time from pathlib import Path from typing import Any, Dict, List, Optional import numpy as np import pandas as pd import threading # Reuse all shared utilities from rk_tracker — no extra dependencies from rk_tracker import ( YFCache, append_jsonl, enrich_classification_from_cache, ensure_dir, fetch_history, fetch_info, filter_common_stock_rows, json_dumps_safe, load_universe_from_csv, nanmean_rows, pct_rank, prefetch_histories, read_jsonl_latest, read_jsonl_tickers, read_yaml, safe_float, safe_text, weighted_score_from_available_pillars, ) # --------------------------------------------------------------------------- # Feature engineering (Kulamägi-specific) # --------------------------------------------------------------------------- def compute_returns(hist: pd.DataFrame) -> tuple: """Trailing 1m / 3m / 6m price returns from daily history.""" if hist is None or hist.empty: return None, None, None close = hist["Close"].dropna() cur = float(close.iloc[-1]) if not close.empty else None if cur is None: return None, None, None def _ret(n): if len(close) > n: past = float(close.iloc[-(n + 1)]) return (cur / past - 1.0) if past > 0 else None return None return _ret(21), _ret(63), _ret(126) # ~1m, 3m, 6m in trading days def compute_ma20(hist: pd.DataFrame) -> Optional[float]: """20-day simple moving average of close.""" if hist is None or hist.empty: return None close = hist["Close"].dropna() if len(close) < 20: return None return float(close.tail(20).mean()) def compute_vol_ratio(hist: pd.DataFrame) -> Optional[float]: """20-day vs 60-day realised volatility ratio. < 1 = contracting = good.""" import math if hist is None or hist.empty or len(hist) < 65: return None close = hist["Close"].dropna() ret = close.pct_change().dropna() if len(ret) < 65: return None vol20 = float(ret.tail(20).std(ddof=1) * math.sqrt(252)) vol60 = float(ret.tail(60).std(ddof=1) * math.sqrt(252)) return vol20 / vol60 if vol60 > 0 else None # --------------------------------------------------------------------------- # 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")) row["average_volume"] = safe_float(info.get("averageVolume")) row["week52_high"] = safe_float(info.get("fiftyTwoWeekHigh")) row["week52_low"] = safe_float(info.get("fiftyTwoWeekLow")) row["return_12m"] = safe_float(info.get("52WeekChange")) row["fifty_day_avg"] = safe_float(info.get("fiftyDayAverage")) # Dollar volume (liquidity proxy: avg daily shares × price) if row["average_volume"] and row["current_price"]: row["dollar_volume"] = row["average_volume"] * row["current_price"] else: row["dollar_volume"] = None # ── % from 52-week high ──────────────────────────────────────────── if row["current_price"] and row["week52_high"] and row["week52_high"] > 0: row["pct_from_52w_high"] = (row["current_price"] / row["week52_high"]) - 1.0 else: row["pct_from_52w_high"] = None # ── Fundamentals ─────────────────────────────────────────────────── row["revenue_growth"] = safe_float(info.get("revenueGrowth")) row["earnings_growth"] = safe_float( info.get("earningsGrowth") or info.get("earningsQuarterlyGrowth") ) # ── Price history metrics ────────────────────────────────────────── try: hist = fetch_history(cache, ticker_u) row["return_1m"], row["return_3m"], row["return_6m"] = compute_returns(hist) ma20 = compute_ma20(hist) row["ma20"] = ma20 row["above_20ma"] = bool(row["current_price"] and ma20 and row["current_price"] > ma20) row["above_50ma"] = bool( row["current_price"] and row["fifty_day_avg"] and row["current_price"] > row["fifty_day_avg"] ) row["vol_ratio_20_60"] = compute_vol_ratio(hist) except Exception: row["return_1m"] = row["return_3m"] = row["return_6m"] = None row["ma20"] = row["vol_ratio_20_60"] = None row["above_20ma"] = row["above_50ma"] = False return row # --------------------------------------------------------------------------- # Scoring — 4 Kulamägi pillars # --------------------------------------------------------------------------- def build_scores( df: pd.DataFrame, weights: Dict[str, float], min_dollar_volume: float = 20_000_000, ) -> pd.DataFrame: out = filter_common_stock_rows(df.copy()) # Hard liquidity filter — remove stocks with insufficient dollar volume mask_liq = out["dollar_volume"].fillna(0) >= min_dollar_volume out = out[mask_liq].copy() if out.empty: return out # ── Pillar 1: Momentum (40%) ─────────────────────────────────────── # Higher returns across all timeframes = better out["_m_1m"] = pct_rank(out["return_1m"], ascending=True) * 100 out["_m_3m"] = pct_rank(out["return_3m"], ascending=True) * 100 out["_m_6m"] = pct_rank(out["return_6m"], ascending=True) * 100 out["_m_12m"] = pct_rank(out["return_12m"], ascending=True) * 100 out["momentum_score"] = nanmean_rows( out["_m_1m"], out["_m_3m"], out["_m_6m"], out["_m_12m"] ) # ── Pillar 2: Structure (35%) ────────────────────────────────────── # Closer to 52W high = better (pct_from_52w_high is negative; less negative = higher rank) out["_s_52wh"] = pct_rank(out["pct_from_52w_high"], ascending=True) * 100 # Boolean flags → 0 or 100 out["_s_20ma"] = out["above_20ma"].fillna(False).astype(float) * 100 out["_s_50ma"] = out["above_50ma"].fillna(False).astype(float) * 100 # Lower vol ratio = more contracted base = better out["_s_vol"] = (1.0 - pct_rank(out["vol_ratio_20_60"], ascending=True)) * 100 out["structure_score"] = nanmean_rows( out["_s_52wh"], out["_s_20ma"], out["_s_50ma"], out["_s_vol"] ) # ── Pillar 3: Quality (15%) ──────────────────────────────────────── out["_q_rev"] = pct_rank(out["revenue_growth"], ascending=True) * 100 out["_q_earn"] = pct_rank(out["earnings_growth"], ascending=True) * 100 out["quality_score"] = nanmean_rows(out["_q_rev"], out["_q_earn"]) # ── Pillar 4: Liquidity (10%) ────────────────────────────────────── out["liquidity_score"] = pct_rank(out["dollar_volume"], ascending=True) * 100 w = weights out = weighted_score_from_available_pillars(out, [ ("momentum_score", "Momentum", float(w.get("momentum", 40))), ("structure_score", "Structure", float(w.get("structure", 35))), ("quality_score", "Quality", float(w.get("quality", 15))), ("liquidity_score", "Liquidity", float(w.get("liquidity", 10))), ]) out["total_score"] = out["total_score"].clip(lower=0, upper=100) # Drop internal 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", "momentum_score", "structure_score", "quality_score", "liquidity_score", "sector", "industry", "quote_type", "type_disp", "country", "return_1m", "return_3m", "return_6m", "return_12m", "pct_from_52w_high", "above_20ma", "above_50ma", "vol_ratio_20_60", "revenue_growth", "earnings_growth", "dollar_volume", "average_volume", "current_price", "market_cap", "week52_high", "week52_low", ] def write_outputs(scored: pd.DataFrame, outputs_dir: Path, suffix: str = "") -> None: scored.to_csv(outputs_dir / f"kulamagi{suffix}.csv", index=False) cols = [c for c in COMPACT_COLS if c in scored.columns] scored[cols].to_csv(outputs_dir / f"kulamagi_compact{suffix}.csv", index=False) # --------------------------------------------------------------------------- # Pipeline orchestration # --------------------------------------------------------------------------- def _jsonl_path(cfg: Dict, outputs_dir: Path) -> Path: return outputs_dir / cfg.get("kulamagi", {}).get("results_jsonl", "kulamagi_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) kul_cfg = cfg.get("kulamagi", {}) weights = kul_cfg.get("scoring_weights", { "momentum": 40, "structure": 35, "quality": 15, "liquidity": 10 }) min_dv = float(kul_cfg.get("min_dollar_volume", 20_000_000)) scored = build_scores(df, weights=weights, min_dollar_volume=min_dv) scored = scored.sort_values(["total_score", "ticker"], ascending=[False, True]) write_outputs(scored, outputs_dir) out_path = outputs_dir / "kulamagi_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")) kul_cfg = cfg.get("kulamagi", {}) weights = kul_cfg.get("scoring_weights", { "momentum": 40, "structure": 35, "quality": 15, "liquidity": 10 }) min_dv = float(kul_cfg.get("min_dollar_volume", 20_000_000)) 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) # Bulk prefetch histories prefetch_histories(todo, cache_dir) 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, min_dollar_volume=min_dv) 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="Kulamägi-style breakout 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)