# rk_tracker.py  —  v2, yfinance edition
# Roaring Kitty-style contrarian value screener
#
# Data source: yfinance (Yahoo Finance) — no API key required
#
# Scoring pillars:
#   1. Value                (35%) — P/B, P/S, EV/EBITDA, net cash / market cap
#   2. Financial Health     (25%) — gross margin, FCF margin, debt/equity, current ratio
#   3. Sentiment & Crowding (25%) — short % of float, 52-week drawdown, put/call OI ratio, beta
#   4. Quality & Momentum   (15%) — ROE, revenue growth, analyst consensus, relative 52-week return
#
# Guardrails:
#   - Distress flag  (D/E > 200 or current ratio < 0.8) → score capped at 60
#   - Dilution flag  (shares +30% over 2 fiscal years)  → score minus 15 points
#
# Install:
#   python -m venv .venv && source .venv/bin/activate
#   pip install -U pip yfinance pandas numpy pyyaml
#
# Run:
#   python rk_tracker.py --config config.yaml --universe_csv us_universe.csv --resume
#   python rk_tracker.py --config config.yaml --finalize_only

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import math
import re
import threading
import time
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional

import numpy as np
import pandas as pd
import yaml
import yfinance as yf


# ---------------------------------------------------------------------------
# Utilities
# ---------------------------------------------------------------------------

def _sha1(s: str) -> str:
    return hashlib.sha1(s.encode()).hexdigest()

def ensure_dir(p: str | Path) -> Path:
    p = Path(p)
    p.mkdir(parents=True, exist_ok=True)
    return p

def read_yaml(path: str | Path) -> Dict[str, Any]:
    with open(path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f) or {}

def safe_float(x: Any) -> Optional[float]:
    try:
        if x is None:
            return None
        if isinstance(x, (int, float, np.integer, np.floating)):
            v = float(x)
            return None if (math.isnan(v) or math.isinf(v)) else v
        s = str(x).strip()
        if s.lower() in {"", "nan", "none", "null", "inf", "-inf"}:
            return None
        return float(s)
    except Exception:
        return None

def safe_text(x: Any) -> Optional[str]:
    if x is None:
        return None
    s = str(x).strip()
    return s if s and s.lower() not in {"nan", "none", "null"} else None

def pct_rank(series: pd.Series, ascending: bool = True) -> pd.Series:
    """Percentile rank in [0, 1]; NaN entries stay NaN."""
    s = series.copy()
    mask = s.notna()
    out = pd.Series(np.nan, index=s.index, dtype=float)
    if mask.sum() < 2:
        return out
    out.loc[mask] = s.loc[mask].rank(pct=True, ascending=ascending)
    return out

def bell_score(p: float, target: float = 0.75, sigma: float = 0.15) -> float:
    """Gaussian-shaped 0-100 score peaked at target percentile."""
    if p is None or not np.isfinite(p):
        return np.nan
    z = (p - target) / sigma
    return float(100.0 * math.exp(-(z * z)))

def nanmean_rows(*series: pd.Series) -> pd.Series:
    """Element-wise nanmean across multiple equal-length series."""
    mat = np.vstack([s.values for s in series]).astype(float)
    valid = np.isfinite(mat)
    counts = valid.sum(axis=0)
    sums = np.where(valid, mat, 0.0).sum(axis=0)
    means = np.divide(
        sums,
        counts,
        out=np.full_like(sums, np.nan, dtype=float),
        where=counts > 0,
    )
    return pd.Series(means, index=series[0].index)

def security_exclusion_reason(
    ticker: Any,
    name: Any = None,
    quote_type: Any = None,
    type_disp: Any = None,
) -> Optional[str]:
    """Return a reason when a symbol is clearly not common stock."""
    t = str(safe_text(ticker) or "").upper().strip()
    n = str(safe_text(name) or "").upper()
    qt = str(safe_text(quote_type) or "").upper().strip()
    td = str(safe_text(type_disp) or "").upper().strip()

    if not t:
        return "missing_ticker"
    if re.search(r"-(WT|WS|W)$", t):
        return "warrant"
    if re.search(r"-(UN|U)$", t):
        return "unit"
    if re.search(r"-(RT|R)$", t):
        return "right"
    if re.search(r"-P[A-Z]?$", t):
        return "preferred"

    if qt and qt != "EQUITY":
        return f"quote_type:{qt.lower()}"
    if td in {"ETF", "FUND", "MUTUAL FUND"}:
        return f"type:{td.lower().replace(' ', '_')}"

    if re.search(r"\bWARRANTS?\b", n):
        return "warrant"
    if re.search(r"\bRIGHTS?\b", n):
        return "right"
    if re.search(r"\bUNITS?\b", n):
        return "unit"
    if re.search(r"\bPREFERRED\b|\bPFD\b|\bCAPITAL TRUST\b|\bTRUST PREFERRED\b", n):
        return "preferred"
    if re.search(r"\bETF\b|\bEXCHANGE TRADED FUND\b", n):
        return "etf"
    # Exchange-traded debt: "7.875% Notes due 2029", "6.00% Senior Notes"...
    # yfinance labels these EQUITY, but no common stock has a coupon in its name.
    if re.search(r"\d(\.\d+)?\s*%", n):
        return "note_or_preferred"
    if re.search(r"\bNOTES?\s+DUE\b|\bSENIOR NOTES?\b|\bSUBORDINATED NOTES?\b|\bDEBENTURES?\b", n):
        return "note"
    # Pre-merger SPAC shells: pure cash boxes that score as fake net-cash
    # bargains. Any company with ACQUISITION in its legal name is a SPAC
    # ("Chenghe Acquisition III Co.", "Pioneer Acquisition I Corp"...).
    if re.search(r"\bACQUISITION\b", n):
        return "spac_shell"
    # Serial shell naming: "... Corp III", "... Investment Corp II"
    if re.search(r"\b(CORP|CORPORATION|CO|COMPANY)\.?\s+(I{1,3}|IV|V|VI{0,3}|IX|X)$", n):
        return "spac_shell"

    return None

def filter_common_stock_rows(df: pd.DataFrame) -> pd.DataFrame:
    if df.empty or "ticker" not in df.columns:
        return df
    out = df.copy()
    name_col = "long_name" if "long_name" in out.columns else "name" if "name" in out.columns else None
    quote_col = "quote_type" if "quote_type" in out.columns else None
    type_col = "type_disp" if "type_disp" in out.columns else None

    reasons = out.apply(
        lambda r: security_exclusion_reason(
            r.get("ticker"),
            r.get(name_col) if name_col else None,
            r.get(quote_col) if quote_col else None,
            r.get(type_col) if type_col else None,
        ),
        axis=1,
    )
    keep = reasons.isna()

    # No market cap → cannot be ranked (and in practice these are preferred
    # lines, exchange-traded notes, warrants, or SPAC units of companies whose
    # common stock is listed separately WITH a market cap).
    if "market_cap" in out.columns:
        keep = keep & out["market_cap"].notna()

    return out[keep].copy()

def weighted_score_from_available_pillars(
    out: pd.DataFrame,
    pillars: list[tuple[str, str, float]],
) -> pd.DataFrame:
    total_weight = sum(weight for _, _, weight in pillars)
    numerator = pd.Series(0.0, index=out.index)
    denominator = pd.Series(0.0, index=out.index)

    for col, _label, weight in pillars:
        valid = out[col].notna()
        numerator += out[col].fillna(0.0) * weight
        denominator += valid.astype(float) * weight

    out["total_score_raw"] = numerator.div(denominator).where(denominator > 0)
    out["total_score"] = out["total_score_raw"].copy()
    out["score_confidence"] = (
        denominator / total_weight
        if total_weight > 0
        else pd.Series(np.nan, index=out.index)
    )

    def _missing(row: pd.Series) -> str:
        labels = [label for col, label, _weight in pillars if pd.isna(row[col])]
        return ", ".join(labels)

    out["missing_pillars"] = out.apply(_missing, axis=1)
    out["unscorable_reason"] = np.where(
        denominator <= 0,
        "No usable scoring pillars",
        "",
    )
    return out

def json_dumps_safe(obj: Any) -> str:
    def _default(o: Any) -> str:
        try:
            return str(o)
        except Exception:
            return "<unserializable>"
    return json.dumps(obj, default=_default, ensure_ascii=False)

def append_jsonl(path: Path, record: Dict[str, Any]) -> None:
    ensure_dir(path.parent)
    with open(path, "a", encoding="utf-8") as f:
        f.write(json_dumps_safe(record) + "\n")
        f.flush()

def _has_error(rec: Dict[str, Any]) -> bool:
    err = rec.get("error")
    if err is None:
        return False
    if isinstance(err, float) and math.isnan(err):
        return False
    return str(err).strip().lower() not in {"", "none", "nan", "null"}

def read_jsonl_tickers(path: Path, include_errors: bool = False) -> set:
    tickers: set = set()
    if not path.exists():
        return tickers
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                rec = json.loads(line)
                t = str(rec.get("ticker", "")).upper().strip()
                if t and (include_errors or not _has_error(rec)):
                    tickers.add(t)
            except Exception:
                continue
    return tickers

def read_jsonl_latest(path: Path) -> pd.DataFrame:
    df = pd.read_json(path, lines=True)
    if df.empty or "ticker" not in df.columns:
        return df
    df["ticker"] = df["ticker"].astype(str).str.upper().str.strip()
    df = df[df["ticker"].str.len() > 0].copy()
    return df.drop_duplicates(subset=["ticker"], keep="last")

def enrich_classification_from_cache(df: pd.DataFrame, cache_dir: Path) -> pd.DataFrame:
    if df.empty or "ticker" not in df.columns:
        return df

    out = df.copy()
    fields = ["sector", "industry", "quote_type", "type_disp", "long_name", "country"]
    for field in fields:
        if field not in out.columns:
            out[field] = None

    tickers = out["ticker"].astype(str).str.upper().str.strip()
    missing = {
        field: out[field].isna() | (out[field].astype(str).str.strip() == "")
        for field in fields
    }
    need_mask = pd.Series(False, index=out.index)
    for mask in missing.values():
        need_mask = need_mask | mask
    if not need_mask.any():
        return out

    meta: Dict[str, Dict[str, Optional[str]]] = {}
    for ticker in tickers[need_mask].unique():
        path = cache_dir / f"info_{ticker}.json"
        rec = {field: None for field in fields}
        if path.exists():
            try:
                raw = json.loads(path.read_text(encoding="utf-8"))
                if isinstance(raw, dict):
                    rec["sector"] = safe_text(raw.get("sector"))
                    rec["industry"] = safe_text(raw.get("industry"))
                    rec["quote_type"] = safe_text(raw.get("quoteType"))
                    rec["type_disp"] = safe_text(raw.get("typeDisp"))
                    rec["long_name"] = safe_text(raw.get("longName") or raw.get("shortName"))
                    rec["country"] = safe_text(raw.get("country"))
            except Exception:
                pass
        meta[ticker] = rec

    for field in fields:
        fill = tickers.map(lambda t, field=field: meta.get(t, {}).get(field))
        out.loc[missing[field], field] = fill[missing[field]]
    return out

def is_retryable_error(exc: Exception) -> bool:
    msg = str(exc).lower()
    return any(token in msg for token in [
        "too many requests",
        "rate limited",
        "rate limit",
        "429",
    ])


# ---------------------------------------------------------------------------
# Cache — file-based TTL cache wrapping yfinance calls
# ---------------------------------------------------------------------------

class YFCache:
    """
    Wraps yfinance with file-based caching so reruns are fast and
    rate-limiting risk is reduced. JSON for scalar data, CSV for DataFrames.
    """

    def __init__(self, cache_dir: Path, min_delay_seconds: float = 0.5):
        self.cache_dir = ensure_dir(cache_dir)
        self.min_delay = min_delay_seconds
        self._last_call = 0.0
        self._lock = threading.Lock()

    def _throttle(self) -> None:
        with self._lock:
            now = time.time()
            base_time = max(self._last_call, now)
            next_call = base_time + self.min_delay
            self._last_call = next_call
        
        sleep_time = next_call - now - self.min_delay
        if sleep_time > 0:
            time.sleep(sleep_time)

    def _fetch_with_retries(self, fetch_fn: Callable):
        backoffs = [5.0, 15.0, 45.0]
        for attempt in range(len(backoffs) + 1):
            self._throttle()
            try:
                return fetch_fn()
            except Exception as e:
                if attempt >= len(backoffs) or not is_retryable_error(e):
                    raise
                wait_s = backoffs[attempt]
                print(f"  → rate limited; retrying in {wait_s:.0f}s")
                time.sleep(wait_s)

    def get_json(self, key: str, ttl_hours: float, fetch_fn: Callable) -> Any:
        path = self.cache_dir / f"{key}.json"
        if path.exists():
            age_h = (time.time() - path.stat().st_mtime) / 3600.0
            if age_h <= ttl_hours:
                return json.loads(path.read_text(encoding="utf-8"))
        data = self._fetch_with_retries(fetch_fn)
        path.write_text(json.dumps(data, default=str), encoding="utf-8")
        return data


# ---------------------------------------------------------------------------
# Universe loading
# ---------------------------------------------------------------------------

def load_universe_from_csv(csv_path: str | Path) -> List[str]:
    df = pd.read_csv(csv_path)
    cols = {c.lower().strip(): c for c in df.columns}

    ticker_col = next((cols[c] for c in ["ticker", "symbol"] if c in cols), None)
    if ticker_col is None:
        raise ValueError("CSV must have a 'ticker' or 'symbol' column")

    mcap_col = next((cols[c] for c in ["market_cap", "marketcap", "mcap"] if c in cols), None)
    name_col = next((cols[c] for c in ["name", "company", "company_name"] if c in cols), None)

    df["_ticker"] = df[ticker_col].astype(str).str.upper().str.strip()
    df = df[df["_ticker"].str.len() > 0].copy()
    if name_col:
        df["_exclude_reason"] = df.apply(
            lambda r: security_exclusion_reason(r["_ticker"], r.get(name_col)),
            axis=1,
        )
    else:
        df["_exclude_reason"] = df["_ticker"].apply(security_exclusion_reason)
    df = df[df["_exclude_reason"].isna()].copy()

    # Remove preferred stocks (-PA/-PB/…), warrants (-W*), rights (-R*), units (-U*).
    # Dual-class shares like BRK-B and BF-A do NOT match this pattern and are kept.
    special = df["_ticker"].str.contains(r"-(P|W|R|U)[A-Z]{0,2}$", regex=True)
    df = df[~special].copy()

    if mcap_col:
        df["_mcap"] = df[mcap_col].apply(safe_float)
        df = df.sort_values(["_mcap", "_ticker"], ascending=[False, True], na_position="last")
    else:
        df = df.sort_values("_ticker")

    return df["_ticker"].tolist()


# ---------------------------------------------------------------------------
# yfinance data fetchers (all cached)
# ---------------------------------------------------------------------------

def fetch_info(cache: YFCache, ticker: str, yf_ticker=None) -> Dict[str, Any]:
    """
    yfinance .info dict — contains fundamentals, valuation ratios,
    short interest, analyst consensus, and price metadata. TTL 24h.
    Pass yf_ticker to reuse an existing Ticker object (saves object creation
    and shares the underlying HTTP session).
    """
    key = f"info_{ticker}"
    def _fetch():
        t = yf_ticker if yf_ticker is not None else yf.Ticker(ticker)
        return t.info
    raw = cache.get_json(key, ttl_hours=24, fetch_fn=_fetch)
    return raw if isinstance(raw, dict) else {}


def prefetch_histories(tickers: List[str], cache_dir: Path, batch_size: int = 400) -> None:
    """
    Downloads historical price data in bulk for tickers that do not have
    cached history or whose cached history is older than 12 hours.
    Saves them directly to raw/hist_{ticker}.csv, bypassing single-ticker requests.

    batch_size=400 balances round-trip efficiency against the risk of Yahoo
    rejecting an oversized request. Tune down if you see frequent batch errors.
    """
    needed = []
    for t in tickers:
        t_u = str(t).upper().strip()
        if not t_u:
            continue
        path = cache_dir / f"hist_{t_u}.csv"
        if not path.exists() or (time.time() - path.stat().st_mtime) / 3600.0 > 12:
            needed.append(t_u)

    if not needed:
        return

    print(f"  → Prefetching histories for {len(needed)} tickers in batches of {batch_size}...")

    for i in range(0, len(needed), batch_size):
        batch = needed[i : i + batch_size]
        print(f"    Batch {i // batch_size + 1}/{-(-len(needed) // batch_size)} ({len(batch)} tickers)...")
        try:
            df = yf.download(batch, period="2y", group_by="ticker", auto_adjust=True, progress=False)
            if df.empty:
                continue

            is_multi = isinstance(df.columns, pd.MultiIndex)

            if is_multi:
                # Use get_level_values().unique() — df.columns.levels[0] retains ALL level
                # values seen since the MultiIndex was created, including tickers that were
                # dropped/filtered out, which causes spurious misses when checking membership.
                available = set(df.columns.get_level_values(0).unique())

            for t in batch:
                try:
                    if is_multi:
                        if t in available:
                            t_df = df[t].dropna(subset=["Close"])
                            if not t_df.empty:
                                path = cache_dir / f"hist_{t}.csv"
                                t_df.index.name = "Date"
                                t_df.reset_index().to_csv(path, index=False)
                    else:
                        t_df = df.dropna(subset=["Close"])
                        if not t_df.empty:
                            path = cache_dir / f"hist_{t}.csv"
                            t_df.index.name = "Date"
                            t_df.reset_index().to_csv(path, index=False)
                except Exception:
                    pass
        except Exception as e:
            print(f"    Error downloading batch: {e}")
        time.sleep(1.0)


def fetch_history(cache: YFCache, ticker: str, period: str = "2y", yf_ticker=None) -> pd.DataFrame:
    """
    Daily OHLCV price history. Stored as CSV; TTL 12h.
    Returns DataFrame with DatetimeIndex and columns: Open, High, Low, Close, Volume.
    Pass yf_ticker to reuse an existing Ticker object (avoids a redundant object creation
    on the fallback path when bulk prefetch missed this ticker).
    """
    key = f"hist_{ticker}"
    path = cache.cache_dir / f"{key}.csv"

    if path.exists():
        age_h = (time.time() - path.stat().st_mtime) / 3600.0
        if age_h <= 12:
            df = pd.read_csv(path, parse_dates=["Date"], index_col="Date")
            return df

    t = yf_ticker if yf_ticker is not None else yf.Ticker(ticker)
    df = cache._fetch_with_retries(
        lambda: t.history(period=period, auto_adjust=True)
    )
    if not df.empty:
        df.index.name = "Date"
        df.reset_index().to_csv(path, index=False)
    return df


def fetch_put_call_ratio(cache: YFCache, ticker: str, yf_ticker=None) -> Optional[float]:
    """
    Put/call open-interest ratio from the nearest-term options expiry.
    Higher ratio = more bearish bets outstanding = contrarian opportunity.
    Returns None when no options market exists (common for small caps).
    TTL 6h.
    Pass yf_ticker to reuse an existing Ticker object (shares HTTP session).
    """
    key = f"pcr_{ticker}"

    def _fetch():
        t = yf_ticker if yf_ticker is not None else yf.Ticker(ticker)
        dates = t.options
        if not dates:
            return None
        chain = t.option_chain(dates[0])
        calls_oi = float(chain.calls["openInterest"].fillna(0).sum())
        puts_oi  = float(chain.puts["openInterest"].fillna(0).sum())
        if calls_oi <= 0:
            return None
        return puts_oi / calls_oi

    try:
        result = cache.get_json(key, ttl_hours=6, fetch_fn=_fetch)
        return safe_float(result)
    except Exception:
        return None


def fetch_shares_growth_2y(cache: YFCache, ticker: str, yf_ticker=None) -> Optional[float]:
    """
    Fractional change in shares outstanding over ~2 fiscal years,
    from the annual balance sheet. Used for the dilution guardrail. TTL 48h.
    Pass yf_ticker to reuse an existing Ticker object (shares HTTP session).
    """
    key = f"bs_{ticker}"

    def _fetch():
        t = yf_ticker if yf_ticker is not None else yf.Ticker(ticker)
        bs = t.balance_sheet
        if bs is None or bs.empty:
            return None
        for row_name in ["Ordinary Shares Number", "Share Issued", "Common Stock"]:
            if row_name in bs.index:
                shares = bs.loc[row_name].dropna().sort_index(ascending=False)
                if len(shares) >= 3:
                    latest       = float(shares.iloc[0])
                    two_years_ago = float(shares.iloc[2])
                    if two_years_ago > 0:
                        return (latest / two_years_ago) - 1.0
        return None

    try:
        result = cache.get_json(key, ttl_hours=48, fetch_fn=_fetch)
        return safe_float(result)
    except Exception:
        return None


# ---------------------------------------------------------------------------
# Feature engineering
# ---------------------------------------------------------------------------

def compute_drawdown_52w(hist: pd.DataFrame) -> Optional[float]:
    """Drawdown from 52-week high to current close. Negative = below peak."""
    if hist is None or hist.empty or len(hist) < 30:
        return None
    close = hist["Close"].dropna().tail(252)
    if len(close) < 30:
        return None
    peak = float(close.max())
    last = float(close.iloc[-1])
    return (last / peak) - 1.0 if peak > 0 else None


def compute_vol_ratio_20_60(hist: pd.DataFrame) -> Optional[float]:
    """20-day vs 60-day realised volatility ratio (annualised)."""
    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_for_ticker(cache: YFCache, ticker: str) -> Dict[str, Any]:
    ticker_u = ticker.upper()
    row: Dict[str, Any] = {"ticker": ticker_u}

    # Create one Ticker object and reuse it across all sub-fetches.
    # This shares the underlying HTTP session, avoids redundant object
    # instantiation, and is the single biggest per-ticker efficiency win.
    yf_t = yf.Ticker(ticker_u)

    # --- Fundamentals & metadata from yfinance info ---
    info = fetch_info(cache, ticker_u, yf_ticker=yf_t)

    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["market_cap"]       = safe_float(info.get("marketCap"))
    row["price_to_book"]    = safe_float(info.get("priceToBook"))
    row["price_to_sales"]   = safe_float(info.get("priceToSalesTrailing12Months"))
    row["ev_to_ebitda"]     = safe_float(info.get("enterpriseToEbitda"))
    row["gross_margin"]     = safe_float(info.get("grossMargins"))
    row["fcf_reported"]     = safe_float(info.get("freeCashflow"))
    row["total_revenue"]    = safe_float(info.get("totalRevenue"))
    row["debt_to_equity"]   = safe_float(info.get("debtToEquity"))
    row["current_ratio"]    = safe_float(info.get("currentRatio"))
    row["return_on_equity"] = safe_float(info.get("returnOnEquity"))
    row["revenue_growth"]   = safe_float(info.get("revenueGrowth"))
    row["short_pct_float"]  = safe_float(info.get("shortPercentOfFloat"))
    row["beta"]             = safe_float(info.get("beta"))
    # recommendationMean: 1 = strong buy, 5 = strong sell
    # For contrarian scoring, higher (sell) = more neglected = more opportunity
    row["analyst_mean"]     = safe_float(info.get("recommendationMean"))
    row["week52_change"]    = safe_float(info.get("52WeekChange"))
    row["sp52_change"]      = safe_float(info.get("SandP52WeekChange"))

    # Net cash position and ratio
    total_cash = safe_float(info.get("totalCash"))
    total_debt = safe_float(info.get("totalDebt"))
    net_cash = None
    if total_cash is not None or total_debt is not None:
        net_cash = (total_cash or 0.0) - (total_debt or 0.0)
    row["net_cash"] = net_cash
    row["net_cash_to_market_cap"] = (
        net_cash / row["market_cap"]
        if net_cash is not None and row["market_cap"] and row["market_cap"] > 0
        else None
    )

    # FCF margin (free cash flow / revenue)
    row["fcf_margin"] = (
        row["fcf_reported"] / row["total_revenue"]
        if row["fcf_reported"] is not None
        and row["total_revenue"] is not None
        and row["total_revenue"] > 0
        else None
    )

    # Relative 52-week performance vs S&P 500
    # Negative = underperformed the market = beaten-down = contrarian signal
    if row["week52_change"] is not None and row["sp52_change"] is not None:
        row["relative_52w"] = row["week52_change"] - row["sp52_change"]
    else:
        row["relative_52w"] = None

    # --- Price metrics from history ---
    try:
        hist = fetch_history(cache, ticker_u, yf_ticker=yf_t)
        row["drawdown_52w"]    = compute_drawdown_52w(hist)
        row["vol_ratio_20_60"] = compute_vol_ratio_20_60(hist)
        row["price_latest"]    = safe_float(hist["Close"].iloc[-1]) if not hist.empty else None
    except Exception:
        row["drawdown_52w"]    = None
        row["vol_ratio_20_60"] = None
        row["price_latest"]    = None

    # --- Sentiment: put/call open-interest ratio ---
    # Skip the options API call for non-equity instruments and micro-caps (< $50M
    # market cap) — these never have a liquid options market, so the fetch is
    # guaranteed to return None and just wastes a throttled request slot.
    quote_type = (safe_text(info.get("quoteType")) or "").upper()
    mcap = row.get("market_cap") or 0.0
    _has_options_market = (quote_type == "EQUITY" and mcap >= 50_000_000)
    if _has_options_market:
        row["put_call_ratio"] = fetch_put_call_ratio(cache, ticker_u, yf_ticker=yf_t)
    else:
        row["put_call_ratio"] = None

    # --- Dilution check from balance sheet ---
    row["shares_growth_2y"] = fetch_shares_growth_2y(cache, ticker_u, yf_ticker=yf_t)

    # --- Guardrail flags ---
    de = row.get("debt_to_equity")
    cr = row.get("current_ratio")
    row["high_leverage_flag"] = bool(de is not None and de > 200)
    row["low_liquidity_flag"] = bool(cr is not None and cr < 0.8)
    row["distress_flag"]      = row["high_leverage_flag"] or row["low_liquidity_flag"]

    return row


# ---------------------------------------------------------------------------
# Scoring — 4 pillars
# ---------------------------------------------------------------------------

def build_scores(df: pd.DataFrame, weights: Dict[str, float], guardrails: Dict[str, Any]) -> pd.DataFrame:
    out = filter_common_stock_rows(df.copy())

    if out.empty:
        return out

    # ------------------------------------------------------------------
    # Pillar 1: Value
    # Lower P/B, P/S, EV/EBITDA = better (cheap); higher net cash = better
    # ------------------------------------------------------------------
    out["_v_pb"]   = (1.0 - pct_rank(out["price_to_book"],  ascending=True)) * 100.0
    out["_v_ps"]   = (1.0 - pct_rank(out["price_to_sales"], ascending=True)) * 100.0
    out["_v_eveb"] = (1.0 - pct_rank(out["ev_to_ebitda"],   ascending=True)) * 100.0
    out["_v_ncm"]  = pct_rank(out["net_cash_to_market_cap"], ascending=True)  * 100.0
    out["value_score"] = nanmean_rows(out["_v_pb"], out["_v_ps"], out["_v_eveb"], out["_v_ncm"])

    # ------------------------------------------------------------------
    # Pillar 2: Financial Health
    # Higher margins and liquidity = better; lower leverage = better
    # Current ratio: bell curve (2.0 is ideal; < 1 or very high is unusual)
    # ------------------------------------------------------------------
    _cr_pct = pct_rank(out["current_ratio"], ascending=True)
    out["_h_gm"]  = pct_rank(out["gross_margin"],     ascending=True) * 100.0
    out["_h_fcf"] = pct_rank(out["fcf_margin"],       ascending=True) * 100.0
    out["_h_de"]  = (1.0 - pct_rank(out["debt_to_equity"], ascending=True)) * 100.0
    out["_h_cr"]  = _cr_pct.apply(
        lambda p: bell_score(p, target=0.65, sigma=0.22) if pd.notna(p) else np.nan
    )
    out["health_score"] = nanmean_rows(out["_h_gm"], out["_h_fcf"], out["_h_de"], out["_h_cr"])

    # ------------------------------------------------------------------
    # Pillar 3: Sentiment & Crowding
    # Short %: bell curve peaked at 75th pct (highly shorted but not extreme)
    # Drawdown: more beaten-down = higher contrarian score
    # Put/call ratio: more puts outstanding = more fear = opportunity
    # Beta: bell curve at 70th pct (moderately high = active/volatile name)
    # ------------------------------------------------------------------
    _short_pct = pct_rank(out["short_pct_float"], ascending=True)
    _beta_pct  = pct_rank(out["beta"],            ascending=True)
    out["_sc_short"] = _short_pct.apply(
        lambda p: bell_score(p, target=0.75, sigma=0.18) if pd.notna(p) else np.nan
    )
    out["_sc_draw"]  = pct_rank(out["drawdown_52w"],    ascending=True) * 100.0
    out["_sc_pcr"]   = pct_rank(out["put_call_ratio"],  ascending=True) * 100.0
    out["_sc_beta"]  = _beta_pct.apply(
        lambda p: bell_score(p, target=0.70, sigma=0.20) if pd.notna(p) else np.nan
    )
    out["sentiment_crowding_score"] = nanmean_rows(
        out["_sc_short"], out["_sc_draw"], out["_sc_pcr"], out["_sc_beta"]
    )

    # ------------------------------------------------------------------
    # Pillar 4: Quality & Momentum
    # ROE: higher is better
    # Revenue growth: bell curve at 60th pct (some growth, not hyped)
    # Analyst consensus: contrarian — more "sell" ratings = more neglected
    # Relative 52w return: underperformed S&P = beaten-down opportunity
    # ------------------------------------------------------------------
    _rev_pct = pct_rank(out["revenue_growth"], ascending=True)
    out["_q_roe"] = pct_rank(out["return_on_equity"], ascending=True) * 100.0
    out["_q_rev"] = _rev_pct.apply(
        lambda p: bell_score(p, target=0.60, sigma=0.25) if pd.notna(p) else np.nan
    )
    out["_q_ana"] = pct_rank(out["analyst_mean"],  ascending=True) * 100.0
    out["_q_rel"] = (1.0 - pct_rank(out["relative_52w"], ascending=True)) * 100.0
    out["quality_momentum_score"] = nanmean_rows(
        out["_q_roe"], out["_q_rev"], out["_q_ana"], out["_q_rel"]
    )

    # ------------------------------------------------------------------
    # Weighted total score
    # ------------------------------------------------------------------
    w = weights
    def _w(k: str) -> float:
        return float(w.get(k, 0.0))

    out = weighted_score_from_available_pillars(out, [
        ("value_score", "Value", _w("value")),
        ("health_score", "Financial health", _w("financial_health")),
        ("sentiment_crowding_score", "Sentiment/crowding", _w("sentiment_crowding")),
        ("quality_momentum_score", "Quality/momentum", _w("quality_momentum")),
    ])

    # ------------------------------------------------------------------
    # Guardrails
    # ------------------------------------------------------------------
    distress_cap = float(guardrails.get("distress_cap", 60))
    mask_distress = out.get("distress_flag", pd.Series(False, index=out.index)).astype(bool)
    out.loc[mask_distress, "total_score"] = np.minimum(
        out.loc[mask_distress, "total_score"], distress_cap
    )

    thr     = float(guardrails.get("dilution_penalty_threshold", 0.30))
    penalty = float(guardrails.get("dilution_penalty_points", 15))
    if "shares_growth_2y" in out.columns:
        mask_dilution = out["shares_growth_2y"].notna() & (out["shares_growth_2y"] >= thr)
        out.loc[mask_dilution, "total_score"] -= penalty

    out["total_score"] = out["total_score"].clip(lower=0, upper=100)

    # Drop internal scoring columns
    drop_cols = [c for c in out.columns if c.startswith("_")]
    out = out.drop(columns=drop_cols)

    return out


# ---------------------------------------------------------------------------
# Output writing
# ---------------------------------------------------------------------------

def generate_dashboard(scored: pd.DataFrame, outputs_dir: Path) -> None:
    """
    Bake scored data directly into dashboard.html (project root) so the user
    only ever needs to open that one file — no separate output copy needed.

    Strips any previous data injection before writing the new one, so the
    file stays clean across repeated runs.
    """
    dash_path = Path(__file__).parent.resolve() / "dashboard.html"
    if not dash_path.exists():
        print("  [dashboard] dashboard.html not found — skipping embed.")
        return

    html = dash_path.read_text(encoding="utf-8")

    # Strip any previous injection block so reruns don't stack up
    import re as _re
    html = _re.sub(
        r'<!-- RK_DATA_START -->.*?<!-- RK_DATA_END -->',
        '',
        html,
        flags=_re.DOTALL,
    )

    # Convert scored DataFrame to JSON-safe records
    compact_cols = [
        "ticker", "total_score", "score_confidence", "missing_pillars", "unscorable_reason",
        "value_score", "health_score", "sentiment_crowding_score", "quality_momentum_score",
        "sector", "industry", "quote_type", "type_disp", "country",
        "market_cap",
        "price_to_book", "price_to_sales", "ev_to_ebitda", "net_cash_to_market_cap",
        "gross_margin", "fcf_margin", "debt_to_equity", "current_ratio",
        "short_pct_float", "drawdown_52w", "put_call_ratio", "beta",
        "return_on_equity", "revenue_growth", "analyst_mean", "relative_52w",
        "distress_flag", "high_leverage_flag", "low_liquidity_flag", "shares_growth_2y",
        "error",
    ]
    cols = [c for c in compact_cols if c in scored.columns]
    records = (
        scored[cols]
        .where(scored[cols].notna(), other=None)
        .to_dict(orient="records")
    )
    for rec in records:
        for flag in ("distress_flag", "high_leverage_flag", "low_liquidity_flag"):
            if flag in rec and rec[flag] is not None:
                rec[flag] = bool(rec[flag])

    data_json = json.dumps(records, allow_nan=False, default=str)
    inject = (
        '<!-- RK_DATA_START -->\n'
        '<script>\n'
        f'(function(){{ var d={data_json};\n'
        'if(typeof ingestStocks==="function"){ ingestStocks(d); }'
        'else{ window.PRELOADED_DATA=d; } })();\n'
        '</script>\n'
        '<!-- RK_DATA_END -->\n'
    )

    # Insert just before </body>
    if "</body>" in html:
        html = html.replace("</body>", inject + "</body>", 1)
    else:
        html += "\n" + inject

    dash_path.write_text(html, encoding="utf-8")
    print(f"  [dashboard] Wrote {len(records)} stocks → {dash_path}")


def write_outputs(scored: pd.DataFrame, outputs_dir: Path, suffix: str = "") -> None:
    scored.to_csv(outputs_dir / f"scores{suffix}.csv", index=False)

    compact_cols = [
        "ticker", "total_score", "score_confidence", "missing_pillars", "unscorable_reason",
        "value_score", "health_score", "sentiment_crowding_score", "quality_momentum_score",
        "sector", "industry", "quote_type", "type_disp", "country",
        "market_cap",
        "price_to_book", "price_to_sales", "ev_to_ebitda", "net_cash_to_market_cap",
        "gross_margin", "fcf_margin", "debt_to_equity", "current_ratio",
        "short_pct_float", "drawdown_52w", "put_call_ratio", "beta",
        "return_on_equity", "revenue_growth", "analyst_mean", "relative_52w",
        "distress_flag", "high_leverage_flag", "low_liquidity_flag", "shares_growth_2y",
        "error",
    ]
    compact_cols = [c for c in compact_cols if c in scored.columns]
    scored[compact_cols].to_csv(outputs_dir / f"scores_compact{suffix}.csv", index=False)


# ---------------------------------------------------------------------------
# Pipeline orchestration
# ---------------------------------------------------------------------------

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  = outputs_dir / cfg.get("results_jsonl", "raw_rows.jsonl")

    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("scoring_weights", {})
    guardrails = cfg.get("guardrails", {})
    scored     = build_scores(df, weights=weights, guardrails=guardrails)
    scored     = scored.sort_values(["total_score", "ticker"], ascending=[False, True])
    write_outputs(scored, outputs_dir)
    return outputs_dir / "scores.csv"


def run(
    config_path: str,
    universe_csv: Optional[str] = None,
    resume: bool = False,
    delay_override: Optional[float] = None,
    workers_override: Optional[int] = None,
) -> Path:
    """
    delay_override: if set, overrides min_delay_seconds from config (seconds between API calls).
    workers_override: if set, overrides max_workers from config.
    """
    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("scoring_weights", {})
    guardrails  = cfg.get("guardrails", {})
    checkpoint_every = int(cfg.get("checkpoint_every", 50))
    results_jsonl    = outputs_dir / cfg.get("results_jsonl", "raw_rows.jsonl")
    min_delay        = float(delay_override if delay_override is not None else cfg.get("min_delay_seconds", 0.75))
    max_workers      = int(workers_override if workers_override is not None else cfg.get("max_workers", 8))

    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. Set 'universe' in config.yaml or pass --universe_csv.")

    if resume:
        done = read_jsonl_tickers(results_jsonl)
    else:
        done = set()
        if results_jsonl.exists():
            results_jsonl.write_text("")  # truncate — start fresh when not resuming

    todo = [t.upper().strip() for t in universe if t.upper().strip() and t.upper().strip() not in done]

    if not todo:
        final_path = finalize_from_jsonl(config_path)
        print(f"Wrote final scores: {final_path}")
        return final_path

    # Bulk-prefetch histories first to populate cache
    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_for_ticker(cache, t)
        except Exception as e:
            row = {"ticker": t, "error": str(e)}
        row["processed_utc"] = dt.datetime.utcnow().isoformat()

        # Thread-safe append to JSONL
        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, guardrails=guardrails)
                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")
                    generate_dashboard(sc, outputs_dir)
                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}")

    final_path = finalize_from_jsonl(config_path)
    print(f"Wrote final scores: {final_path}")
    return final_path


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Roaring Kitty-style contrarian value screener (yfinance edition)"
    )
    parser.add_argument("--config",       default="config.yaml",
                        help="Path to config YAML (default: config.yaml)")
    parser.add_argument("--universe_csv", default=None,
                        help="CSV with ticker column; sorted by market cap desc")
    parser.add_argument("--resume",       action="store_true",
                        help="Skip tickers already in raw_rows.jsonl")
    parser.add_argument("--finalize_only", action="store_true",
                        help="Rebuild scores.csv from existing raw_rows.jsonl, no fetching")
    parser.add_argument("--delay",   type=float, default=None,
                        help="Override min_delay_seconds from config (e.g. 0.5 for faster runs)")
    parser.add_argument("--workers", type=int,   default=None,
                        help="Override max_workers from config (e.g. 10 for more parallelism)")
    args = parser.parse_args()

    if args.finalize_only:
        path = finalize_from_jsonl(args.config)
        print(f"Wrote scoring table to: {path}")
    else:
        run(
            args.config,
            universe_csv=args.universe_csv,
            resume=args.resume,
            delay_override=args.delay,
            workers_override=args.workers,
        )