"""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()