量化回测数据复权实战：3种复权方法对比与Python代码实现

在量化交易的世界里，数据质量直接决定了策略回测的可靠性。当我们分析一只股票的历史走势时，经常会发现价格图表上出现突然的"断崖式"下跌或上涨，这通常是由于公司进行了分红、送股或配股等行为导致的除权除息。如果不进行适当处理，这些价格断层会严重扭曲技术指标，导致回测结果失真。本文将深入探讨三种主流复权方法，并提供可直接用于生产的Python实现代码。

1. 复权基础：为什么我们需要处理价格断层

想象一下，你正在回测一个基于均线交叉的策略。某只股票在实施"10送10"分红方案后，股价从20元直接变为10元，但你的持股数量也相应增加了一倍。如果不做任何处理，策略会误判为股价暴跌50%，从而可能错误触发止损。这就是复权要解决的核心问题。

复权的本质是通过数学调整，消除因公司行为导致的价格不连续，还原真实的投资回报曲线。从工程角度看，复权处理需要解决三个关键问题：

1. 价格连续性 ：确保K线走势真实反映投资回报，避免技术指标失真
2. 成本基准统一 ：使不同时期的买入成本具有可比性
3. 交易模拟准确性 ：回测中的买卖价格必须与实际交易价格一致

在实际操作中，我们会遇到三种价格类型：

• 原始价格 ：交易所实际成交的价格，包含所有除权缺口
• 前复权价格 ：以最新价格为基准，向前调整历史价格
• 后复权价格 ：以上市首日为基准，向后调整最新价格

# 基础复权因子计算公式示例
def calculate_adjust_factor(div_cash, div_stock, last_close):
    """
    计算复权因子
    :param div_cash: 每股现金分红(元)
    :param div_stock: 每股送转股数(股)
    :param last_close: 除权前一日收盘价(元)
    :return: 复权因子
    """
    return (last_close - div_cash) / (last_close * (1 + div_stock))

2. 三种复权方法深度对比

2.1 前复权：以现在看过去

前复权是最常用的复权方式，其核心特点是保持当前价格不变，调整历史价格。这种方法的优势在于：

• 直观反映当前成本 ：最新价格与市场实时报价一致
• 技术分析友好 ：均线、MACD等指标连续性好
• 回测实操性强 ：买卖信号基于实际交易价格

但前复权也有明显缺陷：长期使用会导致早期价格被过度压缩，甚至出现负值（如贵州茅台前复权价格在2005年为-30元）。

# 前复权价格计算示例
def forward_adjust(price_series, adjust_series):
    """
    前复权计算
    :param price_series: 原始价格序列(Series)
    :param adjust_series: 复权因子序列(Series)
    :return: 前复权价格序列
    """
    latest_adjust = adjust_series.iloc[-1]
    return price_series * latest_adjust / adjust_series

2.2 后复权：从过去看现在

后复权采用相反的思路——保持上市首日价格不变，调整后续所有价格。这种方法特别适合：

• 长期收益分析 ：清晰展示股票自上市以来的真实涨幅
• 价值投资评估 ：消除分红再投资的影响，反映完整回报
• 学术研究 ：便于计算长期复合收益率

后复权的缺点是当前价格可能与市场报价差异巨大（如万科A后复权价格已超过5000元），不适用于短线交易策略。

# 后复权价格计算示例
def backward_adjust(price_series, adjust_series):
    """
    后复权计算
    :param price_series: 原始价格序列(Series)
    :param adjust_series: 复权因子序列(Series)
    :return: 后复权价格序列
    """
    base_adjust = adjust_series.iloc[0]
    return price_series * base_adjust / adjust_series

2.3 动态复权：回测专用解决方案

动态复权是量化领域的创新方法，以前复权为基础，但以回测当前日期为基准点动态调整。这种机制完美解决了传统前复权的"未来函数"问题：

特性	动态复权	传统前复权
基准点	回测当日	最新日期
是否含未来信息	否	是
价格范围	合理	可能负值
适用场景	回测	实时分析

# 动态复权实现核心逻辑
def dynamic_adjust(df):
    """
    动态复权计算(DataFrame版)
    :param df: 包含date,close,adjust_factor的DataFrame
    :return: 动态复权价格Series
    """
    return df.apply(
        lambda row: row['close'] * row['adjust_factor'] / df[df['date']<=row['date']]['adjust_factor'].iloc[-1],
        axis=1
    )

3. 实战：基于Baostock的完整复权流程

下面我们通过Baostock金融数据接口，演示从数据获取到复权计算的全流程。

3.1 数据获取与预处理

首先安装必要库并获取基础数据：

import baostock as bs
import pandas as pd

# 登录Baostock系统
lg = bs.login()
if lg.error_code != '0':
    print(f"登录失败：{lg.error_msg}")
    exit()

# 查询复权因子和行情数据
rs = bs.query_history_k_data_plus(
    "sh.600000",  # 浦发银行示例
    "date,close,preclose,adjustfactor",
    start_date="2020-01-01",
    end_date="2023-12-31",
    frequency="d",
    adjustflag="3"  # 不复权
)

# 转换为DataFrame
data_list = []
while (rs.error_code == '0') & rs.next():
    data_list.append(rs.get_row_data())
df = pd.DataFrame(data_list, columns=rs.fields)
df['date'] = pd.to_datetime(df['date'])
df[['close','preclose','adjustfactor']] = df[['close','preclose','adjustfactor']].astype(float)

# 计算累计复权因子
df['cum_adjust'] = df['adjustfactor'] / df['adjustfactor'].iloc[-1]

3.2 三种复权价格计算

基于获取的数据计算各类复权价格：

# 前复权计算
df['forward_adj'] = df['close'] * df['adjustfactor'].iloc[-1] / df['adjustfactor']

# 后复权计算
df['backward_adj'] = df['close'] * df['adjustfactor'].iloc[0] / df['adjustfactor']

# 动态复权计算
df['dynamic_adj'] = df.apply(
    lambda x: x['close'] * x['adjustfactor'] / df[df['date']<=x['date']]['adjustfactor'].iloc[-1],
    axis=1
)

# 结果展示
print(df[['date','close','forward_adj','backward_adj','dynamic_adj']].tail())

3.3 复权效果可视化

使用Matplotlib直观比较三种复权方式的差异：

import matplotlib.pyplot as plt

plt.figure(figsize=(14, 7))
plt.plot(df['date'], df['close'], label='原始价格')
plt.plot(df['date'], df['forward_adj'], label='前复权')
plt.plot(df['date'], df['backward_adj'], label='后复权')
plt.plot(df['date'], df['dynamic_adj'], label='动态复权')
plt.title('浦发银行不同复权方式价格对比')
plt.xlabel('日期')
plt.ylabel('价格(元)')
plt.legend()
plt.grid()
plt.show()

4. 工程实践中的关键问题

4.1 复权时点的精确处理

除权除息日(T日)的价格处理需要特别注意：

• T-1日收盘价：除权前的最后价格
• T日开盘价：已除权价格
• 复权因子生效时点：T日开盘

# 处理除权日的正确方法
def handle_ex_date(df):
    ex_dates = df[df['adjustfactor'] != df['adjustfactor'].shift(-1)].index
    for idx in ex_dates:
        # 除权日前一天收盘价
        pre_close = df.loc[idx, 'preclose']
        # 除权日收盘价
        ex_close = df.loc[idx+1, 'close']
        # 计算实际复权因子
        actual_factor = ex_close / pre_close
        df.loc[idx+1:, 'adjustfactor'] *= actual_factor
    return df

4.2 复权与交易成本的关系

复权处理必须与交易成本计算协调一致：

成本类型	前复权处理	后复权处理
现金分红	从成本价中扣除	加入累计收益
送转股	调整持股成本价	调整当前股价
配股	需单独计算配股支出	反映在复权因子中

4.3 多数据源的一致性校验

不同数据源的复权方法可能不同，需要进行交叉验证：

def validate_data_source(df1, df2, tolerance=0.01):
    """
    验证两个数据源的复权一致性
    :param df1: 数据源1(包含date,close,adjustfactor)
    :param df2: 数据源2(包含date,close,adjustfactor)
    :param tolerance: 允许误差
    :return: 不一致的日期列表
    """
    merged = pd.merge(df1, df2, on='date', suffixes=('_1','_2'))
    discrep = merged[
        (abs(merged['adjustfactor_1'] - merged['adjustfactor_2']) > tolerance) |
        (abs(merged['close_1'] - merged['close_2']) > tolerance)
    ]
    return discrep['date'].tolist()

5. 复权在量化策略中的应用实例

5.1 均线策略中的复权处理

以20/60日均线交叉策略为例，展示不同复权方式的影响：

def ma_cross_strategy(df):
    """均线交叉策略回测"""
    df['ma20'] = df['dynamic_adj'].rolling(20).mean()
    df['ma60'] = df['dynamic_adj'].rolling(60).mean()
    df['signal'] = np.where(df['ma20'] > df['ma60'], 1, 0)
    df['return'] = df['dynamic_adj'].pct_change() * df['signal'].shift(1)
    return df['return'].cumsum()

5.2 多因子选股中的复权一致性

构建多因子模型时，需确保所有因子使用相同的复权基准：

因子类型	复权建议	注意事项
价格类因子	动态复权	避免未来函数
财务因子	不需复权	使用原始财务数据
量价关系因子	与前复权成交量匹配	确保价量一致性

5.3 高频交易中的Tick数据复权

对于Tick级策略，复权处理更为复杂：

def tick_level_adjust(tick_df, daily_adjust):
    """
    Tick数据复权处理
    :param tick_df: Tick数据(包含datetime,price)
    :param daily_adjust: 每日复权因子(包含date,adjustfactor)
    """
    tick_df['date'] = tick_df['datetime'].dt.date
    merged = pd.merge(tick_df, daily_adjust, on='date')
    merged['adj_price'] = merged['price'] * merged['adjustfactor']
    return merged

6. 复权进阶：特殊场景处理

6.1 配股情况的特殊处理

配股既不是现金分红也不是送转股，需要单独计算：

def handle_rights_issue(price, issue_price, ratio):
    """
    处理配股复权
    :param price: 配股前价格
    :param issue_price: 配股价
    :param ratio: 配股比例(每10股配X股)
    :return: 理论除权价
    """
    return (price + issue_price * ratio/10) / (1 + ratio/10)

6.2 多次除权的叠加效应

当股票在短期内多次除权时，复权因子需要累乘：

def cumulative_adjust(df):
    """处理连续除权情况"""
    df['adjust_change'] = df['adjustfactor'] / df['adjustfactor'].shift(1)
    df.loc[df['adjust_change'].isna(), 'adjust_change'] = 1
    df['cum_adjust'] = df['adjust_change'].cumprod()
    return df

6.3 退市股票的后复权处理

对于已退市股票，需要特别注意最后交易日的价格处理：

def handle_delisted(stock_df, delist_date, delist_price):
    """
    处理退市股票复权
    :param stock_df: 股票历史数据
    :param delist_date: 退市日期
    :param delist_price: 退市价格
    """
    last_day = stock_df[stock_df['date'] <= delist_date].iloc[-1]
    adjust_ratio = delist_price / last_day['close']
    stock_df['adjustfactor'] *= adjust_ratio
    return stock_df