feat: H3 replay — premarket-high break at 9:30 vs ORB, paired comparison
The pre-registered premarket hypothesis H3 (the source trader's actual entry) made executable: both legs run per symbol-day over the year of labeled gapper sessions, sharing the deployed exit engine (config parity) and +1-tick adverse fills so the paired delta isolates ENTRY choice. Comparison metric locked in the module docstring before the first run: paired mean R delta (H3-ORB) where both fire, pass at >= +0.05R with n >= 300 pairs, plus a stop-unit guard (median stop-distance % and captured-move % printed for both legs — R parity from a tighter stop is a fail, the vwap-reclaim lesson). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
momentum devandClaude Fable 5 committed
1 parent
37757d3d47
commit
3fe0cb0891
1 file changed
+191
@@ -0,0 +1,191 @@
|
||||
"""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())
|
||||
Reference in new issue
Block a user