trumpsignal-backend/app/services/scanners/_archive/weekly_rsi_reversal.py

"""
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")