""" setup_engine.py [SIMULATION-ONLY 2026-09-05] Detects the five concrete entry patterns regime_engine.py's allowed_setups already references by name but nothing until now actually implements: PULLBACK_CONTINUATION, VWAP_RECLAIM, BREAKOUT_RETEST, HIGHER_LOW_REVERSAL, COMPRESSION_EXPANSION. Not consumed anywhere in the live bot yet -- see config.json's setup_engine._note. Each detector is a stateless checklist evaluated against the CURRENT trend/structure/stream_features snapshot -- same pattern entry_engine.evaluate_entry() already uses (a list of checks, each either passed or not, with an explanation), not a multi-call tracked state machine. This keeps every function here pure (no state, no I/O), consistent with every other module built this session. The project's original design brief describes each setup as a temporal SEQUENCE (e.g. "impulse -> pullback -> volume contracts -> higher low -> EMA9 reclaim -> ENTRY") -- what's implemented here checks that the CURRENT snapshot is consistent with being at the end of that sequence (the structural/ momentum fingerprint a completed sequence leaves behind), not that the sequence was traced step-by-step over time. That's a deliberate simplification worth knowing about, not a hidden gap: a true multi-call temporal tracker is a bigger, different kind of module (it would need caller-held per-symbol history, like stream_features.py's `prior` parameter) and isn't needed to get useful signal out of this layer yet. """ from dataclasses import dataclass, field from config_loader import get_config PULLBACK_CONTINUATION = "PULLBACK_CONTINUATION" VWAP_RECLAIM = "VWAP_RECLAIM" BREAKOUT_RETEST = "BREAKOUT_RETEST" HIGHER_LOW_REVERSAL = "HIGHER_LOW_REVERSAL" COMPRESSION_EXPANSION = "COMPRESSION_EXPANSION" ALL_SETUPS = [PULLBACK_CONTINUATION, VWAP_RECLAIM, BREAKOUT_RETEST, HIGHER_LOW_REVERSAL, COMPRESSION_EXPANSION] DEFAULT_CONFIG = { "vwap_reclaim_min_time_below_pct": 30.0, "retest_buffer_pct": 0.5, "min_volume_acceleration": 1.0, "compression_expansion_min_range_expansion_pct": 40.0, } @dataclass class SetupResult: setup: str valid: bool confidence: float # 0-100, % of this setup's checks that passed checks_passed: list = field(default_factory=list) checks_failed: list = field(default_factory=list) def _result(setup: str, checks: list) -> SetupResult: """checks: list of (passed: bool, description: str). valid requires ALL checks to pass (a setup is either confirmed or it isn't -- no partial credit toward triggering an entry), but confidence reports the fraction anyway so a near-miss is visible in logs rather than looking identical to a setup with nothing going for it at all.""" passed = [desc for ok, desc in checks if ok] failed = [desc for ok, desc in checks if not ok] confidence = round(len(passed) / len(checks) * 100.0, 1) if checks else 0.0 return SetupResult(setup=setup, valid=len(failed) == 0 and len(passed) > 0, confidence=confidence, checks_passed=passed, checks_failed=failed) def detect_pullback_continuation(features, trend, structure, cfg: dict) -> SetupResult: lows = [s for s in structure.swings if s.kind == "LOW" and s.label is not None] last_low_is_hl = bool(lows) and lows[-1].label == "HL" ema9 = features.ema.get("ema9") price = features.price.get("last") checks = [ (trend.state in ("STRONG_UPTREND", "WEAK_UPTREND"), "established uptrend (trend_engine)"), (structure.structure == "BULLISH_STRUCTURE", "confirmed bullish swing structure (the prior impulse)"), (last_low_is_hl, "most recent low swing is a higher low, not a lower low"), (ema9 is not None and price is not None and price > ema9, "price has reclaimed EMA9"), (trend.trend_slope > 0, "momentum turning back positive after the pullback"), ((features.volume.get("volume_acceleration") or 0) >= cfg["min_volume_acceleration"], "volume re-expanding, not still contracting"), ] return _result(PULLBACK_CONTINUATION, checks) def detect_vwap_reclaim(features, trend, structure, cfg: dict) -> SetupResult: lows = [s for s in structure.swings if s.kind == "LOW" and s.label is not None] last_low_is_hl = bool(lows) and lows[-1].label == "HL" price_vs_vwap = features.vwap.get("price_vs_vwap_pct", 0) time_below = features.vwap.get("time_below_pct", 0) checks = [ (price_vs_vwap > 0, "price currently above VWAP"), (time_below >= cfg["vwap_reclaim_min_time_below_pct"], f"spent real time below VWAP recently ({time_below:.0f}% of window), this is a genuine reclaim"), (last_low_is_hl, "held a higher low since reclaiming (didn't immediately lose it again)"), (trend.trend_slope > 0, "momentum turning positive"), ((features.volume.get("volume_acceleration") or 0) >= cfg["min_volume_acceleration"], "volume accelerating into the reclaim"), ] return _result(VWAP_RECLAIM, checks) def detect_breakout_retest(features, trend, structure, bars: list, cfg: dict) -> SetupResult: resistance = structure.last_swing_high price = features.price.get("last") broke_above = False retesting = False holding = False if resistance is not None and price is not None and bars: broke_above = any(b["c"] > resistance for b in bars) buffer = resistance * (cfg["retest_buffer_pct"] / 100.0) retesting = abs(price - resistance) <= buffer or (resistance <= price <= resistance + buffer) holding = price >= resistance checks = [ (resistance is not None, "a confirmed swing-high resistance level exists"), (broke_above, "price has broken above that resistance at some point"), (retesting, "price has pulled back to retest the former resistance level"), (holding, "resistance is holding as new support on the retest (not failing back below it)"), (trend.direction == "UP", "underlying trend is still up"), ] return _result(BREAKOUT_RETEST, checks) def detect_higher_low_reversal(features, trend, structure, cfg: dict) -> SetupResult: lows = [s for s in structure.swings if s.kind == "LOW" and s.label is not None] bottomed_then_hl = len(lows) >= 2 and lows[-2].label == "LL" and lows[-1].label == "HL" ema9 = features.ema.get("ema9") price = features.price.get("last") checks = [ (bottomed_then_hl, "made a lower low, then a higher low (bottoming pattern, not a fresh downtrend)"), (trend.trend_slope > 0, "price velocity has turned positive"), (ema9 is not None and price is not None and price > ema9, "price has reclaimed EMA9"), ((features.velocity.get("velocity_sub") or 0) > 0, "very recent velocity is positive"), ] return _result(HIGHER_LOW_REVERSAL, checks) def detect_compression_expansion(features, trend, structure, cfg: dict) -> SetupResult: lows = [s for s in structure.swings if s.kind == "LOW"] holding_structure = structure.structure in ("BULLISH_STRUCTURE", "RANGE_STRUCTURE") last_low_ok = (not lows) or lows[-1].label in (None, "HL") range_expansion = features.volatility.get("range_expansion_pct") checks = [ (range_expansion is not None and range_expansion >= cfg["compression_expansion_min_range_expansion_pct"], f"range expanding out of a prior compression (range_expansion_pct={range_expansion})"), (holding_structure, "structure held (bullish or range) through the compression, no breakdown"), (last_low_ok, "most recent low did not undercut the prior one during compression"), ((features.volume.get("volume_acceleration") or 0) >= cfg["min_volume_acceleration"], "volume expanding alongside the range"), (trend.direction != "DOWN", "not breaking down as it expands"), ] return _result(COMPRESSION_EXPANSION, checks) def evaluate_setups(features, trend, structure, bars: list = None, cfg: dict = None) -> dict: """Returns {setup_name: SetupResult} for all five setups. Callers (opportunity_ranker.py, eventually) filter this against regime_engine's allowed_setups for the current regime before treating any of them as tradeable -- this function doesn't know or care what regime is active, it just reports what patterns the current snapshot is consistent with.""" cfg = {**DEFAULT_CONFIG, **(cfg or get_config().get("setup_engine", {}))} bars = bars or [] if (getattr(features, "insufficient_data", False) or getattr(trend, "insufficient_data", False) or getattr(structure, "insufficient_data", False)): return {name: SetupResult(setup=name, valid=False, confidence=0.0, checks_failed=["insufficient upstream data"]) for name in ALL_SETUPS} return { PULLBACK_CONTINUATION: detect_pullback_continuation(features, trend, structure, cfg), VWAP_RECLAIM: detect_vwap_reclaim(features, trend, structure, cfg), BREAKOUT_RETEST: detect_breakout_retest(features, trend, structure, bars, cfg), HIGHER_LOW_REVERSAL: detect_higher_low_reversal(features, trend, structure, cfg), COMPRESSION_EXPANSION: detect_compression_expansion(features, trend, structure, cfg), } def best_setup(setups: dict) -> SetupResult: """The single highest-confidence VALID setup, or None if nothing validated. Ties broken by ALL_SETUPS priority order (see regime_engine.py's original design-brief priority note -- pullback continuation first, recovery-style reversal last), not arbitrary dict order.""" valid = [setups[name] for name in ALL_SETUPS if setups.get(name) and setups[name].valid] if not valid: return None return max(valid, key=lambda r: r.confidence)