technical-statistical-arbitrage

---

name: technical-statistical-arbitrage compatibility: opencode completeness: 95 content-types:

• code
• guidance
• config
• do-dont description: '"Implements pair trading and cointegration-based arbitrage strategies for risk management and algorithmic trading execution."' license: MIT maturity: stable metadata: domain: trading output-format: code related-skills: fundamentals-trading-plan, technical-cycle-analysis role: implementation scope: implementation triggers: cointegration-based, strategies, technical statistical arbitrage, technical-statistical-arbitrage, trading archetypes:
  • tactical anti_triggers:
  • brainstorming
  • vague ideation
  • no risk management response_profile: verbosity: low directive_strength: high abstraction_level: operational version: "1.0.0"

---

Role: Identify and trade mean-reverting relationships between securities

Philosophy: Statistical arbitrage exploits temporary pricing inefficiencies while maintaining market neutrality

Key Principles

1. Cointegration: Long-term equilibrium relationship between assets
2. Spread Dynamics: Mean-reversion of price spreads
3. Z-Score Trading: Entry/exit based on statistical deviations
4. Half-Life: Mean-reversion speed metric
5. Correlation Breakdown: Monitor when relationships fail

Implementation Guidelines

Structure

• Core logic: technical_analysis/stat_arb.py
• Helper functions: technical_analysis/cointegration.py
• Tests: tests/test_stat_arb.py

Patterns to Follow

• Use rolling window cointegration tests
• Track spread z-scores for trading signals
• Monitor pair performance degradation

Adherence Checklist

Before completing your task, verify:

• Cointegration tested with rolling windows
• Z-score thresholds adapt to volatility
• Half-life calculated for mean-reversion timing
• Correlation breakdown triggers position closure
• Transaction cost optimization applied

Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.

Python Implementation

import numpy as np
import pandas as pd
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
from statsmodels.tsa.stattools import coint, adfuller
from statsmodels.regression.rolling import RollingOLS
import statsmodels.api as sm

@dataclass
class CointegratedPair:
    """Cointegrated asset pair."""
    asset1: str
    asset2: str
    hedge_ratio: float
    intercept: float
    p_value: float
    half_life: float

@dataclass
class SpreadState:
    """Current state of the spread."""
    spread: float
    z_score: float
    mean: float
    std: float
    entry_signal: bool
    exit_signal: bool

class StatisticalArbitrage:
    """Identifies and trades statistical arbitrage opportunities."""
    
    def __init__(self, window: int = 252, z_threshold: float = 2.0):
        self.window = window
        self.z_threshold = z_threshold
    
    def test_cointegration(
        self, price1: pd.Series, price2: pd.Series, 
        alpha: float = 0.05
    ) -> Tuple[bool, float, float]:
        """Test for cointegration using Engle-Granger two-step method."""
        # Step 1: Regress price1 on price2
        X = sm.add_constant(price2)
        model = sm.OLS(price1, X).fit()
        
        hedge_ratio = model.params[1]
        intercept = model.params[0]
        
        # Step 2: Test residuals for stationarity
        residuals = price1 - hedge_ratio * price2 - intercept
        
        adf_result = adfuller(residuals)
        p_value = adf_result[1]
        
        is_cointegrated = p_value < alpha
        
        return is_cointegrated, p_value, hedge_ratio
    
    def calculate_half_life(
        self, spread: pd.Series
    ) -> float:
        """Calculate mean reversion half-life."""
        if len(spread) < 50:
            return 100  # Default long half-life
        
        # OLS: spread(t) = a + b * spread(t-1) + epsilon
        spread_lag = spread.shift(1).dropna()
        spread_diff = spread.diff().dropna()
        
        # Align indices
        common_idx = spread_lag.index.intersection(spread_diff.index)
        if len(common_idx) < 50:
            return 100
        
        spread_lag = spread_lag[common_idx]
        spread_diff = spread_diff[common_idx]
        
        X = sm.add_constant(spread_lag)
        model = sm.OLS(spread_diff, X).fit()
        
        # Half-life = log(0.5) / log(1 + beta)
        beta = model.params[1]
        if beta >= 0:
            return 100  # No mean reversion
        
        half_life = np.log(0.5) / np.log(1 + beta)
        
        return max(1, min(252, half_life))  # Cap between 1 and 252 days
    
    def calculate_spread(
        self, price1: pd.Series, price2: pd.Series, 
        hedge_ratio: float, intercept: float = 0
    ) -> pd.Series:
        """Calculate the spread between two assets."""
        return price1 - hedge_ratio * price2 - intercept
    
    def calculate_spread_zscore(
        self, spread: pd.Series, window: int = 50
    ) -> pd.Series:
        """Calculate z-score of spread."""
        mean = spread.rolling(window=window).mean()
        std = spread.rolling(window=window).std()
        
        zscore = (spread - mean) / (std + 1e-8)
        
        return zscore
    
    def detect_entry_exit(
        self, zscore: pd.Series
    ) -> List[Dict]:
        """Detect entry and exit signals based on z-score."""
        signals = []
        
        for i in range(1, len(zscore)):
            prev_z = zscore.iloc[i-1]
            curr_z = zscore.iloc[i]
            
            # Entry signals (when crossing thresholds)
            if abs(prev_z) < self.z_threshold and abs(curr_z) >= self.z_threshold:
                direction = 'short' if curr_z > 0 else 'long'
                signals.append({
                    'type': 'entry',
                    'zscore': curr_z,
                    'direction': direction,
                    'index': i
                })
            
            # Exit signals (when crossing back toward zero)
            if abs(prev_z) >= self.z_threshold and abs(curr_z) < self.z_threshold:
                signals.append({
                    'type': 'exit',
                    'zscore': curr_z,
                    'index': i
                })
        
        return signals
    
    def find_best_pairs(
        self, price_data: pd.DataFrame, top_n: int = 5
    ) -> List[CointegratedPair]:
        """Find best cointegrated pairs from multiple assets."""
        assets = list(price_data.columns)
        pairs = []
        
        for i in range(len(assets)):
            for j in range(i + 1, len(assets)):
                asset1, asset2 = assets[i], assets[j]
                
                is_cointegrated, p_value, hedge_ratio = self.test_cointegration(
                    price_data[asset1], price_data[asset2]
                )
                
                if is_cointegrated and p_value < 0.01:
                    # Calculate half-life
                    spread = self.calculate_spread(
                        price_data[asset1], price_data[asset2], hedge_ratio
                    )
                    half_life = self.calculate_half_life(spread)
                    
                    pairs.append(CointegratedPair(
                        asset1=asset1,
                        asset2=asset2,
                        hedge_ratio=hedge_ratio,
                        intercept=0,
                        p_value=p_value,
                        half_life=half_life
                    ))
        
        # Sort by p-value (most cointegrated first) and half-life (faster mean-reversion preferred)
        pairs.sort(key=lambda x: (x.p_value, x.half_life))
        
        return pairs[:top_n]
    
    def rolling_cointegration_test(
        self, price1: pd.Series, price2: pd.Series, 
        window: int = 100
    ) -> pd.Series:
        """Perform rolling cointegration test."""
        p_values = pd.Series(index=price1.index, dtype=float)
        
        for i in range(window, len(price1)):
            window_price1 = price1.iloc[i-window:i]
            window_price2 = price2.iloc[i-window:i]
            
            _, p_value, _ = self.test_cointegration(window_price1, window_price2)
            p_values.iloc[i] = p_value
        
        return p_values

---

---

Pattern 2: Risk-Managed Trading Logic with Validation

from __future__ import annotations

import logging
from dataclasses import dataclass
from typing import Optional


logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class TradeSignal:
    """Immutable trade signal with all required validation constraints."""
    symbol: str
    side: str  # "buy" or "sell"
    price: float
    quantity: float
    confidence: float  # 0.0 to 1.0
    reason: str

    def validate(self) -> bool:
        """Validate that the trade signal meets all business constraints."""
        if self.quantity <= 0:
            raise ValueError(f"Quantity must be positive, got {self.quantity}")
        if self.price <= 0:
            raise ValueError(f"Price must be positive, got {self.price}")
        if not 0.0 <= self.confidence <= 1.0:
            raise ValueError(f"Confidence must be between 0 and 1, got {self.confidence}")
        return True


def generate_trade_signal(
    symbol: str,
    side: str,
    price: float,
    quantity: float,
    confidence: float,
    reason: str,
) -> TradeSignal:
    """Generate a validated trade signal with guard clause checks."""
    if side not in ("buy", "sell"):
        raise ValueError(f"Invalid side '{side}', must be 'buy' or 'sell'")

    signal = TradeSignal(
        symbol=symbol,
        side=side,
        price=price,
        quantity=quantity,
        confidence=confidence,
        reason=reason,
    )
    signal.validate()
    logger.info("Trade signal generated: %s %s %.4f @ %.2f (confidence=%.2f)",
                 symbol, side, quantity, price, confidence)
    return signal


def execute_with_risk_check(signal: TradeSignal, max_position_pct: float = 0.05) -> dict:
    """Execute a trade signal after applying risk management checks."""
    adjusted_quantity = signal.quantity
    if signal.side == "buy" and signal.quantity > max_position_pct:
        logger.warning("Position %s exceeds max %.1f%% — capping to %.4f",
                        signal.symbol, max_position_pct * 100, max_position_pct)
        adjusted_quantity = max_position_pct

    return {
        "symbol": signal.symbol,
        "side": signal.side,
        "price": signal.price,
        "quantity": adjusted_quantity,
        "capped": adjusted_quantity < signal.quantity,
        "confidence": signal.confidence,
        "status": "submitted",
    }

Constraints

MUST DO

• Implement indicator calculations using rolling windows with explicit lookback periods; never use full-history data for online indicators
• Validate signal generation by confirming alignment across multiple independent indicators before acting on a single signal
• Calculate all price-based indicators (SMA, EMA, RSI) on closing prices unless specifically designed for tick data
• Include proper handling of missing/NaN candles in indicator pipelines — forward-fill only within session boundaries
• Log signal generation with the full context window of indicator values that led to each signal

MUST NOT DO

• Do not use look-ahead bias: never reference future bars or prices when calculating indicators during backtesting
• Avoid recalculating all indicators from scratch on every tick — maintain running state for efficiency
• Never combine indicators with different timeframes without explicit resampling and clear documentation of the alignment logic
• Do not generate signals based on a single indicator crossover; require confirmation from price action or volume
• Avoid hardcoding parameter values (e.g., RSI period = 14) without testing regime-specific optima

Live References

Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.

• Statistical Arbitrage Overview
• Pairs Trading Strategies
• Cointegration in Trading
• Mean Reversion Trading Strategies
• Quantitative Arbitrage Methods