"""Regime detector for market regime classification. Classifies the current market regime for each ticker based on EMA trend indicators and volatility ratios. Adjusts scoring thresholds and contradiction penalties per regime. Requirements: 7.1, 7.2, 7.3, 7.4, 7.5, 7.6, 7.7, 7.9 """ from __future__ import annotations import math import statistics from dataclasses import dataclass from enum import Enum class MarketRegime(str, Enum): """Market regime classification categories.""" TREND_FOLLOWING = "trend_following" PANIC = "panic" MEAN_REVERSION = "mean_reversion" UNCERTAINTY = "uncertainty" @dataclass(frozen=True) class RegimeClassification: """Result of regime detection for a ticker.""" regime: MarketRegime trend_indicator: float # R = sign(EMA_20 - EMA_100) volatility_ratio: float # V_r = σ_20 / σ_100 bullish_threshold: float # Adjusted ±threshold for direction bearish_threshold: float contradiction_penalty_multiplier: float # 0.4 default, 0.6 for uncertainty @dataclass(frozen=True) class RegimeConfig: """Configuration parameters for regime detection.""" ema_short_period: int = 20 ema_long_period: int = 100 vol_short_period: int = 20 vol_long_period: int = 100 panic_vol_ratio: float = 1.5 trend_vol_ratio: float = 1.2 mean_reversion_vol_ratio: float = 1.0 default_threshold: float = 0.15 panic_threshold: float = 0.10 mean_reversion_threshold: float = 0.20 uncertainty_contradiction_multiplier: float = 0.6 # Default uncertainty classification used when data is insufficient _DEFAULT_UNCERTAINTY = RegimeClassification( regime=MarketRegime.UNCERTAINTY, trend_indicator=0.0, volatility_ratio=1.0, bullish_threshold=0.15, bearish_threshold=-0.15, contradiction_penalty_multiplier=0.6, ) def compute_ema(values: list[float], period: int) -> float: """Compute exponential moving average over the last ``period`` values. Uses the standard EMA formula with multiplier = 2 / (period + 1). Iterates through the values, seeding the EMA with the first value. Raises ``ValueError`` when *values* is empty or *period* < 1. """ if not values or period < 1: raise ValueError("values must be non-empty and period must be >= 1") # Use only the last `period` values (or all if fewer) data = values[-period:] if len(values) >= period else values multiplier = 2.0 / (period + 1) ema = data[0] for value in data[1:]: ema = (value - ema) * multiplier + ema return ema def _sign(x: float) -> float: """Return -1.0, 0.0, or 1.0 for the sign of *x*.""" if x > 0.0: return 1.0 if x < 0.0: return -1.0 return 0.0 def classify_regime( closing_prices: list[float], returns: list[float], config: RegimeConfig = RegimeConfig(), ) -> RegimeClassification: """Classify market regime from price and return history. Requires at least ``config.ema_long_period`` days of price history for EMA_100. Falls back to UNCERTAINTY when data is insufficient or standard deviations are zero. Requirements: 7.1, 7.2, 7.3, 7.4, 7.5, 7.6, 7.7, 7.9 """ # Insufficient price data → uncertainty if len(closing_prices) < config.ema_long_period: return _DEFAULT_UNCERTAINTY # Insufficient return data → uncertainty if len(returns) < config.vol_long_period: return _DEFAULT_UNCERTAINTY # --- Trend indicator: R = sign(EMA_short - EMA_long) --- ema_short = compute_ema(closing_prices, config.ema_short_period) ema_long = compute_ema(closing_prices, config.ema_long_period) trend_indicator = _sign(ema_short - ema_long) # --- Volatility ratio: V_r = σ_short / σ_long --- short_returns = returns[-config.vol_short_period:] long_returns = returns[-config.vol_long_period:] # Guard against zero or near-zero standard deviations if len(short_returns) < 2 or len(long_returns) < 2: return _DEFAULT_UNCERTAINTY sigma_short = statistics.stdev(short_returns) sigma_long = statistics.stdev(long_returns) if sigma_long == 0.0 or sigma_short == 0.0: return _DEFAULT_UNCERTAINTY if math.isnan(sigma_short) or math.isnan(sigma_long): return _DEFAULT_UNCERTAINTY volatility_ratio = sigma_short / sigma_long # --- Classification rules (Req 7.3) --- # Panic takes priority: V_r > 1.5 if volatility_ratio > config.panic_vol_ratio: regime = MarketRegime.PANIC threshold = config.panic_threshold # ±0.10 contradiction_mult = 0.4 # Trend-following: R ≠ 0 AND V_r < 1.2 elif trend_indicator != 0.0 and volatility_ratio < config.trend_vol_ratio: regime = MarketRegime.TREND_FOLLOWING threshold = config.default_threshold # ±0.15 contradiction_mult = 0.4 # Mean-reversion: R = 0 AND V_r < 1.0 elif trend_indicator == 0.0 and volatility_ratio < config.mean_reversion_vol_ratio: regime = MarketRegime.MEAN_REVERSION threshold = config.mean_reversion_threshold # ±0.20 contradiction_mult = 0.4 # Uncertainty: all other cases else: regime = MarketRegime.UNCERTAINTY threshold = config.default_threshold # ±0.15 contradiction_mult = config.uncertainty_contradiction_multiplier # 0.6 return RegimeClassification( regime=regime, trend_indicator=trend_indicator, volatility_ratio=volatility_ratio, bullish_threshold=threshold, bearish_threshold=-threshold, contradiction_penalty_multiplier=contradiction_mult, ) # --------------------------------------------------------------------------- # V3 Regime Detection — Calibrated Evidence Engine # Requirements: 6.1, 6.2, 6.3, 6.4, 6.5, 6.6, 6.7, 6.8 # --------------------------------------------------------------------------- @dataclass(frozen=True) class V3RegimeClassification: """V3 regime classification result with calibrated parameters. Attributes: regime: Market regime category. trend_z: ATR-normalized trend indicator (EMA_20 - EMA_100) / ATR_20. vol_ratio: Volatility ratio sigma_20 / sigma_100. evidence_multiplier: Regime-specific gamma for posterior LLR scaling. confidence_multiplier: Regime-specific confidence scaling factor. phi_decay: Evidence state decay factor for projection. atr_multiplier: Regime-specific ATR multiplier for stop computation. """ regime: MarketRegime trend_z: float vol_ratio: float evidence_multiplier: float confidence_multiplier: float phi_decay: float atr_multiplier: float # Regime parameter lookup: (gamma, confidence_mult, phi, ATR_mult, min_edge) _V3_REGIME_PARAMS: dict[MarketRegime, tuple[float, float, float, float, float]] = { MarketRegime.PANIC: (0.70, 0.70, 0.35, 2.5, 0.0100), MarketRegime.TREND_FOLLOWING: (1.10, 1.00, 0.80, 1.8, 0.0035), MarketRegime.MEAN_REVERSION: (0.90, 0.95, 0.55, 1.4, 0.0050), MarketRegime.UNCERTAINTY: (0.80, 0.85, 0.50, 2.0, 0.0075), } # Default uncertainty classification for v3 when data is insufficient (Req 6.8) _DEFAULT_V3_UNCERTAINTY = V3RegimeClassification( regime=MarketRegime.UNCERTAINTY, trend_z=0.0, vol_ratio=1.0, evidence_multiplier=0.80, confidence_multiplier=0.85, phi_decay=0.50, atr_multiplier=2.0, ) def _compute_ema_full(values: list[float], span: int) -> float: """Compute EMA over the full values list with given span. Uses standard EMA formula: alpha = 2 / (span + 1), iterating from the beginning of the list. Seeds EMA with the first value. This differs from ``compute_ema`` which only uses the last ``period`` values. V3 requires iterating over the full history to produce a stable EMA_100. """ if not values or span < 1: raise ValueError("values must be non-empty and span must be >= 1") alpha = 2.0 / (span + 1) ema = values[0] for value in values[1:]: ema = alpha * value + (1.0 - alpha) * ema return ema def classify_regime_v3( closing_prices: list[float], daily_returns: list[float], atr_20: float, ) -> V3RegimeClassification: """Classify market regime using v3 ATR-normalized indicators. Computes trend_z = (EMA_20 - EMA_100) / ATR_20 and vol_ratio = sigma_20 / sigma_100 to determine the market regime. Classification priority (Req 6.2–6.5): 1. Panic: vol_ratio > 1.5 OR |trend_z| > 2.5 2. Trend following: |trend_z| >= 0.75 AND vol_ratio < 1.3 3. Mean reversion: |trend_z| < 0.50 AND vol_ratio < 1.0 4. Uncertainty: all other cases Falls back to uncertainty when data is insufficient (Req 6.8): - Fewer than 100 closing prices for EMA_100 - ATR_20 <= 0 (insufficient bars for ATR) - Fewer than 100 daily returns for sigma_100 Requirements: 6.1, 6.2, 6.3, 6.4, 6.5, 6.6, 6.7, 6.8 """ # --- Data sufficiency check (Req 6.8) --- if len(closing_prices) < 100: return _DEFAULT_V3_UNCERTAINTY if atr_20 <= 0.0: return _DEFAULT_V3_UNCERTAINTY if len(daily_returns) < 100: return _DEFAULT_V3_UNCERTAINTY # --- Compute trend_z (Req 6.1) --- ema_20 = _compute_ema_full(closing_prices, span=20) ema_100 = _compute_ema_full(closing_prices, span=100) trend_z = (ema_20 - ema_100) / atr_20 # --- Compute vol_ratio (Req 6.1) --- sigma_20 = statistics.stdev(daily_returns[-20:]) if len(daily_returns) >= 20 else 0.0 sigma_100 = statistics.stdev(daily_returns[-100:]) # Guard against zero sigma_100 if sigma_100 <= 0.0 or math.isnan(sigma_100): return _DEFAULT_V3_UNCERTAINTY if math.isnan(sigma_20): return _DEFAULT_V3_UNCERTAINTY vol_ratio = sigma_20 / sigma_100 # --- Classification rules (Req 6.2–6.5) --- # Priority 1: Panic (Req 6.2) if vol_ratio > 1.5 or abs(trend_z) > 2.5: regime = MarketRegime.PANIC # Priority 2: Trend following (Req 6.3) elif abs(trend_z) >= 0.75 and vol_ratio < 1.3: regime = MarketRegime.TREND_FOLLOWING # Priority 3: Mean reversion (Req 6.4) elif abs(trend_z) < 0.50 and vol_ratio < 1.0: regime = MarketRegime.MEAN_REVERSION # Priority 4: Uncertainty (Req 6.5) else: regime = MarketRegime.UNCERTAINTY # --- Assign regime parameters (Req 6.6, 6.7) --- gamma, conf_mult, phi, atr_mult, _min_edge = _V3_REGIME_PARAMS[regime] return V3RegimeClassification( regime=regime, trend_z=trend_z, vol_ratio=vol_ratio, evidence_multiplier=gamma, confidence_multiplier=conf_mult, phi_decay=phi, atr_multiplier=atr_mult, )