feat: premarket-structure labeler (Phase 1 of premarket axis, observe-only)

Distills each gapper's premarket tape (first-hit price/time, pm high/low,
run %, open-vs-pm-high, VWAP distance, jackknife count) into a
premarket_structure table for entry-conversion analysis. Motivated by
FCUV 2026-07-31: our DB captured the full $1.98->$10.49 premarket ramp
while the ORB entered at $11.89 into the reversal.

Pre-registered hypotheses H1-H3 locked in the module docstring before any
data was examined; report command prints descriptive buckets only, with
n>=30/bucket floor and Phase-2 shadow bar required for promotion. Gates
nothing, trades nothing.

Wired into the nightly scheduler after premarket_eval. One session scan
per build (per-symbol queries were ~100x slower); partial session index
created post-close only (mid-market CONCURRENTLY build deadlocked against
live ingest).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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"

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

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