mean reversion implementation - statistical edge in practice

import numpy as np
import pandas as pd
from scipy import stats
from dataclasses import dataclass, field
from typing import Optional, Tuple, List
from enum import Enum
import asyncio
from datetime import datetime, timedelta

class SignalType(Enum):
    LONG = 1
    SHORT = -1
    FLAT = 0

@dataclass
class MeanReversionConfig:
    # lookback windows
    fast_period: int = 10
    slow_period: int = 50
    vol_period: int = 20

    # z-score thresholds
    entry_z: float = 2.0
    exit_z: float = 0.5
    stop_z: float = 3.5

    # statistical filters
    min_halflife: int = 3
    max_halflife: int = 20
    min_hurst: float = 0.0
    max_hurst: float = 0.4  # mean reverting < 0.5

    # risk parameters
    max_holding_periods: int = 10
    position_size_pct: float = 0.01

@dataclass
class MeanReversionState:
    position: SignalType = SignalType.FLAT
    entry_price: Optional[float] = None
    entry_time: Optional[datetime] = None
    entry_z: Optional[float] = None
    periods_held: int = 0

class MeanReversionStrategy:
    def __init__(self, config: MeanReversionConfig):
        self.config = config
        self.state = MeanReversionState()
        self.trade_history: List[dict] = []

    def calculate_halflife(self, prices: pd.Series) -> float:
        """
        Ornstein-Uhlenbeck halflife estimation.
        Measures speed of mean reversion.
        """
        lagged = prices.shift(1).dropna()
        delta = prices.diff().dropna()

        # align series
        lagged = lagged.iloc[1:]
        delta = delta.iloc[1:]

        if len(lagged) < 20:
            return float('inf')

        # regression: delta = alpha + beta * lagged + epsilon
        try:
            slope, intercept, r_value, p_value, std_err = stats.linregress(
                lagged.values, delta.values
            )
            if slope >= 0:
                return float('inf')  # not mean reverting
            halflife = -np.log(2) / slope
            return max(0.1, halflife)
        except Exception:
            return float('inf')

    def calculate_hurst(self, prices: pd.Series, max_lag: int = 20) -> float:
        """
        Hurst exponent using rescaled range method.
        H < 0.5 = mean reverting
        H = 0.5 = random walk
        H > 0.5 = trending
        """
        if len(prices) < max_lag * 2:
            return 0.5

        lags = range(2, max_lag)
        rs_values = []

        for lag in lags:
            # subdivide into chunks
            chunks = len(prices) // lag
            if chunks < 2:
                continue

            rs_chunk = []
            for i in range(chunks):
                chunk = prices.iloc[i * lag:(i + 1) * lag]
                if len(chunk) < lag:
                    continue

                # cumulative deviate
                mean_chunk = chunk.mean()
                deviate = (chunk - mean_chunk).cumsum()
                range_val = deviate.max() - deviate.min()
                std_val = chunk.std()

                if std_val > 0:
                    rs_chunk.append(range_val / std_val)

            if rs_chunk:
                rs_values.append((lag, np.mean(rs_chunk)))

        if len(rs_values) < 3:
            return 0.5

        # log-log regression
        x = np.log([v[0] for v in rs_values])
        y = np.log([v[1] for v in rs_values])

        try:
            slope, _, _, _, _ = stats.linregress(x, y)
            return np.clip(slope, 0, 1)
        except Exception:
            return 0.5

    def calculate_zscore(self, prices: pd.Series) -> Tuple[float, float, float]:
        """
        Calculate z-score relative to slow moving average.
        Returns: (zscore, fast_ma, slow_ma)
        """
        fast_ma = prices.tail(self.config.fast_period).mean()
        slow_ma = prices.tail(self.config.slow_period).mean()

        if len(prices) < self.config.vol_period:
            return 0.0, fast_ma, slow_ma

        # use rolling std for volatility normalization
        rolling_std = prices.tail(self.config.vol_period).std()

        if rolling_std == 0:
            return 0.0, fast_ma, slow_ma

        zscore = (fast_ma - slow_ma) / rolling_std
        return zscore, fast_ma, slow_ma

    def check_statistical_validity(self, prices: pd.Series) -> Tuple[bool, dict]:
        """
        Verify mean reversion is statistically valid right now.
        """
        halflife = self.calculate_halflife(prices)
        hurst = self.calculate_hurst(prices)

        valid = (
            self.config.min_halflife <= halflife <= self.config.max_halflife
            and self.config.min_hurst <= hurst <= self.config.max_hurst
        )

        metrics = {
            'halflife': halflife,
            'hurst': hurst,
            'valid': valid
        }

        return valid, metrics

    def generate_signal(self, prices: pd.Series,
                        current_time: datetime) -> Tuple[SignalType, dict]:
        """
        Main signal generation logic.
        """
        if len(prices) < self.config.slow_period + 10:
            return SignalType.FLAT, {'reason': 'insufficient_data'}

        # calculate metrics
        zscore, fast_ma, slow_ma = self.calculate_zscore(prices)
        valid, stats_metrics = self.check_statistical_validity(
            prices.tail(100)
        )

        signal_info = {
            'zscore': zscore,
            'fast_ma': fast_ma,
            'slow_ma': slow_ma,
            **stats_metrics
        }

        # check existing position
        if self.state.position != SignalType.FLAT:
            self.state.periods_held += 1

            # exit conditions
            should_exit = False
            exit_reason = None

            # z-score crossed back to neutral
            if self.state.position == SignalType.LONG and zscore > -self.config.exit_z:
                should_exit = True
                exit_reason = 'zscore_exit'
            elif self.state.position == SignalType.SHORT and zscore < self.config.exit_z:
                should_exit = True
                exit_reason = 'zscore_exit'

            # stop loss (mean reversion failed)
            if abs(zscore) > self.config.stop_z:
                should_exit = True
                exit_reason = 'stop_loss'

            # time stop
            if self.state.periods_held >= self.config.max_holding_periods:
                should_exit = True
                exit_reason = 'time_stop'

            if should_exit:
                signal_info['exit_reason'] = exit_reason
                self._record_trade(prices.iloc[-1], current_time, exit_reason)
                self.state = MeanReversionState()  # reset
                return SignalType.FLAT, signal_info

            return self.state.position, signal_info

        # entry logic (only if flat)
        if not valid:
            signal_info['reason'] = 'statistical_filter'
            return SignalType.FLAT, signal_info

        # long entry: price below mean
        if zscore < -self.config.entry_z:
            self.state = MeanReversionState(
                position=SignalType.LONG,
                entry_price=prices.iloc[-1],
                entry_time=current_time,
                entry_z=zscore
            )
            signal_info['reason'] = 'long_entry'
            return SignalType.LONG, signal_info

        # short entry: price above mean
        if zscore > self.config.entry_z:
            self.state = MeanReversionState(
                position=SignalType.SHORT,
                entry_price=prices.iloc[-1],
                entry_time=current_time,
                entry_z=zscore
            )
            signal_info['reason'] = 'short_entry'
            return SignalType.SHORT, signal_info

        signal_info['reason'] = 'no_signal'
        return SignalType.FLAT, signal_info

    def _record_trade(self, exit_price: float, exit_time: datetime,
                      exit_reason: str) -> None:
        """Record completed trade for analysis."""
        if self.state.entry_price is None:
            return

        pnl_pct = (exit_price - self.state.entry_price) / self.state.entry_price
        if self.state.position == SignalType.SHORT:
            pnl_pct = -pnl_pct

        self.trade_history.append({
            'entry_time': self.state.entry_time,
            'exit_time': exit_time,
            'entry_price': self.state.entry_price,
            'exit_price': exit_price,
            'entry_z': self.state.entry_z,
            'direction': self.state.position.name,
            'periods_held': self.state.periods_held,
            'pnl_pct': pnl_pct,
            'exit_reason': exit_reason
        })

    def get_performance_stats(self) -> dict:
        """Calculate strategy performance metrics."""
        if not self.trade_history:
            return {}

        trades_df = pd.DataFrame(self.trade_history)
        wins = trades_df[trades_df['pnl_pct'] > 0]
        losses = trades_df[trades_df['pnl_pct'] <= 0]

        return {
            'total_trades': len(trades_df),
            'win_rate': len(wins) / len(trades_df) if len(trades_df) > 0 else 0,
            'avg_win': wins['pnl_pct'].mean() if len(wins) > 0 else 0,
            'avg_loss': losses['pnl_pct'].mean() if len(losses) > 0 else 0,
            'profit_factor': abs(wins['pnl_pct'].sum() / losses['pnl_pct'].sum())
                if len(losses) > 0 and losses['pnl_pct'].sum() != 0 else float('inf'),
            'avg_holding_periods': trades_df['periods_held'].mean(),
            'sharpe': trades_df['pnl_pct'].mean() / trades_df['pnl_pct'].std()
                * np.sqrt(252) if trades_df['pnl_pct'].std() > 0 else 0
        }