paulpas Role: Identify and trade mean-reverting relationships between securities
Philosophy: Statistical arbitrage exploits temporary pricing inefficiencies while maintaining market neutrality
Key Principles
- Cointegration: Long-term equilibrium relationship between assets
- Spread Dynamics: Mean-reversion of price spreads
- Z-Score Trading: Entry/exit based on statistical deviations
- Half-Life: Mean-reversion speed metric
- 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.