""" Weekly RSI extreme + recovery scanner. What it catches: Severe weekly oversold conditions that flush capitulating holders, followed by the first sign of strength. Historical examples this would have hit: BTC 2022-11 ($15.5k bottom) ETH 2022-06 ($900 bottom) SOL 2022-12 ($8 bottom) AAVE 2022-06 ($45 bottom) Trigger logic (intentionally strict): PRE-CONDITION: ≥ 4 consecutive completed weekly bars with RSI(14) < 30 (this is the capitulation phase — sustained extreme weakness) TRIGGER: current completed week's RSI(14) >= 35 (i.e. the FIRST recovery week — RSI has lifted off the floor) COOLDOWN: 60 days — these events happen ~once per year per asset. No re-firing on the same setup. Companion exit parameters (passed to the bot via signal payload — wider stops to match the longer holding period and higher single-trade conviction): SL = 8% TRAILING_ACTIVATE = 15% TRAILING_STOP = 6% MAX_HOLD = 60 days This is a high-conviction, low-frequency signal. Expect 0-3 fires per asset per year. Don't tune it for higher frequency — that defeats the point. """ from __future__ import annotations import logging from datetime import datetime, timedelta, timezone from typing import Optional import httpx from app.config import settings from app.services.market_data import REVERSAL_BASKET, for_asset, drop_in_progress_bar from app.services import scanner_state SCANNER_NAME = "weekly_rsi_reversal" scanner_state.register(SCANNER_NAME) logger = logging.getLogger(__name__) # ─── Tunables ─────────────────────────────────────────────────────────────── RSI_PERIOD = 14 RSI_OVERSOLD = 30 # "deeply oversold" floor (long setup) RSI_RECOVERY_TRIGGER = 35 # bounce-off line for the long RSI_OVERBOUGHT = 70 # "euphoric" ceiling (short setup) RSI_ROLLOVER_TRIGGER = 65 # roll-off line for the short WEEKS_OF_OVERSOLD = 4 # consecutive extreme weeks required (both sides) WEEKS_TO_FETCH = 40 # enough history for stable Wilder's RSI COOLDOWN_DAYS = 60 # Exit profile passed to /signals/ingest. Caller can override on the bot side. PAYLOAD_CONFIDENCE = 88 # high conviction — top of the regime-filter score PAYLOAD_EXPECTED_MOVE = 30.0 # historical median move after capitulation # Cooldown is now DB-backed via scanner_state.in_cooldown() — survives # restart. No more in-memory _last_signal_at dict. # ─── RSI math (Wilder's smoothed) ─────────────────────────────────────────── def calculate_rsi(closes: list[float], period: int = 14) -> list[Optional[float]]: """Wilder's RSI. Returns a list aligned with `closes` where the first `period` entries are None (not enough history yet). Implementation note: the first valid RSI uses a SIMPLE mean of the first `period` gains/losses, subsequent values use exponential smoothing with alpha = 1/period. This matches TradingView, Binance UI, and most charting tools — the "Cutler's RSI" (pure SMA) variant would give different results. """ n = len(closes) rsi: list[Optional[float]] = [None] * n if n < period + 1: return rsi gains = [max(closes[i] - closes[i - 1], 0) for i in range(1, n)] losses = [max(closes[i - 1] - closes[i], 0) for i in range(1, n)] # First valid RSI = simple mean over first `period` deltas avg_gain = sum(gains[:period]) / period avg_loss = sum(losses[:period]) / period if avg_loss == 0: rsi[period] = 100.0 else: rs = avg_gain / avg_loss rsi[period] = 100 - 100 / (1 + rs) # Subsequent values use Wilder's smoothing for i in range(period + 1, n): avg_gain = (avg_gain * (period - 1) + gains[i - 1]) / period avg_loss = (avg_loss * (period - 1) + losses[i - 1]) / period if avg_loss == 0: rsi[i] = 100.0 else: rs = avg_gain / avg_loss rsi[i] = 100 - 100 / (1 + rs) return rsi # ─── Signal logic ─────────────────────────────────────────────────────────── def evaluate_rsi_reversal(weekly_candles: list[dict]) -> tuple[bool, dict]: """Pure function — no side effects, fully testable. Returns (is_signal, debug_info). """ if len(weekly_candles) < RSI_PERIOD + WEEKS_OF_OVERSOLD + 2: return False, {"reason": "insufficient_data", "bars": len(weekly_candles)} closes = [c["close"] for c in weekly_candles] rsi = calculate_rsi(closes, RSI_PERIOD) # The MOST RECENT weekly bar is `closes[-1]`. We treat it as the "current" # recovery candidate. Look BACKWARDS for WEEKS_OF_OVERSOLD prior bars all # with RSI < RSI_OVERSOLD. current_rsi = rsi[-1] if current_rsi is None: return False, {"reason": "rsi_not_computable"} window = rsi[-(WEEKS_OF_OVERSOLD + 1):-1] # the N weeks before current if any(r is None for r in window): return False, {"reason": "history_gaps_in_window"} this_close = weekly_candles[-1]["close"] prev_close = weekly_candles[-2]["close"] # ── LONG: capitulation bottom ────────────────────────────────────────── # Sustained RSI<30, now lifting ≥35, price turning up. if (current_rsi >= RSI_RECOVERY_TRIGGER and all(r < RSI_OVERSOLD for r in window) and this_close > prev_close): low_w = min(c["low"] for c in weekly_candles[-(WEEKS_OF_OVERSOLD + 1):]) return True, { "direction": "buy", "current_rsi": round(current_rsi, 1), "window_rsi": [round(r, 1) for r in window], "move_pct": round((this_close - low_w) / low_w * 100, 2) if low_w > 0 else 0.0, "trigger_close": round(this_close, 4), } # ── SHORT: euphoric top ──────────────────────────────────────────────── # Sustained RSI>70, now rolling ≤65, price turning down. Symmetric mirror. if (current_rsi <= RSI_ROLLOVER_TRIGGER and all(r > RSI_OVERBOUGHT for r in window) and this_close < prev_close): high_w = max(c["high"] for c in weekly_candles[-(WEEKS_OF_OVERSOLD + 1):]) return True, { "direction": "short", "current_rsi": round(current_rsi, 1), "window_rsi": [round(r, 1) for r in window], "move_pct": round((high_w - this_close) / high_w * 100, 2) if high_w > 0 else 0.0, "trigger_close": round(this_close, 4), } # Neither side qualified — report the closest miss for the log. return False, { "reason": "no_setup", "current_rsi": round(current_rsi, 1), "window_rsi": [round(r, 1) for r in window], } # ─── Ingest emission ──────────────────────────────────────────────────────── async def _emit_signal(asset: str, debug: dict) -> None: if not settings.ingest_api_key: logger.warning("RSI-reversal would fire on %s but INGEST_API_KEY empty", asset) return direction = debug["direction"] side_word = "capitulation bottom" if direction == "buy" else "euphoric top" payload = { "source": "rsi_reversal", "external_id": f"rsi-{asset}-{direction}-{datetime.now(timezone.utc).strftime('%Y%W')}", "text": ( f"Weekly RSI {side_word} on {asset}: RSI {debug['current_rsi']} after " f"{WEEKS_OF_OVERSOLD}wk extreme (window: {debug['window_rsi']}). " f"{debug['move_pct']}% move off the extreme @ ${debug['trigger_close']}." ), "signal": direction, "target_asset": asset, "confidence": PAYLOAD_CONFIDENCE, "category": "rsi_extreme_reversal", "expected_move_pct": PAYLOAD_EXPECTED_MOVE, } async with httpx.AsyncClient(timeout=10) as client: resp = await client.post( "http://localhost:8000/api/signals/ingest", json=payload, headers={"X-Ingest-Key": settings.ingest_api_key}, ) if resp.status_code >= 400: logger.error("RSI-reversal ingest failed (%d): %s", resp.status_code, resp.text[:200]) else: logger.info("RSI-reversal signal emitted for %s: %s", asset, resp.json()) # ─── Scheduler entry point ────────────────────────────────────────────────── async def scan_once() -> None: """One scan pass over REVERSAL_BASKET. Called by APScheduler. Kill-switch aware: short-circuits if the operator disabled this scanner. Cooldown is DB-backed (queries posts table), so restarts don't reset state. """ if not scanner_state.is_enabled(SCANNER_NAME): logger.debug("RSI-reversal scanner disabled — skipping run") return fired_any = False error_msg: Optional[str] = None for asset in REVERSAL_BASKET: # Cooldown — DB-backed. if await scanner_state.in_cooldown("rsi_reversal", asset, COOLDOWN_DAYS): continue provider = for_asset(asset) try: candles = await provider.fetch_1w(asset, weeks=WEEKS_TO_FETCH) except Exception as exc: error_msg = f"{asset}: {exc}" logger.error("RSI-reversal fetch failed for %s via %s: %s", asset, provider.name, exc) continue candles = drop_in_progress_bar(candles, "1w") if not candles: logger.warning("RSI-reversal: %s returned 0 weekly bars from %s", asset, provider.name) continue is_signal, debug = evaluate_rsi_reversal(candles) if is_signal: logger.info("RSI-reversal scan %s [%s]: FIRE — %s", asset, provider.name, debug) await _emit_signal(asset, debug) fired_any = True else: logger.debug("RSI-reversal scan %s [%s]: no — %s", asset, provider.name, debug) if error_msg: scanner_state.record_run(SCANNER_NAME, "error", error_msg) elif fired_any: scanner_state.record_run(SCANNER_NAME, "fired") else: scanner_state.record_run(SCANNER_NAME, "ok")