momentum-system/research/premarket_structure.py

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