💡 风险管理实战技巧

本章讲解了风险管理，这里分享实战中最有效的风险控制方法：

1. 凯利公式仓位管理

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. 波动率目标仓位

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. 止损止盈策略

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风险度量

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. 风险预算分配

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. 风险监控仪表盘

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. 风险管理黄金法则