momentum-system/research/live_scorecard.py

"""Live-test scorecard: is the frozen strategy making money on the real account?

The test was pre-registered on 2026-09-25, after the commission leak was closed
(first clean session 2026-09-24):

  PASS  after >= 80 round trips since the freeze, mean R > 0 AND realized P&L > 0.
  FAIL  the equity-floor kill switch halts a session.
  Otherwise the test is still running. Any change to a strategy setting in
  research/live_test_freeze.json restarts the count, so this report flags drift.

Why 80: per-trade R ranges roughly -1 to +10 with a stdev of ~1.4, so under ~80
trades a positive mean cannot be told apart from zero. Why no filters were
changed at the freeze: the 2026-09-25 study (239 real fills, re-simulated under
the current exit) found that capping the entry chase lost R at every cap, and
that no break-time factor, including the L1/L2 book, held its sign across both
halves of the sample.

Run against the LIVE (tradier) database:
    DATABASE_URL=<tradier dsn> PG_SKIP_SCHEMA=1 python -m research.live_scorecard
"""

from __future__ import annotations

import argparse
import json
import random
from pathlib import Path

from research.multi_schema import open_research_db

FREEZE_FILE = Path(__file__).with_name("live_test_freeze.json")
TARGET_TRADES = 80
# safety knobs: changing these does not change the strategy under test
SAFETY_KEYS = {"TRADING_EQUITY_FLOOR", "TRADING_DEFAULT_EQUITY"}


def load_freeze(path: Path = FREEZE_FILE) -> dict:
    return json.loads(path.read_text())


def config_drift(frozen: dict, current: dict) -> dict:
    """{key: (frozen, current)} for strategy settings that differ from the freeze."""
    keys = set(frozen) | {k for k in current if k.startswith(("TRADING_", "TRIGGER_"))}
    return {k: (frozen.get(k), current.get(k)) for k in sorted(keys - SAFETY_KEYS)
            if frozen.get(k) != current.get(k)}


def bootstrap_mean_ci(xs: list[float], n: int = 4000, seed: int = 7) -> tuple[float, float]:
    if not xs:
        return (float("nan"), float("nan"))
    rng = random.Random(seed)
    means = sorted(sum(rng.choices(xs, k=len(xs))) / len(xs) for _ in range(n))
    return means[int(0.05 * n)], means[int(0.95 * n) - 1]


def verdict(r_values: list[float], pnl: float, floor_halts: int,
            target: int = TARGET_TRADES) -> str:
    if floor_halts:
        return "FAIL: the equity floor halted trading"
    if len(r_values) < target:
        return f"RUNNING: {len(r_values)}/{target} trades"
    mean_r = sum(r_values) / len(r_values)
    if mean_r > 0 and pnl > 0:
        return "PASS: positive per-trade R and positive P&L"
    return "FAIL: not profitable after the full sample"


def journal_trades(con, since: str) -> list[dict]:
    """Closed round trips the event store journaled since ``since``, with R measured
    against the stop the order was submitted with (position_closed carries the
    final, trailed stop)."""
    closes = con.execute(
        "SELECT timestamp, payload_json FROM events WHERE event_type='position_closed' "
        "AND timestamp >= ? ORDER BY timestamp", [since]).fetchall()
    out = []
    for ts, payload in closes:
        p = json.loads(payload or "{}")
        sym = p.get("symbol")
        row = con.execute(
            "SELECT payload_json FROM events WHERE event_type='order_approval_requested' "
            "AND payload_json LIKE ? AND timestamp::date = ?::date AND timestamp < ? "
            "ORDER BY timestamp DESC LIMIT 1", [f'%"symbol": "{sym}"%', ts, ts]).fetchone()
        stop0 = None
        if row:
            stop0 = (json.loads(row[0]).get("execution_request") or {}).get("stop_loss_price")
        entry, exit_p = p.get("entry_price"), p.get("exit_price")
        r = None
        if stop0 and entry and exit_p is not None and float(entry) > float(stop0):
            r = (float(exit_p) - float(entry)) / (float(entry) - float(stop0))
        out.append({"ts": ts, "symbol": sym, "pnl": float(p.get("realized_pnl") or 0.0),
                    "r": r, "reason": p.get("exit_reason"), "source": "journal"})
    return out


# The broker's order list is polled into account_orders_updated snapshots. Each
# row below is one FILLED order, taken from the latest snapshot that showed it
# (prices are final there) with the time it was FIRST seen filled as its fill time.
_BROKER_FILLS_SQL = """
SELECT ts, first_seen, sym, side, typ, fq, fpx, stop_px FROM (
  SELECT e.timestamp AS ts,
         min(e.timestamp) OVER (PARTITION BY o->>'broker_order_id') AS first_seen,
         o->>'symbol' AS sym, o->>'side' AS side, o->>'type' AS typ,
         (o->>'filled_quantity')::float AS fq, (o->>'filled_avg_price')::float AS fpx,
         (o->>'stop_price')::float AS stop_px,
         row_number() OVER (PARTITION BY o->>'broker_order_id' ORDER BY e.timestamp DESC) AS rn
  FROM events e, jsonb_array_elements(e.payload_json::jsonb->'orders') o
  WHERE e.event_type='account_orders_updated' AND e.timestamp >= ?
    AND o->>'status'='filled' AND o->>'type' IS NOT NULL
    AND (o->>'filled_quantity')::float > 0 AND o->>'filled_avg_price' IS NOT NULL
) x WHERE rn = 1 ORDER BY first_seen
"""


def broker_fills(con, since: str) -> list[dict]:
    """Filled broker orders since ``since`` (one row per broker order id)."""
    rows = con.execute(_BROKER_FILLS_SQL, [since]).fetchall()
    return [{"ts": first_seen, "symbol": sym, "side": side, "type": typ, "qty": float(fq),
             "price": float(fpx), "stop": float(stop_px) if stop_px is not None else None}
            for _ts, first_seen, sym, side, typ, fq, fpx, stop_px in rows]


def reconstruct_missing(fills: list[dict], journal: list[dict]) -> list[dict]:
    """Round trips the broker filled but the journal never closed.

    TNON on 2026-10-02 was bought and stopped out 49 s later; the stop filled
    before a position record existed, so no position_closed was ever written and
    the scorecard under-counted the test by one full -1R loser. For every
    (session date, symbol) with a filled buy AND a filled sell at the broker but
    no journaled close, this synthesises the trade from the fills: entry/exit are
    the fill VWAPs, the stop is the sell-stop order's stop price (so R is measured
    the same way as journaled trades). Pure: no DB, unit-tested.
    """
    journaled = {(str(t["ts"])[:10], t["symbol"]) for t in journal}
    by_day: dict[tuple[str, str], list[dict]] = {}
    for f in fills:
        by_day.setdefault((str(f["ts"])[:10], f["symbol"]), []).append(f)
    out = []
    for key, fs in sorted(by_day.items()):
        if key in journaled:
            continue
        buys = [f for f in fs if f["side"] == "buy"]
        sells = [f for f in fs if f["side"] == "sell"]
        if not buys or not sells:
            continue
        qty = min(sum(f["qty"] for f in buys), sum(f["qty"] for f in sells))
        if qty <= 0:
            continue
        entry = sum(f["qty"] * f["price"] for f in buys) / sum(f["qty"] for f in buys)
        exit_p = sum(f["qty"] * f["price"] for f in sells) / sum(f["qty"] for f in sells)
        stops = [f["stop"] for f in sells if f["type"] == "stop" and f["stop"]]
        stop0 = stops[0] if stops else None
        r = (exit_p - entry) / (entry - stop0) if stop0 and entry > stop0 else None
        out.append({"ts": max(f["ts"] for f in sells), "symbol": key[1],
                    "pnl": round(qty * (exit_p - entry), 2), "r": r,
                    "reason": ("stop_loss" if stops else "market_exit") + " [broker fill; journal missed]",
                    "source": "broker"})
    return out


def load_trades(con, since: str, broker_check: bool = True) -> list[dict]:
    """All round trips since ``since``: the journaled ones plus any the broker
    filled that the journal missed (see reconstruct_missing). ``broker_check``
    is for the LIVE Tradier store; the paper store's order snapshots run to
    gigabytes and its Alpaca journal is reconciled by alpaca_paper/sync instead."""
    journal = journal_trades(con, since)
    if not broker_check:
        return journal
    try:
        extra = reconstruct_missing(broker_fills(con, since), journal)
    except Exception as exc:  # noqa: BLE001 — the cross-check must never hide the journal
        print(f"(broker cross-check unavailable: {str(exc)[:120]})")
        extra = []
    return sorted(journal + extra, key=lambda t: str(t["ts"]))


def equity_path(con, since: str) -> tuple[float | None, float | None]:
    rows = con.execute(
        "SELECT payload_json FROM events WHERE event_type='account_summary_updated' "
        "AND timestamp >= ? ORDER BY timestamp", [since]).fetchall()
    eq = [json.loads(r[0]).get("total_equity") for r in rows]
    eq = [float(e) for e in eq if e]
    return (eq[0], eq[-1]) if eq else (None, None)


def floor_halts(con, since: str) -> int:
    """Sessions the kill switch halted during market hours (the midnight re-check
    that fires while equity is still below the floor is not a new halt)."""
    rows = con.execute(
        "SELECT timestamp, message FROM events WHERE event_type='risk_rule_triggered' "
        "AND timestamp >= ? AND message LIKE 'EQUITY FLOOR%'", [since]).fetchall()
    return len({ts.date() for ts, _ in rows if 9 <= ts.hour < 16})


def main() -> None:
    ap = argparse.ArgumentParser(description="live-test scorecard since the freeze")
    ap.add_argument("--since", help="override the freeze date (YYYY-MM-DD)")
    args = ap.parse_args()
    freeze = load_freeze()
    since = args.since or freeze["frozen_on"]
    con = open_research_db("market")

    trades = load_trades(con, since)
    rs = [t["r"] for t in trades if t["r"] is not None]
    pnl = sum(t["pnl"] for t in trades)
    halts = floor_halts(con, since)
    eq0, eq1 = equity_path(con, since)
    current = dict(con.execute("SELECT key, value FROM config_overrides").fetchall())
    drift = config_drift(freeze["settings"], current)

    print(f"Live test since {since}  ({len(trades)} round trips)\n")
    for t in trades:
        r = f"{t['r']:+.2f}R" if t["r"] is not None else "   n/a"
        print(f"  {str(t['ts'])[:16]}  {t['symbol']:6s} {r}  ${t['pnl']:+8.2f}  {t['reason']}")
    if rs:
        lo, hi = bootstrap_mean_ci(rs)
        wins = sum(1 for r in rs if r > 0)
        print(f"\nmean R {sum(rs) / len(rs):+.3f}  (90% CI {lo:+.3f} .. {hi:+.3f})"
              f"  win rate {wins}/{len(rs)}")
    print(f"realized P&L ${pnl:+.2f}")
    missed = [t for t in trades if t.get("source") == "broker"]
    if missed:
        print(f"journal missed {len(missed)} round trip(s) the broker filled; counted from the fills: "
              + ", ".join(f"{str(t['ts'])[:10]} {t['symbol']}" for t in missed))
    if eq0 is not None:
        print(f"equity ${eq0:.2f} -> ${eq1:.2f}  (difference from realized P&L = fees + deposits/refunds)")
    print(f"floor halts: {halts}")
    if drift:
        print("\nSETTINGS CHANGED SINCE THE FREEZE (the count restarts from the change):")
        for k, (a, b) in drift.items():
            print(f"  {k}: {a!r} -> {b!r}")
    print(f"\n{verdict(rs, pnl, halts)}")


if __name__ == "__main__":
    main()