4. 数据采集与预处理：规则命中日志采集、特征数据标准化、时间窗口划分

规则引擎的漂移监控依赖高质量、低延迟的输入数据。本章聚焦于从原始规则命中日志到标准化监控特征的完整预处理链路，涵盖日志采集、特征标准化与时间窗口划分三个核心环节。只有经过严格预处理的数据，才能支撑后续的分布对比与漂移检测算法。

4.1 规则命中日志采集

规则引擎每执行一次判定，都会产生一条命中日志。日志采集的完整性、时效性与结构化程度，直接决定漂移监控的覆盖范围与计算精度。

4.1.1 日志字段设计

一条完整的规则命中日志至少应包含以下字段：

4.1.2 采集架构

推荐采用 日志代理 + 消息队列 + 流处理 的轻量级采集架构，避免对规则引擎主链路产生性能干扰。

4.1.3 采集避坑要点

• 避免阻塞：日志写入必须异步，禁止在规则执行线程内同步写日志。
• 避免丢失：生产环境应开启消息队列的持久化与ACK机制。
• 避免重复：在流处理侧基于 request_id + rule_id 做幂等去重。

4.2 特征数据标准化

原始日志中的特征值往往存在量纲差异、缺失值、异常值等问题，必须经过标准化才能用于漂移计算。标准化策略因特征类型而异。

4.2.1 数值型特征标准化

4.2.2 类别型特征编码

• Label Encoding：适用于有序类别（如风险等级 low / medium / high）。
• One-Hot Encoding：适用于无序类别（如渠道来源 app / web / h5）。
• Target Encoding：适用于高基数类别，用目标变量均值替代原始值，需配合交叉验证防止过拟合。

4.2.3 缺失值处理

4.3 时间窗口划分

漂移监控的核心是比较两个时间窗口内的特征分布差异。窗口划分方式决定了监控的灵敏度与计算开销。

4.3.1 固定窗口

固定窗口按自然时间边界（每小时、每天、每周）切分，窗口长度恒定，不重叠。

4.3.2 滑动窗口

滑动窗口以固定步长向前推进，窗口内数据持续更新，适合对时效性要求高的场景。

• 窗口长度：决定参与计算的数据量，通常设为 1 小时或 24 小时。
• 滑动步长：决定监控粒度，通常设为 5 分钟或 1 小时。
• 参考窗口：与当前窗口对比的历史窗口，长度通常为当前窗口的 3~7 倍。

4.3.3 窗口划分对比

4.4 数据预处理流程总览

下图展示了从原始日志到标准化窗口特征的完整数据流：

4.5 预处理代码示例

以下 Python 代码展示了从原始日志到标准化窗口特征的完整预处理流程：

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

def preprocess_logs(raw_logs: pd.DataFrame, window_type: str = 'fixed', window_size: str = '1h'):
    """
    规则命中日志预处理流水线
    :param raw_logs: 原始日志DataFrame，必须包含 event_time, rule_id, features 列
    :param window_type: 'fixed' 或 'sliding'
    :param window_size: 窗口长度，如 '1h', '24h'
    :return: 标准化后的窗口特征DataFrame
    """
    # 1. 时间戳解析与排序
    raw_logs['event_time'] = pd.to_datetime(raw_logs['event_time'])
    raw_logs = raw_logs.sort_values('event_time')

    # 2. 特征展开：将json格式的features拆分为多列
    features_df = pd.json_normalize(raw_logs['features'])
    logs = pd.concat([raw_logs.drop('features', axis=1), features_df], axis=1)

    # 3. 数值特征标准化（Z-Score）
    numeric_cols = ['amount', 'age', 'credit_score']  # 示例特征
    for col in numeric_cols:
        if col in logs.columns:
            mean = logs[col].mean()
            std = logs[col].std()
            logs[f'{col}_zscore'] = (logs[col] - mean) / (std + 1e-8)

    # 4. 类别特征编码（Label Encoding）
    categorical_cols = ['channel', 'risk_level']
    for col in categorical_cols:
        if col in logs.columns:
            logs[f'{col}_encoded'] = logs[col].astype('category').cat.codes

    # 5. 时间窗口划分
    if window_type == 'fixed':
        logs['window'] = logs['event_time'].dt.floor(window_size)
    elif window_type == 'sliding':
        # 滑动窗口：以当前时间向前回溯window_size
        logs['window'] = logs['event_time'].apply(
            lambda x: x - timedelta(hours=1) if window_size == '1h' else x - timedelta(days=1)
        )

    # 6. 按窗口聚合统计特征分布
    window_features = logs.groupby(['rule_id', 'window']).agg(
        count=('request_id', 'count'),
        amount_mean=('amount_zscore', 'mean'),
        amount_std=('amount_zscore', 'std'),
        channel_mode=('channel_encoded', lambda x: x.mode().iloc[0] if not x.mode().empty else -1)
    ).reset_index()

    return window_features

4.6 预处理质量检查清单