第四章：动量因子回测框架搭建

4.1 回测框架的核心思想

做量化交易这些年，我见过太多人一上来就写回测代码。结果呢？回测曲线漂亮得不行，实盘一跑就崩。说白了，回测框架不是让你自我陶醉的，它是帮你发现问题的。

我个人习惯把回测框架拆成三块：

• 数据层：行情数据的获取、清洗、对齐
• 策略层：信号生成、仓位管理、交易执行
• 分析层：绩效统计、风险归因、对比分析

你想想看，如果数据层就有问题，后面再漂亮的策略也是白搭。我在项目中遇到过好几次，因为复权数据没处理好，回测结果完全失真。

4.2 数据准备与清洗

动量因子回测，数据是地基。我一般用日线数据，频率太低信号滞后，频率太高噪声太多。日线刚刚好。

先看数据获取的代码：

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def load_stock_data(symbol, start_date, end_date):
    """
    加载股票日线数据
    我习惯用本地CSV文件，速度快，不受网络影响
    """
    df = pd.read_csv(f'data/{symbol}.csv', parse_dates=['date'])
    df = df[(df['date'] >= start_date) & (df['date'] <= end_date)]
    df = df.sort_values('date').reset_index(drop=True)
    
    # 数据清洗：去掉停牌日和涨跌停日
    # 嗯，这里要注意，涨跌停日的价格是失真的
    df = df[df['volume'] > 0]  # 成交量不为0
    df = df[df['close'] != df['pre_close']]  # 排除涨跌停
    
    return df

数据清洗这一步，我曾经吃过亏。有一次回测一个动量策略，收益高得离谱。后来发现是因为包含了新股上市首日的数据——那天的涨幅是44%，动量信号全被带偏了。

4.3 动量因子计算

动量因子有很多种定义方式。我个人最常用的是过去N日的累计收益率，简单有效。但要注意，动量因子对计算周期很敏感。

来看核心计算逻辑：

def calculate_momentum_factor(df, lookback=20):
    """
    计算动量因子
    lookback: 回溯期，我一般用20个交易日（约1个月）
    """
    df = df.copy()
    
    # 计算过去N日的累计收益率
    df['momentum'] = df['close'].pct_change(lookback)
    
    # 剔除极端值（我习惯用3倍标准差）
    mean = df['momentum'].mean()
    std = df['momentum'].std()
    df['momentum'] = df['momentum'].clip(mean - 3*std, mean + 3*std)
    
    # 标准化处理，方便不同股票之间比较
    df['momentum_zscore'] = (df['momentum'] - mean) / std
    
    return df

为什么会做标准化？你想想看，茅台涨5%和小盘股涨5%，背后的动量含义完全不同。标准化后，我们才能公平地比较不同股票的动量强度。

4.4 回测引擎实现

回测引擎是整个框架的心脏。我设计的原则是：事件驱动 + 向量化计算。事件驱动能模拟真实交易，向量化计算保证速度。

先看框架结构图：

回测引擎的核心代码：

class MomentumBacktest:
    def __init__(self, data, lookback=20, top_pct=0.2):
        """
        动量因子回测引擎
        data: 多只股票的DataFrame，列包括 date, symbol, close
        lookback: 动量计算周期
        top_pct: 选股比例（前20%）
        """
        self.data = data
        self.lookback = lookback
        self.top_pct = top_pct
        self.positions = None
        self.returns = None
        
    def generate_signals(self):
        """生成交易信号"""
        # 计算每只股票的动量因子
        self.data['momentum'] = self.data.groupby('symbol')['close'].transform(
            lambda x: x.pct_change(self.lookback)
        )
        
        # 每月调仓：取动量最强的top_pct
        # 我习惯用月末最后一个交易日作为调仓日
        monthly = self.data[self.data['date'].dt.is_month_end].copy()
        monthly['rank'] = monthly.groupby('date')['momentum'].rank(ascending=False)
        monthly['signal'] = (monthly['rank'] <= monthly['rank'].quantile(1 - self.top_pct)).astype(int)
        
        self.signals = monthly[['date', 'symbol', 'signal']]
        return self.signals
    
    def run_backtest(self):
        """执行回测"""
        signals = self.generate_signals()
        
        # 计算每日持仓
        # 这里用了前向填充，模拟持有到下个调仓日
        daily_signals = signals.set_index('date').groupby('symbol').resample('D').ffill()
        daily_signals = daily_signals.dropna().reset_index()
        
        # 合并收益率
        merged = pd.merge(daily_signals, self.data, on=['date', 'symbol'])
        merged['daily_return'] = merged['close'].pct_change()
        merged['strategy_return'] = merged['signal'] * merged['daily_return']
        
        # 组合收益（等权配置）
        portfolio = merged.groupby('date')['strategy_return'].mean()
        
        self.portfolio_returns = portfolio
        return portfolio

4.5 绩效评估与风险分析

回测跑完了，别急着高兴。绩效评估才是见真章的时候。我一般看这几个指标：

指标	计算公式	我的经验阈值
年化收益率	累计收益^(252/N) - 1	15%以上算合格
夏普比率	(年化收益 - 无风险利率) / 年化波动率	1.5以上才敢实盘
最大回撤	峰值到谷底的最大跌幅	不超过20%
胜率	盈利天数 / 总交易天数	55%以上
盈亏比	平均盈利 / 平均亏损	2:1以上

绩效评估代码：

def evaluate_performance(returns, risk_free_rate=0.03):
    """
    回测绩效评估
    returns: 策略日收益率序列
    """
    # 年化收益率
    total_return = (1 + returns).prod() - 1
    n_years = len(returns) / 252
    annual_return = (1 + total_return) ** (1 / n_years) - 1
    
    # 年化波动率
    annual_vol = returns.std() * np.sqrt(252)
    
    # 夏普比率
    sharpe = (annual_return - risk_free_rate) / annual_vol
    
    # 最大回撤
    cumulative = (1 + returns).cumprod()
    running_max = cumulative.expanding().max()
    drawdown = (cumulative - running_max) / running_max
    max_drawdown = drawdown.min()
    
    # 胜率
    win_rate = (returns > 0).mean()
    
    # 盈亏比
    avg_win = returns[returns > 0].mean()
    avg_loss = abs(returns[returns < 0].mean())
    profit_loss_ratio = avg_win / avg_loss if avg_loss != 0 else np.inf
    
    results = {
        '年化收益率': f'{annual_return:.2%}',
        '年化波动率': f'{annual_vol:.2%}',
        '夏普比率': f'{sharpe:.2f}',
        '最大回撤': f'{max_drawdown:.2%}',
        '胜率': f'{win_rate:.2%}',
        '盈亏比': f'{profit_loss_ratio:.2f}'
    }
    
    return results

4.6 回测中的常见陷阱

做回测这么多年，我踩过的坑可以写本书了。这里挑几个最常见的：

1. 幸存者偏差：只用现在还在交易的股票做回测。那些退市的、暴跌的都被忽略了。我习惯用全量历史数据，包括已退市的股票。
2. 交易成本忽略：动量策略换手率高，佣金和滑点影响巨大。我一般按双边千分之三估算成本。
3. 未来函数：用未来的数据计算当前的信号。比如用当天的收盘价计算当天的收益率。这个必须严格避免。
4. 参数过拟合：在历史数据上反复调参，直到曲线完美。这样的策略一到实盘就露馅。

4.7 完整回测流程示例

最后，我把整个流程串起来，给你看一个完整的回测示例：

# 加载数据
symbols = ['000001.SZ', '600036.SH', '601318.SH']  # 平安银行、招商银行、中国平安
all_data = []
for sym in symbols:
    df = load_stock_data(sym, '2018-01-01', '2023-12-31')
    all_data.append(df)

data = pd.concat(all_data, ignore_index=True)

# 初始化回测引擎
bt = MomentumBacktest(data, lookback=20, top_pct=0.2)

# 执行回测
returns = bt.run_backtest()

# 绩效评估
results = evaluate_performance(returns)
for key, value in results.items():
    print(f'{key}: {value}')

# 输出示例：
# 年化收益率: 18.35%
# 年化波动率: 22.10%
# 夏普比率: 1.62
# 最大回撤: -15.43%
# 胜率: 56.78%
# 盈亏比: 2.15

你看，一个完整的动量因子回测框架，其实没那么复杂。关键是每个环节都要严谨，别偷懒。数据清洗多花10分钟，可能帮你省下10天的无效调试时间。

框架搭好了，后面的事情就顺了。你可以换不同的动量定义、调不同的参数、加不同的风控规则。但记住：框架的可靠性永远比策略的收益率更重要。