""" Couche 7 — Détection de contexte de marché (per-token) ====================================================== Identifie le régime structurel d'un token donné parmi 6 régimes : - PHOENIX : rebond depuis ATL récent → neutralise partiellement RSI surachat - BULL_RUN : tendance haussière saine → grille standard - PARABOLIC_EXTENSION : tendance verticale, pas encore de capitulation → prudence accrue, pas de blocage - PUMP_TERMINAL : pic terminal avec capitulation visible → blocage long + suggestion profit-taking - CONSOLIDATION : prix stable, faible volatilité, pas baissier → grille standard, watch breakout - BEAR_RANGE : tendance baissière nette → blocage long NOTE : couche per-token, complémentaire de `market_regime.py` (macro BTC+alts). Cf. PREDICTIVE_SYSTEM_V2.md — section "COUCHE 7". Calibration v2 (post-test 12/05/2026) : - PUMP_TERMINAL : trend_7d 100% → 40% (seuil initial inatteignable) - PHOENIX : scoring par points (≥3/4 conditions) au lieu de ET strict - Nouveaux régimes intermédiaires : PARABOLIC_EXTENSION, CONSOLIDATION """ from __future__ import annotations import json import logging import os import threading import time from dataclasses import dataclass, field, asdict from typing import Dict, List, Optional, Tuple import numpy as np import requests logger = logging.getLogger('MarketContext') BINANCE_API_URL = "https://api.binance.com/api/v3" # ────────────────────────────────────────────────────────────────────────────── # Config flags — contrôlables depuis market_spy.py / config.py # ────────────────────────────────────────────────────────────────────────────── # Activer la couche 7 (False = aucun effet, le module peut être importé sans risque) COUCHE7_ENABLED = False # Shadow mode : log la décision mais ne bloque rien même si COUCHE7_ENABLED=True COUCHE7_SHADOW_MODE = True # Log JSONL des décisions (pour backtest a posteriori) COUCHE7_LOG_ENABLED = True _SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) COUCHE7_LOG_FILE = os.path.join(_SCRIPT_DIR, 'data', 'couche7_decisions.log') # ────────────────────────────────────────────────────────────────────────────── # Data structures # ────────────────────────────────────────────────────────────────────────────── @dataclass class ContextMetrics: """Sous-indicateurs utilisés pour la classification.""" current_price: float atl_60d: float ath_180d: float distance_from_atl_60d_pct: float # (price - atl) / atl distance_from_ath_180d_pct: float # price / ath days_since_atl_60d: int days_since_ath_180d: int trend_7d_pct: float trend_30d_pct: float volume_ratio_7d_30d: float capitulation_candles_3d: int # mèches hautes longues récentes realized_volatility_30d: float # std des log-returns daily @dataclass class ContextResult: """Résultat de la détection pour un token.""" regime: str confidence: float # 0..1 metrics: ContextMetrics implications: Dict notes: List[str] = field(default_factory=list) def to_dict(self) -> Dict: return asdict(self) # ────────────────────────────────────────────────────────────────────────────── # Thresholds (calibration v2) # ────────────────────────────────────────────────────────────────────────────── PHOENIX_THRESHOLDS = { 'dist_atl_min': 0.30, 'dist_atl_max': 3.00, 'trend_30d_min': 0.30, 'volume_ratio_min': 1.30, 'days_since_atl_max': 45, 'min_conditions_met': 3, } PUMP_TERMINAL_THRESHOLDS = { 'dist_ath_min': 0.75, 'trend_7d_min': 0.40, 'trend_30d_min': 0.60, 'capitulation_candles_min': 1, 'min_conditions_met': 3, } PARABOLIC_EXTENSION_THRESHOLDS = { 'trend_30d_min': 0.60, 'trend_7d_min': 0.20, } BULL_RUN_THRESHOLDS = { 'trend_30d_min': 0.15, 'trend_30d_max': 0.80, } CONSOLIDATION_THRESHOLDS = { 'trend_30d_abs_max': 0.08, # |trend| ≤ 8% pour être vraiment neutre 'volatility_max': 0.06, } IMPLICATIONS = { 'PHOENIX': { 'rsi_overbought_threshold': 85, 'stage_filter': 'neutralized', 'max_position_size_multiplier': 1.5, 'target_upside_pct': 100, 'use_unlock_long_term': False, 'block_long_entries': False, }, 'BULL_RUN': { 'rsi_overbought_threshold': 75, 'stage_filter': 'standard', 'max_position_size_multiplier': 1.0, 'target_upside_pct': 30, 'use_unlock_long_term': True, 'block_long_entries': False, }, 'PARABOLIC_EXTENSION': { 'rsi_overbought_threshold': 70, 'stage_filter': 'cautious', 'max_position_size_multiplier': 0.6, 'target_upside_pct': 25, 'use_unlock_long_term': True, 'block_long_entries': False, 'suggest_tight_stop': True, }, 'PUMP_TERMINAL': { 'rsi_overbought_threshold': 60, 'stage_filter': 'amplified', 'block_long_entries': True, 'suggest_profit_taking': True, 'use_unlock_long_term': True, 'max_position_size_multiplier': 0.0, }, 'CONSOLIDATION': { 'rsi_overbought_threshold': 75, 'stage_filter': 'standard', 'max_position_size_multiplier': 0.8, 'target_upside_pct': 20, 'use_unlock_long_term': True, 'block_long_entries': False, 'watch_for_breakout': True, }, 'BEAR_RANGE': { 'block_long_entries': True, 'watch_for_phoenix': True, 'rsi_overbought_threshold': 65, 'use_unlock_long_term': True, 'max_position_size_multiplier': 0.0, }, } # ────────────────────────────────────────────────────────────────────────────── # Indicateurs auxiliaires # ────────────────────────────────────────────────────────────────────────────── def _detect_capitulation_candles(klines_1d: List[List]) -> int: """Compte les bougies de capitulation acheteuse sur les 3 derniers jours. Heuristique : upper_wick > 1.5× corps ET range > 1.5× moyenne 14j. """ if not klines_1d or len(klines_1d) < 14: return 0 try: ranges = [float(k[2]) - float(k[3]) for k in klines_1d[-14:]] avg_range = float(np.mean(ranges)) if ranges else 0.0 if avg_range <= 0: return 0 count = 0 for k in klines_1d[-3:]: o, h, l, c = float(k[1]), float(k[2]), float(k[3]), float(k[4]) body = abs(c - o) upper_wick = h - max(o, c) day_range = h - l if upper_wick > 1.5 * max(body, 1e-12) and day_range > 1.5 * avg_range: count += 1 return count except Exception as e: logger.debug(f"capitulation detection error: {e}") return 0 def _safe_pct_change(end: float, start: float) -> float: if start <= 0: return 0.0 return (end - start) / start def _realized_volatility(closes: np.ndarray, window: int = 30) -> float: """Std des log-returns sur `window` derniers jours.""" if len(closes) < 2: return 0.0 w = min(window, len(closes) - 1) sub = closes[-(w + 1):] rets = np.diff(np.log(np.clip(sub, 1e-12, None))) return float(np.std(rets)) if len(rets) else 0.0 # ────────────────────────────────────────────────────────────────────────────── # Construction des métriques # ────────────────────────────────────────────────────────────────────────────── def compute_metrics_from_klines(klines_1d: List[List]) -> Optional[ContextMetrics]: """Construit ContextMetrics depuis des klines 1d (>= 30 bougies, idéal 180).""" if not klines_1d or len(klines_1d) < 30: logger.warning("Pas assez de klines pour compute_metrics (min 30 jours)") return None try: highs = np.array([float(k[2]) for k in klines_1d]) lows = np.array([float(k[3]) for k in klines_1d]) closes = np.array([float(k[4]) for k in klines_1d]) volumes = np.array([float(k[5]) for k in klines_1d]) current_price = float(closes[-1]) win_60 = min(60, len(klines_1d)) win_180 = min(180, len(klines_1d)) atl_idx = int(np.argmin(lows[-win_60:])) atl_60d = float(lows[-win_60:][atl_idx]) days_since_atl = (win_60 - 1) - atl_idx ath_idx = int(np.argmax(highs[-win_180:])) ath_180d = float(highs[-win_180:][ath_idx]) days_since_ath = (win_180 - 1) - ath_idx dist_atl_pct = _safe_pct_change(current_price, atl_60d) dist_ath_pct = current_price / ath_180d if ath_180d > 0 else 0.0 trend_7d = _safe_pct_change(current_price, float(closes[-8])) if len(closes) >= 8 else 0.0 trend_30d = _safe_pct_change(current_price, float(closes[-31])) if len(closes) >= 31 else 0.0 vol_7d = float(np.mean(volumes[-7:])) if len(volumes) >= 7 else 0.0 vol_30d = float(np.mean(volumes[-30:])) if len(volumes) >= 30 else 0.0 volume_ratio = vol_7d / vol_30d if vol_30d > 0 else 1.0 cap_candles = _detect_capitulation_candles(klines_1d) rv = _realized_volatility(closes, window=30) return ContextMetrics( current_price=current_price, atl_60d=atl_60d, ath_180d=ath_180d, distance_from_atl_60d_pct=dist_atl_pct, distance_from_ath_180d_pct=dist_ath_pct, days_since_atl_60d=days_since_atl, days_since_ath_180d=days_since_ath, trend_7d_pct=trend_7d, trend_30d_pct=trend_30d, volume_ratio_7d_30d=volume_ratio, capitulation_candles_3d=cap_candles, realized_volatility_30d=rv, ) except Exception as e: logger.error(f"compute_metrics error: {e}", exc_info=True) return None # ────────────────────────────────────────────────────────────────────────────── # Classification # ────────────────────────────────────────────────────────────────────────────── def _phoenix_score(m: ContextMetrics) -> Tuple[int, Dict[str, bool]]: ph = PHOENIX_THRESHOLDS conds = { 'dist_atl_in_range': ph['dist_atl_min'] <= m.distance_from_atl_60d_pct <= ph['dist_atl_max'], 'trend_30d_strong': m.trend_30d_pct >= ph['trend_30d_min'], 'volume_surge': m.volume_ratio_7d_30d >= ph['volume_ratio_min'], 'atl_recent': m.days_since_atl_60d <= ph['days_since_atl_max'], } return sum(conds.values()), conds def _pump_terminal_score(m: ContextMetrics) -> Tuple[int, Dict[str, bool]]: pt = PUMP_TERMINAL_THRESHOLDS conds = { 'near_ath': m.distance_from_ath_180d_pct >= pt['dist_ath_min'], 'trend_7d_vertical': m.trend_7d_pct >= pt['trend_7d_min'], 'trend_30d_extended': m.trend_30d_pct >= pt['trend_30d_min'], 'capitulation_visible': m.capitulation_candles_3d >= pt['capitulation_candles_min'], } return sum(conds.values()), conds def classify_regime(m: ContextMetrics) -> Tuple[str, float, List[str]]: """Classifie le régime à partir des métriques. Ordre : PUMP_TERMINAL > PHOENIX (si rebond ATL récent fort) > PARABOLIC_EXTENSION > BULL_RUN > CONSOLIDATION > BEAR_RANGE """ notes: List[str] = [] pt_count, pt_detail = _pump_terminal_score(m) ph_count, ph_detail = _phoenix_score(m) # 1) PUMP_TERMINAL — 3/4 conditions ET capitulation visible obligatoire if ( pt_count >= PUMP_TERMINAL_THRESHOLDS['min_conditions_met'] and pt_detail['capitulation_visible'] ): conf = float(np.clip(0.55 + 0.10 * pt_count, 0.55, 0.95)) notes.append( f"PT {pt_count}/4 + capitulation: dist_ATH={m.distance_from_ath_180d_pct:.0%}, " f"trend7d={m.trend_7d_pct:+.0%}, trend30d={m.trend_30d_pct:+.0%}, " f"cap_candles={m.capitulation_candles_3d}" ) return 'PUMP_TERMINAL', conf, notes # 2) PHOENIX — rebond depuis ATL récent (priorité sur Parabolic si ATL récent) if ph_count >= PHOENIX_THRESHOLDS['min_conditions_met'] and ph_detail['atl_recent']: conf = 0.55 + 0.10 * ph_count conf += 0.10 * min(1.0, (m.trend_30d_pct - PHOENIX_THRESHOLDS['trend_30d_min']) / 0.5) conf += 0.05 * min(1.0, (m.volume_ratio_7d_30d - 1.0) / 2.0) conf = float(np.clip(conf, 0.55, 0.95)) notes.append( f"Phoenix {ph_count}/4: ATL+{m.distance_from_atl_60d_pct:+.0%} " f"({m.days_since_atl_60d}d), trend30d={m.trend_30d_pct:+.0%}, " f"vol×{m.volume_ratio_7d_30d:.2f}" ) return 'PHOENIX', conf, notes # 3) PARABOLIC_EXTENSION — vertical sans capitulation (et pas Phoenix) pe = PARABOLIC_EXTENSION_THRESHOLDS if ( m.trend_30d_pct >= pe['trend_30d_min'] and m.trend_7d_pct >= pe['trend_7d_min'] and m.capitulation_candles_3d == 0 ): conf = float(np.clip(0.60 + 0.15 * min(1.0, m.trend_30d_pct / 1.5), 0.60, 0.90)) notes.append( f"Parabolic: trend30d={m.trend_30d_pct:+.0%}, trend7d={m.trend_7d_pct:+.0%}, " f"no capitulation yet" ) return 'PARABOLIC_EXTENSION', conf, notes # 4) PHOENIX faible (ATL pas récent mais autres conditions OK) if ph_count >= PHOENIX_THRESHOLDS['min_conditions_met']: conf = 0.50 + 0.05 * ph_count conf = float(np.clip(conf, 0.50, 0.75)) notes.append( f"Phoenix weak {ph_count}/4 (ATL too old {m.days_since_atl_60d}d): " f"trend30d={m.trend_30d_pct:+.0%}, vol×{m.volume_ratio_7d_30d:.2f}" ) return 'PHOENIX', conf, notes # 5) BULL_RUN — tendance saine br = BULL_RUN_THRESHOLDS if ( br['trend_30d_min'] <= m.trend_30d_pct <= br['trend_30d_max'] and m.capitulation_candles_3d == 0 ): conf = float(np.clip( 0.60 + 0.20 * min(1.0, (m.trend_30d_pct - br['trend_30d_min']) / 0.5), 0.60, 0.90, )) notes.append(f"Bull steady: trend30d={m.trend_30d_pct:+.0%}, no capitulation") return 'BULL_RUN', conf, notes # 6) CONSOLIDATION cs = CONSOLIDATION_THRESHOLDS if ( abs(m.trend_30d_pct) <= cs['trend_30d_abs_max'] and m.realized_volatility_30d <= cs['volatility_max'] ): notes.append( f"Consolidation: trend30d={m.trend_30d_pct:+.0%}, " f"vol={m.realized_volatility_30d:.2%}" ) return 'CONSOLIDATION', 0.65, notes # 7) BEAR_RANGE — défaut notes.append( f"Bear/range: trend30d={m.trend_30d_pct:+.0%}, trend7d={m.trend_7d_pct:+.0%}" ) return 'BEAR_RANGE', 0.70, notes def detect_market_context(klines_1d: List[List]) -> Optional[ContextResult]: """Pipeline complet : klines 1d → ContextResult.""" metrics = compute_metrics_from_klines(klines_1d) if metrics is None: return None regime, confidence, notes = classify_regime(metrics) return ContextResult( regime=regime, confidence=round(confidence, 3), metrics=metrics, implications=IMPLICATIONS[regime], notes=notes, ) # ────────────────────────────────────────────────────────────────────────────── # Helpers I/O (fetch klines Binance public) # ────────────────────────────────────────────────────────────────────────────── def fetch_daily_klines(symbol: str, limit: int = 180, timeout: int = 10) -> Optional[List[List]]: """Récupère `limit` bougies 1d depuis l'API publique Binance.""" try: r = requests.get( f"{BINANCE_API_URL}/klines", params={'symbol': symbol, 'interval': '1d', 'limit': limit}, timeout=timeout, ) if r.status_code == 200: data = r.json() return data if isinstance(data, list) else None logger.debug(f"Binance {symbol} HTTP {r.status_code}: {r.text[:120]}") return None except Exception as e: logger.debug(f"fetch_daily_klines({symbol}) error: {e}") return None # ────────────────────────────────────────────────────────────────────────────── # Singleton avec cache per-symbole (TTL 1h) # ────────────────────────────────────────────────────────────────────────────── _CACHE_TTL = 3600 # secondes class MarketContextDetector: """Singleton qui maintient un cache per-symbole de ContextResult (TTL 1h). Usage : from market_context import get_market_context_detector detector = get_market_context_detector() result = detector.get_context('SAGAUSDC') # → ContextResult | None """ def __init__(self) -> None: self._cache: Dict[str, tuple] = {} # symbol → (ContextResult, timestamp) self._lock = threading.Lock() # ── Résolution du symbole ──────────────────────────────────────────────── @staticmethod def _resolve_symbol(symbol: str) -> Optional[str]: """Essaie le symbole tel quel, puis le fallback USDT si USDC absent. Ex : SAGAUSDC → essaie SAGAUSDC d'abord, puis SAGAUSDT. """ klines = fetch_daily_klines(symbol, limit=30) if klines and len(klines) >= 30: return symbol if symbol.endswith('USDC'): fallback = symbol[:-4] + 'USDT' klines2 = fetch_daily_klines(fallback, limit=30) if klines2 and len(klines2) >= 30: logger.debug(f"Couche7: fallback {symbol} → {fallback}") return fallback return None # ── Entrée principale ──────────────────────────────────────────────────── def get_context(self, symbol: str, force_refresh: bool = False) -> Optional[ContextResult]: """Retourne le ContextResult pour le symbole, depuis le cache si frais. - Si données insuffisantes (< 30j) → retourne None (régime UNKNOWN) - Ne lève jamais d'exception — toujours fail-safe. """ now = time.monotonic() with self._lock: if not force_refresh and symbol in self._cache: result, ts = self._cache[symbol] if now - ts < _CACHE_TTL: return result try: resolved = self._resolve_symbol(symbol) if resolved is None: logger.debug(f"Couche7: {symbol} non résolvable (< 30j de données)") return None klines = fetch_daily_klines(resolved, limit=180) if not klines or len(klines) < 30: return None result = detect_market_context(klines) if result is not None: with self._lock: self._cache[symbol] = (result, now) return result except Exception as e: logger.warning(f"Couche7 get_context({symbol}): {e}") return None def invalidate(self, symbol: str) -> None: with self._lock: self._cache.pop(symbol, None) def cache_stats(self) -> Dict: with self._lock: return {'cached_symbols': len(self._cache), 'ttl_seconds': _CACHE_TTL} # Singleton global _detector_instance: Optional[MarketContextDetector] = None _detector_lock = threading.Lock() def get_market_context_detector() -> MarketContextDetector: """Retourne le singleton MarketContextDetector (pattern identique à market_regime.py).""" global _detector_instance if _detector_instance is None: with _detector_lock: if _detector_instance is None: _detector_instance = MarketContextDetector() return _detector_instance # ────────────────────────────────────────────────────────────────────────────── # Logger JSONL pour backtest a posteriori # ────────────────────────────────────────────────────────────────────────────── def log_couche7_decision( symbol: str, context: Optional[ContextResult], bot_decision: str, # ex: 'BUY', 'SKIP', 'CONFIRM' reason: str = '', extra: Optional[Dict] = None, ) -> None: """Enregistre une entrée JSONL dans data/couche7_decisions.log. Appelé depuis market_spy.py à chaque évaluation d'un surge. Permet de rejouer offline et comparer décision actuelle vs Couche7. """ if not COUCHE7_LOG_ENABLED: return try: os.makedirs(os.path.dirname(COUCHE7_LOG_FILE), exist_ok=True) entry = { 'ts': time.time(), 'dt': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), 'symbol': symbol, 'bot_decision': bot_decision, 'reason': reason, 'couche7_regime': context.regime if context else 'N/A', 'couche7_confidence': context.confidence if context else 0.0, 'couche7_implications': context.implications if context else {}, 'couche7_notes': context.notes if context else [], 'metrics': asdict(context.metrics) if context else {}, 'shadow_mode': COUCHE7_SHADOW_MODE, 'enabled': COUCHE7_ENABLED, } if extra: entry.update(extra) with open(COUCHE7_LOG_FILE, 'a', encoding='utf-8') as f: f.write(json.dumps(entry, ensure_ascii=False) + '\n') except Exception as e: logger.debug(f"log_couche7_decision error: {e}") # ────────────────────────────────────────────────────────────────────────────── # Matrice macro × per-token (Phase B) # ────────────────────────────────────────────────────────────────────────────── # # Règle conservative : un token PHOENIX en marché macro BEAR reste bloqué. # Un token PUMP_TERMINAL en marché BULL_STRONG reste bloqué. # # Matrice proposée (à implémenter dans Phase B) : # # macro_regime \ token_regime │ PHOENIX │ BULL_RUN │ PARABOLIC │ PUMP_TERM │ CONSOL │ BEAR # ───────────────────────────────────────────────────────────────────────────────────────── # BULL_STRONG │ LONG+ │ LONG │ WATCH │ AVOID │ WATCH │ AVOID # BULL_WEAK │ LONG │ LONG │ WATCH │ AVOID │ AVOID │ AVOID # NEUTRAL │ WATCH │ WATCH │ AVOID │ AVOID │ AVOID │ AVOID # CORRECTION │ WATCH │ AVOID │ AVOID │ AVOID │ AVOID │ AVOID # EARLY_RECOVERY │ LONG+ │ LONG │ WATCH │ AVOID │ WATCH │ AVOID # BEAR │ AVOID │ AVOID │ AVOID │ AVOID │ AVOID │ AVOID # # LONG+ = taille position × 1.5 (Phoenix confirmé par macro haussière) # LONG = décision normale (grille standard) # WATCH = réduire taille × 0.6, TP plus agressif # AVOID = bloquer l'entrée MACRO_CONTEXT_MATRIX: Dict[str, Dict[str, str]] = { 'BULL_STRONG': {'PHOENIX': 'LONG_PLUS', 'BULL_RUN': 'LONG', 'PARABOLIC_EXTENSION': 'WATCH', 'PUMP_TERMINAL': 'AVOID', 'CONSOLIDATION': 'WATCH', 'BEAR_RANGE': 'AVOID'}, 'BULL_WEAK': {'PHOENIX': 'LONG', 'BULL_RUN': 'LONG', 'PARABOLIC_EXTENSION': 'WATCH', 'PUMP_TERMINAL': 'AVOID', 'CONSOLIDATION': 'AVOID', 'BEAR_RANGE': 'AVOID'}, 'NEUTRAL': {'PHOENIX': 'WATCH', 'BULL_RUN': 'WATCH', 'PARABOLIC_EXTENSION': 'AVOID', 'PUMP_TERMINAL': 'AVOID', 'CONSOLIDATION': 'AVOID', 'BEAR_RANGE': 'AVOID'}, 'CORRECTION': {'PHOENIX': 'WATCH', 'BULL_RUN': 'AVOID', 'PARABOLIC_EXTENSION': 'AVOID', 'PUMP_TERMINAL': 'AVOID', 'CONSOLIDATION': 'AVOID', 'BEAR_RANGE': 'AVOID'}, 'EARLY_RECOVERY': {'PHOENIX': 'LONG_PLUS', 'BULL_RUN': 'LONG', 'PARABOLIC_EXTENSION': 'WATCH', 'PUMP_TERMINAL': 'AVOID', 'CONSOLIDATION': 'WATCH', 'BEAR_RANGE': 'AVOID'}, 'BEAR': {'PHOENIX': 'AVOID', 'BULL_RUN': 'AVOID', 'PARABOLIC_EXTENSION': 'AVOID', 'PUMP_TERMINAL': 'AVOID', 'CONSOLIDATION': 'AVOID', 'BEAR_RANGE': 'AVOID'}, } def get_combined_signal(macro_regime: str, token_regime: str) -> str: """Retourne le signal combiné macro × per-token depuis la matrice. Retourne 'LONG' si macro_regime inconnu (fail-safe — ne pas bloquer si données manquantes). """ row = MACRO_CONTEXT_MATRIX.get(macro_regime) if row is None: return 'LONG' # macro inconnue → ne pas interférer return row.get(token_regime, 'WATCH')