4. 样本构建方法论：流程、时间窗口与回溯机制

异常交易样本的构建是风控模型训练的基石。一个高质量的样本集直接决定了模型的上限。本节将系统讲解从原始数据到标注样本的完整流程，并重点剖析时间窗口与回溯机制这两个核心设计要点。

4.1 样本构建核心流程

标准样本构建遵循“数据采集 → 数据清洗 → 特征工程 → 样本标注”的四阶段流水线。每个阶段都有明确的输入、输出与质量检查点。

4.2 数据采集与清洗

数据采集阶段需要处理 Kafka 中的实时交易事件、MySQL 中的账户画像、Redis 中的设备指纹缓存。清洗阶段的核心任务是保证时间序列的单调性与字段的合法性。

// Flink SQL 示例：实时交易流清洗
CREATE TEMPORARY TABLE clean_transaction AS
SELECT
    transaction_id,
    account_id,
    CAST(amount AS DECIMAL(18, 2)) AS amount,
    TO_TIMESTAMP(transaction_time, 'yyyy-MM-dd HH:mm:ss') AS event_time,
    COALESCE(device_id, 'UNKNOWN') AS device_id,
    CASE WHEN amount <= 0 THEN 'INVALID' ELSE status END AS status
FROM raw_transaction
WHERE transaction_id IS NOT NULL
  AND amount IS NOT NULL
  AND amount < 1000000;  -- 异常值截断

4.3 特征工程：时间窗口的三种模式

时间窗口是特征工程中最关键的维度。不同的窗口模式决定了特征对交易行为的刻画粒度。以下是三种最常用的窗口模式及其适用场景。

4.4 样本标注与回溯机制

样本标注的核心难点在于“标签时间对齐”。异常交易往往在发生后一段时间才被确认（如 72 小时后的拒付、30 天后的逾期）。回溯机制正是为了解决这种标签延迟问题而设计。

下图展示了标准回溯标注的流程：

4.5 Flink 中的回溯实现方案

在 Flink 实时流水线中，回溯机制通常通过 KeyedState + 事件时间定时器 实现。核心思路是：将交易事件存入状态，等待指定延迟后，再根据收到的标签事件进行关联。

// Flink DataStream API 回溯标注伪代码
DataStream<Transaction> transactions = ...;
DataStream<LabelEvent> labels = ...;

transactions
    .keyBy(tx -> tx.getTransactionId())
    .connect(labels.keyBy(lb -> lb.getTransactionId()))
    .process(new KeyedCoProcessFunction<String, Transaction, LabelEvent, Sample>() {

        private ValueState<Transaction> txState;

        @Override
        public void processElement1(Transaction tx, Context ctx, Collector<Sample> out) {
            // 设置 72 小时定时器
            ctx.timerService().registerEventTimeTimer(tx.getEventTime() + 72 * 3600 * 1000L);
            txState.update(tx);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<Sample> out) {
            // 超时未收到标签，按正常样本处理
            Transaction tx = txState.value();
            if (tx != null) {
                out.collect(new Sample(tx, Label.NORMAL));
                txState.clear();
            }
        }

        @Override
        public void processElement2(LabelEvent label, Context ctx, Collector<Sample> out) {
            // 收到标签，立即关联输出
            Transaction tx = txState.value();
            if (tx != null) {
                out.collect(new Sample(tx, label.getLabel()));
                txState.clear();
            }
        }
    });

4.6 样本构建质量评估指标

完成样本构建后，需要通过以下指标评估样本集的质量，确保其满足模型训练要求。