"""drawdown_control.py — Dynamic Drawdown Control & Position Sizing Implements CPPI (Constant Proportion Portfolio Insurance), fractional Kelly criterion, dynamic leverage based on current drawdown, and volatility targeting. Essential for survival-first quant strategies. References: - Perold & Sharpe 1988: "Dynamic Strategies for Asset Allocation" (CPPI) - Thorp 2006: "The Kelly Criterion in Blackjack, Sports Betting, and the Stock Market" - Grossman & Zhou 1993: "Optimal Investment Strategies for Controlling Drawdowns" """ import numpy as np, pandas as pd class DrawdownController: """Controls drawdown via CPPI, Kelly, and dynamic leverage.""" def __init__(self, max_dd=0.15, target_vol=0.10, kelly_fraction=0.5): self.max_dd = max_dd self.target_vol = target_vol self.kelly_frac = kelly_fraction def cppi_weights(self, prices, floor_ratio=0.8, multiplier=3.0): """CPPI: allocate to risky asset based on cushion above floor. Returns equity exposure fraction [0, 1]. """ nav = prices / prices.iloc[0] peak = nav.expanding().max() floor = peak * floor_ratio cushion = nav - floor exposure = multiplier * cushion / nav return np.clip(exposure, 0, 1.0) def kelly_leverage(self, returns, window=252): """Fractional Kelly optimal leverage. f* = mu / sigma^2 (full Kelly) f = fraction * f* (half Kelly = safer) """ r = returns.dropna() if len(r) < 30: return 0.5 mu = r.tail(window).mean() * 252 sigma2 = (r.tail(window).std() * np.sqrt(252)) ** 2 if sigma2 < 1e-10: return 0.5 f_star = mu / sigma2 return np.clip(self.kelly_frac * f_star, 0.0, 2.0) def dynamic_leverage(self, prices, returns, current_dd=None): """Dynamic leverage: reduce when in drawdown, increase when at peak.""" nav = prices / prices.iloc[0] peak = nav.expanding().max() dd = (nav - peak) / peak if current_dd is None: current_dd = dd.iloc[-1] # Vol targeting vol = returns.tail(63).std() * np.sqrt(252) vol_scalar = self.target_vol / (vol + 1e-10) # Drawdown guard dd_scalar = max(0, 1 - abs(current_dd) / self.max_dd) return np.clip(vol_scalar * dd_scalar, 0.0, 2.0) def position_size(self, capital, atr, risk_per_trade=0.01): """ATR-based position sizing (Turtle-style). Position = (Capital * Risk%) / ATR """ if atr <= 0: return 0 return capital * risk_per_trade / atr def optimal_f(self, returns, window=100): """Optimal-F (Ralph Vince): geometric growth maximizing fraction. f = argmax G(f) where G(f) = product(1 + f * (-W/R)) """ r = returns.tail(window).dropna() if len(r) < 20: return 0.25 # Simplified: use Kelly as proxy mu = r.mean(); sigma = r.std() if sigma < 1e-10: return 0.25 return np.clip(mu / (sigma ** 2 + 1e-10), 0.0, 1.0) def report(self, prices, returns): """Generate position sizing recommendations.""" nav = prices / prices.iloc[0] current_dd = (nav.iloc[-1] - nav.expanding().max().iloc[-1]) / nav.expanding().max().iloc[-1] cppi = self.cppi_weights(prices) kelly = self.kelly_leverage(returns) dyn = self.dynamic_leverage(prices, returns, current_dd) optf = self.optimal_f(returns) vol = returns.tail(63).std() * np.sqrt(252) return f"""## Position Sizing & Drawdown Control | Metric | Value | |--------|-------| | Current Drawdown | {current_dd*100:.1f}% | | Target Max Drawdown | {self.max_dd*100:.1f}% | | Current Volatility (3M ann) | {vol*100:.1f}% | | Target Volatility | {self.target_vol*100:.1f}% | **Recommended Leverage / Exposure:** | Strategy | Exposure | Rationale | |----------|----------|-----------| | CPPI (multiplier=3) | {cppi.iloc[-1]*100:.0f}% | Cushion-based insurance | | Half-Kelly | {kelly*100:.0f}% | Growth-optimal, halved for safety | | Vol-Target + DD-Guard | {dyn*100:.0f}% | Scales with vol and drawdown | | Optimal-F | {optf*100:.0f}% | Geometric growth maximizing | **Composite Recommendation:** {(cppi.iloc[-1] * 0.3 + kelly * 0.3 + dyn * 0.4)*100:.0f}% equity exposure """ if __name__ == '__main__': np.random.seed(42) returns = pd.Series(np.random.normal(0.0005, 0.015, 500), index=pd.date_range('2022-01-01', periods=500, freq='B')) prices = (1 + returns).cumprod() ctrl = DrawdownController(max_dd=0.15, target_vol=0.10, kelly_fraction=0.5) print(ctrl.report(prices, returns))