""" exit_resistance_stall.py [2026-09-23] "Stalled at resistance -> sell at the ceiling" exit layer, specified by the user after walking through OPTX (bought $7.61, then tested a $7.70-7.72 ceiling ~10 times in 50 minutes on thin volume without a breakout, while the 15c giveback-room gate never armed). Every poll, for an open position: 1. Find the ceiling. Among the 1-min bars since entry (and since the last confirmed breakout, if any), the ceiling is the HIGHEST bar high h that has at least min_touches bar highs within [h - band, h], band = max(band_cents, band_pct of price). A single wick above a real cluster doesn't count (it has no touches of its own), so the cluster under it stays the ceiling. zone_top = h, zone_bottom = the lowest high in that cluster. 2. Proof of breakout: a COMPLETED bar closing above zone_top + breakout_buffer on volume >= breakout_volume_ratio x the average completed-bar volume. That turns the stall clock off: only bars after the breakout bar are considered from then on, so a new ceiling (if any) has to form above it and starts its own clock. 3. The clock starts at the ceiling's FIRST TOUCH after entry (the earliest bar in the cluster). After stall_minutes with no breakout, "building" is checked once: rising lows (the minimum low of each third of the bars since first touch strictly increasing) OR rising volume on the touches (later touches' average volume >= rising_volume_ratio x earlier touches'). If building, the deadline moves out by extension_minutes (up to max_extensions times); otherwise the rule arms. 4. Armed: the first poll where price >= zone_bottom sells (market order, via the caller). Applies whether the position is above or below entry at that point -- at the ceiling is the best exit the stall is going to offer. Pure function of its inputs; caller holds state_in/state_out ({"breakout_after", "extensions", "armed"}), same convention as exit_giveback_room.py. """ from dataclasses import dataclass, field from datetime import datetime, timedelta from config_loader import get_config STATE_HOLD = "HOLD" STATE_EXIT = "EXIT" DEFAULT_CONFIG = { "enabled": False, "min_touches": 3, "band_cents": 2.0, "band_pct": 0.3, "stall_minutes": 60.0, "extension_minutes": 10.0, "max_extensions": 1, "breakout_buffer_cents": 1.0, "breakout_volume_ratio": 1.5, "rising_volume_ratio": 1.3, # [2026-09-25] User rule (EMBC: armed at a 5.845-5.86 ceiling while price # had settled into a lower 5.81-5.83 box). Once armed, also find the real # ceiling of the LAST low_ceiling_minutes. If no slot is free, or the # position is no longer healthy, sell when price reaches that lower # ceiling. With a free slot and a healthy position, keep holding for the # original ceiling. "low_ceiling_enabled": False, "low_ceiling_minutes": 15, } @dataclass class ResistanceStallDecision: symbol: str state: str = STATE_HOLD should_exit: bool = False reason: str = "" metrics: dict = field(default_factory=dict) state_out: dict = field(default_factory=dict) def _merge_cfg(cfg): return {**DEFAULT_CONFIG, **(cfg or get_config().get("resistance_stall", {}))} def _as_dt(t): return datetime.fromisoformat(t) if isinstance(t, str) else t def _find_ceiling(bars: list, band: float, min_touches: int, max_bars_above: int = 1): """The ceiling has to be the top of the range: at most max_bars_above bars (a stray wick) may print above it. Without this, a stock trending away from an old consolidation would read that consolidation as its 'ceiling'. Bars above only grow as h drops, so the search stops there.""" highs = sorted({b["h"] for b in bars}, reverse=True) for h in highs: if sum(1 for b in bars if b["h"] > h) > max_bars_above: break cluster = [b for b in bars if h - band <= b["h"] <= h] if len(cluster) >= min_touches: return h, min(b["h"] for b in cluster), cluster return None, None, [] def _rising_lows(bars: list) -> bool: if len(bars) < 6: return False n = len(bars) // 3 thirds = [bars[:n], bars[n:2 * n], bars[2 * n:]] lows = [min(b["l"] for b in part) for part in thirds] return lows[0] < lows[1] < lows[2] def _rising_touch_volume(cluster: list, ratio: float) -> bool: if len(cluster) < 4: return False half = len(cluster) // 2 early = sum(b["v"] for b in cluster[:half]) / half late = sum(b["v"] for b in cluster[half:]) / (len(cluster) - half) return early > 0 and late >= early * ratio def evaluate(symbol: str, bars: list, entry_time, now_ts, state_in: dict, cfg: dict = None, slots_free: bool = True, healthy: bool = True) -> ResistanceStallDecision: """ bars: the symbol's 1-min bars, oldest first, forming bar last (same list exit.py gets). entry_time/now_ts: datetimes or ISO strings. """ cfg = _merge_cfg(cfg) state = {"breakout_after": None, "extensions": 0, "armed": False, **(state_in or {})} entry_dt, now = _as_dt(entry_time), _as_dt(now_ts) entry_minute = entry_dt.replace(second=0, microsecond=0) if not bars: return ResistanceStallDecision(symbol, reason="no bars", state_out=state) price = bars[-1]["c"] band = max(cfg["band_cents"] / 100.0, price * cfg["band_pct"] / 100.0) start = _as_dt(state["breakout_after"]) if state["breakout_after"] else entry_minute window = [b for b in bars if _as_dt(b["t"]) >= start] completed = window[:-1] zone_top, zone_bottom, cluster = _find_ceiling(window, band, cfg["min_touches"]) metrics = {"price": price, "band": round(band, 4)} if zone_top is None: return ResistanceStallDecision(symbol, reason="no ceiling yet", metrics=metrics, state_out=state) # Proof of breakout: a completed bar closing above the zone on real volume. all_completed = [b for b in bars[:-1]] avg_vol = (sum(b["v"] for b in all_completed[-60:]) / len(all_completed[-60:])) if all_completed else 0.0 first_touch = _as_dt(cluster[0]["t"]) for b in completed: if (_as_dt(b["t"]) > first_touch and b["c"] > zone_top + cfg["breakout_buffer_cents"] / 100.0 and avg_vol and b["v"] >= avg_vol * cfg["breakout_volume_ratio"]): new_state = {"breakout_after": (_as_dt(b["t"]) + timedelta(minutes=1)).isoformat(), "extensions": 0, "armed": False} return ResistanceStallDecision( symbol, reason=(f"breakout confirmed: {_as_dt(b['t']).strftime('%H:%M')} bar closed " f"{b['c']:.4f} > ceiling {zone_top:.4f} on {b['v']:.0f} sh " f"({b['v'] / avg_vol:.1f}x avg) -- stall clock reset"), metrics={**metrics, "zone_top": zone_top, "zone_bottom": zone_bottom}, state_out=new_state) since_first = [b for b in window if _as_dt(b["t"]) >= first_touch] rising_lows = _rising_lows([b for b in since_first[:-1]] or since_first) rising_vol = _rising_touch_volume(cluster, cfg["rising_volume_ratio"]) stalled_min = (now - first_touch).total_seconds() / 60.0 deadline = cfg["stall_minutes"] + state["extensions"] * cfg["extension_minutes"] metrics.update({"zone_top": round(zone_top, 4), "zone_bottom": round(zone_bottom, 4), "touches": len(cluster), "first_touch": first_touch.isoformat(), "stalled_minutes": round(stalled_min, 1), "deadline_minutes": deadline, "rising_lows": rising_lows, "rising_touch_volume": rising_vol, "armed": state["armed"]}) if not state["armed"] and stalled_min >= deadline: if (rising_lows or rising_vol) and state["extensions"] < cfg["max_extensions"]: state["extensions"] += 1 return ResistanceStallDecision( symbol, reason=(f"stalled {stalled_min:.0f}m under {zone_bottom:.4f}-{zone_top:.4f} but building " f"(rising lows={rising_lows}, rising test volume={rising_vol}) -- " f"+{cfg['extension_minutes']:.0f}m"), metrics=metrics, state_out=state) state["armed"] = True metrics["armed"] = True if state["armed"] and price >= zone_bottom: return ResistanceStallDecision( symbol, state=STATE_EXIT, should_exit=True, reason=(f"resistance stall: {len(cluster)} touches of {zone_bottom:.4f}-{zone_top:.4f} over " f"{stalled_min:.0f}m, no breakout -- sold at the ceiling ({price:.4f})"), metrics=metrics, state_out=state) if state["armed"] and cfg["low_ceiling_enabled"]: cutoff = now - timedelta(minutes=cfg["low_ceiling_minutes"]) recent = [b for b in completed if _as_dt(b["t"]) >= cutoff] lc_top, lc_bottom, lc_cluster = _find_ceiling(recent, band, cfg["min_touches"]) if recent else (None, None, []) if lc_top is not None and lc_top < zone_bottom: metrics.update({"low_ceiling_top": round(lc_top, 4), "low_ceiling_bottom": round(lc_bottom, 4), "low_ceiling_touches": len(lc_cluster), "slots_free": slots_free, "healthy": healthy}) if (not slots_free or not healthy) and price >= lc_bottom: why = "no free slot" if not slots_free else "position no longer healthy" return ResistanceStallDecision( symbol, state=STATE_EXIT, should_exit=True, reason=(f"resistance stall, low ceiling: armed at {zone_bottom:.4f}-{zone_top:.4f} but the last " f"{cfg['low_ceiling_minutes']:.0f}m topped at {lc_bottom:.4f}-{lc_top:.4f} " f"({len(lc_cluster)} touches); {why} -- sold at {price:.4f}"), metrics=metrics, state_out=state) if slots_free and healthy: return ResistanceStallDecision( symbol, reason=(f"armed at {zone_bottom:.4f}; last {cfg['low_ceiling_minutes']:.0f}m ceiling " f"{lc_bottom:.4f}-{lc_top:.4f}, holding (slot free, healthy)"), metrics=metrics, state_out=state) return ResistanceStallDecision( symbol, reason=(f"armed at {zone_bottom:.4f}; will sell at the last-{cfg['low_ceiling_minutes']:.0f}m " f"ceiling {lc_bottom:.4f} ({'no free slot' if not slots_free else 'not healthy'})"), metrics=metrics, state_out=state) reason = (f"armed, waiting for price to reach {zone_bottom:.4f}" if state["armed"] else f"ceiling {zone_bottom:.4f}-{zone_top:.4f}, {len(cluster)} touches, " f"{stalled_min:.0f}/{deadline:.0f}m") return ResistanceStallDecision(symbol, reason=reason, metrics=metrics, state_out=state)