k
54884f3e24
Feed-health pass over KOL_FEEDS: - raoulpal: stale Substack (last 2024-05) → Real Vision podcast feed - dampedspring: paywalled (0 entries) → free "Damped Spring 101" Substack - unchained: Cloudflare 403 → canonical Megaphone podcast feed - lynalden: Cloudflare 202 → FeedBurner mirror - glassnode: recovered via httpx http2=True (was 403 on HTTP/1.1) - browser User-Agent + Accept headers on feed fetch - removed dead feeds with no active replacement: placeholder, dragonfly, niccarter, eugene - pin h2==4.3.0 (required by http2=True) All 25 remaining feeds verified fetching real body content; newest post per feed ≤88d. Bundles in-flight KOL-module work already in the working tree (kol_x ingest, migration 027, tests). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
261 lines
10 KiB
Python
261 lines
10 KiB
Python
"""
|
|
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)
|