319 lines
11 KiB
Python
319 lines
11 KiB
Python
"""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,
|
||
)
|