research/cybernetic_ensemble.py

"""
Cybernetic Ensemble: PID + RL + Differentiable Optimization
Three control loops working together:

1. Inner loop (PID): milliseconds to seconds - volatility targeting
2. Middle loop (Differentiable Opt): daily - portfolio optimization  
3. Outer loop (Dreamer RL): weekly/monthly - strategy adaptation

This is the control hierarchy that Wiener imagined.
"""

import numpy as np
import pandas as pd
import torch
from typing import Dict, Tuple, Optional


class CyberneticPortfolioController:
    """
    Three-layer control system for portfolio management.
    
    Layer 1 (Fast): PID controls instant risk exposure
    Layer 2 (Medium): Differentiable optimizer sets target weights
    Layer 3 (Slow): Dreamer RL adapts the optimizer's parameters
    
    Each layer operates on a different timescale.
    Faster layers respond to immediate conditions.
    Slower layers learn from the faster layers' performance.
    """
    
    def __init__(
        self,
        n_assets: int,
        pid_config: dict = None,
        dreamer_agent = None,  # Your AgenticForecaster instance
        device: str = "cpu"
    ):
        self.n_assets = n_assets
        self.device = device
        
        # Layer 1: PID Controller (fastest)
        from .cybernetic import PIDController
        self.pid = PIDController(**(pid_config or {}))
        
        # Layer 2: Differentiable Optimizer weights will come from solver
        # We'll store the last optimization result
        self.last_opt_weights = None
        
        # Layer 3: Dreamer agent (slowest)
        self.dreamer = dreamer_agent
        
        # State tracking for homeostasis
        self.performance_buffer = []
        self.adaptation_step = 0
        
    def compute_weights(
        self,
        opt_weights: pd.Series,           # From your CVXPY solver
        portfolio_returns: pd.Series,     # Historical returns
        regime: Optional[str] = None,     # From HMM detector
        explore: bool = False
    ) -> pd.Series:
        """
        Three-stage control:
        1. Dreamer suggests adjustment to optimizer's parameters
        2. Optimizer produces target weights (already done by solver)
        3. PID scales weights based on realized volatility
        """
        
        # ─────────────────────────────────────────────────────────────
        # LAYER 3: Dreamer RL Adaptation (slow)
        # ─────────────────────────────────────────────────────────────
        
        if self.dreamer is not None and self.adaptation_step % 50 == 0:
            # Dreamer observes recent performance and suggests
            # adjustments to the optimization objective
            dreamer_adjustment = self._compute_dreamer_adjustment(portfolio_returns)
        else:
            dreamer_adjustment = 1.0
            
        # ─────────────────────────────────────────────────────────────
        # LAYER 2: Differentiable Optimization (medium)
        # Already handled by your solver.build_and_optimize()
        # We just store the result
        # ─────────────────────────────────────────────────────────────
        
        self.last_opt_weights = opt_weights.copy()
        
        # Apply Dreamer's learned adjustment to weights
        if dreamer_adjustment != 1.0:
            risky = opt_weights.drop(labels=['CASH'], errors='ignore')
            risky_adjusted = risky * dreamer_adjustment
            # Re-normalize
            risky_adjusted = risky_adjusted / risky_adjusted.sum()
            adjusted_weights = risky_adjusted.copy()
            adjusted_weights['CASH'] = 1.0 - risky_adjusted.sum()
        else:
            adjusted_weights = opt_weights.copy()
            
        # ─────────────────────────────────────────────────────────────
        # LAYER 1: PID Control (fastest)
        # ─────────────────────────────────────────────────────────────
        
        if len(portfolio_returns) > self.pid.lookback_days:
            leverage = self.pid.compute_leverage(portfolio_returns)
            
            # Adjust target volatility based on regime (outer homeostasis)
            if regime and hasattr(self.pid, 'target_vol'):
                from .cybernetic import AdaptiveRiskController
                adaptive = AdaptiveRiskController()
                self.pid.target_vol = adaptive.get_target_vol(regime)
                
            final_weights = self.pid.apply_to_weights(adjusted_weights, leverage)
        else:
            final_weights = adjusted_weights
            leverage = 1.0
            
        # Store performance for Dreamer's next adaptation
        self.performance_buffer.append({
            'step': self.adaptation_step,
            'leverage': leverage,
            'dreamer_adj': dreamer_adjustment,
            'regime': regime
        })
        
        # Trim buffer
        if len(self.performance_buffer) > 1000:
            self.performance_buffer = self.performance_buffer[-500:]
            
        self.adaptation_step += 1
        
        return final_weights
    
    def _compute_dreamer_adjustment(self, returns: pd.Series) -> float:
        """
        Dreamer observes recent returns and outputs a scalar adjustment
        to the optimizer's risk aversion or expected returns.
        
        This is where RL learns to override the optimizer
        when it's consistently wrong.
        """
        if self.dreamer is None:
            return 1.0
            
        with torch.no_grad():
            # Convert recent returns to observation tensor
            # This assumes Dreamer is already trained
            obs = self._returns_to_observation(returns)
            obs_tensor = torch.FloatTensor(obs).to(self.device)
            
            # Get Dreamer's action (portfolio weights suggestion)
            if hasattr(self.dreamer, 'act'):
                # Use Dreamer's policy
                action, _ = self.dreamer.act(
                    obs_tensor.unsqueeze(0), 
                    self.dreamer.rssm.initial_state(1, self.device),
                    torch.zeros(1, self.dreamer.action_dim).to(self.device)
                )
                # Convert action to adjustment scalar
                # Action is portfolio weights; we extract implied risk level
                adjustment = float(action.mean().cpu().numpy())
                # Map from [0,1] to [0.5, 1.5]
                adjustment = 0.5 + adjustment
            else:
                adjustment = 1.0
                
        return np.clip(adjustment, 0.5, 1.5)
    
    def _returns_to_observation(self, returns: pd.Series) -> np.ndarray:
        """Convert return series to Dreamer observation format"""
        # Simple feature engineering
        obs = np.array([
            returns.mean(),
            returns.std(),
            returns.skew(),
            returns.kurtosis(),
            returns.iloc[-1],  # last return
            returns.iloc[-5:].mean(),  # 5-day mean
            returns.iloc[-21:].mean(),  # 21-day mean
        ])
        return obs
    
    def get_control_state(self) -> dict:
        """Diagnostic information for debugging"""
        return {
            'pid_target_vol': self.pid.target_vol,
            'pid_integral': self.pid.integral,
            'pid_prev_error': self.pid.prev_error,
            'adaptation_step': self.adaptation_step,
            'recent_leverages': [p['leverage'] for p in self.performance_buffer[-10:]] if self.performance_buffer else [],
        }


class MetaController:
    """
    The highest level of control: adjusts the Cybernetic Ensemble itself.
    
    This implements Ashby's Law of Requisite Variety:
    "Only variety can absorb variety."
    
    If the market becomes more complex, this controller adds complexity
    to the control system (more PID terms, different RL hyperparameters).
    """
    
    def __init__(self, base_controller: CyberneticPortfolioController):
        self.controller = base_controller
        self.performance_history = []
        self.complexity_level = 1  # 1=simple, 5=complex
        
    def monitor_and_adapt(self, portfolio_value: float, benchmark_value: float):
        """
        Track tracking error between portfolio and benchmark.
        If error grows beyond threshold, increase control complexity.
        """
        tracking_error = abs(portfolio_value - benchmark_value) / benchmark_value
        
        self.performance_history.append(tracking_error)
        
        # Rolling average of last 20 days
        if len(self.performance_history) > 20:
            recent_error = np.mean(self.performance_history[-20:])
            
            if recent_error > 0.05 and self.complexity_level < 5:
                self.complexity_level += 1
                self._increase_complexity()
            elif recent_error < 0.01 and self.complexity_level > 1:
                self.complexity_level -= 1
                self._decrease_complexity()
                
    def _increase_complexity(self):
        """Add more sophisticated control"""
        print(f"Increasing control complexity to level {self.complexity_level}")
        
        # Example: Make PID more aggressive
        self.controller.pid.kp *= 1.2
        self.controller.pid.ki *= 1.1
        
        # Example: Reduce Dreamer's exploration
        if hasattr(self.controller.dreamer, 'actor'):
            # Reduce exploration noise
            pass
            
    def _decrease_complexity(self):
        """Simplify control when market is calm"""
        print(f"Decreasing control complexity to level {self.complexity_level}")
        
        # Return to baseline
        self.controller.pid.kp = max(1.0, self.controller.pid.kp / 1.2)
        self.controller.pid.ki = max(0.3, self.controller.pid.ki / 1.1)