📚 课程目录第4章时间序列切分策略

4. 时间序列切分策略：滑动窗口法、扩展窗口法、时间点分割法在交易数据中的应用与代码实现

在金融交易数据中，样本的切分不能采用随机打乱的方式，因为交易数据具有严格的时间依赖性和序列自相关性。错误的切分会导致未来信息泄露，使模型在回测中表现优异，但在实盘交易中失效。本章详细讲解三种主流的时间序列切分策略，并给出其在异常交易检测场景下的具体实现。

核心原则： 任何切分策略都必须保证训练集的时间戳全部早于验证集/测试集的时间戳，杜绝“用未来预测过去”的数据泄露。

4.1 三种切分策略对比

策略名称	核心思想	训练集规模	适用场景	主要风险
滑动窗口法	固定窗口长度，按步长滑动	恒定	高频交易、短期模式识别	窗口长度选择敏感
扩展窗口法	起始点固定，终点逐步后移	递增	趋势跟踪、长期策略回测	早期样本权重过大
时间点分割法	按固定时间点（如月末、季末）切分	阶梯式增长	监管报告、定期模型重训	切分点可能错过市场突变

4.2 滑动窗口法（Sliding Window）

滑动窗口法维护一个固定长度的训练窗口，每次按步长向前滑动，丢弃最旧的数据，加入最新的数据。该方法适用于概念漂移较快的交易场景，如日内高频异常交易检测。

  关键参数： 窗口长度（window_size）决定模型记忆的历史深度；步长（step_size）控制模型更新频率。步长越小，模型更新越频繁，但计算开销越大。

代码实现：滑动窗口切分

import pandas as pd
import numpy as np
from typing import List, Tuple

def sliding_window_split(
    df: pd.DataFrame,
    window_size: int = 1000,
    step_size: int = 200,
    test_size: int = 200
) -> List[Tuple[pd.DataFrame, pd.DataFrame]]:
    """
    滑动窗口法切分交易数据
    
    Parameters:
    -----------
    df : pd.DataFrame
        按时间排序的交易数据，必须包含 'timestamp' 列
    window_size : int
        训练窗口长度（样本数）
    step_size : int
        滑动步长（样本数）
    test_size : int
        验证集长度（样本数）
    
    Returns:
    --------
    splits : List[Tuple[train_df, test_df]]
        训练集与验证集的列表
    """
    df = df.sort_values('timestamp').reset_index(drop=True)
    total_len = len(df)
    splits = []
    
    start = 0
    while start + window_size + test_size <= total_len:
        train_end = start + window_size
        test_end = train_end + test_size
        
        train_df = df.iloc[start:train_end].copy()
        test_df = df.iloc[train_end:test_end].copy()
        
        splits.append((train_df, test_df))
        start += step_size
    
    return splits

# 使用示例
# df = pd.read_parquet('trade_data.parquet')
# splits = sliding_window_split(df, window_size=5000, step_size=1000, test_size=500)
# print(f"生成 {len(splits)} 个训练/验证对")

4.3 扩展窗口法（Expanding Window）

扩展窗口法的训练集起始点固定，终点不断向后扩展。该方法保留了全部历史信息，适合市场结构相对稳定、需要长期记忆的场景，如信用风险模型或低频异常交易检测。

代码实现：扩展窗口切分

def expanding_window_split(
    df: pd.DataFrame,
    min_train_size: int = 1000,
    step_size: int = 500,
    test_size: int = 200
) -> List[Tuple[pd.DataFrame, pd.DataFrame]]:
    """
    扩展窗口法切分交易数据
    
    Parameters:
    -----------
    df : pd.DataFrame
        按时间排序的交易数据
    min_train_size : int
        初始训练集最小样本数
    step_size : int
        每次扩展的样本数
    test_size : int
        验证集长度
    
    Returns:
    --------
    splits : List[Tuple[train_df, test_df]]
    """
    df = df.sort_values('timestamp').reset_index(drop=True)
    total_len = len(df)
    splits = []
    
    train_end = min_train_size
    while train_end + test_size <= total_len:
        train_df = df.iloc[:train_end].copy()
        test_df = df.iloc[train_end:train_end + test_size].copy()
        
        splits.append((train_df, test_df))
        train_end += step_size
    
    return splits

# 使用示例
# splits = expanding_window_split(df, min_train_size=3000, step_size=1000, test_size=500)
# print(f"生成 {len(splits)} 个训练/验证对，训练集规模从 {len(splits[0][0])} 到 {len(splits[-1][0])}")

4.4 时间点分割法（Time Point Split）

时间点分割法根据预定义的日历时间点（如每月1日、每季度初）进行切分。该方法与业务周期高度对齐，常用于监管报送、定期模型验证等场景。

注意： 时间点分割法要求切分点之间的数据量不均衡。例如，2月交易天数少，训练集可能明显小于3月。建议在切分后检查各分区的样本量，避免出现极小的验证集。

代码实现：时间点分割

def time_point_split(
    df: pd.DataFrame,
    split_dates: List[str],
    date_column: str = 'timestamp'
) -> List[Tuple[pd.DataFrame, pd.DataFrame]]:
    """
    按指定时间点分割交易数据
    
    Parameters:
    -----------
    df : pd.DataFrame
        交易数据
    split_dates : List[str]
        分割日期列表，格式 'YYYY-MM-DD'
    date_column : str
        日期列名
    
    Returns:
    --------
    splits : List[Tuple[train_df, test_df]]
        每个元素为 (训练集, 验证集)
    """
    df = df.sort_values(date_column).reset_index(drop=True)
    split_dates = sorted(pd.to_datetime(split_dates))
    splits = []
    
    prev_date = df[date_column].min()
    for split_date in split_dates:
        train_mask = df[date_column] < split_date
        test_mask = (df[date_column] >= split_date) & \
                    (df[date_column] < split_date + pd.Timedelta(days=30))
        
        train_df = df[train_mask].copy()
        test_df = df[test_mask].copy()
        
        if len(train_df) > 0 and len(test_df) > 0:
            splits.append((train_df, test_df))
    
    return splits

# 使用示例
# split_dates = ['2023-04-01', '2023-07-01', '2023-10-01', '2024-01-01']
# splits = time_point_split(df, split_dates)
# for i, (train, test) in enumerate(splits):
#     print(f"切分 {i+1}: 训练 {len(train)} 条, 验证 {len(test)} 条")

4.5 策略选择指南

高频异常交易检测（毫秒/秒级）：优先使用滑动窗口法，窗口长度设为 5000~10000 笔交易，步长设为窗口的 20%~50%。
日频/周频策略回测：使用扩展窗口法，初始训练集至少包含 2 年数据，每次扩展 1 个季度。
监管合规模型验证：使用时间点分割法，切分点对齐监管报告截止日期（如每月末、每季末）。
市场结构突变期：避免使用扩展窗口法，改用滑动窗口法并缩短窗口长度，减少历史陈旧数据的影响。

  实战建议： 在异常交易检测项目中，建议同时使用滑动窗口法和时间点分割法进行交叉验证。滑动窗口法评估模型对近期模式的捕捉能力，时间点分割法评估模型在固定时间周期上的稳定性。两种策略的验证结果差异较大时，说明模型存在过拟合或概念漂移问题。