""" compression_engine.py [Phase 3 -- new] Detects range/volatility compression ("coiling") ahead of a potential breakout, and scores how much directional pressure is building during that compression. Pure function, no state, no I/O -- same contract as every other engine in this pipeline (stream_features.py, trend_engine.py, structure_engine.py): given a bars window plus already-computed features/ structure, returns a score. Nothing here fetches data or stores anything. Two related but distinct scores: COMPRESSION_SCORE -- how tight/quiet the recent range is, relative to its own recent history (direction-agnostic: a symbol coiling before a breakdown scores just as high on this alone as one coiling before a breakout). BREAKOUT_PRESSURE_SCORE -- COMPRESSION_SCORE scaled up only when the tightening is happening in a bullish context (VWAP rising, higher lows holding, price still near its highs, volume not collapsing). A quiet range with deteriorating structure is NOT "pressure building toward a breakout" -- it's just quiet, and this keeps that distinction instead of conflating the two. Both `features` (stream_features.StreamFeatures) and `structure` (structure_engine.StructureReading) are optional -- if the caller doesn't have them yet, breakout_pressure_score degrades to compression_score with no bonus (multiplier 1.0) rather than failing, consistent with this project's "missing secondary confirmation reduces confidence, doesn't hard-block" principle (see the project's own design brief section 38). Not consumed anywhere in the live bot yet -- config.compression_engine follows the same enabled:false / SIMULATION-ONLY convention every other engine added to this project uses until it's wired in and backtested. """ from dataclasses import dataclass, field from statistics import pstdev from config_loader import get_config from indicators import atr as calc_atr, distance_from_high_pct DEFAULT_CONFIG = { "lookback_bars": 8, "weights": { "range_contraction": 30, "atr_contraction": 25, "close_to_close_contraction": 15, "holding_near_highs": 15, "volume_not_collapsing": 15, }, "near_high_max_distance_pct": 3.0, "volume_not_collapsing_min_ratio": 0.7, "pressure_bonus_vwap_slope": 0.15, "pressure_bonus_higher_lows": 0.20, "pressure_bonus_near_resistance": 0.15, "pressure_max_multiplier": 1.5, "near_resistance_max_distance_pct": 2.0, } @dataclass class CompressionReading: symbol: str compression_score: float = 0.0 breakout_pressure_score: float = 0.0 range_contraction_ratio: float = None # <1 = tighter than the earlier window atr_contraction_ratio: float = None close_to_close_contraction_ratio: float = None breakdown: dict = field(default_factory=dict) insufficient_data: bool = False missing: list = field(default_factory=list) def _clamp01(x: float) -> float: return max(0.0, min(1.0, x)) def _pct_changes(closes: list) -> list: return [(closes[i] - closes[i - 1]) / closes[i - 1] * 100.0 for i in range(1, len(closes)) if closes[i - 1]] def _range_pct(bars: list) -> float: if not bars: return 0.0 price = bars[-1]["c"] or 1.0 return (max(b["h"] for b in bars) - min(b["l"] for b in bars)) / price * 100.0 def analyze_compression(symbol: str, bars: list, features=None, structure=None, cfg: dict = None) -> CompressionReading: """ bars: 1-min bars, oldest first (stream.StreamManager.get_bars()) -- needs at least 2*lookback_bars+1 to compare a "recent" window against the "earlier" window immediately preceding it. features: optional stream_features.StreamFeatures -- used only for the VWAP-slope bonus on breakout_pressure_score. Omit to skip that bonus. structure: optional structure_engine.StructureReading -- used only for the higher-lows / near-resistance bonuses. Omit to skip those. """ cfg = {**DEFAULT_CONFIG, **(cfg or get_config().get("compression_engine", {}))} lookback = cfg["lookback_bars"] min_bars = 2 * lookback + 1 if len(bars) < min_bars: return CompressionReading(symbol=symbol, insufficient_data=True, missing=[f"need at least {min_bars} bars, have {len(bars)}"]) recent = bars[-lookback:] earlier = bars[-2 * lookback:-lookback] recent_range = _range_pct(recent) earlier_range = _range_pct(earlier) range_contraction_ratio = recent_range / earlier_range if earlier_range > 0 else 1.0 recent_atr = calc_atr(recent, period=max(1, len(recent) - 1)) earlier_atr = calc_atr(earlier, period=max(1, len(earlier) - 1)) atr_contraction_ratio = recent_atr / earlier_atr if earlier_atr > 0 else 1.0 recent_std = pstdev(_pct_changes([b["c"] for b in recent])) if len(recent) > 1 else 0.0 earlier_std = pstdev(_pct_changes([b["c"] for b in earlier])) if len(earlier) > 1 else 0.0 close_to_close_contraction_ratio = recent_std / earlier_std if earlier_std > 0 else 1.0 current_price = bars[-1]["c"] dist_from_high = distance_from_high_pct(current_price, max(b["h"] for b in recent)) recent_vol_avg = sum(b["v"] for b in recent) / len(recent) if recent else 0.0 earlier_vol_avg = sum(b["v"] for b in earlier) / len(earlier) if earlier else 0.0 volume_ratio = recent_vol_avg / earlier_vol_avg if earlier_vol_avg > 0 else 1.0 w = cfg["weights"] breakdown = { # lower ratio = more contraction = more credit; a ratio >= 1 (range # EXPANDING, not contracting) earns zero credit here rather than a # negative one -- this score answers "how compressed," not "how # expanded," those are different questions this module isn't asked. "range_contraction": w["range_contraction"] * _clamp01(1 - range_contraction_ratio), "atr_contraction": w["atr_contraction"] * _clamp01(1 - atr_contraction_ratio), "close_to_close_contraction": w["close_to_close_contraction"] * _clamp01(1 - close_to_close_contraction_ratio), "holding_near_highs": w["holding_near_highs"] * _clamp01( (cfg["near_high_max_distance_pct"] - dist_from_high) / cfg["near_high_max_distance_pct"]), "volume_not_collapsing": w["volume_not_collapsing"] * _clamp01( volume_ratio / cfg["volume_not_collapsing_min_ratio"]), } max_possible = sum(w.values()) compression_score = round(_clamp01(sum(breakdown.values()) / max_possible) * 100.0, 2) # ---- breakout pressure: same compression, scaled by bullish context ---- multiplier = 1.0 if features is not None and not getattr(features, "insufficient_data", False): if (features.vwap.get("slope") or 0) > 0: multiplier += cfg["pressure_bonus_vwap_slope"] if structure is not None and not getattr(structure, "insufficient_data", False): lows = [s for s in structure.swings if s.kind == "LOW" and s.label is not None] if lows and lows[-1].label == "HL": multiplier += cfg["pressure_bonus_higher_lows"] last_high = structure.last_swing_high if last_high is not None and current_price: room_pct = (last_high - current_price) / current_price * 100.0 if 0 <= room_pct <= cfg["near_resistance_max_distance_pct"]: multiplier += cfg["pressure_bonus_near_resistance"] multiplier = min(multiplier, cfg["pressure_max_multiplier"]) breakout_pressure_score = round(min(100.0, compression_score * multiplier), 2) return CompressionReading( symbol=symbol, compression_score=compression_score, breakout_pressure_score=breakout_pressure_score, range_contraction_ratio=round(range_contraction_ratio, 3), atr_contraction_ratio=round(atr_contraction_ratio, 3), close_to_close_contraction_ratio=round(close_to_close_contraction_ratio, 3), breakdown={k: round(v, 2) for k, v in breakdown.items()}, insufficient_data=False, )