research/cybernetic.py

"""
Cybernetic Control Systems for Portfolio Management
PID Controller, Homeostasis, and Hybrid RL-Control
"""

import numpy as np
import pandas as pd
from collections import deque
from dataclasses import dataclass, field


@dataclass
class PIDController:
    """
    Proportional-Integral-Derivative controller for volatility targeting.
    
    How it works:
    1. Measure current volatility (20-day rolling)
    2. Compare to target (e.g., 15% annualized)
    3. Compute error = target - current
    4. Adjust portfolio leverage based on P, I, D terms
    
    The genius: This works even when return predictions are wrong.
    Volatility is much more predictable than returns.
    """
    
    target_vol: float = 0.15  # 15% annualized target
    kp: float = 2.0           # Proportional gain (responds to current error)
    ki: float = 0.5           # Integral gain (corrects persistent bias)
    kd: float = 0.3           # Derivative gain (anticipates future error)
    min_leverage: float = 0.3  # Never go below 30% exposure
    max_leverage: float = 1.5   # Never exceed 150% exposure
    lookback_days: int = 21    # Rolling window for volatility
    
    def __post_init__(self):
        self.integral = 0.0
        self.prev_error = 0.0
        self.vol_history = deque(maxlen=self.lookback_days)
        self.trading_days = 252
        
    def current_volatility(self, returns: pd.Series) -> float:
        """Calculate annualized volatility from recent returns"""
        if len(returns) < self.lookback_days:
            return self.target_vol
            
        recent = returns.iloc[-self.lookback_days:]
        daily_vol = recent.std()
        return daily_vol * np.sqrt(self.trading_days)
    
    def compute_leverage(self, portfolio_returns: pd.Series) -> float:
        """
        Returns a leverage multiplier (e.g., 0.8 means reduce exposure by 20%)
        
        PID formula:
        leverage = 1 + kp * error + ki * integral(error) + kd * derivative(error)
        """
        current_vol = self.current_volatility(portfolio_returns)
        
        # Error = target - actual (positive = need MORE risk)
        error = self.target_vol - current_vol
        
        # Proportional term: immediate response
        p = self.kp * error
        
        # Integral term: accumulates persistent errors
        # Prevents the system from constantly lagging
        self.integral += error * 0.1  # 0.1 = time constant
        # Anti-windup: clamp integral to prevent explosion
        self.integral = np.clip(self.integral, -0.5, 0.5)
        i = self.ki * self.integral
        
        # Derivative term: anticipates where error is going
        d = self.kd * (error - self.prev_error)
        self.prev_error = error
        
        # Base leverage = 1.0 (normal exposure)
        leverage = 1.0 + p + i + d
        
        # Clamp to safe bounds
        leverage = np.clip(leverage, self.min_leverage, self.max_leverage)
        
        return float(leverage)
    
    def apply_to_weights(self, weights: pd.Series, leverage: float) -> pd.Series:
        """
        Scale all risky weights by leverage factor.
        Cash absorbs the difference.
        """
        risky = weights.drop(labels=['CASH'], errors='ignore').copy()
        cash_weight = weights.get('CASH', 0.0)
        
        # Scale risky assets
        risky_scaled = risky * leverage
        
        # Adjust cash to maintain sum = 1.0
        new_cash = 1.0 - risky_scaled.sum()
        
        # Preserve direction (shorts stay short)
        result = risky_scaled.copy()
        result['CASH'] = new_cash
        
        return result


@dataclass
class AdaptiveRiskController:
    """
    Homeostatic risk controller with multiple setpoints.
    
    Different market regimes have different volatility targets.
    This creates a nested control loop:
    - Inner loop: PID targets current volatility
    - Outer loop: Adjusts target based on VIX/regime
    """
    
    base_target: float = 0.15
    
    # Regime-specific targets (lower vol in crisis)
    regime_multipliers = field(default_factory=lambda: {
        "Bull / Low Volatility": 1.2,    # 18% target (take more risk)
        "Normal / Chop": 1.0,            # 15% target
        "Crash / High Volatility": 0.5,  # 7.5% target (protect capital)
    })
    
    def get_target_vol(self, regime: str) -> float:
        multiplier = self.regime_multipliers.get(regime, 1.0)
        return self.base_target * multiplier