quant-trading/agent_review.py

"""
Daily loop of the agent-reviewed strategy ``overnight_recovery_agent_top10``
(strategies/agent_review.py): build the candidate pack, store the agent's
scores with a point-in-time stamp, and measure the scores against what
happened.

    uv run python agent_review.py candidates [--refresh] [--analyst-refresh]
    uv run python agent_review.py submit data/agent/us/draft_<as_of>.json
    uv run python agent_review.py evaluate                  # IC, vetoes, book vs control

Run ``candidates`` after the US close (the monitor's evening phase stores the
bars at 16:35 New York time); ``submit`` must finish before 16:00 New York
time of the next session for the scores to steer that session's close.
The skill .claude/skills/agent-stock-review drives the research step.
"""

from __future__ import annotations

import argparse
import glob
import json
import os
import sys
from datetime import datetime, timezone

import numpy as np
import pandas as pd

from strategies import agent_review as ar

NAME = "overnight_recovery_agent_top10"
CONTROL = "overnight_recovery_top10"
ROOT = os.path.dirname(os.path.abspath(__file__))
LINKS = os.path.join(ROOT, "data", "research", "sec_10k_links.csv")
SECTORS = os.path.join(ROOT, "data", "us_sectors.csv")
SIC = [os.path.join(ROOT, "data", "research", f) for f in ("sec_sic.csv", "mid_sec_sic.csv")]   # names of newer members


def _next_session(as_of: pd.Timestamp) -> pd.Timestamp:
    return as_of + pd.offsets.BDay(1)          # exchange holidays make the real one later


def _latest_scores_raw() -> dict | None:
    files = sorted(glob.glob(os.path.join(ar.SCORE_DIR, "scores_*.json")))
    if not files:
        return None
    with open(files[-1]) as fh:
        return json.load(fh)


def _links(tickers: list[str]) -> dict[str, dict[str, list[str]]]:
    """Customers each candidate names in its latest 10-K, and the S&P 500
    filers that name it as a customer (research/lab/fetch_supply_links.py)."""
    if not os.path.exists(LINKS):
        return {}
    d = pd.read_csv(LINKS, usecols=["filer", "filed", "target", "n_customer"])
    d = d[d["n_customer"] > 0]
    latest = d.groupby("filer")["filed"].transform("max")
    d = d[d["filed"] == latest]
    out = {}
    for t in tickers:
        out[t] = {"customers_named": sorted(d.loc[d["filer"] == t, "target"].unique())[:8],
                  "named_as_customer_by": sorted(d.loc[d["target"] == t, "filer"].unique())[:8]}
    return out


def cmd_candidates(args) -> None:
    import trader
    tickers, bench = trader.strategy_universe("us", NAME)
    close, open_, close_raw = trader.load_strategy_prices("us", NAME, tickers, bench, with_open=True,
                                                          refresh=args.refresh)
    tickers = trader.filter_tradable_tickers(close, tickers)
    as_of = trader.completed_sessions(close_raw, "us", bench)[-1]
    close, open_ = close.loc[:as_of], open_.loc[:as_of]
    bars = trader.strategy_bars("us", NAME, tickers, close, open_)
    strat = trader.STRATEGY_REGISTRY[NAME]()
    cand = strat.candidates(bars)
    from research.lab.analyst_estimates import collect, candidate_context
    if args.analyst_refresh:
        collect(list(cand.index))
    generated_at = datetime.now(timezone.utc).isoformat()
    analyst = candidate_context(list(cand.index), generated_at)

    c = close[cand.index].ffill()
    r = lambda k: (c.iloc[-1] / c.iloc[-1 - k] - 1).round(4)                     # noqa: E731
    vol = c.pct_change(fill_method=None).iloc[-63:].std() * np.sqrt(252)
    off_high = c.iloc[-1] / c.iloc[-252:].max() - 1
    sec = pd.read_csv(SECTORS).set_index("ticker") if os.path.exists(SECTORS) else pd.DataFrame()
    frames = [pd.read_csv(p) for p in SIC if os.path.exists(p)]
    sic = (pd.concat(frames).drop_duplicates("ticker").set_index("ticker") if frames
           else pd.DataFrame(columns=["name", "sicDescription"]))
    held = {}
    for label, name in (("agent_strategy", NAME), ("control", CONTROL)):
        st = trader.load_state("us", name)
        last = st.get("daily_log", [{}])[-1] if st.get("daily_log") else {}
        eq = last.get("equity") or 1.0
        held[label] = {t: round(h["value"] / eq, 4) for t, h in last.get("holdings", {}).items()}
    prev = _latest_scores_raw()
    prev_rows = prev.get("scores", {}) if prev else {}
    links = _links(list(cand.index))

    rows = []
    for t, row in cand.iterrows():
        p = prev_rows.get(t)
        rows.append({
            "ticker": t, "rank": int(row["rank"]), "quant_pct": round(float(row["quant_pct"]), 4),
            "name": sec["Security"].get(t) if "Security" in sec and t in sec.index else sic["name"].get(t),
            "sec_industry": sic["sicDescription"].get(t),
            "sector": sec["GICS Sector"].get(t) if "GICS Sector" in sec else None,
            "sub_industry": sec["GICS Sub-Industry"].get(t) if "GICS Sub-Industry" in sec else None,
            "ret_5d": float(r(5)[t]), "ret_21d": float(r(21)[t]), "ret_63d": float(r(63)[t]),
            "ret_252d": float(r(252)[t]), "vol_63d": round(float(vol[t]), 4),
            "off_52w_high": round(float(off_high[t]), 4),
            "overnight_mom_12m": float(row["overnight_mom_12m"]), "recovery_126d": float(row["recovery_126d"]),
            "held_weight": {k: v.get(t, 0.0) for k, v in held.items()},
            **links.get(t, {}),
            "analyst_estimates": analyst.get(t),
            "previous": ({"as_of": prev["as_of"], "score": p.get("score"), "veto": p.get("veto", False),
                          "summary": p.get("summary", "")} if p else None),
        })
    nxt = _next_session(as_of)
    k = len(close) - 1                        # pre-shift row of as_of: its weights trade at the next close
    until = (-(k - strat.warmup)) % strat.rebal_freq
    pack = {
        "strategy": NAME, "as_of": str(as_of.date()), "next_session": str(nxt.date()),
        "deadline_utc": str(ar.session_deadline(nxt)),
        # 0: the next session rebalances on these scores; otherwise only vetoes act until then
        "sessions_until_rebalance": int(until),
        "generated_at": generated_at,
        "analyst_estimates": {"covered": len(analyst),
                              "eligible": sum(r["eligible"] for r in analyst.values()),
                              "note": "Known only after snapshot collection; stale fiscal metadata is excluded from the research overlay."},
        "rule": {"agent_weight": ar.AGENT_WEIGHT, "score_range": list(ar.SCORE_RANGE),
                 "veto_reasons": list(ar.VETO_REASONS), "top_n": strat.top_n, "pool": strat.pool,
                 "rebal_freq": strat.rebal_freq},
        "holdings": held,
        "candidates": rows,
    }
    os.makedirs(ar.SCORE_DIR, exist_ok=True)
    path = os.path.join(ar.SCORE_DIR, f"candidates_{as_of.date()}.json")
    with open(path, "w") as fh:
        json.dump(pack, fh, indent=1, ensure_ascii=False, default=str)
    print(f"{len(rows)} candidates for the session after {as_of.date()} -> {path}")
    print(f"submit before {pack['deadline_utc']} (16:00 New York on {nxt.date()}) to steer that close; "
          f"{'the next session rebalances' if until == 0 else f'next rebalance in {until} sessions (vetoes act daily)'}")


def _source_day(src) -> pd.Timestamp | None:
    """A source's publication day (UTC), or None when it has no readable date."""
    try:
        ts = pd.Timestamp(src["date"])
    except (TypeError, KeyError, ValueError):
        return None
    if pd.isna(ts):
        return None
    if ts.tzinfo is not None:
        ts = ts.tz_convert("UTC").tz_localize(None)
    return ts.normalize()


def validate(sub: dict, pack: dict, now: pd.Timestamp) -> tuple[dict, list[str]]:
    """Normalized scores and warnings; raises ValueError on a bad file.
    ``now`` is the submission stamp: no source may be dated after its day (UTC)."""
    today = now.tz_convert("UTC").tz_localize(None).normalize()
    want = [c["ticker"] for c in pack["candidates"]]
    rows = sub.get("scores")
    if not isinstance(rows, dict):
        raise ValueError('"scores" must map ticker -> {score, veto, ...}')
    missing, extra = sorted(set(want) - set(rows)), sorted(set(rows) - set(want))
    if missing or extra:
        raise ValueError(f"tickers must match the {len(want)} candidates; missing {missing}, extra {extra}")
    lo, hi = ar.SCORE_RANGE
    out, warn = {}, []
    for t in want:
        v = rows[t]
        s = v.get("score")
        if not isinstance(s, int) or isinstance(s, bool) or not lo <= s <= hi:
            raise ValueError(f"{t}: score must be an integer in [{lo}, {hi}], got {s!r}")
        veto = bool(v.get("veto", False))
        reason = v.get("veto_reason")
        if veto and reason not in ar.VETO_REASONS:
            raise ValueError(f"{t}: veto_reason must be one of {ar.VETO_REASONS}, got {reason!r}")
        if veto and not str(v.get("risks", "")).strip():
            raise ValueError(f"{t}: a veto needs the evidence in 'risks'")
        if not str(v.get("summary", "")).strip():
            raise ValueError(f"{t}: 'summary' is required")
        sources = v.get("sources", [])
        if not isinstance(sources, list):
            raise ValueError(f"{t}: 'sources' must be a list of {{url, date}}")
        for src in sources:
            day = _source_day(src)
            if day is None:
                raise ValueError(f"{t}: source {src!r} needs a publication 'date' (YYYY-MM-DD)")
            if day > today:
                raise ValueError(f"{t}: source {src.get('url')} is dated {src['date']}, "
                                 f"after the submission day {today.date()} (UTC)")
        out[t] = {"score": s, "veto": veto, "veto_reason": reason if veto else None,
                  "confidence": v.get("confidence"),
                  **{k: v.get(k, "") for k in ("earnings", "fundamentals", "logic", "risks", "summary")},
                  "sources": sources}
    extreme = sum(abs(v["score"]) == hi for v in out.values())
    if extreme > 10:
        warn.append(f"{extreme} names at +/-{hi}; the rubric expects at most about 10")
    return out, warn


def cmd_submit(args) -> None:
    with open(args.file) as fh:
        sub = json.load(fh)
    as_of = sub.get("as_of")
    pack_path = os.path.join(ar.SCORE_DIR, f"candidates_{as_of}.json")
    if not os.path.exists(pack_path):
        sys.exit(f"no candidate pack for as_of={as_of!r} ({pack_path})")
    with open(pack_path) as fh:
        pack = json.load(fh)
    now = pd.Timestamp.now(tz="UTC")
    try:
        scores, warn = validate(sub, pack, now)
    except ValueError as exc:
        sys.exit(f"rejected: {exc}")
    deadline = pd.Timestamp(pack["deadline_utc"])
    out = {"strategy": NAME, "as_of": as_of, "created_at": now.isoformat(timespec="seconds"),
           "model": sub.get("model", ""), "n_candidates": len(scores), "scores": scores}
    path = os.path.join(ar.SCORE_DIR, f"scores_{as_of}_{now:%H%M%S}.json")
    with open(path, "w") as fh:
        json.dump(out, fh, indent=1, ensure_ascii=False)
    for w in warn:
        print(f"warning: {w}")
    if now > deadline:
        print(f"warning: stamped after the {pack['next_session']} deadline ({deadline}); "
              f"these scores steer later sessions only")
    ups = sorted((t for t, v in scores.items() if v["score"] > 0), key=lambda t: -scores[t]["score"])
    vetoes = [t for t, v in scores.items() if v["veto"]]
    print(f"stored {path}: {len(ups)} raised, {sum(v['score'] < 0 for v in scores.values())} lowered, "
          f"vetoes {vetoes or 'none'}")


def cmd_evaluate(args) -> None:
    import trader
    latest = {}                                   # the last submission of each as_of
    for p in sorted(glob.glob(os.path.join(ar.SCORE_DIR, "scores_*.json"))):
        latest[os.path.basename(p)[7:17]] = p
    files = [latest[k] for k in sorted(latest)]
    if not files:
        sys.exit("no score files yet")
    close = pd.read_csv(os.path.join(ROOT, "data", "us.csv"), index_col=0, parse_dates=True).sort_index()
    rows, vetoes = [], []
    for path in files:
        with open(path) as fh:
            d = json.load(fh)
        with open(os.path.join(ar.SCORE_DIR, f"candidates_{d['as_of']}.json")) as fh:
            pack = json.load(fh)
        after = close.index[close.index > pd.Timestamp(d["as_of"])]
        for h in (5, 10):
            if len(after) <= h:
                continue
            s0, s1 = after[0], after[h]
            q = {c["ticker"]: c["quant_pct"] for c in pack["candidates"]}
            df = pd.DataFrame({"agent": {t: v["score"] for t, v in d["scores"].items()}, "quant": q})
            df["fwd"] = (close.loc[s1] / close.loc[s0] - 1).reindex(df.index)
            df = df.dropna()
            rows.append({"as_of": d["as_of"], "h": h, "n": len(df),
                         "ic_agent": df["agent"].rank().corr(df["fwd"].rank()),
                         "ic_quant": df["quant"].rank().corr(df["fwd"].rank()),
                         "top_minus_rest": df.loc[df["agent"] > 0, "fwd"].mean() - df.loc[df["agent"] <= 0, "fwd"].mean()})
            if h == 10:
                for t, v in d["scores"].items():
                    if v.get("veto") and t in df.index:
                        vetoes.append({"as_of": d["as_of"], "ticker": t, "reason": v["veto_reason"],
                                       "fwd10_vs_candidates": df.at[t, "fwd"] - df["fwd"].mean()})
    R = pd.DataFrame(rows)
    print(f"{len(files)} scored sessions ({min(latest)} .. {max(latest)})")
    for h, g in (R.groupby("h") if len(R) else []):
        def t(x):
            x = x.dropna()
            return x.mean() / x.std() * np.sqrt(len(x)) if len(x) > 2 and x.std() > 0 else float("nan")
        print(f"  {h:>2}-day: files {len(g)} | IC agent {g['ic_agent'].mean():+.3f} (t {t(g['ic_agent']):.2f}) "
              f"| IC quant {g['ic_quant'].mean():+.3f} (t {t(g['ic_quant']):.2f}) "
              f"| raised minus rest {g['top_minus_rest'].mean()*100:+.2f}%  (overlapping windows inflate t)")
    if vetoes:
        V = pd.DataFrame(vetoes)
        print(f"  vetoes: {len(V)}, 10-day return vs candidate mean {V['fwd10_vs_candidates'].mean()*100:+.2f}% "
              f"(negative = the veto helped)")
    first = pd.Timestamp(json.load(open(files[0]))["as_of"])
    eq = {}
    for name in (NAME, CONTROL):
        e = pd.Series(trader.load_state("us", name).get("daily_equity", {}), dtype=float)
        e.index = pd.to_datetime(e.index)
        eq[name] = e.sort_index().loc[first:]
    E = pd.DataFrame(eq).dropna()
    if len(E) > 1:
        ret = E.iloc[-1] / E.iloc[0] - 1
        print(f"  books since {E.index[0].date()}: agent {ret[NAME]*100:+.2f}% vs control {ret[CONTROL]*100:+.2f}% "
              f"({len(E)} sessions)")


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd", required=True)
    p = sub.add_parser("candidates", help="write the candidate pack for the next session")
    p.add_argument("--refresh", action="store_true", help="download prices first (the monitor normally has)")
    p.add_argument("--analyst-refresh", action="store_true", help="archive current Yahoo analyst estimates before writing the pack")
    p = sub.add_parser("submit", help="validate and store an agent scores file")
    p.add_argument("file")
    sub.add_parser("evaluate", help="score IC, veto outcomes, and the book against its control")
    args = ap.parse_args()
    {"candidates": cmd_candidates, "submit": cmd_submit, "evaluate": cmd_evaluate}[args.cmd](args)


if __name__ == "__main__":
    main()