regime detection filtering framework - how i adapt to market conditions

import numpy as np
import pandas as pd
from typing import Tuple, Dict
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class RegimeState:
    """Market regime state representation"""
    vix: float
    correlation: float
    volume_pct: float
    confidence: float
    regime_type: str
    timestamp: datetime

class RegimeDetector:
    """
    Detects market regimes and calculates trade acceptance thresholds.

    Strategy thrives in:
    - VIX 14-19 (optimal volatility)
    - Correlation 0.4-0.7 (diversification exists)
    - Volume >80% of 20-day avg (liquidity present)
    """

    def __init__(
        self,
        vix_optimal_range: Tuple[float, float] = (14.0, 19.0),
        vix_acceptable_range: Tuple[float, float] = (12.0, 25.0),
        corr_optimal_range: Tuple[float, float] = (0.4, 0.7),
        corr_acceptable_range: Tuple[float, float] = (0.3, 0.8),
        volume_threshold: float = 0.80,
        lookback_days: int = 20
    ):
        self.vix_optimal = vix_optimal_range
        self.vix_acceptable = vix_acceptable_range
        self.corr_optimal = corr_optimal_range
        self.corr_acceptable = corr_acceptable_range
        self.volume_threshold = volume_threshold
        self.lookback_days = lookback_days

        # Historical regime states
        self.regime_history: list[RegimeState] = []

    def calculate_vix_score(self, vix: float) -> float:
        """
        Calculate VIX component score.

        Returns:
            1.0 if in optimal range
            0.5-1.0 if in acceptable range
            0.0-0.5 if outside acceptable range
        """
        if self.vix_optimal[0] <= vix <= self.vix_optimal[1]:
            return 1.0

        if self.vix_acceptable[0] <= vix <= self.vix_acceptable[1]:
            # Linear interpolation in acceptable range
            if vix < self.vix_optimal[0]:
                dist = self.vix_optimal[0] - vix
                max_dist = self.vix_optimal[0] - self.vix_acceptable[0]
            else:
                dist = vix - self.vix_optimal[1]
                max_dist = self.vix_acceptable[1] - self.vix_optimal[1]

            return 0.5 + (0.5 * (1 - dist / max_dist))

        # Outside acceptable range - severe penalty
        if vix < self.vix_acceptable[0]:
            # Too low VIX
            dist = self.vix_acceptable[0] - vix
            return max(0.0, 0.5 - (dist / 10))
        else:
            # Too high VIX
            dist = vix - self.vix_acceptable[1]
            return max(0.0, 0.5 - (dist / 10))

    def calculate_correlation_score(self, correlation: float) -> float:
        """
        Calculate correlation component score.

        Returns:
            1.0 if in optimal range
            0.5-1.0 if in acceptable range
            0.0-0.5 if outside acceptable range
        """
        if self.corr_optimal[0] <= correlation <= self.corr_optimal[1]:
            return 1.0

        if self.corr_acceptable[0] <= correlation <= self.corr_acceptable[1]:
            if correlation < self.corr_optimal[0]:
                dist = self.corr_optimal[0] - correlation
                max_dist = self.corr_optimal[0] - self.corr_acceptable[0]
            else:
                dist = correlation - self.corr_optimal[1]
                max_dist = self.corr_acceptable[1] - self.corr_optimal[1]

            return 0.5 + (0.5 * (1 - dist / max_dist))

        # Outside acceptable range
        if correlation < self.corr_acceptable[0]:
            # Too low correlation (no diversification)
            return max(0.0, 0.5 - (self.corr_acceptable[0] - correlation))
        else:
            # Too high correlation (no edge)
            return max(0.0, 0.5 - (correlation - self.corr_acceptable[1]))

    def calculate_volume_score(
        self,
        current_volume: float,
        historical_volumes: np.ndarray
    ) -> float:
        """
        Calculate volume component score.

        Args:
            current_volume: Today's volume
            historical_volumes: Last N days volumes

        Returns:
            0.0-1.0 score based on volume relative to average
        """
        avg_volume = np.mean(historical_volumes)
        volume_ratio = current_volume / avg_volume

        if volume_ratio >= 1.0:
            # Above average volume = good
            return 1.0
        elif volume_ratio >= self.volume_threshold:
            # Acceptable volume range
            dist = 1.0 - volume_ratio
            max_dist = 1.0 - self.volume_threshold
            return 0.5 + (0.5 * (1 - dist / max_dist))
        else:
            # Below threshold - severe penalty
            return max(0.0, volume_ratio / self.volume_threshold * 0.5)

    def detect_regime(
        self,
        vix: float,
        correlation: float,
        volume: float,
        historical_volumes: np.ndarray,
        timestamp: datetime = None
    ) -> RegimeState:
        """
        Detect current market regime and calculate confidence.

        Args:
            vix: Current VIX level
            correlation: Current market correlation
            volume: Current trading volume
            historical_volumes: Historical volume data
            timestamp: Current timestamp

        Returns:
            RegimeState with confidence and regime type
        """
        # Calculate component scores
        vix_score = self.calculate_vix_score(vix)
        corr_score = self.calculate_correlation_score(correlation)
        vol_score = self.calculate_volume_score(volume, historical_volumes)

        # Weighted average (VIX and volume matter more)
        confidence = (
            0.4 * vix_score +
            0.3 * vol_score +
            0.3 * corr_score
        )

        # Determine regime type
        if confidence >= 0.75:
            regime_type = "optimal"
        elif confidence >= 0.60:
            regime_type = "acceptable"
        elif confidence >= 0.40:
            regime_type = "challenging"
        else:
            regime_type = "hostile"

        # Create regime state
        state = RegimeState(
            vix=vix,
            correlation=correlation,
            volume_pct=volume / np.mean(historical_volumes),
            confidence=confidence,
            regime_type=regime_type,
            timestamp=timestamp or datetime.now()
        )

        # Store in history
        self.regime_history.append(state)

        return state

    def calculate_acceptance_threshold(self, regime_confidence: float) -> float:
        """
        Calculate trade acceptance threshold based on regime confidence.

        Higher confidence = lower threshold (accept more trades)
        Lower confidence = higher threshold (accept fewer trades)

        Args:
            regime_confidence: 0-1 regime confidence score

        Returns:
            Acceptance threshold for trade quality score
        """
        # Inverse relationship: low confidence = high threshold
        # Optimal regime (0.8+ confidence) = 0.5 threshold (accept 50%+)
        # Hostile regime (0.3 confidence) = 0.8 threshold (accept 20%)

        if regime_confidence >= 0.75:
            # Optimal conditions - accept more trades
            return 0.50
        elif regime_confidence >= 0.60:
            # Acceptable conditions - normal acceptance
            return 0.60
        elif regime_confidence >= 0.40:
            # Challenging conditions - selective
            return 0.70
        else:
            # Hostile conditions - extremely selective
            return 0.80

    def should_accept_trade(
        self,
        trade_quality_score: float,
        regime_state: RegimeState
    ) -> Tuple[bool, str]:
        """
        Determine if trade should be accepted based on regime and quality.

        Args:
            trade_quality_score: 0-1 quality score for this trade
            regime_state: Current regime state

        Returns:
            (accept: bool, reason: str)
        """
        threshold = self.calculate_acceptance_threshold(regime_state.confidence)

        if trade_quality_score >= threshold:
            return (
                True,
                f"ACCEPT: quality {trade_quality_score:.2f} >= threshold {threshold:.2f} "
                f"(regime: {regime_state.regime_type}, confidence: {regime_state.confidence:.2f})"
            )
        else:
            return (
                False,
                f"REJECT: quality {trade_quality_score:.2f} < threshold {threshold:.2f} "
                f"(regime: {regime_state.regime_type}, confidence: {regime_state.confidence:.2f})"
            )

    def get_regime_summary(self, lookback_days: int = 7) -> Dict:
        """
        Get summary statistics for recent regime history.

        Args:
            lookback_days: Number of days to summarize

        Returns:
            Dictionary with regime statistics
        """
        if not self.regime_history:
            return {}

        cutoff = datetime.now() - timedelta(days=lookback_days)
        recent = [s for s in self.regime_history if s.timestamp >= cutoff]

        if not recent:
            return {}

        return {
            'avg_confidence': np.mean([s.confidence for s in recent]),
            'avg_vix': np.mean([s.vix for s in recent]),
            'avg_correlation': np.mean([s.correlation for s in recent]),
            'avg_volume_pct': np.mean([s.volume_pct for s in recent]),
            'regime_distribution': {
                regime: sum(1 for s in recent if s.regime_type == regime) / len(recent)
                for regime in ['optimal', 'acceptable', 'challenging', 'hostile']
            },
            'days_analyzed': len(recent)
        }


# Example usage
if __name__ == "__main__":
    detector = RegimeDetector()

    # August 2024 conditions
    august_vix = 16.0
    august_corr = 0.54
    august_volume = 850000  # 30% below normal
    historical_vol = np.array([1200000] * 20)  # Normal volume

    # Detect regime
    state = detector.detect_regime(
        vix=august_vix,
        correlation=august_corr,
        volume=august_volume,
        historical_volumes=historical_vol
    )

    print(f"Regime Type: {state.regime_type}")
    print(f"Confidence: {state.confidence:.2f}")
    print(f"VIX: {state.vix}")
    print(f"Correlation: {state.correlation}")
    print(f"Volume %: {state.volume_pct:.1%}")

    # Calculate acceptance threshold
    threshold = detector.calculate_acceptance_threshold(state.confidence)
    print(f"\nAcceptance Threshold: {threshold:.2f}")

    # Test trade acceptance
    trade_score = 0.65
    accept, reason = detector.should_accept_trade(trade_score, state)
    print(f"\nTrade Decision: {reason}")