| """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 = returns.tail(63).std() * np.sqrt(252) |
| vol_scalar = self.target_vol / (vol + 1e-10) |
| |
| 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 |
| |
| 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)) |
|
|
