momentum-system/research/premarket_eval.py

"""Accumulate-then-re-validate the premarket (pm1) shadow track.

The pre-registered verdict on 2026-07-01 was PROMISING-BUT-UNDERPOWERED: the forward moves are
REAL (avg +11-14% upside, unlike the RTH levers' tight-stop artifacts) but n=21 is far too small
for the +1R Wilson bound to conclude. The live loop ingests premarket bars every trading day, so
this job rebuilds pm1 from the (growing) DB and appends a timestamped metrics row to a history
file. Run it on a schedule (systemd timer) or by hand; watch `n` climb and the Wilson LB tighten
until the locked bar is decisively cleared or failed.

  python -m research.premarket_eval
"""

import os
from datetime import datetime, timezone

from dotenv import load_dotenv

from research import labeler
from research.multi_schema import open_research_db

# config parity with the live loop: DB-backed knobs (config_overrides) beat
# compose env — a bare env would replay under DEFAULT exits (the 4.5x-baseline
# landmine, orderflow-shelf-verdicts 2026-07-17)
from runtime.env_overrides import load as _load_cfg_overrides  # noqa: E402
_load_cfg_overrides()

load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env"))

HIST = os.path.join(os.path.dirname(__file__), "..", "data", "premarket_eval_history.tsv")
_HEADER = "utc_ts\tn\tdays\trate_1r\twilson_lb\tmedian_up_60m\tmedian_dd_60m\tbig_gap_1r\tpassed\n"


def _refresh_shadow_enrichment(con, session_date) -> None:
    """Shadow enrichment: float cache (FMP) + SEC dilution filings for the
    session's actively-traded names. SHADOW ONLY — neither gates entries
    (float refuted as a gate 2026-06-29; filings veto needs a pre-registered
    bar first, plan 016). Two INDEPENDENT try-blocks: float failing must
    never skip dilution, and vice versa (previously they shared one
    try-block, so an UndefinedTable on float_cache silently skipped the
    dilution refresh too)."""
    syms = [r[0] for r in con.execute(
        "SELECT symbol FROM minute_bars WHERE session_date = ? "
        "GROUP BY symbol ORDER BY sum(volume) DESC LIMIT 30", [session_date]).fetchall()]
    if not syms:
        return
    try:
        from research.ingestion.float_cache import refresh_float_cache
        print("float cache:", refresh_float_cache(con, syms))
    except Exception as exc:  # noqa: BLE001 — enrichment never breaks the eval
        print(f"float cache enrichment skipped: {exc}")
    try:
        from research.ingestion.sec_dilution import refresh_dilution_events
        print("dilution filings:", refresh_dilution_events(con, syms))
    except Exception as exc:  # noqa: BLE001 — enrichment never breaks the eval
        print(f"dilution enrichment skipped: {exc}")


def _metrics(con) -> dict:
    rows = con.execute(
        "SELECT l.max_upside_next_60m, l.max_drawdown_next_60m, "
        "CASE WHEN l.reached_1r_before_minus_1r THEN 1 ELSE 0 END, se.gap_pct "
        "FROM outcome_labels l JOIN setup_events se ON se.setup_id = l.setup_id "
        "WHERE se.setup_version=?", [labeler.PREMARKET_SETUP_VERSION]).fetchall()
    n = len(rows)
    days = con.execute("SELECT count(DISTINCT session_date) FROM setup_events WHERE setup_version=?",
                       [labeler.PREMARKET_SETUP_VERSION]).fetchone()[0]
    if not n:
        return {"n": 0, "days": days, "rate_1r": 0.0, "wilson_lb": 0.0,
                "median_up_60m": 0.0, "median_dd_60m": 0.0, "big_gap_1r": 0.0, "passed": False}
    r1 = [r[2] for r in rows]
    big = [r[2] for r in rows if (r[3] or 0) >= 0.5]
    med_up = labeler._median([r[0] for r in rows]) or 0.0
    med_dd = labeler._median([r[1] for r in rows]) or 0.0
    rate_1r = sum(r1) / n
    rate_big = (sum(big) / len(big)) if big else 0.0
    passed = (med_up >= labeler.PREMARKET_PROMO_MEDIAN_FWDMAX
              and med_dd > labeler.PREMARKET_PROMO_MEDIAN_ADVERSE
              and rate_1r > labeler.PREMARKET_PROMO_MIN_1R
              and rate_big > labeler.PREMARKET_PROMO_MIN_1R)
    return {"n": n, "days": days, "rate_1r": rate_1r,
            "wilson_lb": labeler._wilson_lower(sum(r1), n),
            "median_up_60m": med_up, "median_dd_60m": med_dd,
            "big_gap_1r": rate_big, "passed": passed}


def main() -> None:
    con = open_research_db("market")
    # upgrade today's bars to the SIP consolidated tape first (live ingest is IEX-only,
    # ~2-5% of true volume) so the pm1 rebuild + every downstream signal sees real volume
    try:
        from alpaca_paper.client import AlpacaPaperClient
        from alpaca_paper.settings import AlpacaPaperSettings
        from datetime import date as _date
        from research.sip_backfill import refetch_minute_session
        _client = AlpacaPaperClient(AlpacaPaperSettings.from_env())
        today = _date.today()
        # estimator audit (pre): today's stored volumes are ratio-SCALED IEX estimates —
        # snapshot them before the SIP truth overwrites, then measure the error.
        est = dict(con.execute(
            "SELECT symbol, sum(volume) FROM minute_bars WHERE session_date=? "
            "AND source_provider='alpaca_iex_scaled' GROUP BY symbol", [today]).fetchall())
        n, rows = refetch_minute_session(con, _client, today)
        print(f"SIP refetch today: {n} symbols -> {rows} rows")
        if est:
            errs = []
            for sym, ev in est.items():
                tv = con.execute("SELECT sum(volume) FROM minute_bars WHERE symbol=? "
                                 "AND session_date=?", [sym, today]).fetchone()[0]
                if tv:
                    errs.append((sym, float(ev) / float(tv) - 1.0))
            if errs:
                errs.sort(key=lambda x: abs(x[1]))
                med = errs[len(errs) // 2][1]
                worst = max(errs, key=lambda x: abs(x[1]))
                line = (f"{today}\t{len(errs)}\t{med:+.3f}\t{worst[0]}\t{worst[1]:+.3f}")
                print(f"volume-estimator audit: n={len(errs)} median_err={med:+.1%} "
                      f"worst={worst[0]} {worst[1]:+.1%}")
                with open(os.path.join(os.path.dirname(__file__), "..", "data",
                                       "volume_estimator_audit.tsv"), "a") as f:
                    f.write(line + "\n")
    except Exception as exc:  # noqa: BLE001 — never let the refetch break the eval
        print(f"SIP refetch skipped: {exc}")
    # nightly DB housekeeping: observability-spam events (module ticks, per-symbol criteria
    # evaluations, account snapshots) are only useful for a few days of debugging but were
    # ~80% of the events table (3.5 GB before the 2026-07-04 purge). Trading-decision events
    # are kept forever. Autovacuum (tuned to 2%) reclaims the space.
    try:
        spam = ("module_tick", "criteria_evaluated", "signal_blocked",
                "account_summary_updated", "account_positions_updated", "account_orders_updated",
                "entry_gate_evaluated")
        ph = ",".join(["?"] * len(spam))
        cur = con.execute(
            f"DELETE FROM events WHERE event_type IN ({ph}) "
            "AND timestamp < now() - interval '14 days'", list(spam))
        print(f"events retention: purged {getattr(cur, 'rowcount', '?')} spam rows >14d")
    except Exception as exc:  # noqa: BLE001 — housekeeping must never break the eval
        print(f"events retention skipped: {exc}")
    # log rotation: systemd units append to data/*.log with no bound (live_run.log hit 12 MB
    # in 3 weeks). Copy-then-truncate is safe with O_APPEND writers; keep one generation.
    try:
        import glob
        import shutil
        for lp in glob.glob(os.path.join(os.path.dirname(__file__), "..", "data", "*.log")):
            if os.path.getsize(lp) > 20 * 1024 * 1024:
                shutil.copy2(lp, lp + ".1")
                with open(lp, "r+") as f:
                    f.truncate(0)
                print(f"rotated {os.path.basename(lp)}")
    except Exception as exc:  # noqa: BLE001
        print(f"log rotation skipped: {exc}")
    # shadow enrichment: float cache (FMP) + SEC dilution filings for the day's
    # actively-traded names. SHADOW ONLY — neither gates entries (float refuted
    # as a gate 2026-06-29; filings veto needs a pre-registered bar first).
    try:
        from zoneinfo import ZoneInfo
        session_date = datetime.now(ZoneInfo("America/New_York")).date()
        _refresh_shadow_enrichment(con, session_date)
    except Exception as exc:  # noqa: BLE001 — enrichment never breaks the eval
        print(f"float/dilution enrichment skipped: {exc}")
    labeler.build_premarket(con, rebuild=True)          # rebuild from the grown DB
    m = _metrics(con)
    ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
    row = (f"{ts}\t{m['n']}\t{m['days']}\t{m['rate_1r']:.3f}\t{m['wilson_lb']:.3f}\t"
           f"{m['median_up_60m']:.4f}\t{m['median_dd_60m']:.4f}\t{m['big_gap_1r']:.3f}\t{m['passed']}")
    new = not os.path.exists(HIST)
    with open(HIST, "a") as f:
        if new:
            f.write(_HEADER)
        f.write(row + "\n")
    print("appended:", row)
    labeler.validate_premarket(con)
    # DUAL-WRITE (plan 006): also emit pm1's verdict through the experiment
    # harness (the TSV append above stays until the owner retires it). Fully
    # guarded — the harness must never break the nightly eval.
    try:
        from research.experiments import adjudicate, registry
        from storage.event_store import EventStore
        store = EventStore(os.environ.get("WATCHER_EVENT_DB_PATH", "momentum"))
        try:
            registry.apply_schema(store.con)
            pm1_json = os.path.join(os.path.dirname(__file__), "..", "experiments", "pm1_v1.json")
            if registry.get(store.con, "pm1_v1") is None and os.path.exists(pm1_json):
                registry.register(store.con, pm1_json)
            for v in adjudicate.adjudicate(store):
                if v["experiment_id"] == "pm1_v1":
                    print(f"harness pm1_v1 verdict: {v['verdict']} (n={v['n']})")
        finally:
            store.close()
    except Exception as exc:  # noqa: BLE001 — harness bridge never breaks the eval
        print(f"experiment harness bridge skipped: {exc}")


if __name__ == "__main__":
    main()