""" signal_aggregator.py — Module de signaux macro/on-chain (observation mode) Processus séparé : lit les APIs, écrit un cache JSON toutes les 60s. Démarrage : nohup python3 signal_aggregator.py > logs/signal_aggregator.log 2>&1 & Adaptations vs version originale : - Symboles USDC → USDT pour tous les appels fapi (bot trade USDC, perps sont USDT) - NETFLOW_INFLOW_HIGH_USD = 500_000 (calibré mid-caps, pas large-caps) - unlock_proximity = 0 toujours (TokenUnlocks n'a pas d'API publique) - Watchlist dynamique : lit market_spy_watchlist.json si disponible """ import os import json import time import logging import tempfile import requests from datetime import datetime, timezone # ── Config ────────────────────────────────────────────────────────────────── SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) CACHE_FILE = os.path.join(SCRIPT_DIR, "signal_cache.json") WATCHLIST_FILE = os.path.join(SCRIPT_DIR, "market_spy_watchlist.json") UPDATE_INTERVAL = 60 # secondes entre deux mises à jour # Seuils calibrés mid-caps (pas BTC/ETH) NETFLOW_INFLOW_HIGH_USD = 500_000 # 500K$ sur 1h = pression acheteuse forte pour mid-cap NETFLOW_INFLOW_MED_USD = 100_000 # 100K$ = pression modérée NETFLOW_OUTFLOW_WARN_USD = -200_000 # -200K$ = sortie notable # Funding rate seuils FUNDING_BULL_THRESHOLD = 0.01 # 0.01% / 8h → longs surchargés (bearish pour nous) FUNDING_BEAR_THRESHOLD = -0.005 # -0.005% / 8h → shorts surchargés (bullish) # Fallback : watchlist statique si le fichier dynamique n'existe pas STATIC_WATCHLIST = [ "BIOUSDC", "WLFIUSDC", "PENDLEUSDC", "BANANAS31USDC", "XVGUSDC", "DASHUSDC", "TSTUSDC", "PARTIUSDC", "ORCAUSDC", "APEUSD", "MANTRAUSDC", "ORDIUSDC", "MEGAUSDC", "GIGGLEUSDC", ] BINANCE_FAPI = "https://fapi.binance.com" BINANCE_SPOT = "https://api.binance.com" logging.basicConfig( level=logging.INFO, format="%(asctime)s [SIGNAL] %(levelname)s %(message)s", datefmt="%H:%M:%S", ) log = logging.getLogger("signal_aggregator") # ── Helpers ────────────────────────────────────────────────────────────────── def usdt_symbol(sym: str) -> str: """Convertit BIOUSDC → BIOUSDT pour les appels fapi (perps USDT).""" if sym.endswith("USDC"): return sym[:-4] + "USDT" if sym.endswith("USDT"): return sym return sym + "USDT" def _get(url: str, params: dict = None, timeout: int = 8) -> dict | list | None: """GET avec timeout, retourne None si échec.""" try: r = requests.get(url, params=params, timeout=timeout) r.raise_for_status() return r.json() except Exception as e: log.debug(f"GET failed {url}: {e}") return None def load_watchlist() -> list[str]: """Charge la watchlist depuis le fichier JSON du bot, fallback statique.""" try: if os.path.exists(WATCHLIST_FILE): with open(WATCHLIST_FILE) as f: data = json.load(f) if isinstance(data, list) and len(data) > 0: return data except Exception: pass return STATIC_WATCHLIST # ── Signaux ────────────────────────────────────────────────────────────────── def compute_funding_oi_score(symbols: list[str]) -> dict: """ Score 0-10 basé sur funding rate + open interest agrégés. > 7 : momentum haussier fort < 3 : funding négatif / OI faible = pression baissière """ total_funding = 0.0 total_oi_usd = 0.0 valid_count = 0 for sym in symbols: fsym = usdt_symbol(sym) # Funding rate data = _get(f"{BINANCE_FAPI}/fapi/v1/premiumIndex", {"symbol": fsym}) if data and "lastFundingRate" in data: try: total_funding += float(data["lastFundingRate"]) * 100 # → % valid_count += 1 except (ValueError, TypeError): pass # Open Interest en USD price_data = _get(f"{BINANCE_FAPI}/fapi/v1/ticker/price", {"symbol": fsym}) oi_data = _get(f"{BINANCE_FAPI}/fapi/v1/openInterest", {"symbol": fsym}) if price_data and oi_data: try: price = float(price_data["price"]) oi = float(oi_data["openInterest"]) total_oi_usd += price * oi except (ValueError, TypeError): pass if valid_count == 0: return {"score": 5.0, "avg_funding_pct": 0.0, "total_oi_usd": 0.0, "valid": 0} avg_funding = total_funding / valid_count avg_oi_usd = total_oi_usd / max(valid_count, 1) # Score funding : centré sur 0%, symétrique if avg_funding >= FUNDING_BULL_THRESHOLD: funding_score = 2.0 # Longs surchargés → bearish (attention) elif avg_funding <= FUNDING_BEAR_THRESHOLD: funding_score = 8.0 # Shorts surchargés → bullish momentum else: # Interpolation linéaire entre -0.005% et +0.01% t = (avg_funding - FUNDING_BEAR_THRESHOLD) / (FUNDING_BULL_THRESHOLD - FUNDING_BEAR_THRESHOLD) funding_score = 8.0 - 6.0 * t # Score OI : purement indicatif (pas de veto) if avg_oi_usd > 50_000_000: oi_score = 8.0 elif avg_oi_usd > 10_000_000: oi_score = 6.0 elif avg_oi_usd > 1_000_000: oi_score = 4.0 else: oi_score = 2.0 combined = 0.65 * funding_score + 0.35 * oi_score return { "score": round(combined, 2), "avg_funding_pct": round(avg_funding, 5), "total_oi_usd": round(total_oi_usd, 0), "valid": valid_count, } def compute_netflow_pressure(symbols: list[str]) -> dict: """ Score 0-10 basé sur taker buy/sell ratio (proxy netflow). Seuil adapté mid-caps : 500K$ / 100K$ au lieu de 5M$ / 1M$. """ total_buy_vol = 0.0 total_sell_vol = 0.0 valid_count = 0 for sym in symbols: fsym = usdt_symbol(sym) # Taker long/short ratio (15min) data = _get( f"{BINANCE_FAPI}/futures/data/takerlongshortRatio", {"symbol": fsym, "period": "15m", "limit": 4} ) if not data or not isinstance(data, list): continue # Prix pour convertir en USD price_data = _get(f"{BINANCE_FAPI}/fapi/v1/ticker/price", {"symbol": fsym}) price = float(price_data["price"]) if price_data else 1.0 try: for bar in data: buy_vol = float(bar.get("buyVol", 0)) sell_vol = float(bar.get("sellVol", 0)) total_buy_vol += buy_vol * price total_sell_vol += sell_vol * price valid_count += 1 except (ValueError, TypeError, KeyError): pass if valid_count == 0 or (total_buy_vol + total_sell_vol) == 0: return {"score": 5.0, "net_flow_usd": 0.0, "buy_ratio": 0.5, "valid": 0} net_flow = total_buy_vol - total_sell_vol buy_ratio = total_buy_vol / (total_buy_vol + total_sell_vol) # Score basé sur netflow absolu (seuils mid-caps) if net_flow >= NETFLOW_INFLOW_HIGH_USD: score = 9.0 elif net_flow >= NETFLOW_INFLOW_MED_USD: score = 7.0 elif net_flow >= 0: score = 5.5 elif net_flow >= NETFLOW_OUTFLOW_WARN_USD: score = 3.5 else: score = 1.5 # Ajustement buy_ratio if buy_ratio >= 0.60: score = min(10, score + 1.0) elif buy_ratio <= 0.40: score = max(0, score - 1.0) return { "score": round(score, 2), "net_flow_usd": round(net_flow, 0), "buy_ratio": round(buy_ratio, 4), "valid": valid_count, } def compute_unlock_proximity(_symbols: list[str]) -> dict: """ DÉSACTIVÉ — TokenUnlocks n'a pas d'API publique. Retourne toujours 0 (signal neutre, pas de pénalité). """ return {"score": 0.0, "note": "disabled - no public API", "valid": 0} def _funding_to_score(funding_pct: float) -> float: """Convertit un funding (%) en score 0-10 (plus haut = plus favorable long squeeze).""" if funding_pct >= FUNDING_BULL_THRESHOLD: return 2.0 if funding_pct <= FUNDING_BEAR_THRESHOLD: return 8.0 t = (funding_pct - FUNDING_BEAR_THRESHOLD) / (FUNDING_BULL_THRESHOLD - FUNDING_BEAR_THRESHOLD) return 8.0 - 6.0 * t def _netflow_to_score(net_flow_usd: float, buy_ratio: float) -> float: """Convertit netflow + buy_ratio en score 0-10.""" if net_flow_usd >= NETFLOW_INFLOW_HIGH_USD: score = 9.0 elif net_flow_usd >= NETFLOW_INFLOW_MED_USD: score = 7.0 elif net_flow_usd >= 0: score = 5.5 elif net_flow_usd >= NETFLOW_OUTFLOW_WARN_USD: score = 3.5 else: score = 1.5 if buy_ratio >= 0.60: score = min(10, score + 1.0) elif buy_ratio <= 0.40: score = max(0, score - 1.0) return score def compute_symbol_signal_map(symbols: list[str]) -> tuple[dict, dict]: """ Construit les signaux par symbole pour identifier des setups 'explosifs' minutes. Retourne (symbol_map, explosive_summary). """ per_symbol = {} for sym in symbols: fsym = usdt_symbol(sym) # funding funding_pct = None data = _get(f"{BINANCE_FAPI}/fapi/v1/premiumIndex", {"symbol": fsym}) if data and "lastFundingRate" in data: try: funding_pct = float(data["lastFundingRate"]) * 100 except (ValueError, TypeError): funding_pct = None # netflow proxy (1h) net_flow_usd = 0.0 buy_ratio = 0.5 taker_ok = False taker = _get( f"{BINANCE_FAPI}/futures/data/takerlongshortRatio", {"symbol": fsym, "period": "15m", "limit": 4} ) price_data = _get(f"{BINANCE_FAPI}/fapi/v1/ticker/price", {"symbol": fsym}) if taker and isinstance(taker, list) and price_data: try: price = float(price_data["price"]) buy_sum = 0.0 sell_sum = 0.0 for bar in taker: buy_sum += float(bar.get("buyVol", 0)) * price sell_sum += float(bar.get("sellVol", 0)) * price total = buy_sum + sell_sum if total > 0: net_flow_usd = buy_sum - sell_sum buy_ratio = buy_sum / total taker_ok = True except (ValueError, TypeError, KeyError): pass # oi usd oi_usd = 0.0 oi_data = _get(f"{BINANCE_FAPI}/fapi/v1/openInterest", {"symbol": fsym}) if oi_data and price_data: try: oi_usd = float(oi_data["openInterest"]) * float(price_data["price"]) except (ValueError, TypeError, KeyError): oi_usd = 0.0 if funding_pct is None and not taker_ok: continue f_score = _funding_to_score(funding_pct if funding_pct is not None else 0.0) n_score = _netflow_to_score(net_flow_usd, buy_ratio) # Score explosif 0-10: priorité au déclencheur court terme (netflow), # puis surcharge contrarienne via funding, puis liquidité via OI. oi_score = 2.0 if oi_usd > 50_000_000: oi_score = 8.0 elif oi_usd > 10_000_000: oi_score = 6.0 elif oi_usd > 1_000_000: oi_score = 4.0 explosive_score = (0.50 * n_score) + (0.35 * f_score) + (0.15 * oi_score) # Qualitatif if explosive_score >= 7.5: setup = "SETUP_FORT" elif explosive_score >= 6.0: setup = "SETUP_MOYEN" elif explosive_score >= 4.5: setup = "MIXTE" else: setup = "FAIBLE" # Risque de bull trap trap_risk = "LOW" if (funding_pct is not None and funding_pct >= FUNDING_BULL_THRESHOLD) and net_flow_usd < 0: trap_risk = "HIGH" elif funding_pct is not None and funding_pct >= 0.006: trap_risk = "MEDIUM" per_symbol[sym] = { "symbol": sym, "fapi_symbol": fsym, "funding_pct": round(funding_pct if funding_pct is not None else 0.0, 5), "net_flow_usd": round(net_flow_usd, 0), "buy_ratio": round(buy_ratio, 4), "oi_usd": round(oi_usd, 0), "funding_score": round(f_score, 2), "netflow_score": round(n_score, 2), "explosive_score": round(explosive_score, 2), "setup": setup, "trap_risk": trap_risk, } ranked = sorted(per_symbol.values(), key=lambda x: x["explosive_score"], reverse=True) top_candidates = ranked[:5] avg_explosive = 0.0 if ranked: avg_explosive = sum(x["explosive_score"] for x in ranked) / len(ranked) summary = { "avg_explosive_score": round(avg_explosive, 2), "symbols_ranked": len(ranked), "top_candidates": top_candidates, "bull_trap_alerts": [x["symbol"] for x in ranked if x["trap_risk"] == "HIGH"], } return per_symbol, summary # ── Agrégation ─────────────────────────────────────────────────────────────── def compute_all_signals(symbols: list[str]) -> dict: """Calcule les 3 composantes et le score agrégé (0-20, unlock désactivé).""" funding_oi = compute_funding_oi_score(symbols) netflow = compute_netflow_pressure(symbols) unlock = compute_unlock_proximity(symbols) symbol_map, explosive = compute_symbol_signal_map(symbols) # Score agrégé sur 20 (funding_oi/10 + netflow/10, unlock ignoré) raw_score = funding_oi["score"] + netflow["score"] # 0-20 # Interprétation if raw_score >= 15: interpretation = "BULLISH_STRONG" elif raw_score >= 12: interpretation = "BULLISH" elif raw_score >= 9: interpretation = "NEUTRAL" elif raw_score >= 6: interpretation = "BEARISH" else: interpretation = "BEARISH_STRONG" # Veto actif si score très bas (< 5/20) — DÉSACTIVÉ en observation mode veto_active = False # ← Mettre True pour activer après validation veto_reason = "" if raw_score < 5: veto_active = False # Toujours False en obs mode, mais on le trace veto_reason = f"Score {raw_score:.1f}/20 < 5 — serait veto en prod" # Modulateur de taille 0.5-1.0 — IGNORÉ en observation mode if raw_score >= 14: size_mod = 1.0 elif raw_score >= 10: size_mod = 0.85 elif raw_score >= 7: size_mod = 0.70 else: size_mod = 0.50 return { "timestamp": datetime.now(timezone.utc).isoformat(), "symbols_count": len(symbols), "funding_oi": funding_oi, "netflow": netflow, "unlock": unlock, "score_total": round(raw_score, 2), "score_max": 20, "interpretation": interpretation, "veto_active": veto_active, "veto_reason": veto_reason, "size_modulator": round(size_mod, 2), "obs_mode": True, # Toujours True tant que non validé "symbol_signals": symbol_map, "explosive": explosive, } def write_cache(payload: dict) -> None: """Écriture atomique du cache JSON (tempfile + rename).""" tmp = CACHE_FILE + ".tmp" try: with open(tmp, "w") as f: json.dump(payload, f, indent=2) os.replace(tmp, CACHE_FILE) log.info( f"Cache écrit — score={payload['score_total']:.1f}/20 " f"({payload['interpretation']}) | " f"funding={payload['funding_oi']['avg_funding_pct']:+.5f}% | " f"netflow={payload['netflow']['net_flow_usd']:+.0f}$ " f"(ratio={payload['netflow']['buy_ratio']:.3f})" ) except Exception as e: log.error(f"Échec écriture cache: {e}") try: os.remove(tmp) except OSError: pass # ── Main loop ──────────────────────────────────────────────────────────────── def main(): log.info("=== signal_aggregator démarré (observation mode) ===") log.info(f"Cache → {CACHE_FILE}") log.info(f"Mise à jour toutes les {UPDATE_INTERVAL}s") log.info("VETO désactivé, size_modulator ignoré — mode observation uniquement") consecutive_errors = 0 while True: try: symbols = load_watchlist() log.debug(f"Watchlist: {len(symbols)} symboles") payload = compute_all_signals(symbols) write_cache(payload) consecutive_errors = 0 except KeyboardInterrupt: log.info("Arrêt demandé") break except Exception as e: consecutive_errors += 1 log.error(f"Erreur cycle #{consecutive_errors}: {e}") if consecutive_errors >= 5: log.error("5 erreurs consécutives — pause 5min") time.sleep(300) consecutive_errors = 0 time.sleep(UPDATE_INTERVAL) if __name__ == "__main__": main()