momentum-system/research/daily_report.py

"""Daily after-close report: the two things that matter right now, plus the plumbing they depend on.

  1. the live opening-range bot's frozen 80-trade test (research/live_test_freeze.json)
  2. the judge v2 paper trial (judge/PAPER_TRIAL.md): paper vs simulator, trade by trade
  plus the old-vs-new exit A/B (paper stack 5433 vs live 5434) and stack / recorder health.

ALERTS at the top list anything the owner has to act on. Runs from Task Scheduler "Momentum\\DailyReport" on
weekdays after the close; read-only everywhere (PG_SKIP_SCHEMA, plain SELECTs).

usage: python -m research.daily_report [YYYY-MM-DD]
out:   data/reports/daily_<date>.md
"""
from __future__ import annotations

import json
import os
import random
import statistics as st
import sys
from datetime import datetime
from pathlib import Path
from urllib.parse import urlparse
from zoneinfo import ZoneInfo

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
os.environ["PG_SKIP_SCHEMA"] = "1"
from dotenv import dotenv_values  # noqa: E402

from research.live_scorecard import (TARGET_TRADES, bootstrap_mean_ci, config_drift, load_freeze,  # noqa: E402
                                     load_trades, verdict)

ET = ZoneInfo("America/New_York")
OUT = ROOT / "data" / "reports"
TRIAL_START = "2026-10-02"
AB_SINCE = "2026-09-14"


class Con:
    """Just enough of the research-DB interface for live_scorecard's helpers ('?' placeholders)."""

    def __init__(self, port: int):
        import psycopg2
        u = urlparse(dotenv_values(ROOT / ".env")["DATABASE_URL"])
        self.c = psycopg2.connect(host=u.hostname, port=port, user=u.username, password=u.password,
                                  dbname=u.path[1:], connect_timeout=10)
        self.c.autocommit = True
        with self.c.cursor() as cur:
            cur.execute("SET statement_timeout = 120000")

    def execute(self, sql, params=()):
        cur = self.c.cursor()
        cur.execute(sql.replace("%", "%%").replace("?", "%s"), list(params))
        return cur


def live_section(con, today, alerts) -> list[str]:
    freeze = load_freeze()
    since = freeze["frozen_on"]
    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 = len({ts.date() for ts, _ in con.execute(
        "SELECT timestamp, message FROM events WHERE event_type='risk_rule_triggered' AND timestamp >= ? "
        "AND message LIKE 'EQUITY FLOOR%'", [since]).fetchall() if 9 <= ts.hour < 16})
    eq = [json.loads(r[0]).get("total_equity") for r in con.execute(
        "SELECT payload_json FROM events WHERE event_type='account_summary_updated' AND timestamp >= ? "
        "ORDER BY timestamp DESC LIMIT 1", [today]).fetchall()]
    current = dict(con.execute("SELECT key, value FROM config_overrides").fetchall())
    drift = config_drift(freeze["settings"], current)
    v = verdict(rs, pnl, halts)
    lines = ["## 1. Live opening-range bot: frozen 80-trade test", "",
             f"- **{v}**  (since {since})"]
    if rs:
        lo, hi = bootstrap_mean_ci(rs)
        lines.append(f"- mean R {st.mean(rs):+.3f} (90% CI {lo:+.2f} .. {hi:+.2f}), win {sum(r > 0 for r in rs)}/{len(rs)}, "
                     f"realized P&L ${pnl:+.2f}")
    if eq and eq[0]:
        floor = float(freeze["settings"].get("TRADING_EQUITY_FLOOR", 215))
        lines.append(f"- equity ${float(eq[0]):.2f} (floor ${floor:.0f}, headroom ${float(eq[0]) - floor:.2f})")
        if float(eq[0]) - floor < 5:
            alerts.append(f"Live equity ${float(eq[0]):.2f} is within $5 of the ${floor:.0f} floor.")
    todays = [t for t in trades if str(t["ts"])[:10] == today]
    lines.append(f"- today: {len(todays)} trade(s)" + "".join(
        f"\n  - {t['symbol']} {t['r']:+.2f}R ${t['pnl']:+.2f} ({t['reason']})" if t["r"] is not None else
        f"\n  - {t['symbol']} ${t['pnl']:+.2f} ({t['reason']})" for t in todays))
    for d in freeze.get("known_deviations", []):
        if not d.get("to"):
            lines.append(f"- open deviation since {d['from']}: {d['what'][:140]}...")
    if drift:
        alerts.append("Live settings drifted from the freeze: " + ", ".join(f"{k} {a!r}->{b!r}" for k, (a, b) in drift.items()))
    if v.startswith("FAIL") or v.startswith("PASS"):
        alerts.append(f"Live test verdict: {v}")
    return lines + [""]


def ab_section(paper, live, alerts) -> list[str]:
    P, L = load_trades(paper, AB_SINCE), load_trades(live, AB_SINCE)
    used, diff = set(), []
    for x in L:
        for i, p in enumerate(P):
            if i not in used and str(p["ts"])[:10] == str(x["ts"])[:10] and p["symbol"] == x["symbol"] \
                    and p["r"] is not None and x["r"] is not None:
                used.add(i); diff.append(x["r"] - p["r"]); break
    lines = ["## Exit A/B: live 12% trail vs paper old tight exits (same stock, same day)", ""]
    if diff:
        rng = random.Random(1)
        bs = sorted(st.mean(rng.choices(diff, k=len(diff))) for _ in range(10000))
        lines.append(f"- pairs {len(diff)}; live minus old {st.mean(diff):+.3f}R (90% CI {bs[500]:+.2f} .. {bs[9500]:+.2f})")
        if len(diff) >= 50:
            alerts.append(f"Exit A/B reached {len(diff)} pairs: decision point on the live exits (after the 80-trade test).")
    return lines + [""]


def judge_section(today, alerts) -> list[str]:
    from judge.review import review
    jdir = ROOT / "data" / "judge"
    days = sorted(p.stem for p in (jdir / "journal").glob("*.jsonl") if p.stem >= TRIAL_START)
    lines = ["## 2. Judge v2 paper trial (judge/PAPER_TRIAL.md)", ""]
    all_f, today_df = [], None
    for d in days:
        if not any((jdir / "ticks").glob(f"{d}_*.pkl")):
            continue
        df, md = review(d)
        (jdir / "reviews").mkdir(exist_ok=True)
        (jdir / "reviews" / f"{d}.md").write_text(md, encoding="utf-8")
        if d == today:
            today_df = df
        if len(df):
            all_f.append(df.assign(day=d))
    import pandas as pd
    allx = pd.concat(all_f, ignore_index=True) if all_f else pd.DataFrame()
    fl = allx[allx.status == "filled"] if len(allx) else allx
    n = len(fl)
    lines.append(f"- counted trades {n}/60 (since {TRIAL_START}); orders {len(allx)}, unfilled "
                 f"{int((allx.status == 'unfilled').sum()) if len(allx) else 0}")
    if n:
        gap = (fl.paper_r - fl.sim_r).dropna().tolist()
        if gap:
            rng = random.Random(2)
            bs = sorted(st.mean(rng.choices(gap, k=len(gap))) for _ in range(10000))
            lines.append(f"- execution parity: paper minus sim {st.mean(gap):+.3f}R per trade "
                         f"(90% CI {bs[500]:+.2f} .. {bs[9500]:+.2f}); target >= -0.15R")
            if n >= 60:
                ok = st.mean(gap) >= -0.15 and bs[9500] >= 0
                alerts.append(f"Judge paper trial reached {n} trades: parity {'HOLDS' if ok else 'FAILS'} "
                              f"({st.mean(gap):+.3f}R). Decision per judge/PAPER_TRIAL.md.")
        lines.append(f"- paper R sum {fl.paper_r.sum():+.2f} (mean {fl.paper_r.mean():+.2f}); "
                     f"sim on the same fills {fl.sim_r.sum():+.2f}")
    if today_df is not None and len(today_df):
        tf = today_df[today_df.status == "filled"]
        flagged = tf[tf["flags"] != ""] if len(tf) else tf
        lines.append(f"- today: {len(tf)} filled, {len(today_df) - len(tf)} unfilled, {len(flagged)} flagged "
                     f"(details: data/judge/reviews/{today}.md)")
        for r in flagged.itertuples():
            lines.append(f"  - {r.sym}: paper {r.paper_r} vs sim {r.sim_r}  [{r.flags}]")
    j = jdir / "journal" / f"{today}.jsonl"
    if j.exists():
        rows = [json.loads(x) for x in j.read_text(encoding="utf-8").splitlines() if x.strip()]
        errs = [r for r in rows if r.get("kind") == "error"]
        starts = [r for r in rows if r.get("kind") == "start"]
        stopped = any(r.get("kind") == "stop" for r in rows)
        lines.append(f"- trader: {len(starts)} start(s), {len(errs)} error row(s), "
                     f"{'ended cleanly' if stopped else 'NO clean stop row'}")
        if len(errs) >= 10:
            from collections import Counter
            top = Counter((e.get("where"), str(e.get("err", ""))[:60]) for e in errs).most_common(1)[0]
            alerts.append(f"Judge trader logged {len(errs)} errors today (top: {top[0]} x{top[1]}). Possible loop.")
        if len(starts) > 1:
            alerts.append(f"Judge trader restarted {len(starts) - 1} time(s) today (crash or closed window).")
    elif datetime.now(ET).weekday() < 5:
        alerts.append("Judge trader wrote no journal today: did it run?")
    return lines + [""]


def health_section(paper, live, today, alerts) -> list[str]:
    lines = ["## Health", ""]
    for name, con in (("live 5434", live), ("paper 5433", paper)):
        try:
            first, last, n = con.execute(
                "SELECT min(timestamp), max(timestamp), count(*) FROM events WHERE event_type='module_tick' "
                "AND timestamp >= ? AND timestamp < ?", [f"{today} 09:30", f"{today} 16:00"]).fetchone()
            if first is not None and first.strftime("%H:%M") > "09:40":
                alerts.append(f"The {name} stack was down at the open: first heartbeat {first:%H:%M} ET.")
            gap = con.execute(
                "SELECT max(g) FROM (SELECT timestamp - lag(timestamp) OVER (ORDER BY timestamp) AS g FROM events "
                "WHERE event_type='module_tick' AND timestamp >= ? AND timestamp < ?) x",
                [f"{today} 09:30", f"{today} 16:00"]).fetchone()[0]
            gap_min = gap.total_seconds() / 60 if gap is not None else None
            lines.append(f"- {name}: {n} heartbeats in market hours, last {last}"
                         + (f", longest gap {gap_min:.0f} min" if gap_min is not None else ""))
            if gap_min is not None and gap_min >= 10:
                alerts.append(f"The {name} stack was silent for {gap_min:.0f} min during market hours.")
            if n == 0:
                alerts.append(f"The {name} stack wrote no heartbeats during market hours.")
            elif last and last.strftime("%H:%M") < "15:45":
                alerts.append(f"The {name} stack went silent at {last:%H:%M} ET.")
        except Exception as e:  # noqa: BLE001
            lines.append(f"- {name}: ERROR {str(e)[:120]}")
            alerts.append(f"Could not read the {name} database: {str(e)[:80]}")
    pw = ROOT / "data" / "prwire" / f"{today}.jsonl"
    if pw.exists():
        rows = [json.loads(x) for x in pw.read_text(encoding="utf-8").splitlines() if x.strip()]
        items = [r for r in rows if r.get("guid")]
        lines.append(f"- press-release recorder: {len(items)} items today, last seen {items[-1]['seen_utc'] if items else '-'}")
        if not items:
            alerts.append("The press-release recorder logged nothing today.")
    else:
        alerts.append("The press-release recorder has no file for today.")
    return lines + [""]


def main(argv=None) -> str:
    argv = sys.argv[1:] if argv is None else argv
    today = argv[0] if argv else datetime.now(ET).strftime("%Y-%m-%d")
    alerts: list[str] = []
    body: list[str] = []
    live = paper = None
    try:
        live = Con(5434)
        body += live_section(live, today, alerts)
    except Exception as e:  # noqa: BLE001
        alerts.append(f"Live database unreachable: {str(e)[:100]}")
    body += judge_section(today, alerts)
    try:
        paper = Con(5433)
        if live:
            body += ab_section(paper, live, alerts)
    except Exception as e:  # noqa: BLE001
        alerts.append(f"Paper database unreachable: {str(e)[:100]}")
    if live and paper:
        body += health_section(paper, live, today, alerts)
    head = [f"# Daily report {today}", "", "### ALERTS" if alerts else "### No alerts: nothing needs action.", ""]
    head += [f"- {a}" for a in alerts] + ([""] if alerts else [])
    md = "\n".join(head + body)
    OUT.mkdir(parents=True, exist_ok=True)
    (OUT / f"daily_{today}.md").write_text(md, encoding="utf-8")
    print(md)
    return md


if __name__ == "__main__":
    main()