""" exit_resistance_timer_experimental.py [2026-09-19] NOT wired into monitor.py's live loop -- backtest-only. User's proposed design (2026-09-19), deliberately simpler than exit_prf_experimental.py's multi-stage confirm-count state machine: once price stops printing new highs, check whether trade imbalance is still positive (buyers still active). If so, that's "at resistance" -- start a fixed real-time clock (recovery_window_seconds). If price hasn't made a new high by the time that clock runs out, exit. No confirm-read voting, no structure (lower-highs/lower-lows) layer, no strong-imbalance floor -- just "resistance detected -> wait N seconds for a new high -> exit if none came." Time-boxed in wall-clock seconds (not poll counts), since poll cadence isn't guaranteed constant (config's poll_interval_seconds=5 is the target, but real logged poll times drift -- see simulate.py's load_real_poll_times docstring). state_in carries the resistance-start timestamp as an ISO string so evaluate() stays a pure function of its inputs, same convention as exit_prf_experimental.py's state_in/state_out. Reversible exactly like PRF: any poll that prints a fresh peak high cancels the timer outright, no matter how much of the window had elapsed. """ from dataclasses import dataclass, field from datetime import datetime STATE_NORMAL = "NORMAL" STATE_AT_RESISTANCE = "AT_RESISTANCE" STATE_HOLD = "HOLD" STATE_EXIT = "EXIT" DEFAULT_CONFIG = { "min_mfe_pct": 3.0, # same "after +3% MFE" arm gate PRF uses -- don't evaluate pre-move noise "resistance_band_pct": 0.15, # within this % of peak still counts as "at the peak", not a new high "resistance_imbalance_threshold": 0.0, # imbalance strictly above this counts as "still positive" "recovery_window_seconds": 60.0, # real-time clock started the poll resistance is first detected } @dataclass class ResistanceTimerDecision: 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 {})} def evaluate(symbol: str, bars: list, entry_price: float, trade_imbalance, now_ts: datetime, state_in: dict, cfg: dict = None) -> ResistanceTimerDecision: """ bars: same short rolling 1-min-bar window exit.py/exit_prf_experimental.py take. trade_imbalance: stream.py's get_trade_imbalance() live read, None on a cold buffer (fails open -- holds whatever state was already in progress). now_ts: this poll's timestamp (UTC datetime) -- the recovery window is measured against this, not a poll count. state_in: caller-held state fed back every poll -- {"peak_price", "resistance_since" (ISO string or None), "peak_at_resistance"}. Fresh {} at entry. """ cfg = _merge_cfg(cfg) state_in = dict(state_in or {}) if len(bars) < 3: return ResistanceTimerDecision(symbol=symbol, reason="insufficient bar data", state_out=state_in) price = bars[-1]["c"] prior_peak = state_in.get("peak_price", price) peak_price = max(prior_peak, price) made_new_high = price > prior_peak resistance_since = state_in.get("resistance_since") peak_at_resistance = state_in.get("peak_at_resistance") mfe_pct = (peak_price - entry_price) / entry_price * 100.0 if entry_price else 0.0 armed = mfe_pct >= cfg["min_mfe_pct"] metrics = { "price": price, "peak_price": round(peak_price, 4), "mfe_pct": round(mfe_pct, 3), "armed": armed, "made_new_high": made_new_high, "trade_imbalance": trade_imbalance, "resistance_since_in": resistance_since, } base_state = {"peak_price": peak_price} if not armed: return ResistanceTimerDecision( symbol=symbol, state=STATE_HOLD, reason=f"not armed yet (MFE {mfe_pct:.2f}% < {cfg['min_mfe_pct']}%)", metrics=metrics, state_out={**base_state, "resistance_since": None, "peak_at_resistance": None}) # A fresh peak high is a recovery -- cancels any running timer outright, # however much of the window had already elapsed. if made_new_high: reason = (f"new high ${price:.4f} -- resistance timer cancelled" if resistance_since else f"making new highs (${price:.4f})") return ResistanceTimerDecision( symbol=symbol, state=STATE_HOLD, reason=reason, metrics=metrics, state_out={**base_state, "resistance_since": None, "peak_at_resistance": None}) if trade_imbalance is None: # Cold read -- hold whatever state (ticking or not) steady rather than # start/cancel the timer on a data gap, same fail-open contract PRF uses. return ResistanceTimerDecision( symbol=symbol, state=STATE_AT_RESISTANCE if resistance_since else STATE_HOLD, reason="no imbalance data this poll -- holding", metrics=metrics, state_out={**base_state, "resistance_since": resistance_since, "peak_at_resistance": peak_at_resistance}) imbalance_positive = trade_imbalance > cfg["resistance_imbalance_threshold"] at_resistance_band = price >= peak_price * (1 - cfg["resistance_band_pct"] / 100.0) metrics.update({"imbalance_positive": imbalance_positive, "at_resistance_band": at_resistance_band}) if resistance_since is None: if imbalance_positive and at_resistance_band: # First poll of failed response: price stalled at its own peak # while imbalance is still positive -- start the recovery clock. return ResistanceTimerDecision( symbol=symbol, state=STATE_AT_RESISTANCE, reason=(f"at resistance ${peak_price:.4f}, imbalance {trade_imbalance:+.0%} still positive, " f"no new high -- starting {cfg['recovery_window_seconds']:.0f}s recovery timer"), metrics=metrics, state_out={**base_state, "resistance_since": now_ts.isoformat(), "peak_at_resistance": peak_price}) return ResistanceTimerDecision( symbol=symbol, state=STATE_HOLD, reason=(f"no new high but not at-resistance-with-positive-imbalance yet " f"(imbalance {trade_imbalance:+.0%})"), metrics=metrics, state_out={**base_state, "resistance_since": None, "peak_at_resistance": None}) # Timer already ticking from a prior poll. elapsed = (now_ts - datetime.fromisoformat(resistance_since)).total_seconds() metrics["elapsed_seconds"] = round(elapsed, 1) giveback_pct = (peak_at_resistance - price) / peak_at_resistance * 100.0 if peak_at_resistance else 0.0 metrics["giveback_pct"] = round(giveback_pct, 3) if elapsed >= cfg["recovery_window_seconds"]: return ResistanceTimerDecision( symbol=symbol, state=STATE_EXIT, should_exit=True, reason=(f"no recovery within {cfg['recovery_window_seconds']:.0f}s of hitting resistance at " f"${peak_at_resistance:.4f} ({elapsed:.0f}s elapsed, still no new high, " f"giveback {giveback_pct:.2f}%)"), metrics=metrics, state_out={**base_state, "resistance_since": None, "peak_at_resistance": None}) return ResistanceTimerDecision( symbol=symbol, state=STATE_AT_RESISTANCE, reason=(f"at resistance ${peak_at_resistance:.4f}, {elapsed:.0f}/{cfg['recovery_window_seconds']:.0f}s " f"elapsed waiting for recovery, giveback {giveback_pct:.2f}%"), metrics=metrics, state_out={**base_state, "resistance_since": resistance_since, "peak_at_resistance": peak_at_resistance})