""" Breakout Signal Monitor — ETH + LINK (BTC as trend context) Polls Binance every 5 minutes. When both conditions are met: 1. BB squeeze: BB width in bottom 20% of last 60 candles 2. Volume spike: current volume > 2.5x 20-period average 3. Taker buy ratio > 60% 4. Price closes above upper Bollinger Band Broadcasts alert via WebSocket. Gated by user on/off toggle. """ import collections import json import logging import os from datetime import datetime, timezone from typing import Deque, Optional import httpx from app.config import settings logger = logging.getLogger(__name__) # ── Config ──────────────────────────────────────────────────────────────────── WATCH_SYMBOLS = ["ETHUSDT", "LINKUSDT"] CONTEXT_SYMBOL = "BTCUSDT" ALL_SYMBOLS = [CONTEXT_SYMBOL] + WATCH_SYMBOLS CANDLE_LIMIT = 200 # candles to fetch per poll (200 x 5m = ~16.7h history) BB_PERIOD = 20 BB_SQUEEZE_PCT = 20 # bottom 20% of BB width history → squeeze VOLUME_MULT = 2.5 TBR_THRESH = 0.60 # taker buy ratio threshold BTC_TREND_MA = 288 # 24h trend (needs separate longer fetch for BTC) # ── State ───────────────────────────────────────────────────────────────────── _enabled: bool = False _recent_signals: Deque[dict] = collections.deque(maxlen=50) _last_fired: dict[str, Optional[datetime]] = {s: None for s in WATCH_SYMBOLS} # B52: persist _enabled across process restarts without a DB migration. # The file survives `systemctl restart` but is cleared by OS reboots (which is # acceptable — an operator who reboots the server is expected to re-arm the # monitor). Path is configurable via env so staging vs prod can differ. _STATE_FILE = os.environ.get( "BREAKOUT_MONITOR_STATE_FILE", "/tmp/trumpsignal-breakout-state.json", ) def _load_persisted_state() -> None: """Read _enabled from disk on startup. Called once at module import time.""" global _enabled try: with open(_STATE_FILE) as f: data = json.load(f) _enabled = bool(data.get("enabled", False)) logger.info("Breakout monitor: loaded persisted state enabled=%s", _enabled) except FileNotFoundError: pass # first run — start disabled except Exception as exc: logger.warning("Breakout monitor: failed to load state file: %s", exc) def _persist_state() -> None: """Write _enabled to disk so it survives process restarts.""" try: with open(_STATE_FILE, "w") as f: json.dump({"enabled": _enabled, "updated_at": datetime.now(timezone.utc).isoformat()}, f) except Exception as exc: logger.warning("Breakout monitor: failed to persist state: %s", exc) # Load persisted state at import time (called when main.py registers the scheduler). _load_persisted_state() # ── Public API ──────────────────────────────────────────────────────────────── def set_enabled(value: bool) -> None: global _enabled _enabled = value _persist_state() # B52: survive restarts logger.info("Funding signal monitor: %s", "ENABLED" if value else "DISABLED") def is_enabled() -> bool: return _enabled def get_recent_signals(limit: int = 20) -> list[dict]: return list(reversed(list(_recent_signals)))[:limit] def get_status() -> dict: return { "enabled": _enabled, "symbols": WATCH_SYMBOLS, "recent_signal_count": len(_recent_signals), } # ── Binance fetch ───────────────────────────────────────────────────────────── async def _fetch_klines(client: httpx.AsyncClient, symbol: str, limit: int) -> list[list]: url = f"{settings.binance_rest_url}/api/v3/klines" try: resp = await client.get(url, params={ "symbol": symbol, "interval": "5m", "limit": limit }, timeout=15) resp.raise_for_status() return resp.json() except Exception as exc: logger.warning("Failed to fetch klines for %s: %s", symbol, exc) return [] def _parse_candle(row: list) -> dict: total_vol = float(row[5]) taker_buy = float(row[9]) return { "time": row[0], "open": float(row[1]), "high": float(row[2]), "low": float(row[3]), "close": float(row[4]), "volume": total_vol, "tbr": taker_buy / total_vol if total_vol > 0 else 0.5, } # ── Indicators ──────────────────────────────────────────────────────────────── def _compute_signal(candles: list[dict]) -> Optional[dict]: """ Returns signal dict if breakout signal fires on the most recent candle, else None. Requires at least 60 candles. """ # Drop the last (current incomplete) candle — always use completed candles candles = candles[:-1] if len(candles) < 60: return None closes = [c["close"] for c in candles] volumes = [c["volume"] for c in candles] tbrs = [c["tbr"] for c in candles] # ── Bollinger Bands (last BB_PERIOD candles) ────────────────────────────── window = closes[-BB_PERIOD:] bb_mid = sum(window) / BB_PERIOD variance = sum((x - bb_mid) ** 2 for x in window) / BB_PERIOD bb_std = variance ** 0.5 bb_upper = bb_mid + 2 * bb_std bb_width = (4 * bb_std) / bb_mid if bb_mid > 0 else 0 # BB width percentile over last 60 candles widths = [] for i in range(len(candles) - 60, len(candles)): w_window = closes[max(0, i - BB_PERIOD + 1): i + 1] if len(w_window) < BB_PERIOD: continue m = sum(w_window) / len(w_window) s = (sum((x - m) ** 2 for x in w_window) / len(w_window)) ** 0.5 widths.append((4 * s) / m if m > 0 else 0) if not widths: return None rank = sum(1 for w in widths if w < bb_width) / len(widths) * 100 in_squeeze = rank < BB_SQUEEZE_PCT # ── Volume ──────────────────────────────────────────────────────────────── vol_ma = sum(volumes[-BB_PERIOD - 1:-1]) / BB_PERIOD vol_spike = volumes[-1] > VOLUME_MULT * vol_ma if vol_ma > 0 else False # ── Taker buy ratio ─────────────────────────────────────────────────────── tbr_ok = tbrs[-1] > TBR_THRESH # ── Price above upper BB ────────────────────────────────────────────────── above_bb = closes[-1] > bb_upper if in_squeeze and vol_spike and tbr_ok and above_bb: return { "bb_width_pct": round(rank, 1), "vol_mult": round(volumes[-1] / vol_ma, 2) if vol_ma > 0 else 0, "tbr": round(tbrs[-1], 3), "close": closes[-1], "bb_upper": round(bb_upper, 4), } return None def _btc_trend(candles: list[dict]) -> Optional[bool]: """True = BTC above 24h MA, False = below, None = not enough data.""" if len(candles) < BTC_TREND_MA: return None closes = [c["close"] for c in candles] ma = sum(closes[-BTC_TREND_MA:]) / BTC_TREND_MA return closes[-1] > ma # ── Main poll loop ──────────────────────────────────────────────────────────── async def poll_funding_signal() -> None: """Called by APScheduler every 5 minutes.""" from app.ws.manager import manager # avoid circular import at module load async with httpx.AsyncClient(timeout=20) as client: # Fetch BTC with longer history for 24h trend btc_rows = await _fetch_klines(client, CONTEXT_SYMBOL, BTC_TREND_MA + 10) btc_candles = [_parse_candle(r) for r in btc_rows] btc_up = _btc_trend(btc_candles) # Fetch ETH + LINK for symbol in WATCH_SYMBOLS: rows = await _fetch_klines(client, symbol, CANDLE_LIMIT) if not rows: continue candles = [_parse_candle(r) for r in rows] sig = _compute_signal(candles) if sig is None: continue # Deduplicate: don't fire same symbol twice within 30 minutes now = datetime.now(timezone.utc) last = _last_fired.get(symbol) if last and (now - last).total_seconds() < 1800: logger.info("Signal for %s deduplicated (too soon)", symbol) continue _last_fired[symbol] = now alert = { "type": "funding_signal", "symbol": symbol, "time": now.isoformat(), "close": sig["close"], "tbr": sig["tbr"], "vol_mult": sig["vol_mult"], "bb_pct": sig["bb_width_pct"], "bb_upper": sig["bb_upper"], "btc_trend": "↑ uptrend" if btc_up else ("↓ downtrend" if btc_up is False else "unknown"), "enabled": _enabled, } _recent_signals.append(alert) if _enabled: logger.info("🚨 SIGNAL %s | price=%.4f tbr=%.2f vol=%.1fx btc=%s", symbol, sig["close"], sig["tbr"], sig["vol_mult"], alert["btc_trend"]) await manager.broadcast(alert) else: logger.info("Signal %s (monitor OFF — not broadcast)", symbol)