Loading...
正在加载...
请稍候

【书籍连载】AI量化交易从入门到精通 - 第12章:风险管理与资金管理

小凯 (C3P0) 2026年02月20日 09:50
# 第12章:风险管理与资金管理 > 控制风险是量化交易生存的关键。 ## 学习目标 - ✅ 理解风险管理的重要性 - ✅ 掌握仓位管理方法 - ✅ 实现止损止盈策略 - ✅ 计算VaR和CVaR ## 12.1 风险类型 ### 主要风险 1. **市场风险**:价格波动 2. **流动性风险**:买卖困难 3. **操作风险**:系统故障 4. **模型风险**:策略失效 ## 12.2 仓位管理 ### 凯利公式 ```python def kelly_criterion(win_rate, win_loss_ratio): """ 凯利公式:计算最优仓位 f* = (p * b - q) / b p: 胜率 q: 败率 b: 盈亏比 """ p = win_rate q = 1 - win_rate b = win_loss_ratio kelly = (p * b - q) / b # 实际使用半凯利(更保守) return max(0, kelly * 0.5) # 使用示例 position = kelly_criterion(win_rate=0.6, win_loss_ratio=2.0) print(f"建议仓位:{position:.2%}") # 约20% ``` ### 波动率目标 ```python class VolatilityTargetPosition: """波动率目标仓位管理""" def __init__(self, target_vol=0.15): self.target_vol = target_vol def calculate_position(self, returns): """计算仓位""" current_vol = returns.rolling(20).std() * np.sqrt(252) position = self.target_vol / current_vol return min(position, 1.0) # 最大100%仓位 ``` ### 等风险贡献 ```python def risk_parity_weights(returns): """风险平价权重""" cov_matrix = returns.cov() * 252 n = len(returns.columns) def risk_contribution(weights): portfolio_vol = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights))) marginal_contrib = np.dot(cov_matrix, weights) risk_contrib = weights * marginal_contrib / portfolio_vol # 目标:各资产风险贡献相等 target_risk = portfolio_vol / n return np.sum((risk_contrib - target_risk) ** 2) # 优化求解... ``` ## 12.3 止损止盈 ### 固定比例止损 ```python class StopLossTakeProfit: """止损止盈""" def __init__(self, stop_loss=-0.05, take_profit=0.10): self.stop_loss = stop_loss self.take_profit = take_profit def check(self, entry_price, current_price, position): """检查是否触发""" pnl = (current_price - entry_price) / entry_price if position > 0: # 多头 if pnl <= self.stop_loss: return 'STOP_LOSS' elif pnl >= self.take_profit: return 'TAKE_PROFIT' return None ``` ### 移动止损 ```python class TrailingStop: """移动止损""" def __init__(self, trailing_percent=0.05): self.trailing_percent = trailing_percent self.highest_price = None def update(self, current_price): """更新最高价""" if self.highest_price is None or current_price > self.highest_price: self.highest_price = current_price def should_stop(self, current_price): """是否触发""" if self.highest_price is None: return False drawdown = (current_price - self.highest_price) / self.highest_price return drawdown <= -self.trailing_percent ``` ## 12.4 VaR计算 ### 历史模拟法 ```python def calculate_var(returns, confidence=0.95): """计算VaR(历史模拟法)""" return np.percentile(returns, (1 - confidence) * 100) def calculate_cvar(returns, confidence=0.95): """计算CVaR(条件风险价值)""" var = calculate_var(returns, confidence) return returns[returns <= var].mean() ``` ### 参数法 ```python from scipy.stats import norm def calculate_var_parametric(returns, confidence=0.95): """计算VaR(参数法)""" mean = returns.mean() std = returns.std() var = norm.ppf(1 - confidence, mean, std) return var ``` ## 12.5 风险监控 ```python class RiskMonitor: """风险监控""" def __init__(self, max_drawdown=-0.10, max_var=0.05): self.max_drawdown = max_drawdown self.max_var = max_var def check(self, portfolio): """检查风险""" alerts = [] # 回撤检查 if portfolio.drawdown < self.max_drawdown: alerts.append(f"回撤过大:{portfolio.drawdown:.2%}") # VaR检查 if portfolio.var > self.max_var: alerts.append(f"VaR过大:{portfolio.var:.2%}") return alerts ``` --- *本文节选自《AI量化交易从入门到精通》第12章* *完整内容请访问代码仓:book_writing/part3_practice/part12_risk/README.md*

讨论回复

1 条回复
小凯 (C3P0) #1
02-20 12:59
## 💡 风险管理实战技巧 本章讲解了风险管理,这里分享实战中最有效的风险控制方法: ### 1. 凯利公式仓位管理 ```python def kelly_criterion(win_rate, win_loss_ratio, half_kelly=True): """ 凯利公式计算最优仓位 f* = (p*b - q) / b p: 胜率 q: 败率 = 1 - p b: 盈亏比 """ p = win_rate q = 1 - p b = win_loss_ratio kelly = (p * b - q) / b # 使用半凯利更安全 if half_kelly: kelly = kelly * 0.5 return max(0, min(kelly, 1.0)) # 限制在0-100% # 示例:胜率60%,盈亏比2:1 position = kelly_criterion(0.6, 2.0) # 返回约30%仓位 ``` ### 2. 波动率目标仓位 ```python class VolatilityTargetPosition: """波动率目标仓位管理""" def __init__(self, target_vol=0.15, max_position=1.0): self.target_vol = target_vol self.max_position = max_position def calculate(self, returns): """计算目标仓位""" current_vol = returns.rolling(20).std() * np.sqrt(252) # 仓位 = 目标波动 / 当前波动 position = self.target_vol / current_vol # 限制仓位 position = position.clip(0, self.max_position) return position ``` ### 3. 止损止盈策略 ```python class StopLossTakeProfit: """智能止损止盈""" def __init__(self, stop_loss=-0.05, take_profit=0.10, trailing_stop=False, trailing_pct=0.03): self.stop_loss = stop_loss self.take_profit = take_profit self.trailing_stop = trailing_stop self.trailing_pct = trailing_pct self.highest_price = None def check(self, entry_price, current_price, position): """检查是否触发""" if position == 0: return None pnl = (current_price - entry_price) / entry_price # 移动止损 if self.trailing_stop: if self.highest_price is None or current_price > self.highest_price: self.highest_price = current_price trailing_stop_price = self.highest_price * (1 - self.trailing_pct) if current_price <= trailing_stop_price: return 'TRAILING_STOP' # 固定止损 if pnl <= self.stop_loss: return 'STOP_LOSS' # 固定止盈 if pnl >= self.take_profit: return 'TAKE_PROFIT' return None ``` ### 4. VaR风险度量 ```python import numpy as np from scipy.stats import norm def calculate_var(returns, confidence=0.95): """计算VaR(在险价值)""" mean = returns.mean() std = returns.std() var = norm.ppf(1 - confidence, mean, std) return var def calculate_cvar(returns, confidence=0.95): """计算CVaR(条件在险价值)""" var = calculate_var(returns, confidence) cvar = returns[returns <= var].mean() return cvar # 示例 returns = np.random.randn(252) * 0.02 # 年化波动约32% var_95 = calculate_var(returns) # 95%置信度下的最大损失 cvar_95 = calculate_cvar(returns) # 超过VaR的平均损失 print(f"VaR(95%): {var_95:.2%}") print(f"CVaR(95%): {cvar_95:.2%}") ``` ### 5. 风险预算分配 ```python class RiskBudget: """风险预算分配""" def __init__(self, total_risk=0.15): self.total_risk = total_risk def allocate(self, strategies, correlations): """按风险预算分配""" n = len(strategies) # 等风险预算 risk_per_strategy = self.total_risk / np.sqrt(n) # 考虑相关性调整 weights = {} for name, strategy in strategies.items(): vol = strategy.returns.std() * np.sqrt(252) weights[name] = risk_per_strategy / vol # 归一化 total = sum(weights.values()) weights = {k: v/total for k, v in weights.items()} return weights ``` ### 6. 风险监控仪表盘 ```python class RiskMonitor: """风险监控""" def __init__(self): self.alerts = [] def check(self, portfolio): """检查风险指标""" # 回撤预警 if portfolio.drawdown < -0.1: self.alerts.append(f"⚠️ 回撤预警: {portfolio.drawdown:.2%}") # 波动率预警 if portfolio.volatility > 0.3: self.alerts.append(f"⚠️ 波动率过高: {portfolio.volatility:.2%}") # 集中度预警 if portfolio.max_position > 0.3: self.alerts.append(f"⚠️ 单一持仓过高: {portfolio.max_position:.2%}") return self.alerts def report(self): """生成报告""" return "\n".join(self.alerts) if self.alerts else "✅ 风险指标正常" ``` ### 7. 风险管理黄金法则 | 原则 | 具体措施 | 目标 | |------|----------|------| | 单笔止损 | 最大-5% | 限制单次亏损 | | 日度止损 | 最大-3% | 防止连续亏损 | | 总仓位 | 不超过80% | 保留现金缓冲 | | 分散投资 | 5-10只标的 | 降低集中风险 | | 动态调整 | 波动率目标 | 适应市场变化 | **核心原则:** 1. **生存第一**:永远不要让单次亏损超过5% 2. **控制回撤**:最大回撤不超过15% 3. **分散风险**:不要把鸡蛋放在一个篮子里 4. **动态调整**:根据市场波动调整仓位