Ignition P0: pure detector + tests, and a decisive real-data finding
detect_momentum_ignition (strategy/evaluation/ignition.py): standalone, pure, binary-rule detector for a vertical squeeze (blue-sky fresh-HOD + price velocity + ABSOLUTE volume burst, NOT rvol/float). Stop = last established higher low (per operator). NOT wired into classify_setup (deliberately). 8 unit tests on synthetic verticals: fires with the right trigger/stop, and each gate blocks (not-blue-sky / not-new-HOD / velocity / volume / insufficient-bars / no-ATH). REALITY CHECK (verify, don't assume): ran it over the REAL PLSM 2026-06-24 bars -> 0 fires, CORRECTLY. PLSM's high was at the 9:35 ET open (it ran +400% PRE-MARKET, which we don't ingest), and in RTH it was HALTED 228 of ~390 minutes — a halt-resumption GAP chop (+3.6/-12.5/-17.3/-14.4% jumps), not a smooth vertical. A consecutive-bar detector structurally cannot see that. So the ignition opportunity is a PRE-MARKET + halt-resumption phenomenon -> capturing it needs pre-market bar ingestion + halt-aware gap detection + pre-market/halt trading (into the FCUV halt-down risk), a far bigger lift than an RTH detector. Documented front-and-center in docs/IGNITION_SETUP_SCOPE.md as a pre-P1 operator decision. The P0 detector + tests are a correct foundation regardless. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
1 parent
7ebb503790
commit
1c8b4f4309
3 files changed
+228
No files matched your search
@@ -75,6 +75,25 @@ squeeze logs once, not 1,895×), never emit `signal_ready`, never set a symbol "
|
||||
- **Catalyst tag** needs Ollama running that day — force-enable it for the shadow grading pass so news-backed vs not can be measured?
|
||||
- **Distinct-runner-n ≥30** may be unrealistic given runner sparsity — a Bayesian interval at low n instead of a point rate?
|
||||
|
||||
## ⚠️ P0 reality-check (2026-06-25) — READ BEFORE P1
|
||||
The P0 detector is built + unit-tested (fires correctly on a clean synthetic vertical). But running
|
||||
it over the **real** PLSM 2026-06-24 bars fired **0 times — correctly** — and the *why* reshapes this
|
||||
whole effort:
|
||||
- PLSM's session HIGH (12.85) was made at the **9:35 ET open** (it gapped from ~$4 **pre-market**); RTH
|
||||
was then a day-long **fade**. The "+400% to $20" the trader cites happened **pre-market**, which the
|
||||
bot does not ingest (RTH-only `minute_bars`) or trade.
|
||||
- PLSM was **halted for 228 of ~390 RTH minutes** — its RTH move is **halt-resumption GAPS**
|
||||
(+3.6%, −12.5%, −17.3%, −14.4% …), NOT a smooth consecutive-bar vertical. A detector keyed on
|
||||
consecutive green bars + smooth velocity **structurally cannot** see a halt-gapped move.
|
||||
|
||||
**Implication:** the momentum-ignition opportunity is largely a **pre-market + halt-resumption**
|
||||
phenomenon, not an RTH smooth-vertical one. Capturing it would need (a) **pre-market/extended-hours bar
|
||||
ingestion**, (b) **halt-aware, gap-up detection** (resumption gaps, not consecutive bars), and (c) the
|
||||
bot to **trade pre-market / halt resumptions** — which runs straight into the **halt-down blowup risk**
|
||||
(FCUV) the anti-chase guards exist to prevent. That is a far bigger lift than an RTH detector, and an
|
||||
operator decision before P1: invest in pre-market+halt data/logic, or shelve the ignition lane and bank
|
||||
the ORB/float work instead. The P0 detector + tests remain a correct foundation either way.
|
||||
|
||||
## Hard constraints (non-negotiable)
|
||||
Shadow-only until forward data proves it · never touches the ORB book or anti-chase guards · ships dark
|
||||
(`SHADOW_IGNITION_ENABLED=0`) · binary rules, no fitting · do NOT gate on RVOL or float (both broken for
|
||||
|
||||
@@ -0,0 +1,124 @@
|
||||
"""Momentum-ignition detector — a SECOND setup the pullback-ORB scorer is blind to.
|
||||
|
||||
A vertical squeeze (PLSM-class: blue-sky, volume explosion, NO pullback) makes
|
||||
``strategy.evaluation.structure.classify_setup`` return NONE — it requires a
|
||||
consolidation/pullback phase a true parabola never gives — so the bot blocks it
|
||||
(PLSM 2026-06-24 was evaluated 1,895x and blocked ~30% every time). This is a
|
||||
standalone, pure, binary-rule detector. It is deliberately NOT wired into
|
||||
classify_setup / evaluate_setup (those pullback criteria would reject it, and
|
||||
STRATEGY_SETUPS would filter it). SHADOW-ONLY: the caller logs the signal and
|
||||
NEVER trades it until forward data proves runner-vs-trap separation.
|
||||
|
||||
Gates (ALL required; computable from intraday bars + a prior all-time high):
|
||||
1. BLUE-SKY FRESH-HOD — the latest bar prints a NEW session high AND that
|
||||
high is above the prior all-time high (no overhead resistance, room to run).
|
||||
2. PRICE VELOCITY — +velocity_min over velocity_window bars, with
|
||||
>= green_min of the last green_lookback bars green and rising highs.
|
||||
3. ABSOLUTE VOLUME BURST — last-bar volume >= burst_mult x the prior K-bar
|
||||
mean, AND cumulative session volume >= abs_vol_floor. ABSOLUTE floors, NOT
|
||||
RVOL: scan_gappers RVOL zeros out on thin IPOs (avg-daily-vol = 0) and float
|
||||
is NULL — that blind spot is exactly why PLSM read None mid-ignition.
|
||||
|
||||
stop = the last established HIGHER LOW beneath the ignition bar (ride the up-
|
||||
structure). breakout_level = the fresh-HOD trigger. Catalyst / VWAP / float are
|
||||
TAGS the caller attaches for grading — never gates here. Thresholds are params
|
||||
(env knobs at the call site), seeded from one PLSM window; calibrate only after
|
||||
shadow data accumulates.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
import pandas as pd
|
||||
|
||||
|
||||
@dataclass
|
||||
class IgnitionSignal:
|
||||
is_valid: bool
|
||||
breakout_level: float | None = None # fresh-HOD trigger (entry reference)
|
||||
stop_level: float | None = None # last established higher low
|
||||
reason: str | None = None
|
||||
signal_values: dict = field(default_factory=dict)
|
||||
|
||||
|
||||
def _last_higher_low(lows: list[float]) -> float:
|
||||
"""Ride the stop under the last established HIGHER LOW *beneath* the final
|
||||
(ignition) bar: scan back from the 2nd-to-last bar for the most recent bar
|
||||
whose low is higher than the bar before it. Fallback: the window's min low."""
|
||||
for i in range(len(lows) - 2, 0, -1):
|
||||
if lows[i] > lows[i - 1]:
|
||||
return lows[i]
|
||||
return min(lows) if lows else 0.0
|
||||
|
||||
|
||||
def detect_momentum_ignition(
|
||||
bars: pd.DataFrame,
|
||||
*,
|
||||
prior_ath: float | None = None,
|
||||
velocity_min: float = 0.08,
|
||||
velocity_window: int = 8,
|
||||
green_min: int = 4,
|
||||
green_lookback: int = 5,
|
||||
burst_mult: float = 3.0,
|
||||
burst_window: int = 5,
|
||||
abs_vol_floor: float = 100_000.0,
|
||||
) -> IgnitionSignal:
|
||||
"""Detect a vertical momentum ignition. Pure; no pullback required."""
|
||||
need = max(velocity_window + 1, green_lookback, burst_window + 1)
|
||||
if bars is None or len(bars) < need:
|
||||
return IgnitionSignal(False, reason=f"need >= {need} bars")
|
||||
|
||||
o = bars["open"].astype(float).to_numpy()
|
||||
h = bars["high"].astype(float).to_numpy()
|
||||
low_arr = bars["low"].astype(float).to_numpy()
|
||||
c = bars["close"].astype(float).to_numpy()
|
||||
v = bars["volume"].astype(float).to_numpy()
|
||||
last = len(c) - 1
|
||||
session_high = float(h.max())
|
||||
cum_vol = float(v.sum())
|
||||
|
||||
# gate 1 — blue-sky fresh-HOD
|
||||
new_hod = bool(h[last] >= session_high) # final bar made the high
|
||||
blue_sky = prior_ath is not None and session_high > float(prior_ath)
|
||||
|
||||
# gate 2 — price velocity / verticality
|
||||
base = c[last - velocity_window]
|
||||
velocity_pct = (c[last] / base - 1.0) if base > 0 else 0.0
|
||||
lookback = range(last - green_lookback + 1, last + 1)
|
||||
green_n = sum(1 for i in lookback if c[i] > o[i])
|
||||
hh_n = sum(1 for i in range(last - green_lookback + 1, last + 1) if h[i] > h[i - 1])
|
||||
velocity_ok = (velocity_pct >= velocity_min and green_n >= green_min
|
||||
and hh_n >= green_min)
|
||||
|
||||
# gate 3 — absolute volume burst (NOT rvol)
|
||||
prior_mean = float(v[last - burst_window:last].mean()) if burst_window > 0 else 0.0
|
||||
burst_ratio = (v[last] / prior_mean) if prior_mean > 0 else 0.0
|
||||
volume_ok = burst_ratio >= burst_mult and cum_vol >= abs_vol_floor
|
||||
|
||||
vals = {
|
||||
"velocity_pct": round(velocity_pct, 4),
|
||||
"consecutive_green": green_n,
|
||||
"higher_high_count": hh_n,
|
||||
"volume_burst_ratio": round(burst_ratio, 2),
|
||||
"cum_volume": cum_vol,
|
||||
"session_high": round(session_high, 4),
|
||||
"prior_ath": float(prior_ath) if prior_ath is not None else None,
|
||||
"is_blue_sky": bool(blue_sky),
|
||||
"new_hod": new_hod,
|
||||
}
|
||||
|
||||
if not (new_hod and blue_sky):
|
||||
return IgnitionSignal(False, reason="not blue-sky fresh-HOD", signal_values=vals)
|
||||
if not velocity_ok:
|
||||
return IgnitionSignal(False, reason="velocity gate failed", signal_values=vals)
|
||||
if not volume_ok:
|
||||
return IgnitionSignal(False, reason="volume-burst gate failed", signal_values=vals)
|
||||
|
||||
return IgnitionSignal(
|
||||
True,
|
||||
breakout_level=round(session_high, 4), # the fresh HOD = entry reference
|
||||
stop_level=round(_last_higher_low(list(low_arr)), 4),
|
||||
reason="momentum_ignition",
|
||||
signal_values=vals,
|
||||
)
|
||||
@@ -0,0 +1,85 @@
|
||||
"""Momentum-ignition detector (P0): binary gates on synthetic vertical squeezes.
|
||||
|
||||
Pure + deterministic. A real-PLSM regression fixture is deferred to P2 (when the
|
||||
nightly scorer runs over real shadow signals)."""
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from strategy.evaluation.ignition import _last_higher_low, detect_momentum_ignition
|
||||
|
||||
COLS = ["open", "high", "low", "close", "volume"]
|
||||
|
||||
# a clean vertical ignition: 10 bars, all green, rising, last bar a new HOD on a
|
||||
# big volume burst. open, high, low, close, volume.
|
||||
IGNITION = [
|
||||
(4.00, 4.20, 3.90, 4.10, 5000),
|
||||
(4.10, 4.40, 4.05, 4.35, 6000),
|
||||
(4.35, 4.70, 4.30, 4.65, 7000),
|
||||
(4.65, 5.00, 4.60, 4.95, 8000),
|
||||
(4.95, 5.40, 4.90, 5.35, 9000),
|
||||
(5.35, 5.90, 5.30, 5.85, 10000),
|
||||
(5.85, 6.50, 5.80, 6.45, 12000),
|
||||
(6.45, 7.20, 6.40, 7.10, 14000),
|
||||
(7.10, 8.00, 7.05, 7.90, 16000),
|
||||
(7.90, 9.20, 7.85, 9.10, 60000), # new HOD 9.20, volume burst 60k vs ~12k mean
|
||||
]
|
||||
|
||||
|
||||
def _df(rows):
|
||||
return pd.DataFrame(rows, columns=COLS)
|
||||
|
||||
|
||||
def test_ignition_fires_with_correct_trigger_and_stop():
|
||||
r = detect_momentum_ignition(_df(IGNITION), prior_ath=5.0)
|
||||
assert r.is_valid and r.reason == "momentum_ignition"
|
||||
assert r.breakout_level == 9.20 # the fresh HOD = entry reference
|
||||
assert r.stop_level == 7.05 # last higher low (bar 8 low > bar 7 low)
|
||||
sv = r.signal_values
|
||||
assert sv["is_blue_sky"] and sv["new_hod"]
|
||||
assert sv["velocity_pct"] > 0.08
|
||||
assert sv["volume_burst_ratio"] >= 3.0
|
||||
assert sv["cum_volume"] >= 100_000
|
||||
|
||||
|
||||
def test_not_blue_sky_blocks():
|
||||
# prior ATH above the session high -> not blue sky, even though it's a new HOD
|
||||
r = detect_momentum_ignition(_df(IGNITION), prior_ath=100.0)
|
||||
assert not r.is_valid and r.reason == "not blue-sky fresh-HOD"
|
||||
|
||||
|
||||
def test_not_new_hod_blocks():
|
||||
# final bar fades below the session high (bar 8 = 8.00) -> no fresh HOD
|
||||
fade = IGNITION[:-1] + [(7.90, 7.95, 7.40, 7.50, 60000)]
|
||||
r = detect_momentum_ignition(_df(fade), prior_ath=5.0)
|
||||
assert not r.is_valid and r.reason == "not blue-sky fresh-HOD"
|
||||
assert r.signal_values["new_hod"] is False
|
||||
|
||||
|
||||
def test_velocity_gate_blocks():
|
||||
# blue-sky + volume pass, but require an impossibly steep move -> velocity fails
|
||||
r = detect_momentum_ignition(_df(IGNITION), prior_ath=5.0, velocity_min=2.0)
|
||||
assert not r.is_valid and r.reason == "velocity gate failed"
|
||||
|
||||
|
||||
def test_volume_burst_gate_blocks():
|
||||
# blue-sky + velocity pass, but require an impossible burst -> volume fails
|
||||
r = detect_momentum_ignition(_df(IGNITION), prior_ath=5.0, burst_mult=100.0)
|
||||
assert not r.is_valid and r.reason == "volume-burst gate failed"
|
||||
|
||||
|
||||
def test_insufficient_bars():
|
||||
r = detect_momentum_ignition(_df(IGNITION[:4]), prior_ath=5.0)
|
||||
assert not r.is_valid and "need" in (r.reason or "")
|
||||
|
||||
|
||||
def test_missing_prior_ath_is_not_blue_sky():
|
||||
# no prior ATH -> can't confirm blue sky -> blocked (fail-safe, never fires blind)
|
||||
r = detect_momentum_ignition(_df(IGNITION), prior_ath=None)
|
||||
assert not r.is_valid and r.reason == "not blue-sky fresh-HOD"
|
||||
|
||||
|
||||
def test_last_higher_low_helper():
|
||||
# rising lows -> last established higher low beneath the final bar
|
||||
assert _last_higher_low([3.9, 4.05, 4.3, 4.6]) == 4.3 # excludes final bar (4.6)
|
||||
# no higher low -> fallback to the window min
|
||||
assert _last_higher_low([5.0, 4.0, 3.0, 2.0]) == 2.0
|
||||
Reference in new issue
Block a user