momentum-system/research/premarket_h3.py

"""H3 replay — premarket-high break at 9:30 vs the ORB entry, same names.

The pre-registered premarket hypothesis (research/premarket_structure.py):

  H3  A "premarket-high break at/after 9:30" entry beats the ORB entry on
      the same names.

PRE-REGISTERED COMPARISON (locked before first run):
  - Population: shared live gate (strategy/population.py, $2-20, gap >= 10%,
    35-150% dead band) on the session's open-knowable gap; sessions from
    premarket_structure (the year of labeled gapper days).
  - Both legs use the DEPLOYED exit engine (strategy.exits.simulate_exit under
    config_overrides parity) and +1-tick adverse fills on entry, -1 on exit.
  - ORB leg: range = first 5 RTH minutes; entry on the first later bar whose
    high crosses the range high; stop = range low.
  - H3 leg: entry on the first RTH bar (9:30 onward) whose high crosses the
    PREMARKET high; stop = low of the final 15 premarket minutes (the
    consolidation base under the high). No consolidation low below entry ->
    no trade (structure absent).
  - PRIMARY METRIC: paired mean R delta (H3 - ORB) on sessions where BOTH
    legs fire. H3 passes at >= +0.05R with n >= 300 pairs.
  - STOP-UNIT GUARD (the vwap-reclaim lesson): R parity from tighter stops is
    an artifact, so the report always shows each leg's median stop distance %
    and median captured move % — a "win" that exists only in R units with a
    2x tighter stop and equal %-capture is a FAIL, not a pass.

Read-only against the research DB; writes one TSV of per-trade rows.

  PYTHONPATH=. python -m research.premarket_h3 [--limit-days N]
"""

from __future__ import annotations

import argparse
import os
import sys
from datetime import date
from zoneinfo import ZoneInfo

from dotenv import load_dotenv

from runtime.env_overrides import load as _load_cfg_overrides

_load_cfg_overrides()   # config parity: deployed TRADING_EXIT_* beat bare defaults
load_dotenv()

from research.multi_schema import open_research_db          # noqa: E402
from research.query import query_minute_bars, query_previous_close  # noqa: E402
from strategy.exits import ExitConfig, TRAIL_PRIOR_LOW, parse_profit_tiers  # noqa: E402
from strategy.exits import simulate_exit                    # noqa: E402
from strategy.population import PopulationConfig, population_reject  # noqa: E402

ET = ZoneInfo("America/New_York")
UTC = ZoneInfo("UTC")
ADV = 0.01               # +1 tick on entry, -1 on exit (experiments fidelity floor)
_POP = PopulationConfig(min_entry_gap_pct=0.10)

OUT = os.path.join(os.path.dirname(__file__), "..", "data", "premarket_h3_results.tsv")


def _exit_cfg() -> ExitConfig:
    return ExitConfig(
        target_r=float(os.environ.get("TRADING_REWARD_MULTIPLE", "10")),
        trail_mode=os.environ.get("TRADING_EXIT_TRAIL_MODE", TRAIL_PRIOR_LOW),
        trail_after_r=float(os.environ.get("TRADING_EXIT_TRAIL_AFTER_R", "1")),
        profit_lock_tiers=parse_profit_tiers(os.environ.get("TRADING_EXIT_PROFIT_TIERS", "")))


def _et(ts):
    return ts.replace(tzinfo=UTC).astimezone(ET) if ts.tzinfo is None else ts.astimezone(ET)


def _leg(bars_rth, start_idx, level, stop, cfg):
    """First bar from start_idx whose high crosses ``level`` -> entry; simulate
    the deployed exit on everything after. Returns dict or None."""
    if level is None or stop is None:
        return None
    for i in range(start_idx, len(bars_rth)):
        bar = bars_rth.iloc[i]
        if float(bar["high"]) >= level:
            entry_fill = max(level, float(bar["open"])) + ADV
            if entry_fill <= stop:
                return None
            after = bars_rth.iloc[i + 1:]
            if not len(after):
                return None
            res = simulate_exit(entry_fill, stop, after, cfg)
            risk = entry_fill - stop
            r = sum(fl.frac * ((fl.price - ADV) - entry_fill) / risk for fl in res.fills)
            pct = sum(fl.frac * ((fl.price - ADV) - entry_fill) / entry_fill for fl in res.fills)
            return {"entry": round(entry_fill, 4), "stop": round(stop, 4),
                    "risk_pct": round(risk / entry_fill, 4), "r": round(r, 3),
                    "pct": round(pct, 4), "entry_t": str(_et(bar["timestamp"]))[11:16],
                    "reason": res.reason}
    return None


def run(con, limit_days: int | None = None) -> list[dict]:
    days = con.execute(
        "SELECT DISTINCT session_date FROM premarket_structure ORDER BY session_date DESC"
        + (f" LIMIT {int(limit_days)}" if limit_days else "")).fetchall()
    cfg = _exit_cfg()
    rows_out: list[dict] = []
    for (sess,) in days:
        pm_rows = con.execute(
            "SELECT symbol, pm_high, pm_close FROM premarket_structure "
            "WHERE session_date = ? AND pm_run_pct >= 0.10", [str(sess)]).fetchall()
        for symbol, pm_high, pm_close in pm_rows:
            try:
                prev = query_previous_close(con, symbol, sess if isinstance(sess, date) else date.fromisoformat(str(sess)))
                if not prev or not pm_close:
                    continue
                gap = pm_close / prev - 1.0
                if population_reject(_POP, gap_pct=gap, price=pm_close, volume=None, rvol=0.0) is not None:
                    continue
                bars = query_minute_bars(con, symbol, sess)
                if bars is None or bars.empty:
                    continue
                ets = bars["timestamp"].map(_et)
                is_rth = ets.map(lambda t: (t.hour, t.minute) >= (9, 30) and (t.hour, t.minute) < (16, 0))
                rth = bars[is_rth.values].reset_index(drop=True)
                pm = bars[~is_rth.values & ets.map(lambda t: (t.hour, t.minute) < (9, 30)).values]
                if len(rth) < 10 or len(pm) < 15:
                    continue
                # ORB leg: 5-min range, scan from bar 5
                orb_high = float(rth.iloc[:5]["high"].max())
                orb_low = float(rth.iloc[:5]["low"].min())
                orb = _leg(rth, 5, orb_high, orb_low, cfg)
                # H3 leg: premarket-high break from 9:30, stop = last-15-pm-min low
                h3_stop = float(pm.iloc[-15:]["low"].min())
                h3 = _leg(rth, 0, float(pm_high), h3_stop, cfg)
                if orb is None and h3 is None:
                    continue
                rows_out.append({"session": str(sess), "symbol": symbol, "gap": round(gap, 3),
                                 "orb": orb, "h3": h3})
            except Exception:  # noqa: BLE001 — one bad symbol-day never kills the sweep
                continue
    return rows_out


def report(rows: list[dict]) -> None:
    import statistics as st

    def agg(key):
        legs = [r[key] for r in rows if r[key]]
        if not legs:
            return None
        rs = [x["r"] for x in legs]
        return {"n": len(legs), "win": sum(1 for x in rs if x > 0) / len(rs),
                "expR": st.mean(rs), "medR": st.median(rs),
                "med_stop_pct": st.median(x["risk_pct"] for x in legs),
                "med_pct": st.median(x["pct"] for x in legs),
                "med_entry_t": st.median_low(sorted(x["entry_t"] for x in legs))}

    for key in ("orb", "h3"):
        a = agg(key)
        if a:
            print(f"{key.upper():4s} n={a['n']:5d} win={a['win']*100:4.1f}% expR={a['expR']:+.3f} "
                  f"medR={a['medR']:+.3f} med_stop={a['med_stop_pct']*100:.1f}% "
                  f"med_move={a['med_pct']*100:+.2f}% med_entry={a['med_entry_t']}")
    paired = [r for r in rows if r["orb"] and r["h3"]]
    if paired:
        deltas = [r["h3"]["r"] - r["orb"]["r"] for r in paired]
        import statistics as st2
        print(f"\nPAIRED (both fired) n={len(paired)}  mean R delta (H3-ORB) = "
              f"{st2.mean(deltas):+.3f}  median = {st2.median(deltas):+.3f}")
        print("PRE-REGISTERED BAR: pass requires >= +0.05R at n >= 300 pairs, "
              "AND survives the stop-unit guard above.")


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--limit-days", type=int)
    args = ap.parse_args(argv)
    con = open_research_db("market")
    rows = run(con, limit_days=args.limit_days)
    with open(OUT, "w") as f:
        f.write("session\tsymbol\tgap\tleg\tentry_t\tentry\tstop\trisk_pct\tr\tpct\treason\n")
        for r in rows:
            for leg in ("orb", "h3"):
                if r[leg]:
                    x = r[leg]
                    f.write(f"{r['session']}\t{r['symbol']}\t{r['gap']}\t{leg}\t{x['entry_t']}\t"
                            f"{x['entry']}\t{x['stop']}\t{x['risk_pct']}\t{x['r']}\t{x['pct']}\t{x['reason']}\n")
    print(f"{len(rows)} symbol-days -> {OUT}\n")
    report(rows)
    return 0


if __name__ == "__main__":
    sys.exit(main())