diff --git a/deploy/docker/truenas-compose.yml b/deploy/docker/truenas-compose.yml index 9e5d5d2..0507a15 100644 --- a/deploy/docker/truenas-compose.yml +++ b/deploy/docker/truenas-compose.yml @@ -213,6 +213,10 @@ services: while true; do if [ "$$(date +%H:%M)" = "20:00" ] && [ "$$(date +%u)" -le 5 ]; then python -m research.premarket_eval || echo "[scheduler] eval FAILED" + # Premarket-structure labeler (Phase 1, observe-only): distill each + # gapper's premarket tape for entry-conversion analysis (H1-H3 + # pre-registered in the module docstring). Gates nothing. + python -m research.premarket_structure build || echo "[scheduler] premarket_structure FAILED" # SPY/QQQ ingest -> forward source for the labeler's market-context features (plan 009) python -m research.ingestion.market_context || echo "[scheduler] market_context ingest FAILED" python -m research.eod_reconcile || echo "[scheduler] reconcile FAILED" diff --git a/research/premarket_structure.py b/research/premarket_structure.py new file mode 100644 index 0000000..7e57d82 --- /dev/null +++ b/research/premarket_structure.py @@ -0,0 +1,245 @@ +"""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()) diff --git a/tests/unit/test_premarket_structure.py b/tests/unit/test_premarket_structure.py new file mode 100644 index 0000000..2c71415 --- /dev/null +++ b/tests/unit/test_premarket_structure.py @@ -0,0 +1,46 @@ +"""compute_features: the pure premarket distiller (FCUV-shaped synthetic tape).""" + +from datetime import datetime, timedelta + +from research.premarket_structure import compute_features, MIN_PM_BARS + + +def _bar(i, o, h, low, c, v=50_000): + return {"timestamp": datetime(2026, 7, 31, 12, 13) + timedelta(minutes=i), + "open": o, "high": h, "low": low, "close": c, "volume": v} + + +def test_fcuv_shaped_ramp(): + # $2 -> high $10.49 -> close 9:29 at $10.20, one >=5% jackknife on the way + bars = [ + _bar(0, 2.00, 2.60, 1.98, 2.55), + _bar(1, 2.55, 4.25, 2.50, 4.10), + _bar(2, 4.10, 8.00, 4.00, 7.80), + _bar(3, 7.80, 10.25, 7.40, 9.60), # runs to 10.25 then drops >5% (jackknife) + _bar(4, 9.60, 10.49, 9.40, 10.20), # new high, close near high + ] + f = compute_features(bars) + assert f is not None + assert f["first_price"] == 2.00 + assert f["pm_high"] == 10.49 + assert f["pm_close"] == 10.20 + assert abs(f["pm_run_pct"] - (10.49 - 2.0) / 2.0) < 1e-6 + assert f["open_vs_pm_high_pct"] > -0.05 # consolidating at high (H2 bucket) + assert f["pullbacks_5pct"] >= 1 # the 10.25 -> 9.40 knife + assert f["pm_volume"] == 250_000 + + +def test_faded_open_bucket(): + # ran to 10 early, bled to 6 into the open -> big negative open_vs_pm_high + bars = [_bar(0, 5.0, 10.0, 4.9, 9.8)] + [ + _bar(i, 9.0 - i, 9.2 - i, 8.5 - i, 8.8 - i) for i in range(1, 5) + ] + f = compute_features(bars) + assert f is not None + assert f["open_vs_pm_high_pct"] < -0.15 + + +def test_thin_tape_rejected(): + assert compute_features([_bar(0, 2, 2.2, 1.9, 2.1)] * (MIN_PM_BARS - 1)) is None + thin = [_bar(i, 2, 2.2, 1.9, 2.1, v=100) for i in range(MIN_PM_BARS)] + assert compute_features(thin) is None # volume floor