4. 时间序列数据：OHLCV数据格式、复权因子处理与时间索引对齐

在量化回测中，时间序列数据是构建样本与标签对齐的基础。原始行情数据通常以OHLCV格式存储，但直接使用原始价格进行回测会因分红、送股、配股等事件导致价格缺口，从而产生虚假信号。因此，必须通过复权因子对价格序列进行修正，并确保时间索引严格对齐。

4.1 OHLCV数据格式与标准字段

OHLCV是金融时间序列最基础的5维数据，分别代表：开盘价（Open）、最高价（High）、最低价（Low）、收盘价（Close）和成交量（Volume）。标准日线数据通常还包含成交额（Amount）和涨跌幅（PctChange）。

4.2 复权因子处理：前复权与后复权

股票在存续期内会发生分红、送股、配股、拆细等事件，导致除权除息后价格出现断崖式下跌。这种缺口并非市场真实波动，而是权益分配造成的名义价格调整。复权的目的就是消除这些事件对价格连续性的影响。

4.2.1 复权因子定义

复权因子（Adjust Factor）是一个累积乘数序列，用于将历史价格调整到与当前价格可比的水平。设复权因子序列为 adj_factor[t]，则复权价格计算公式为：

# 复权价格 = 原始价格 × 复权因子
adj_close[t] = close[t] * adj_factor[t]

# 复权因子通常由除权除息事件计算得出
# 对于前复权：adj_factor[t] = 当前复权因子 / t时刻复权因子
# 对于后复权：adj_factor[t] = t时刻复权因子 / 基准日复权因子

4.2.2 前复权 vs 后复权

4.2.3 复权因子计算流程

复权因子的计算依赖于除权除息事件表。以下是一个简化的计算步骤：

1. 获取事件数据： 从数据源获取分红、送股、配股、拆细等事件的除权除息日、每股分红、送股比例、配股比例、配股价等信息。
2. 计算除权参考价： 除权参考价 = (前收盘价 - 每股现金红利 + 配股价 × 配股比例) / (1 + 送股比例 + 配股比例)
3. 计算单次复权因子： 单次因子 = 前收盘价 / 除权参考价
4. 累积复权因子： 从最早交易日开始，将单次因子累积连乘，得到每日的累积复权因子。
5. 生成前复权价格： 前复权价格 = 原始价格 × (最新累积因子 / 当日累积因子)

import pandas as pd
import numpy as np

def calc_forward_adjust_factor(events_df, price_df):
    """
    计算前复权因子
    events_df: 包含除权除息事件的DataFrame，字段：date, cash_div, stock_div, rights_ratio, rights_price
    price_df: 包含原始价格的DataFrame，字段：date, close
    """
    # 按日期排序
    events = events_df.sort_values('date')
    
    # 计算单次复权因子
    events['factor'] = (events['pre_close'] / 
                        (events['pre_close'] - events['cash_div'] + 
                         events['rights_ratio'] * events['rights_price']) /
                        (1 + events['stock_div'] + events['rights_ratio']))
    
    # 累积复权因子（从最早到最晚）
    events['cum_factor'] = events['factor'].cumprod()
    
    # 前复权：最新因子为1，历史因子 = 最新累积因子 / 当日累积因子
    latest_cum = events['cum_factor'].iloc[-1]
    events['adj_factor'] = latest_cum / events['cum_factor']
    
    # 合并到价格数据
    price_df = price_df.merge(events[['date', 'adj_factor']], on='date', how='left')
    price_df['adj_factor'] = price_df['adj_factor'].fillna(1.0)
    
    # 计算前复权价格
    price_df['adj_close'] = price_df['close'] * price_df['adj_factor']
    
    return price_df

4.3 时间索引对齐

在回测样本构建中，时间索引对齐是确保特征、标签、价格三者时间一致性的关键步骤。常见问题包括：交易日历不一致、时区偏移、数据频率不匹配、停牌日缺失等。

4.3.1 对齐原则

• 严格时间戳： 所有数据使用统一的datetime64[ns]或Timestamp类型，避免字符串比较。
• 交易日历统一： 使用统一的交易日历（如A股使用上交所/深交所交易日历），剔除周末和法定节假日。
• 前向填充： 对于停牌日或缺失日，使用前一个交易日的价格进行填充，避免引入未来信息。
• 后向填充： 仅在标签计算时使用，确保标签窗口内数据完整。

4.3.2 对齐流程

def align_time_index(price_df, feature_df, label_df, trade_calendar):
    """
    对齐价格、特征、标签的时间索引
    trade_calendar: 交易日列表（datetime64）
    """
    # 1. 统一索引类型
    price_df.index = pd.to_datetime(price_df.index)
    feature_df.index = pd.to_datetime(feature_df.index)
    label_df.index = pd.to_datetime(label_df.index)
    
    # 2. 对齐到交易日历
    aligned_price = price_df.reindex(trade_calendar, method='ffill')
    aligned_feature = feature_df.reindex(trade_calendar, method='ffill')
    aligned_label = label_df.reindex(trade_calendar, method='ffill')
    
    # 3. 删除非交易日（如果日历包含非交易日）
    aligned_price = aligned_price.loc[aligned_price.index.isin(trade_calendar)]
    aligned_feature = aligned_feature.loc[aligned_feature.index.isin(trade_calendar)]
    aligned_label = aligned_label.loc[aligned_label.index.isin(trade_calendar)]
    
    # 4. 合并为一个DataFrame
    combined = pd.concat({
        'price': aligned_price,
        'feature': aligned_feature,
        'label': aligned_label
    }, axis=1)
    
    # 5. 删除任何包含NaN的行（可选，根据策略决定）
    combined = combined.dropna()
    
    return combined

4.3.3 时间索引对齐流程图

4.4 实战：构建对齐后的回测样本

以下是一个完整的实战流程，将OHLCV数据、复权因子处理和时间索引对齐整合在一起：

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

def build_backtest_sample(raw_price_path, events_path, trade_calendar_path, 
                          start_date='2018-01-01', end_date='2023-12-31'):
    """
    构建对齐后的回测样本
    """
    # 1. 加载原始数据
    price_df = pd.read_csv(raw_price_path, parse_dates=['date'], index_col='date')
    events_df = pd.read_csv(events_path, parse_dates=['date'])
    trade_calendar = pd.read_csv(trade_calendar_path, parse_dates=['date'])['date']
    
    # 2. 过滤日期范围
    price_df = price_df.loc[start_date:end_date]
    events_df = events_df[(events_df['date'] >= start_date) & 
                          (events_df['date'] <= end_date)]
    trade_calendar = trade_calendar[(trade_calendar >= start_date) & 
                                    (trade_calendar <= end_date)]
    
    # 3. 计算前复权因子
    price_df = calc_forward_adjust_factor(events_df, price_df)
    
    # 4. 生成特征（示例：过去5日收益率）
    price_df['ret_5d'] = price_df['adj_close'].pct_change(5)
    price_df['vol_ma5'] = price_df['volume'].rolling(5).mean()
    
    # 5. 生成标签（示例：未来5日收益率）
    price_df['label_fwd_5d'] = price_df['adj_close'].pct_change(-5).shift(-5)
    
    # 6. 时间索引对齐
    feature_cols = ['ret_5d', 'vol_ma5']
    label_cols = ['label_fwd_5d']
    
    aligned = align_time_index(
        price_df[['adj_close', 'adj_factor']], 
        price_df[feature_cols], 
        price_df[label_cols], 
        trade_calendar
    )
    
    # 7. 删除标签为NaN的行（未来数据不可用）
    aligned = aligned.dropna(subset=['label'])
    
    return aligned

# 使用示例
sample = build_backtest_sample(
    raw_price_path='stock_000001.csv',
    events_path='events_000001.csv',
    trade_calendar_path='trade_calendar.csv'
)
print(f"回测样本形状: {sample.shape}")
print(sample.head())

4.5 常见问题与避坑指南