""" 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)