""" structure_engine.py [SIMULATION-ONLY 2026-09-05] Detects swing-point market structure (higher highs / higher lows / lower highs / lower lows) and the failure patterns built on top of it (structure break, failed higher low, support failure, resistance rejection, failed breakout). Not consumed anywhere in the live bot yet -- see config.json's structure_engine._note. DIFFERENT FROM indicators.is_higher_highs_higher_lows() / is_lower_ highs_lower_lows() (already used by intraday_health.py and entry_engine.py): those check whether a FIXED WINDOW of raw bars shows a majority-higher (or majority-lower) step pattern -- useful as a quick health/confirmation signal, but they don't identify discrete swing PIVOTS the way this module does. This module finds actual local turning points (a real "9.62 LOW, 9.81 HIGH, 9.70 HIGHER LOW, 9.94 HIGHER HIGH" sequence, per the project's original design brief) and tracks the relationship between consecutive pivots of the same type. Both modules stay -- they answer different questions at different granularity, and neither is a duplicate of the other. NO LOOK-AHEAD: a bar can only be confirmed as a swing pivot once enough LATER bars exist to know price actually turned there (see pivot_confirm_bars in config) -- there is an inherent, honest lag between "a swing just formed" and "we can confirm it," same as any real swing-trading system. This module never looks past whatever `bars` the caller hands it. """ from dataclasses import dataclass, field from config_loader import get_config DEFAULT_CONFIG = { "pivot_lookback_bars": 2, "pivot_confirm_bars": 2, "min_swings_for_structure": 2, # need at least 2 HIGHs and 2 LOWs to judge structure "resistance_rejection_tolerance_pct": 0.3, "structure_score_lookback_swings": 4, } STRUCTURE_BULLISH = "BULLISH_STRUCTURE" STRUCTURE_BEARISH = "BEARISH_STRUCTURE" STRUCTURE_RANGE = "RANGE_STRUCTURE" STRUCTURE_TRANSITION = "TRANSITION_STRUCTURE" @dataclass class SwingPoint: kind: str # "HIGH" | "LOW" price: float index: int # index into the bars list this was found at t: object label: str = None # "HH"/"LH" for HIGH swings, "HL"/"LL" for LOW swings; None if no prior swing of this kind yet @dataclass class StructureReading: symbol: str structure: str = STRUCTURE_RANGE score: float = 0.0 swings: list = field(default_factory=list) last_swing_high: float = None last_swing_low: float = None flags: dict = field(default_factory=dict) insufficient_data: bool = False missing: list = field(default_factory=list) def find_swing_points(bars: list, lookback: int, confirm: int) -> list: """ Bar i is a swing HIGH if its high is >= every bar's high in the `lookback` bars before it AND every bar's high in the `confirm` bars after it (the "after" requirement is what makes this causal-safe -- a pivot at index i literally cannot be identified until i+confirm bars have arrived). Mirror logic for swing LOWs. Ties (an exact repeat of the extreme) keep the EARLIEST bar as the pivot. """ n = len(bars) swings = [] for i in range(lookback, n - confirm): left = bars[i - lookback:i] right = bars[i + 1:i + 1 + confirm] h = bars[i]["h"] l = bars[i]["l"] if all(h >= b["h"] for b in left) and all(h >= b["h"] for b in right): swings.append(SwingPoint(kind="HIGH", price=h, index=i, t=bars[i]["t"])) if all(l <= b["l"] for b in left) and all(l <= b["l"] for b in right): swings.append(SwingPoint(kind="LOW", price=l, index=i, t=bars[i]["t"])) swings.sort(key=lambda s: s.index) return _dedupe_adjacent(swings) def _dedupe_adjacent(swings: list) -> list: """Collapses consecutive same-kind swings (e.g. two HIGH pivots in a row with no LOW in between -- can happen near the lookback/confirm boundary) down to the single most extreme one, so the sequence alternates HIGH/LOW the way a real zigzag does.""" if not swings: return [] out = [swings[0]] for s in swings[1:]: if s.kind == out[-1].kind: more_extreme = (s.price > out[-1].price) if s.kind == "HIGH" else (s.price < out[-1].price) if more_extreme: out[-1] = s else: out.append(s) return out def _label_swings(swings: list) -> list: """Assigns HH/LH to HIGH swings and HL/LL to LOW swings, each relative to the PRIOR swing of the same kind (not the immediately preceding swing overall, which alternates kind).""" last_of_kind = {} for s in swings: prev = last_of_kind.get(s.kind) if prev is not None: if s.kind == "HIGH": s.label = "HH" if s.price > prev.price else "LH" else: s.label = "HL" if s.price > prev.price else "LL" last_of_kind[s.kind] = s return swings def _classify(swings: list, min_swings: int) -> tuple: """Returns (structure, last_high_label, last_low_label) from the most recent labeled HIGH and LOW swing. Requires at least min_swings of EACH kind to have a real opinion -- otherwise RANGE (an honest 'not enough structure yet' default, not a guess).""" highs = [s for s in swings if s.kind == "HIGH" and s.label is not None] lows = [s for s in swings if s.kind == "LOW" and s.label is not None] if len(highs) < min_swings - 1 or len(lows) < min_swings - 1 or not highs or not lows: return STRUCTURE_RANGE, None, None last_high_label = highs[-1].label last_low_label = lows[-1].label if last_high_label == "HH" and last_low_label == "HL": return STRUCTURE_BULLISH, last_high_label, last_low_label if last_high_label == "LH" and last_low_label == "LL": return STRUCTURE_BEARISH, last_high_label, last_low_label return STRUCTURE_TRANSITION, last_high_label, last_low_label def _structure_score(swings: list, structure: str, lookback_swings: int) -> float: """% of the last `lookback_swings` labeled swings that agree with the declared `structure` -- e.g. for BULLISH_STRUCTURE, an HH or HL agrees, an LH or LL disagrees. RANGE/TRANSITION score by internal consistency instead (how much the recent swings agree with EACH OTHER), since there's no single direction to measure agreement against.""" labeled = [s for s in swings if s.label is not None][-lookback_swings:] if not labeled: return 0.0 if structure == STRUCTURE_BULLISH: agree = sum(1 for s in labeled if s.label in ("HH", "HL")) elif structure == STRUCTURE_BEARISH: agree = sum(1 for s in labeled if s.label in ("LH", "LL")) else: bullish_count = sum(1 for s in labeled if s.label in ("HH", "HL")) bearish_count = sum(1 for s in labeled if s.label in ("LH", "LL")) agree = max(bullish_count, bearish_count) return round(agree / len(labeled) * 100.0, 1) def _detect_flags(bars: list, swings: list, structure: str, cfg: dict) -> dict: flags = { "structure_break": False, "failed_higher_low": False, "failed_breakout": False, "support_failure": False, "resistance_rejection": False, } if not bars: return flags current_price = bars[-1]["c"] highs = [s for s in swings if s.kind == "HIGH" and s.label is not None] lows = [s for s in swings if s.kind == "LOW" and s.label is not None] # structure_break: classifying with the last swing removed gives a # DIFFERENT (established) structure than classifying with it included # -- i.e. the newest swing is what changed the picture. if len(swings) >= 2: prior_structure, _, _ = _classify(swings[:-1], cfg["min_swings_for_structure"]) if prior_structure in (STRUCTURE_BULLISH, STRUCTURE_BEARISH) and structure != prior_structure: flags["structure_break"] = True # failed_higher_low: the most recent LOW swing was labeled HL, but # price has since traded back below that HL's price -- the "higher # low" got invalidated rather than holding as new support. if lows and lows[-1].label == "HL" and current_price < lows[-1].price: flags["failed_higher_low"] = True # support_failure: current price has broken below the most recent # swing LOW outright, regardless of what label it carried. all_lows = [s for s in swings if s.kind == "LOW"] if all_lows and current_price < all_lows[-1].price: flags["support_failure"] = True # resistance_rejection vs failed_breakout: both concern the most # recent swing HIGH. If price got close to it but never closed # above -> rejection. If price DID close above it at some point but # has since fallen back below -> failed breakout (round-tripped). all_highs = [s for s in swings if s.kind == "HIGH"] if all_highs: resistance = all_highs[-1].price tolerance = resistance * (cfg["resistance_rejection_tolerance_pct"] / 100.0) bars_since = [b for b in bars if b["t"] > all_highs[-1].t] broke_above = any(b["c"] > resistance for b in bars_since) if broke_above and current_price < resistance: flags["failed_breakout"] = True elif not broke_above and any(b["h"] >= resistance - tolerance for b in bars_since) \ and current_price < resistance: flags["resistance_rejection"] = True return flags def analyze_structure(symbol: str, bars: list, cfg: dict = None) -> StructureReading: cfg = {**DEFAULT_CONFIG, **(cfg or get_config().get("structure_engine", {}))} lookback = cfg["pivot_lookback_bars"] confirm = cfg["pivot_confirm_bars"] min_needed = lookback + confirm + 1 if len(bars) < min_needed: return StructureReading(symbol=symbol, insufficient_data=True, missing=[f"need at least {min_needed} bars, have {len(bars)}"]) swings = find_swing_points(bars, lookback, confirm) swings = _label_swings(swings) structure, last_high_label, last_low_label = _classify(swings, cfg["min_swings_for_structure"]) score = _structure_score(swings, structure, cfg["structure_score_lookback_swings"]) flags = _detect_flags(bars, swings, structure, cfg) last_highs = [s for s in swings if s.kind == "HIGH"] last_lows = [s for s in swings if s.kind == "LOW"] return StructureReading( symbol=symbol, structure=structure, score=score, swings=swings, last_swing_high=last_highs[-1].price if last_highs else None, last_swing_low=last_lows[-1].price if last_lows else None, flags=flags, insufficient_data=False, )