""" VCP-style breakout scanner — example of an external signal module. What this scans for (Minervini-light, adapted for crypto perps): 1. Volatility contraction: 7-day ATR ≤ 50% of 30-day ATR (the price has been getting tighter — supply is drying up) 2. Breakout confirmation: current 4h close > max(prior 30-day highs) (a real break of the range, not just touching the top) 3. Volume confirmation: last 24h volume ≥ 1.5× 30-day average (institutional participation, not retail noise) 4. Cooldown: no signal for the same asset in the past 48h (avoid stacking) When all 4 conditions hit, the scanner POSTs to /api/signals/ingest. From there the trade goes through the same convex-strategy pipeline as Trump signals — regime filter, sizing, trailing stop, all of it. Why this module exists: Trump signals have 1-2 true catalysts per month. Adding a TA-based source that finds setups continuously means the bot has SOMETHING to act on most weeks, instead of waiting for geopolitical fireworks. The user's hypothesis is that consolidation breakouts have asymmetric upside; the existing trailing-stop logic is purpose-built to capture that. Tunables (don't hardcode in caller — change them HERE if you want to test): ATR_RECENT_DAYS, ATR_BASELINE_DAYS, ATR_RATIO_MAX, BREAKOUT_LOOKBACK_DAYS, VOLUME_MULT_MIN, COOLDOWN_HOURS. Asset list: Default = SOL only. Extend ASSETS_TO_SCAN with more once you've watched this run for a week and confirmed it doesn't fire-hose noise. """ from __future__ import annotations import logging import statistics from datetime import datetime, timedelta, timezone from typing import Optional import httpx from app.config import settings from app.services.market_data import for_asset, drop_in_progress_bar from app.services import scanner_state logger = logging.getLogger(__name__) SCANNER_NAME = "vcp_breakout" scanner_state.register(SCANNER_NAME) # ─── Tunables ─────────────────────────────────────────────────────────────── ATR_RECENT_DAYS = 7 ATR_BASELINE_DAYS = 30 ATR_RATIO_MAX = 0.5 # recent ATR must be ≤ 50% of baseline BREAKOUT_LOOKBACK_DAYS = 30 # break of this many days' high VOLUME_MULT_MIN = 1.5 # current 24h vol vs 30-day avg COOLDOWN_HOURS = 48 # don't re-signal same asset within this window # Assets to scan. Provider selection (Binance vs HL) is automatic via # market_data.for_asset() — HL-native perps like TRUMP/HYPE route to HL # automatically. Add to HL_NATIVE_ASSETS in market_data.py if needed. ASSETS_TO_SCAN = [ "SOL", # "ETH", "LINK", "AVAX", # Binance has them # "TRUMP", "HYPE", # HL-native (routed automatically) ] # Cooldown via scanner_state.in_cooldown — DB-backed, restart-safe. # Data fetching is delegated to market_data.for_asset() — see that module # for Binance vs Hyperliquid routing. # ─── Signal logic ─────────────────────────────────────────────────────────── def _atr_proxy(candles: list[dict]) -> Optional[float]: """Mean of (high-low)/close — simple ATR proxy, dimensionless.""" ranges = [(c["high"] - c["low"]) / c["close"] for c in candles if c["close"] > 0] return statistics.fmean(ranges) if ranges else None def evaluate_breakout(candles: list[dict]) -> tuple[bool, dict]: """Returns (is_signal, debug_info). All 4 gates must pass. Pure function — no side effects, fully testable. """ if len(candles) < ATR_BASELINE_DAYS * 6: return False, {"reason": "insufficient_data", "bars": len(candles)} # 1. Volatility contraction recent_bars = candles[-ATR_RECENT_DAYS * 6:] baseline_bars = candles[-ATR_BASELINE_DAYS * 6:] atr_recent = _atr_proxy(recent_bars) atr_baseline = _atr_proxy(baseline_bars) if not atr_recent or not atr_baseline: return False, {"reason": "atr_calc_failed"} atr_ratio = atr_recent / atr_baseline if atr_ratio > ATR_RATIO_MAX: return False, {"reason": "no_contraction", "atr_ratio": round(atr_ratio, 3)} # 2. Range break — UP through prior high (long) or DOWN through prior # low (short). Lookback excludes the current bar. lookback = candles[-BREAKOUT_LOOKBACK_DAYS * 6 : -1] if not lookback: return False, {"reason": "no_lookback"} prior_high = max(c["high"] for c in lookback) prior_low = min(c["low"] for c in lookback) current_close = candles[-1]["close"] if current_close > prior_high: direction, ref, gap = "buy", prior_high, (current_close - prior_high) / prior_high * 100 elif current_close < prior_low: direction, ref, gap = "short", prior_low, (prior_low - current_close) / prior_low * 100 else: return False, {"reason": "no_range_break", "close": current_close, "prior_high": prior_high, "prior_low": prior_low} # 3. Volume confirmation (same for both directions — a real break needs # participation regardless of which way it goes) recent_vol_24h = sum(c["volume"] for c in candles[-6:]) baseline_avg_24h = sum(c["volume"] for c in candles[-180:]) / 30 if baseline_avg_24h <= 0: return False, {"reason": "no_volume_data"} vol_ratio = recent_vol_24h / baseline_avg_24h if vol_ratio < VOLUME_MULT_MIN: return False, {"reason": "weak_volume", "vol_ratio": round(vol_ratio, 2)} return True, { "direction": direction, "atr_ratio": round(atr_ratio, 3), "vol_ratio": round(vol_ratio, 2), "break_ref": round(ref, 4), "break_at": round(current_close, 4), "headroom_pct": round(gap, 2), } def _confidence_from(debug: dict) -> int: """Map the signal's quality into a 0-100 confidence used by sizing. The tighter the contraction and the bigger the volume spike, the higher the confidence. Capped at 95 — leave headroom so AI signals at 100 stay distinguishable. """ score = 70 # baseline for "all 4 gates passed" # Tighter contraction → higher confidence if debug.get("atr_ratio", 1) < 0.35: score += 10 elif debug.get("atr_ratio", 1) < 0.45: score += 5 # Bigger volume spike → higher confidence if debug.get("vol_ratio", 0) >= 3.0: score += 10 elif debug.get("vol_ratio", 0) >= 2.0: score += 5 return min(score, 95) # ─── Ingest call ──────────────────────────────────────────────────────────── async def _emit_signal(asset: str, debug: dict) -> None: """POST the signal to /api/signals/ingest. Treats the local server as the target — same machine, no network hop in dev. """ if not settings.ingest_api_key: logger.warning("VCP: signal would fire on %s but INGEST_API_KEY is empty", asset) return confidence = _confidence_from(debug) direction = debug["direction"] way = "breakout ↑" if direction == "buy" else "breakdown ↓" rel = "above" if direction == "buy" else "below" payload = { "source": "breakout", "external_id": f"vcp-{asset}-{direction}-{datetime.now(timezone.utc).strftime('%Y%m%d-%H%M')}", "text": ( f"VCP {way} on {asset}: ATR ratio {debug['atr_ratio']} " f"(7d/30d), {debug['vol_ratio']}× normal volume, " f"close ${debug['break_at']} {rel} {BREAKOUT_LOOKBACK_DAYS}-day " f"level ${debug['break_ref']} ({debug['headroom_pct']}%)" ), "signal": direction, "target_asset": asset, "confidence": confidence, "category": "vcp_breakout", "expected_move_pct": 5.0, # historical VCP median expansion } 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("VCP ingest failed (%d): %s", resp.status_code, resp.text[:200]) else: logger.info("VCP signal emitted for %s, confidence=%d: %s", asset, confidence, resp.json()) # ─── Scheduler entry point ────────────────────────────────────────────────── async def scan_once() -> None: """Single scan pass over ASSETS_TO_SCAN. Called every 30 minutes. Kill-switch + DB-backed cooldown. Drops in-progress 4h bar before evaluating — without that, the breakout test ran against a partial bar and could flicker FIRE / no-FIRE within a 4h window. """ if not scanner_state.is_enabled(SCANNER_NAME): logger.debug("VCP scanner disabled — skipping run") return fired_any = False error_msg: Optional[str] = None for asset in ASSETS_TO_SCAN: # Cooldown — DB-backed, restart-safe. cooldown_days = COOLDOWN_HOURS / 24 if await scanner_state.in_cooldown("breakout", asset, cooldown_days): continue provider = for_asset(asset) try: candles = await provider.fetch_4h(asset, days=ATR_BASELINE_DAYS + 1) except Exception as exc: error_msg = f"{asset}: {exc}" logger.error("VCP fetch failed for %s via %s: %s", asset, provider.name, exc) continue candles = drop_in_progress_bar(candles, "4h") if not candles: logger.warning("VCP: %s returned 0 candles from %s — symbol may not exist", asset, provider.name) continue is_signal, debug = evaluate_breakout(candles) logger.info("VCP scan %s [%s]: %s — %s", asset, provider.name, "FIRE" if is_signal else "no", debug) if is_signal: await _emit_signal(asset, debug) fired_any = True 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")