3. 数据加载与预处理

特征回放对比的第一步，是将原始特征数据从存储介质中正确加载到内存，并完成清洗与对齐。原始数据通常存在缺失值、类型不匹配、时间戳不同步等问题，直接影响后续对比分析的准确性。本节将系统讲解使用 Pandas 加载 CSV/Parquet 格式数据、处理缺失值、数据类型转换以及时间戳对齐的标准流程。

3.1 数据加载：CSV 与 Parquet

Pandas 提供 read_csv() 和 read_parquet() 两个核心接口。CSV 是通用文本格式，适合小规模数据交换；Parquet 是列式二进制格式，压缩率高、读取快，适合大规模特征存储。

import pandas as pd

# 加载 CSV，指定日期列和数据类型
df_csv = pd.read_csv(
    'feature_data.csv',
    parse_dates=['timestamp'],
    dtype={'user_id': 'int32', 'score': 'float32'}
)

# 加载 Parquet，自动保留 schema
df_pqt = pd.read_parquet('feature_data.parquet')

print(f'CSV shape: {df_csv.shape}')
print(f'Parquet shape: {df_pqt.shape}')

3.2 缺失值处理

特征数据中的缺失值通常以 NaN、None 或占位符（如 -999）形式存在。处理策略需根据特征含义和下游模型要求决定。

• 删除法：当缺失比例极低（< 1%）且随机分布时，直接删除缺失行。
• 填充法：数值特征常用均值/中位数填充；类别特征用众数或特殊标记填充。
• 前向/后向填充：时间序列特征中，使用 ffill() 或 bfill() 按时间顺序填充。
• 插值法：使用 interpolate() 进行线性或样条插值，适合连续变化特征。

# 检查缺失值
print(df.isnull().sum())

# 策略1：删除缺失比例 < 1% 的行
df_clean = df.dropna(thresh=len(df.columns) - 1)

# 策略2：数值列用中位数填充
num_cols = df.select_dtypes(include='number').columns
df[num_cols] = df[num_cols].fillna(df[num_cols].median())

# 策略3：时间序列前向填充
df['feature_a'] = df['feature_a'].ffill()

# 策略4：线性插值
df['feature_b'] = df['feature_b'].interpolate(method='linear')

3.3 数据类型转换

特征数据加载后，常见类型问题包括：数值列被读为 object、时间戳为字符串、类别列未转为 category 类型。类型转换可降低内存占用并提升计算效率。

# 批量优化数值类型
for col in df.select_dtypes(include='int64').columns:
    df[col] = df[col].astype('int32')

for col in df.select_dtypes(include='float64').columns:
    df[col] = df[col].astype('float32')

# 类别列转换
cat_cols = ['city', 'channel', 'device_type']
for col in cat_cols:
    df[col] = df[col].astype('category')

# 时间戳转换
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')

3.4 时间戳对齐

特征回放对比要求不同来源的特征在相同时间基准上进行比较。时间戳对齐的核心是将多个 DataFrame 按照统一的时间频率（如 1 分钟、1 小时）进行重采样和合并。

# 设置时间索引
df_a = df_a.set_index('timestamp')
df_b = df_b.set_index('timestamp')

# 统一重采样到 1 分钟频率，使用前向填充
df_a_resampled = df_a.resample('1T').ffill()
df_b_resampled = df_b.resample('1T').ffill()

# 按时间索引对齐合并
df_aligned = pd.merge(
    df_a_resampled,
    df_b_resampled,
    left_index=True,
    right_index=True,
    how='outer'
)

# 处理合并后仍存在的缺失时间点
df_aligned = df_aligned.interpolate(method='time')

print(f'对齐后数据范围: {df_aligned.index.min()} ~ {df_aligned.index.max()}')
print(f'对齐后记录数: {len(df_aligned)}')

3.5 完整预处理流水线

将上述步骤整合为可复用的预处理函数，便于在回放对比实验中重复调用。

def load_and_preprocess(csv_path: str, parquet_path: str = None) -> pd.DataFrame:
    """
    加载并预处理特征数据，返回对齐后的 DataFrame。
    """
    # 1. 加载数据
    df = pd.read_csv(csv_path, parse_dates=['timestamp'])
    if parquet_path:
        df_pqt = pd.read_parquet(parquet_path)
        df = pd.concat([df, df_pqt], ignore_index=True)

    # 2. 处理缺失值
    num_cols = df.select_dtypes(include='number').columns
    df[num_cols] = df[num_cols].fillna(df[num_cols].median())

    # 3. 类型转换
    for col in df.select_dtypes(include='int64').columns:
        df[col] = df[col].astype('int32')
    for col in df.select_dtypes(include='float64').columns:
        df[col] = df[col].astype('float32')

    # 4. 时间戳对齐
    df = df.set_index('timestamp')
    df = df.resample('1T').ffill().interpolate(method='time')

    return df.reset_index()

# 使用示例
feature_df = load_and_preprocess('train_features.csv', 'test_features.parquet')
print(feature_df.info())