""" exit_giveback_experimental.py [2026-09-18] NOT wired into monitor.py's live loop -- backtest-only, built to test the giveback-from-peak validation sequence proposed after the GLOO case (entered 4.93, peaked 5.58 at 10:58 ET, exit.py's three existing layers never fired until price had already round-tripped back to 5.225 at 14:17 -- giving back over half the peak gain, because none of VWAP/trend/imbalance-decline reference the position's OWN high since entry, only absolute/session-wide readings). A fourth layer, meant to sit alongside (not replace) exit.py's existing three: only activates once price has given back a meaningful chunk of its own peak (ATR-scaled, so it doesn't trip on ordinary chop), then requires a MAJORITY of four independent votes -- chosen specifically because each one showed a real, early, comparatively clean signal in the GLOO data where the existing signals stayed silent or too noisy: 1. structure -- price lost VWAP, or EMA9 crossed under EMA20 2. demand -- buy-side volume's own rolling regression slope is negative AND sell-side volume isn't fading at least as fast (i.e. real distribution, not just a quiet stretch where nobody's trading either side) 3. momentum -- RSI(14) has faded meaningfully (15+ points) off its own high-since-entry AND sits below 50 4. structure2 -- lower-highs/lower-lows over the recent window (reused from exit.py's own deterioration check) sustained for `exit_confirm_reads` consecutive polls, same hysteresis pattern as exit.py's existing deterioration layer. """ from dataclasses import dataclass, field from indicators import vwap, ema, rsi, linreg_slope, is_lower_highs_lower_lows, atr as calc_atr STATE_HOLD = "HOLD" STATE_EXIT = "EXIT" DEFAULT_CONFIG = { "giveback_atr_multiple": 2.0, # min giveback (in ATR%) before the sequence even activates "min_giveback_pct": 1.5, # floor regardless of ATR (very low-ATR names) "votes_required": 3, # of 4 "confirm_reads": 3, "rsi_period": 14, "rsi_fade_from_peak": 15.0, "rsi_ceiling": 50.0, "vol_history_len": 15, # ~75s at 5s cadence "vol_slope_min_samples": 6, "structure_lookback_bars": 4, } @dataclass class GivebackDecision: 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, buy_vol: float, sell_vol: float, state_in: dict, cfg: dict = None) -> GivebackDecision: """ bars: same short rolling 1-min-bar window exit.py/smart_engine.py take. buy_vol/sell_vol: THIS poll's raw classified volume over the same rolling window smart_engine's imbalance uses (not the ratio -- stream.SymbolBuffer doesn't expose raw volumes, caller pulls them from buf._trade_log the same way get_trade_imbalance() does). state_in: caller-held state, fed back every poll -- {"peak_price", "peak_rsi", "buy_vol_hist": [...], "sell_vol_hist": [...], "confirm_count"}. Fresh {} at entry. """ cfg = _merge_cfg(cfg) state_in = dict(state_in or {}) if len(bars) < 3: return GivebackDecision(symbol=symbol, reason="insufficient bar data", state_out=state_in) price = bars[-1]["c"] vwap_value = vwap(bars) closes = [b["c"] for b in bars] ema9 = ema(closes, 9) ema20 = ema(closes, 20) a = calc_atr(bars, period=min(14, max(2, len(bars) - 1))) atr_pct = (a / price * 100.0) if price else 0.0 rsi_val = rsi(bars, period=cfg["rsi_period"]) lhl = is_lower_highs_lower_lows(bars, lookback=min(cfg["structure_lookback_bars"], len(bars))) peak_price = max(state_in.get("peak_price", price), price) peak_rsi = max(state_in.get("peak_rsi", rsi_val), rsi_val) buy_hist = list(state_in.get("buy_vol_hist", [])) + [buy_vol] sell_hist = list(state_in.get("sell_vol_hist", [])) + [sell_vol] buy_hist = buy_hist[-cfg["vol_history_len"]:] sell_hist = sell_hist[-cfg["vol_history_len"]:] giveback_pct = (peak_price - price) / peak_price * 100.0 if peak_price else 0.0 floor = max(cfg["min_giveback_pct"], cfg["giveback_atr_multiple"] * atr_pct) base_state = {"peak_price": peak_price, "peak_rsi": peak_rsi, "buy_vol_hist": buy_hist, "sell_vol_hist": sell_hist} metrics = {"price": price, "peak_price": round(peak_price, 4), "giveback_pct": round(giveback_pct, 3), "floor_pct": round(floor, 3), "vwap": round(vwap_value, 4), "ema9": round(ema9, 4), "ema20": round(ema20, 4), "rsi": round(rsi_val, 1), "peak_rsi": round(peak_rsi, 1), "lhl": lhl, "buy_vol": buy_vol, "sell_vol": sell_vol} if giveback_pct < floor: return GivebackDecision(symbol=symbol, state=STATE_HOLD, reason=f"giveback {giveback_pct:.2f}% below {floor:.2f}% floor -- not evaluated", metrics=metrics, state_out={**base_state, "confirm_count": 0}) vote_structure = (price < vwap_value) or (ema9 < ema20) vote_demand = False if len(buy_hist) >= cfg["vol_slope_min_samples"]: buy_slope = linreg_slope(buy_hist) sell_slope = linreg_slope(sell_hist) vote_demand = buy_slope < 0 and sell_slope >= buy_slope metrics["buy_vol_slope"] = round(buy_slope, 1) metrics["sell_vol_slope"] = round(sell_slope, 1) vote_momentum = (rsi_val < cfg["rsi_ceiling"]) and (peak_rsi - rsi_val >= cfg["rsi_fade_from_peak"]) vote_structure2 = lhl votes = [vote_structure, vote_demand, vote_momentum, vote_structure2] n_votes = sum(votes) metrics["votes"] = {"structure": vote_structure, "demand": vote_demand, "momentum": vote_momentum, "structure2": vote_structure2, "count": n_votes} if n_votes < cfg["votes_required"]: return GivebackDecision( symbol=symbol, state=STATE_HOLD, reason=f"giveback {giveback_pct:.2f}% but only {n_votes}/4 votes agree", metrics=metrics, state_out={**base_state, "confirm_count": 0}) count = state_in.get("confirm_count", 0) + 1 if count < cfg["confirm_reads"]: return GivebackDecision( symbol=symbol, state=STATE_HOLD, reason=f"giveback {giveback_pct:.2f}%, {n_votes}/4 votes, confirming ({count}/{cfg['confirm_reads']})", metrics=metrics, state_out={**base_state, "confirm_count": count}) return GivebackDecision( symbol=symbol, state=STATE_EXIT, should_exit=True, reason=(f"giveback-from-peak: {giveback_pct:.2f}% off ${peak_price:.4f} peak, " f"{n_votes}/4 votes agreed for {count} reads"), metrics=metrics, state_out={**base_state, "confirm_count": 0})