"""Premarket-structure labeler (Phase 1 of the premarket axis — OBSERVE ONLY). Motivation (2026-07-31, FCUV): the source trader's entry is the 9:30 break of premarket structure; our ORB fires ~10 minutes later into the move he exits. FCUV premarket ($1.98 -> $10.49) was fully captured in minute_bars while the ORB entered at $11.89 and scratched. This job distills each gapper's premarket tape into per-session features so entry-conversion can be analyzed against structure the market drew before the open. It gates NOTHING and trades NOTHING. PRE-REGISTERED HYPOTHESES (locked before any data is examined; Phase-2 shadow verdicts must cite these, not post-hoc slices — the keep-or-zero lesson): H1 Entries filled close to the premarket high (entry <= pm_high * 1.05) convert better than deep-extension entries (entry > pm_high * 1.25). H2 Sessions where the 9:25-9:29 price consolidates near the premarket high (open_vs_pm_high_pct >= -5%) convert better than faded opens (< -15%). H3 (Replay-only, Phase 2) A "premarket-high break at/after 9:30" entry beats the ORB entry on the same names. Minimum n=30 per bucket before ANY conclusion; promotion requires a pre-registered Phase-2 bar (fire-weighted >= +0.05R vs baseline), never a Phase-1 table. Commands: python -m research.premarket_structure build [--since N | --date YYYY-MM-DD] python -m research.premarket_structure report """ from __future__ import annotations import argparse import os import sys from dotenv import load_dotenv from runtime.env_overrides import load as _load_cfg_overrides _load_cfg_overrides() load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env")) MIN_PM_BARS = 5 MIN_PM_VOLUME = 10_000 _DDL = """ CREATE TABLE IF NOT EXISTS premarket_structure ( session_date DATE NOT NULL, symbol VARCHAR NOT NULL, first_ts TIMESTAMP, first_price DOUBLE PRECISION, pm_high DOUBLE PRECISION, pm_high_ts TIMESTAMP, pm_low DOUBLE PRECISION, pm_close DOUBLE PRECISION, -- last premarket close (~9:29 ET) pm_run_pct DOUBLE PRECISION, -- (pm_high - first_price) / first_price open_vs_pm_high_pct DOUBLE PRECISION, -- (pm_close - pm_high) / pm_high; ~0 = consolidating at high retrace_from_high_pct DOUBLE PRECISION, -- (pm_high - pm_close) / pm_high pm_vwap DOUBLE PRECISION, open_vs_pm_vwap_pct DOUBLE PRECISION, pm_volume BIGINT, pm_bars INTEGER, pullbacks_5pct INTEGER, -- jackknife count: >=5% drops from a running high computed_at TIMESTAMP DEFAULT current_timestamp, PRIMARY KEY (session_date, symbol) ); """ def compute_features(bars: list[dict]) -> dict | None: """Distill one symbol-session's premarket bars (chronological) to features. Each bar: {timestamp, open, high, low, close, volume}. Pure function — unit-testable without a database. """ bars = [b for b in bars if b.get("close") and b.get("volume") is not None] if len(bars) < MIN_PM_BARS: return None total_vol = sum(int(b["volume"]) for b in bars) if total_vol < MIN_PM_VOLUME: return None first = bars[0] first_price = float(first["open"] or first["close"]) if first_price <= 0: return None hi_bar = max(bars, key=lambda b: float(b["high"] or 0)) pm_high = float(hi_bar["high"]) pm_low = min(float(b["low"]) for b in bars if b.get("low")) pm_close = float(bars[-1]["close"]) # volume-weighted average price over the premarket session pv = sum(float(b["close"]) * int(b["volume"]) for b in bars) pm_vwap = pv / total_vol if total_vol else pm_close # jackknives: count >=5% drops measured from the running high; re-arm on a # new high so one long fade counts once pullbacks = 0 run_high = 0.0 armed = True for b in bars: h, low = float(b["high"] or 0), float(b["low"] or 0) if h > run_high: run_high = h armed = True if armed and run_high > 0 and low <= run_high * 0.95: pullbacks += 1 armed = False return { "first_ts": first["timestamp"], "first_price": round(first_price, 4), "pm_high": round(pm_high, 4), "pm_high_ts": hi_bar["timestamp"], "pm_low": round(pm_low, 4), "pm_close": round(pm_close, 4), "pm_run_pct": round((pm_high - first_price) / first_price, 4), "open_vs_pm_high_pct": round((pm_close - pm_high) / pm_high, 4), "retrace_from_high_pct": round((pm_high - pm_close) / pm_high, 4), "pm_vwap": round(pm_vwap, 4), "open_vs_pm_vwap_pct": round((pm_close - pm_vwap) / pm_vwap, 4) if pm_vwap > 0 else None, "pm_volume": total_vol, "pm_bars": len(bars), "pullbacks_5pct": pullbacks, } def build(con, session_date: str | None = None, since_days: int | None = None) -> dict: """Compute + upsert features for one session (default: latest) or a range.""" con.execute(_DDL) # session-scan index for this job (partial: premarket rows only). Plain # CREATE (not CONCURRENTLY) — this runs from the post-close nightly, when a # brief write lock on minute_bars is harmless. Building it mid-market # deadlocked against the live loop's ingest (2026-07-31). try: con.execute( "CREATE INDEX IF NOT EXISTS idx_minute_bars_session_pm " "ON minute_bars (session_date) WHERE is_premarket") except Exception: # noqa: BLE001 — index is an optimization, never a blocker pass con.commit() if session_date: dates = [session_date] else: n = since_days or 1 rows = con.execute( "SELECT DISTINCT session_date FROM minute_bars WHERE is_premarket " "ORDER BY session_date DESC LIMIT ?", [n]).fetchall() dates = [str(r[0]) for r in rows] written = 0 for d in dates: # one scan per session (per-symbol queries were ~100x slower on the # multi-GB bars table), grouped in memory — a session's premarket tape # is small (~250 symbols x ~200 bars) rows = con.execute( "SELECT symbol, timestamp, open, high, low, close, volume " "FROM minute_bars WHERE session_date = ? AND is_premarket " "ORDER BY symbol, timestamp", [d]).fetchall() by_sym: dict[str, list[dict]] = {} for r in rows: by_sym.setdefault(r[0], []).append( dict(zip(("timestamp", "open", "high", "low", "close", "volume"), r[1:]))) for sym, bars in by_sym.items(): f = compute_features(bars) if f is None: continue con.execute( """INSERT INTO premarket_structure (session_date, symbol, first_ts, first_price, pm_high, pm_high_ts, pm_low, pm_close, pm_run_pct, open_vs_pm_high_pct, retrace_from_high_pct, pm_vwap, open_vs_pm_vwap_pct, pm_volume, pm_bars, pullbacks_5pct, computed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, current_timestamp) ON CONFLICT (session_date, symbol) DO UPDATE SET first_ts=EXCLUDED.first_ts, first_price=EXCLUDED.first_price, pm_high=EXCLUDED.pm_high, pm_high_ts=EXCLUDED.pm_high_ts, pm_low=EXCLUDED.pm_low, pm_close=EXCLUDED.pm_close, pm_run_pct=EXCLUDED.pm_run_pct, open_vs_pm_high_pct=EXCLUDED.open_vs_pm_high_pct, retrace_from_high_pct=EXCLUDED.retrace_from_high_pct, pm_vwap=EXCLUDED.pm_vwap, open_vs_pm_vwap_pct=EXCLUDED.open_vs_pm_vwap_pct, pm_volume=EXCLUDED.pm_volume, pm_bars=EXCLUDED.pm_bars, pullbacks_5pct=EXCLUDED.pullbacks_5pct, computed_at=current_timestamp""", [d, sym, f["first_ts"], f["first_price"], f["pm_high"], f["pm_high_ts"], f["pm_low"], f["pm_close"], f["pm_run_pct"], f["open_vs_pm_high_pct"], f["retrace_from_high_pct"], f["pm_vwap"], f["open_vs_pm_vwap_pct"], f["pm_volume"], f["pm_bars"], f["pullbacks_5pct"]]) written += 1 con.commit() return {"sessions": len(dates), "rows": written} def report(con) -> None: """Descriptive join of live ORB closes vs premarket structure. NOT a verdict — H1/H2 buckets shown with n; conclusions wait for n>=30/bucket + Phase 2.""" rows = con.execute( """ SELECT (e.payload_json::json->>'entry_price')::float AS entry, (e.payload_json::json->>'realized_pnl')::float AS pnl, p.pm_high, p.open_vs_pm_high_pct FROM events e JOIN premarket_structure p ON p.symbol = e.payload_json::json->>'symbol' AND p.session_date = e.timestamp::date WHERE e.event_type = 'position_closed' """).fetchall() if not rows: print("no joined closes yet — build features, then trade/replay some days") return def bucket_h1(entry, pm_high): x = entry / pm_high return "near_pm_high(<=1.05)" if x <= 1.05 else ("ext_1.05-1.25" if x <= 1.25 else "deep_ext(>1.25)") def bucket_h2(v): return "consol(>=-5%)" if v >= -0.05 else ("pulled(-5..-15%)" if v >= -0.15 else "faded(<-15%)") for name, fn in (("H1 entry vs pm_high", lambda r: bucket_h1(r[0], r[2])), ("H2 open vs pm_high", lambda r: bucket_h2(r[3]))): agg: dict[str, list[float]] = {} for r in rows: if r[0] and r[2]: agg.setdefault(fn(r), []).append(r[1] or 0.0) print(f"--- {name} ---") for k in sorted(agg): pnls = agg[k] wins = sum(1 for x in pnls if x > 0) print(f" {k:22s} n={len(pnls):3d} win={wins/len(pnls)*100:3.0f}% " f"avg=${sum(pnls)/len(pnls):+8.2f}") print(f"(n={len(rows)} joined closes; NO conclusions below n=30/bucket — see module docstring)") def main(argv=None) -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("command", choices=["build", "report"]) ap.add_argument("--date") ap.add_argument("--since", type=int, help="build the last N sessions") args = ap.parse_args(argv) from research.multi_schema import open_research_db con = open_research_db("market") if args.command == "build": out = build(con, session_date=args.date, since_days=args.since) print(f"premarket_structure: {out['rows']} rows across {out['sessions']} session(s)") else: report(con) return 0 if __name__ == "__main__": sys.exit(main())