4. 按时间顺序切分:基于时间戳的固定比例切分

在交易风控场景中,数据天然带有时间顺序属性。与随机切分不同,按时间顺序切分严格保留样本的时间先后关系,避免未来信息泄露(Look-ahead Bias)。这是构建可靠回测与在线风控模型的基础操作。

4.1 为什么必须按时间顺序切分

交易数据具有强时序依赖性:

  • 市场状态漂移:不同年份的波动率、流动性、微观结构存在显著差异
  • 策略过拟合风险:随机切分可能让模型“看到”未来数据,导致回测绩效虚高
  • 业务合规要求:监管机构明确要求风控模型必须基于历史数据训练,不得使用未来信息
核心原则:训练集的时间必须早于验证集,验证集的时间必须早于测试集。任何跨时间段的样本混入都视为数据泄露。

4.2 固定比例切分策略

将数据集按时间顺序排列后,按固定比例划分为三个连续区间:

数据集 比例 时间位置 用途
训练集 70% 最早 模型参数学习、特征工程
验证集 20% 中间 超参数调优、模型选择
测试集 10% 最晚 最终性能评估、泛化能力检验
比例调整建议:对于高频交易数据(如逐笔成交数据),可适当增大训练集比例至80%;对于低频宏观数据(如月度违约率),验证集和测试集至少保留20%以上以保证评估稳定性。

4.3 时间顺序切分流程

下图展示了基于时间戳的固定比例切分完整流程:

基于时间戳的固定比例切分流程 原始交易数据集(按时间戳升序排列) 检查时间戳是否严格递增 计算切分点:70% / 90% 位置索引 训练集(70%) 验证集(20%) 测试集(10%)

4.4 Python 代码实现

以下代码展示了基于时间戳的固定比例切分完整实现:

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

def temporal_train_val_test_split(df, timestamp_col, train_ratio=0.7, val_ratio=0.2):
    """
    基于时间戳的固定比例切分
    
    Parameters:
    -----------
    df : pd.DataFrame
        包含时间戳的交易数据集
    timestamp_col : str
        时间戳列名
    train_ratio : float
        训练集比例,默认0.7
    val_ratio : float
        验证集比例,默认0.2
    
    Returns:
    --------
    train_df, val_df, test_df : 三个切分后的DataFrame
    """
    # 1. 确保数据按时间戳升序排列
    df_sorted = df.sort_values(by=timestamp_col).reset_index(drop=True)
    
    # 2. 计算切分点索引
    n_total = len(df_sorted)
    train_end = int(n_total * train_ratio)
    val_end = int(n_total * (train_ratio + val_ratio))
    
    # 3. 执行切分
    train_df = df_sorted.iloc[:train_end].copy()
    val_df = df_sorted.iloc[train_end:val_end].copy()
    test_df = df_sorted.iloc[val_end:].copy()
    
    # 4. 打印切分信息
    print(f"总样本数: {n_total}")
    print(f"训练集: {len(train_df)} 样本 ({train_ratio*100:.0f}%)")
    print(f"验证集: {len(val_df)} 样本 ({val_ratio*100:.0f}%)")
    print(f"测试集: {len(test_df)} 样本 ({(1-train_ratio-val_ratio)*100:.0f}%)")
    print(f"训练集时间范围: {train_df[timestamp_col].min()} ~ {train_df[timestamp_col].max()}")
    print(f"验证集时间范围: {val_df[timestamp_col].min()} ~ {val_df[timestamp_col].max()}")
    print(f"测试集时间范围: {test_df[timestamp_col].min()} ~ {test_df[timestamp_col].max()}")
    
    return train_df, val_df, test_df

# 使用示例
if __name__ == "__main__":
    # 模拟交易数据
    np.random.seed(42)
    dates = pd.date_range(start='2020-01-01', end='2023-12-31', freq='H')
    n_samples = len(dates)
    
    df = pd.DataFrame({
        'timestamp': dates,
        'price': 100 + np.cumsum(np.random.randn(n_samples) * 0.5),
        'volume': np.random.randint(100, 10000, n_samples),
        'label': np.random.choice([0, 1], n_samples, p=[0.95, 0.05])
    })
    
    # 执行切分
    train, val, test = temporal_train_val_test_split(
        df, 
        timestamp_col='timestamp',
        train_ratio=0.7, 
        val_ratio=0.2
    )
    
    # 验证时间顺序
    assert train['timestamp'].max() < val['timestamp'].min()
    assert val['timestamp'].max() < test['timestamp'].min()
    print("\n时间顺序验证通过:训练集 < 验证集 < 测试集")

4.5 关键注意事项

常见陷阱:
  • 时间戳重复:同一时间戳存在多条记录时,需确保切分边界不将同一时刻的样本分到不同集合
  • 数据稀疏性:如果测试集时间跨度太短(如仅包含1天),评估结果可能不稳定
  • 节假日效应:交易数据存在明显的日历效应,切分时应考虑市场休市日的影响

4.6 切分效果验证

切分完成后,建议进行以下验证:

验证项 检查方法 预期结果
时间顺序 比较各集合的最大/最小时间戳 训练集.max < 验证集.min < 测试集.min
样本比例 计算各集合样本数占比 接近预设比例(允许微小偏差)
数据完整性 检查是否有样本遗漏或重复 三个集合样本数之和等于原始总数
时间连续性 检查切分点附近是否存在时间断层 相邻集合的时间戳应连续

4.7 进阶:基于日期的切分

对于交易日数据,更推荐按自然日进行切分,而非按样本数比例:

def temporal_split_by_date(df, timestamp_col, train_end_date, val_end_date):
    """
    基于日期边界的切分(推荐用于交易日数据)
    
    Parameters:
    -----------
    df : pd.DataFrame
    timestamp_col : str
    train_end_date : str
        训练集截止日期(包含)
    val_end_date : str
        验证集截止日期(包含)
    """
    df_sorted = df.sort_values(by=timestamp_col)
    
    train_df = df_sorted[df_sorted[timestamp_col] <= train_end_date].copy()
    val_df = df_sorted[
        (df_sorted[timestamp_col] > train_end_date) & 
        (df_sorted[timestamp_col] <= val_end_date)
    ].copy()
    test_df = df_sorted[df_sorted[timestamp_col] > val_end_date].copy()
    
    print(f"训练集: {train_end_date} 及之前, 共 {len(train_df)} 样本")
    print(f"验证集: {train_end_date} 之后 ~ {val_end_date}, 共 {len(val_df)} 样本")
    print(f"测试集: {val_end_date} 之后, 共 {len(test_df)} 样本")
    
    return train_df, val_df, test_df

# 使用示例
train, val, test = temporal_split_by_date(
    df, 
    timestamp_col='timestamp',
    train_end_date='2022-12-31',
    val_end_date='2023-06-30'
)
最佳实践:对于风控模型,建议训练集至少覆盖2个完整市场周期(牛熊市),验证集覆盖最近1个季度,测试集覆盖最近1个月。这样既能保证模型学习到足够的历史模式,又能评估其在最新市场环境下的表现。