""" Charter Market Regime Classifier - v1 Mavis 2026-07-20. Classifica il regime di mercato corrente per un symbol, adattato da SQW-023. Regimi (in ordine di priorità): - BREAKOUT_UP / BREAKOUT_DOWN: ADX > 30 + directional > 25 - STRONG_BULL_TREND / STRONG_BEAR_TREND: ADX > 32 + directional - BULL_TREND / BEAR_TREND: ADX 22-32 - VOLATILITY_EXPANSION: ATR ratio > 1.35 - COMPRESSION: BB width < 20mo percentile - SIDEWAYS: tutto il resto (= CHOP) Uso: from charter_core.market_regime import classify_regime regime = classify_regime(bybit_client, "BTCUSDT", timeframe="60") """ from __future__ import annotations import sys from pathlib import Path import numpy as np import pandas as pd # Regimi "TRADABLE" (in cui le strategie producono valore) # Regimi "CHOP" (in cui le strategie producono falsi segnali → SKIP ordini) TRADABLE_REGIMES = { "BREAKOUT_UP", "BREAKOUT_DOWN", "STRONG_BULL_TREND", "STRONG_BEAR_TREND", "BULL_TREND", "BEAR_TREND", "VOLATILITY_EXPANSION", } CHOP_REGIMES = {"SIDEWAYS", "COMPRESSION", "TRANSITION", "INSUFFICIENT_DATA"} def _ema(series, period): return series.ewm(span=period, adjust=False).mean() def _wilder_smooth(series, period): return series.ewm(alpha=1.0 / period, adjust=False, min_periods=period).mean() def _adx(df, period=14): high = df["high"]; low = df["low"]; close = df["close"] prev_close = close.shift(1) tr = pd.concat([ (high - low), (high - prev_close).abs(), (low - prev_close).abs(), ], axis=1).max(axis=1) up_move = high.diff() down_move = -low.diff() plus_dm = pd.Series(np.where((up_move > down_move) & (up_move > 0), up_move, 0.0), index=df.index) minus_dm = pd.Series(np.where((down_move > up_move) & (down_move > 0), down_move, 0.0), index=df.index) atr_w = _wilder_smooth(tr, period) plus_di = 100 * _wilder_smooth(plus_dm, period) / atr_w.replace(0, np.nan) minus_di = 100 * _wilder_smooth(minus_dm, period) / atr_w.replace(0, np.nan) dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan) adx = _wilder_smooth(dx, period) return adx, plus_di, minus_di def classify_regime(bybit_client, symbol: str, timeframe: str = "60", limit: int = 200) -> dict: """Classifica il regime corrente per un symbol. Ritorna dict {"regime": str, "adx": float, "atr_ratio": float, "bb_width": float, "is_chop": bool, "reason": str} """ try: klines = bybit_client.fetch_ohlcv(symbol, timeframe, limit) except Exception as e: return {"regime": "INSUFFICIENT_DATA", "is_chop": True, "reason": f"fetch_ohlcv error: {e}"} if not klines or len(klines) < 80: return {"regime": "INSUFFICIENT_DATA", "is_chop": True, "reason": f"klines insufficienti ({len(klines) if klines else 0})"} df = pd.DataFrame(klines, columns=["ts", "open", "high", "low", "close", "volume"]) df["ts"] = pd.to_datetime(df["ts"], unit="ms", utc=True) df = df.sort_values("ts").set_index("ts") adx, plus_di, minus_di = _adx(df, 14) df["adx"] = adx df["plus_di"] = plus_di df["minus_di"] = minus_di # ATR ratio high = df["high"]; low = df["low"]; close = df["close"] prev_close = close.shift(1) tr = pd.concat([(high - low), (high - prev_close).abs(), (low - prev_close).abs()], axis=1).max(axis=1) df["atr"] = _wilder_smooth(tr, 14) df["atr_median_42"] = df["atr"].rolling(42).median() df["atr_ratio"] = df["atr"] / df["atr_median_42"] # Bollinger width bb_mid = close.rolling(20).mean() bb_sd = close.rolling(20).std(ddof=0) df["bb_width"] = (bb_mid + 2 * bb_sd - (bb_mid - 2 * bb_sd)) / bb_mid df["bb_width_q20"] = df["bb_width"].rolling(120, min_periods=40).quantile(0.20) # Take LAST row (candela più recente) row = df.iloc[-1] cur = { "adx": float(row["adx"]) if not pd.isna(row["adx"]) else None, "plus_di": float(row["plus_di"]) if not pd.isna(row["plus_di"]) else None, "minus_di": float(row["minus_di"]) if not pd.isna(row["minus_di"]) else None, "close": float(row["close"]), "atr_ratio": float(row["atr_ratio"]) if not pd.isna(row["atr_ratio"]) else 1.0, "bb_width": float(row["bb_width"]) if not pd.isna(row["bb_width"]) else 0.0, "bb_width_q20": float(row["bb_width_q20"]) if not pd.isna(row["bb_width_q20"]) else 0.0, } # Classificazione (priorità: BREAKOUT > TREND > VOLATILITY > COMPRESSION > SIDEWAYS) regime = "INSUFFICIENT_DATA" reason = "" if cur["adx"] is None: return {"regime": "INSUFFICIENT_DATA", "is_chop": True, "reason": "ADX None", "adx": cur["adx"], "atr_ratio": cur["atr_ratio"], "bb_width": cur["bb_width"]} if cur["adx"] > 30 and cur["plus_di"] is not None and cur["minus_di"] is not None: if cur["plus_di"] > 25 and cur["minus_di"] < 20: ema_50_val = float(df["ema_50"].iloc[-1]) if "ema_50" in df.columns and not pd.isna(df["ema_50"].iloc[-1]) else 0 regime = "BREAKOUT_UP" if cur["close"] > ema_50_val else "BREAKOUT_DOWN" reason = f"ADX={cur['adx']:.1f} > 30, plus_di={cur['plus_di']:.1f} > 25, minus_di={cur['minus_di']:.1f} < 20" if regime == "INSUFFICIENT_DATA" and cur["adx"] > 32: regime = "STRONG_BULL_TREND" if cur["plus_di"] > cur["minus_di"] else "STRONG_BEAR_TREND" reason = f"ADX={cur['adx']:.1f} > 32 (trend forte)" if regime == "INSUFFICIENT_DATA" and cur["adx"] > 22: regime = "BULL_TREND" if cur["plus_di"] > cur["minus_di"] else "BEAR_TREND" reason = f"ADX={cur['adx']:.1f} 22-32 (trend)" if regime == "INSUFFICIENT_DATA" and cur["atr_ratio"] > 1.35: regime = "VOLATILITY_EXPANSION" reason = f"ATR ratio={cur['atr_ratio']:.2f} > 1.35 (volatile)" if regime == "INSUFFICIENT_DATA" and cur["bb_width"] < cur["bb_width_q20"]: regime = "COMPRESSION" reason = f"BB width={cur['bb_width']:.4f} < q20={cur['bb_width_q20']:.4f} (compression)" if regime == "INSUFFICIENT_DATA": regime = "SIDEWAYS" reason = f"ADX={cur['adx']:.1f} < 22 (chop)" is_chop = regime in CHOP_REGIMES return { "regime": regime, "is_chop": is_chop, "reason": reason, "adx": cur["adx"], "atr_ratio": cur["atr_ratio"], "bb_width": cur["bb_width"], "close": cur["close"], } if __name__ == "__main__": from bybit_demo_client import BybitDemoClient c = BybitDemoClient() for sym in ["BTCUSDT", "ETHUSDT", "SOLUSDT", "ZECUSDT", "DASHUSDT", "WIFUSDT"]: r = classify_regime(c, sym, "60") print(f"{sym}: regime={r['regime']:<25} chop={r['is_chop']} ({r['reason']})")