Entry-quality analysis + gap cap (diagnostic-driven)

diagnose_entries.py buckets backtested trades by feature: cheap stocks ($1-5
+0.5R) crush $10-20 (-0.09R), rvol>=5 (+0.79R) beats <5x (+0.15R), and blow-off
gaps >150% fade. sweep_filters.py verifies price/volume/gap-cap combos
train-vs-test: px<=7 + gap<=60% ~doubles OOS expectancy (+0.76 vs +0.39R, win
86% vs 71%).

Add a gap cap to the armed-trigger eligibility (TRIGGER_GAP_MAX) so blow-off
gappers can be excluded; price cap (WATCHER_PRICE_MAX) and volume floor
(TRIGGER_RVOL_MIN) already exist. Final values pending OOS verification.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

diagnose_entries.py

@@ -0,0 +1,88 @@
+#!/usr/bin/env python3
+"""Diagnose what separates WINNING ORB trades from losers, to find filters that
+raise the edge. Runs the live entry+exit config over the stored gapper sessions
+and buckets each trade's realized R by feature (stop distance, entry time, gap,
+rvol, how-extended-at-entry, price). Big avgR gaps between buckets = a filter.
+
+    PYTHONPATH=. python diagnose_entries.py
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+import pandas as pd
+from dotenv import load_dotenv
+
+sys.path.insert(0, ".")
+load_dotenv()
+os.environ["STRATEGY_SETUPS"] = "opening_range_break"
+
+from config import Config
+from learn_params import load_sessions
+from research.multi_schema import open_research_db
+from strategy.backtest.engine import BacktestEngine
+from strategy.exits import ExitConfig, TRAIL_PRIOR_LOW, parse_profit_tiers
+from strategy.models import SetupCriteria
+
+
+def main():
+    con = open_research_db("market")
+    sessions = load_sessions(con)
+    cfg = ExitConfig(target_r=10.0, trail_mode=TRAIL_PRIOR_LOW, trail_after_r=1.0,
+                     profit_lock_tiers=parse_profit_tiers("8:3,15:9,25:18,40:30"))
+    eng = BacktestEngine(Config(), target_r=10.0, warmup_bars=3, eval_every=1,
+                         ready_score_pct=55, min_bars=3, exit_config=cfg,
+                         criteria=SetupCriteria(gap_pct_min=0.05, relative_volume_min=2.0))
+    trades = []
+    for sym, _sess, bars, pc, adv in sessions:
+        bars_r = bars.reset_index(drop=True)
+        if not len(bars_r):
+            continue
+        first_open = float(bars_r["open"].iloc[0])
+        gap = (first_open / pc - 1) if pc else 0.0
+        rvol = (float(bars_r["volume"].sum()) / adv) if adv else 0.0
+        t0 = pd.Timestamp(bars_r["timestamp"].iloc[0])
+        for t in eng.run(bars, sym, previous_close=pc, avg_daily_volume=adv).trades:
+            entry, stop = t.entry_price, t.stop_price
+            try:
+                em = (pd.Timestamp(t.entry_time) - t0).total_seconds() / 60.0
+            except Exception:
+                em = float("nan")
+            trades.append({
+                "r": t.r_multiple or 0.0,
+                "stop_dist": (entry - stop) / entry * 100 if entry else 0.0,
+                "em": em,
+                "gap": gap * 100,
+                "rvol": rvol,
+                "ext": (entry / first_open - 1) * 100 if first_open else 0.0,
+                "price": entry,
+            })
+
+    n = len(trades)
+    avg = sum(t["r"] for t in trades) / n if n else 0
+    print(f"total trades: {n}   overall avgR: {avg:+.3f}   sumR: {sum(t['r'] for t in trades):+.1f}\n")
+
+    def bucket(name, key, edges):
+        print(f"=== avgR by {name} ===")
+        print(f"  {'bucket':>13} {'n':>4} {'win%':>5} {'avgR':>8} {'sumR':>8}")
+        for lo, hi in edges:
+            grp = [t for t in trades if t[key] == t[key] and lo <= t[key] < hi]
+            if not grp:
+                continue
+            rs = [t["r"] for t in grp]
+            print(f"  {f'[{lo:g},{hi:g})':>13} {len(grp):>4} "
+                  f"{sum(1 for r in rs if r > 0) / len(rs) * 100:>4.0f}% "
+                  f"{sum(rs) / len(rs):>+8.3f} {sum(rs):>+8.1f}")
+        print()
+
+    bucket("STOP DISTANCE %", "stop_dist", [(0, 5), (5, 10), (10, 15), (15, 25), (25, 100)])
+    bucket("ENTRY MINUTE (after open)", "em", [(0, 10), (10, 20), (20, 40), (40, 90), (90, 400)])
+    bucket("GAP %", "gap", [(0, 10), (10, 30), (30, 60), (60, 150), (150, 2000)])
+    bucket("RELATIVE VOLUME", "rvol", [(0, 5), (5, 15), (15, 50), (50, 5000)])
+    bucket("EXTENSION above open % at entry", "ext", [(-100, 0), (0, 5), (5, 15), (15, 50), (50, 2000)])
+    bucket("PRICE $", "price", [(0, 2), (2, 5), (5, 10), (10, 20)])
+
+
+if __name__ == "__main__":
+    main()

run_live_paper.py

@@ -263,6 +263,7 @@ def build_runtime(args: argparse.Namespace) -> dict:
     book = ArmedTriggerBook(
         max_armed=int(os.environ.get("TRIGGER_MAX_ARMED", "6")),
         gap_min=float(os.environ.get("TRIGGER_GAP_MIN", "3.0")),
+        gap_max=float(os.environ.get("TRIGGER_GAP_MAX", "1e9")),
         rvol_min=float(os.environ.get("TRIGGER_RVOL_MIN", "2.0")),
         min_range_pct=float(os.environ.get("TRIGGER_MIN_RANGE_PCT", "0.004")),
     )

runtime/triggers.py

@@ -93,6 +93,7 @@ class ArmedTriggerBook:
 
     max_armed: int = 6
     gap_min: float = 3.0        # min overnight gap % to be eligible to fire
+    gap_max: float = 1e9        # max gap % — skip blow-off gappers (they fade)
     rvol_min: float = 2.0       # min relative volume to be eligible to fire
     min_range_pct: float = 0.004  # opening range must be at least this wide
     triggers: dict[str, ArmedTrigger] = field(default_factory=dict)
@@ -103,7 +104,7 @@ class ArmedTriggerBook:
             t.trigger is not None
             and t.stop is not None
             and t.stop < t.trigger
-            and t.gap_pct >= self.gap_min
+            and self.gap_min <= t.gap_pct <= self.gap_max
             and t.rvol >= self.rvol_min
             and t.range_pct >= self.min_range_pct
         )

sweep_filters.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""Sweep ENTRY filters (price cap / volume floor / gap cap) train-vs-test.
+
+The diagnostic showed clear edges — cheap stocks, high relative volume, and
+non-blowoff gaps win; $10-20 names lose. This verifies which filter combo lifts
+the out-of-sample expectancy without overfitting. Entries+exit are the verified
+live config; only the universe selection (which symbol-days qualify) varies.
+
+Fast: entries are extracted ONCE (with their price/gap/rvol tagged), then each
+filter combo just selects from that list — no re-running the engine per combo.
+
+    PYTHONPATH=. python sweep_filters.py            # ranked table (train+test)
+    PYTHONPATH=. python sweep_filters.py --json     # for the verify workflow
+"""
+from __future__ import annotations
+
+import argparse
+import itertools
+import json
+import os
+import sys
+
+from dotenv import load_dotenv
+
+sys.path.insert(0, ".")
+load_dotenv()
+os.environ["STRATEGY_SETUPS"] = "opening_range_break"
+
+from config import Config
+from learn_params import load_sessions
+from research.multi_schema import open_research_db
+from strategy.backtest.engine import BacktestEngine
+from strategy.exits import ExitConfig, TRAIL_PRIOR_LOW, parse_profit_tiers
+from strategy.models import SetupCriteria
+from sweep_exits import ENTRY, run_variant
+
+CFG = ExitConfig(target_r=10.0, trail_mode=TRAIL_PRIOR_LOW, trail_after_r=1.0,
+                 profit_lock_tiers=parse_profit_tiers("8:3,15:9,25:18,40:30"))
+
+GRID = {
+    "price_max": [5.0, 7.0, 10.0, 20.0],
+    "rvol_min": [2.0, 4.0, 6.0],
+    "gap_max": [60.0, 120.0, 5000.0],   # 5000 = off
+}
+BASE = {"price_max": 20.0, "rvol_min": 2.0, "gap_max": 5000.0}
+
+
+def extract_with_features(sessions):
+    """[{entry, stop, qty, ba, price, gap, rvol}] — engine run ONCE per session."""
+    ref = BacktestEngine(
+        Config(), target_r=2.0, warmup_bars=ENTRY["warmup"], eval_every=ENTRY["eval_every"],
+        ready_score_pct=ENTRY["ready"], min_bars=ENTRY["min_bars"],
+        exit_config=ExitConfig(target_r=2.0),
+        criteria=SetupCriteria(gap_pct_min=ENTRY["gap_min"], relative_volume_min=ENTRY["rvol_min"]))
+    out = []
+    for sym, _sess, bars, pc, adv in sessions:
+        br = bars.reset_index(drop=True)
+        if not len(br):
+            continue
+        gap = (float(br["open"].iloc[0]) / pc - 1) * 100 if pc else 0.0
+        rvol = (float(br["volume"].sum()) / adv) if adv else 0.0
+        for t in ref.run(bars, sym, previous_close=pc, avg_daily_volume=adv).trades:
+            if t.entry_index is None:
+                continue
+            out.append({"entry": t.entry_price, "stop": t.stop_price, "qty": t.quantity,
+                        "ba": br.iloc[t.entry_index + 1:], "gap": gap, "rvol": rvol})
+    return out
+
+
+def run_combo(feats, p):
+    sel = [(f["entry"], f["stop"], f["qty"], f["ba"]) for f in feats
+           if 1.0 <= f["entry"] <= p["price_max"] and 3.0 <= f["gap"] <= p["gap_max"]
+           and f["rvol"] >= p["rvol_min"]]
+    if not sel:
+        return {"trades": 0, "win": 0.0, "avgR": 0.0, "pnl": 0.0}
+    return run_variant(sel, CFG)
+
+
+def split_by_date(sessions, which):
+    days = sorted({s[1] for s in sessions})
+    cut = days[int(len(days) * 0.7)] if len(days) >= 4 else (days[-1] if days else None)
+    if which == "train":
+        return [s for s in sessions if cut and s[1] < cut] or sessions
+    if which == "test":
+        return [s for s in sessions if cut and s[1] >= cut]
+    return sessions
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--json", action="store_true")
+    args = ap.parse_args()
+
+    con = open_research_db("market")
+    sessions = load_sessions(con)
+    feat_train = extract_with_features(split_by_date(sessions, "train"))
+    feat_test = extract_with_features(split_by_date(sessions, "test"))
+
+    rows = []
+    for c in itertools.product(*GRID.values()):
+        p = dict(zip(GRID, c))
+        tr, te = run_combo(feat_train, p), run_combo(feat_test, p)
+        rows.append({
+            "filter": f"px<={p['price_max']:g} rvol>={p['rvol_min']:g} gap<={p['gap_max']:g}",
+            "params": p,
+            "train_avgR": tr["avgR"], "train_trades": tr["trades"],
+            "test_avgR": te["avgR"], "test_trades": te["trades"], "test_win": te["win"],
+        })
+    rows.sort(key=lambda r: r["test_avgR"], reverse=True)
+
+    if args.json:
+        print(json.dumps({"results": rows}))
+        return
+    base = run_combo(feat_test, BASE)
+    base_tr = run_combo(feat_train, BASE)
+    print(f"BASELINE (px<=20 rvol>=2 gap-off): train {base_tr['avgR']:+.3f} ({base_tr['trades']}t)  "
+          f"TEST {base['avgR']:+.3f} ({base['trades']}t) win {base['win'] * 100:.0f}%\n")
+    print(f"{'filter':>32} | {'train avgR':>10} {'n':>4} | {'TEST avgR':>10} {'n':>4} {'win%':>5}")
+    for r in rows[:12]:
+        print(f"{r['filter']:>32} | {r['train_avgR']:>+10.3f} {r['train_trades']:>4} | "
+              f"{r['test_avgR']:>+10.3f} {r['test_trades']:>4} {r['test_win'] * 100:>4.0f}%")
+
+
+if __name__ == "__main__":
+    main()