回测与风险管理

回测和风险管理是量化投资中最重要的两个环节——一个验证策略是否有效，一个确保策略不会让你爆仓。

---

为什么需要回测？

回测就是用历史数据模拟策略的交易过程，验证策略的表现。

不经过回测就实盘 = 拿真金白银赌博。

好的回测需要验证：

• 策略在历史上各种市场环境中表现如何？
• 最大亏损是多少？你能承受吗？
• 策略在最近几年是否失效？

---

通用回测框架

import pandas as pd
import numpy as np

def backtest_strategy(df, initial_capital=100000, commission=0.0003):
    """
    通用策略回测框架
    
    参数:
        df: 包含 close 和 signal 列的 DataFrame
            signal: 持仓信号 (0=空仓, 1=持有)
        initial_capital: 初始资金
        commission: 手续费率
    """
    df = df.copy()
    
    capital = initial_capital
    position = 0        # 持股数量
    trades = []          # 交易记录
    portfolio_values = []  # 每日市值
    
    for i in range(1, len(df)):
        date = df.index[i]
        price = df['close'].iloc[i]
        signal = df['signal'].iloc[i]
        prev_signal = df['signal'].iloc[i - 1]
        
        # 买入信号：从空仓变持仓
        if signal == 1 and prev_signal == 0 and position == 0:
            shares = int(capital * (1 - commission) // price)
            if shares > 0:
                cost = shares * price * (1 + commission)
                capital -= cost
                position = shares
                trades.append({
                    'date': date, 'type': 'BUY',
                    'price': price, 'shares': shares, 'cost': cost
                })
        
        # 卖出信号：从持仓变空仓
        elif signal == 0 and prev_signal == 1 and position > 0:
            revenue = position * price * (1 - commission)
            capital += revenue
            trades.append({
                'date': date, 'type': 'SELL',
                'price': price, 'shares': position, 'revenue': revenue
            })
            position = 0
        
        # 记录每日市值
        total_value = capital + position * price
        portfolio_values.append({'date': date, 'value': total_value})
    
    # 最后强平
    if position > 0:
        last_price = df['close'].iloc[-1]
        capital += position * last_price
    
    # 生成回测报告
    portfolio = pd.DataFrame(portfolio_values).set_index('date')
    final_value = capital
    
    return {
        'portfolio': portfolio,
        'trades': pd.DataFrame(trades),
        'final_capital': final_value,
        'total_return': (final_value - initial_capital) / initial_capital
    }

---

回测绩效指标

def calculate_performance_metrics(portfolio_df, risk_free_rate=0.02):
    """
    计算策略绩效指标
    """
    df = portfolio_df.copy()
    
    # 日收益率
    df['daily_return'] = df['value'].pct_change()
    
    # 1. 累计收益率
    total_return = df['value'].iloc[-1] / df['value'].iloc[0] - 1
    
    # 2. 年化收益率
    n_days = len(df)
    annual_return = (1 + total_return) ** (252 / n_days) - 1
    
    # 3. 年化波动率
    annual_volatility = df['daily_return'].std() * np.sqrt(252)
    
    # 4. 夏普比率
    excess_returns = df['daily_return'] - risk_free_rate / 252
    sharpe_ratio = np.sqrt(252) * excess_returns.mean() / df['daily_return'].std()
    
    # 5. 最大回撤
    cummax = df['value'].cummax()
    drawdown = (df['value'] - cummax) / cummax
    max_drawdown = drawdown.min()
    
    # 6. 卡尔玛比率（收益/最大回撤）
    calmar_ratio = annual_return / abs(max_drawdown) if max_drawdown != 0 else np.inf
    
    # 7. 胜率
    win_rate = (df['daily_return'] > 0).sum() / df['daily_return'].dropna().shape[0]
    
    print("=" * 50)
    print("策略绩效报告")
    print("=" * 50)
    print(f"累计收益率:   {total_return*100:>8.2f}%")
    print(f"年化收益率:   {annual_return*100:>8.2f}%")
    print(f"年化波动率:   {annual_volatility*100:>8.2f}%")
    print(f"夏普比率:     {sharpe_ratio:>8.2f}")
    print(f"最大回撤:     {max_drawdown*100:>8.2f}%")
    print(f"卡尔玛比率:   {calmar_ratio:>8.2f}")
    print(f"胜率:         {win_rate*100:>8.2f}%")
    print("=" * 50)
    
    return {
        'total_return': total_return,
        'annual_return': annual_return,
        'annual_volatility': annual_volatility,
        'sharpe_ratio': sharpe_ratio,
        'max_drawdown': max_drawdown,
        'calmar_ratio': calmar_ratio,
        'win_rate': win_rate
    }

---

风险管理体系

1. 仓位管理

def calculate_position_size(capital, price, risk_per_trade=0.02, 
                             stop_loss_pct=0.05):
    """
    基于风险的仓位计算
    
    参数:
        capital: 总资金
        price: 当前价格
        risk_per_trade: 单笔交易最大亏损比例（如 2%）
        stop_loss_pct: 止损百分比（如 5%）
    
    返回: 建议的买入股数
    """
    max_loss_amount = capital * risk_per_trade  # 最大亏损金额
    loss_per_share = price * stop_loss_pct      # 每股最大亏损
    shares = int(max_loss_amount / loss_per_share)
    
    return shares

2. 止损止盈规则

def apply_stop_loss_take_profit(df, stop_loss=-0.10, take_profit=0.30):
    """
    应用止损止盈规则
    
    参数:
        stop_loss: 止损线（如 -10%）
        take_profit: 止盈线（如 +30%）
    """
    df = df.copy()
    df['cum_return'] = (1 + df['strategy_return']).cumprod()
    df['peak'] = df['cum_return'].cummax()
    df['drawdown'] = (df['cum_return'] - df['peak']) / df['peak']
    
    # 止损：从最高点回撤超过止损线
    df.loc[df['drawdown'] < stop_loss, 'signal'] = 0
    
    # 止盈：从入场点涨幅超过止盈线（需要跟踪入场点）
    # ... 具体实现依赖于如何跟踪入场价
    
    return df

3. 风险管理核心原则

原则	说明
单笔风险上限	任何单笔交易的亏损不超过总资金的 2%
总风险上限	所有持仓的合计风险敞口不超过总资金的 25%
止损必设	每笔交易必须有明确的止损价位
分散化	不把所有资金集中在单只股票或单一策略
杠杆控制	初学者不使用杠杆；经验者杠杆不超过 2 倍

---

回测的常见陷阱

1. 过度拟合（Overfitting）

策略在历史数据上表现极好，但参数太多太复杂，实际上是拟合了噪声而非规律。

预防方法：

• 保持策略简单（参数不超过 3-5 个）
• 用不同的时间段验证（样本外测试）
• 关注夏普比率，而非累计收益率

2. 幸存者偏差

只用了现在还存在的股票数据（退市的股票被排除），导致回测结果失真。

3. 未来函数

在回测中使用了未来才知道的信息。例如：用"当天收盘价"决定"当天开盘时"的交易。

4. 忽略交易成本

手续费、滑点、冲击成本可能占策略收益的 20-50%。

---

学习路径总结

完成本课程的学习后，你已经掌握了：

1. ✅ 金融市场基础知识和交易机制
2. ✅ 基本面和技术面分析方法
3. ✅ 量化投资的核心理念和策略框架
4. ✅ Python 数据科学工具（NumPy、Pandas、Matplotlib）
5. ✅ 经典量化策略实现（双均线、因子选股、均值回归）
6. ✅ 策略回测和风险管理方法

---

下一步学习建议

• 📖 进阶回测框架：Backtrader 或 Zipline
• 📊 更多数据源：AKShare、OpenBB
• 🤖 机器学习策略：将 ML 模型集成到选股和择时中
• 🔗 实盘对接：使用券商 API 实现自动化交易

💡 量化投资有风险，入市需谨慎。本课程内容仅供学习参考，不构成投资建议。