📚 课程目录第1章风控样本概述

1. 交易风控样本概述

1.1 交易风控样本的定义

交易风控样本，是指在金融交易系统中，用于训练、验证和评估风控模型的历史交易记录及其关联特征数据。每一条样本通常对应一次交易行为，包含交易属性、账户行为、设备环境、时序特征以及最终的风险标签。

  核心公式： 样本 = (特征向量 X, 标签 Y) ，其中 X 包含交易金额、频次、设备指纹、地理位置等维度，Y 表示该交易是否为欺诈、盗用或违规操作。

样本维度	典型字段	数据来源
交易属性	交易金额、交易时间、交易类型、商户ID	交易流水表
账户行为	登录频次、历史交易成功率、账户余额	用户行为日志
设备环境	设备指纹、IP地址、操作系统版本	设备指纹库
时序特征	过去1小时交易次数、过去7天交易总额	实时计算引擎
风险标签	0-正常 / 1-欺诈 / 2-盗用 / 3-违规	人工标注 + 事后确认

1.2 样本在风控模型中的重要性

风控模型的本质是“从历史样本中学习风险模式，并泛化到未来交易”。样本的质量、数量、分布直接决定了模型的上限：

样本决定模型能力边界：即使算法再先进，如果样本中欺诈模式覆盖不全，模型也无法识别新型攻击。
样本偏差导致模型失效：如果训练样本只包含特定时间段的交易，模型在业务高峰期或大促期间可能严重误判。
样本时效性影响实时风控：风控场景中，欺诈手法快速演变，过时的样本会让模型“刻舟求剑”。

实战经验： 在金融风控领域，样本工程（Sample Engineering）往往比模型调参更能带来效果提升。一个经过精心切分的样本集，可以让 AUC 提升 5%~15%。

1.3 样本生命周期管理

交易风控样本并非静态数据，而是经历从采集、清洗、标注、切分、训练到归档的完整生命周期。在 Flink 实时风控场景下，样本生命周期管理需要兼顾离线批量与实时流处理。

常见误区： 很多团队只关注模型训练阶段的样本切分，忽略了样本的采集质量监控和过期样本淘汰机制，导致线上模型效果持续衰减。

生命周期阶段	核心任务	Flink 相关技术	常见风险
1. 样本采集	从交易流水、日志、外部数据源抽取原始数据	Flink CDC、Kafka Source	数据延迟、字段缺失
2. 样本清洗	去重、异常值过滤、缺失值填充	Flink ProcessFunction、CEP	过度清洗导致样本量不足
3. 特征工程	生成统计特征、时序特征、交叉特征	Flink SQL、State、Timer	特征穿越（未来信息泄露）
4. 样本标注	基于事后结果打标（欺诈确认、正常交易）	Flink 异步 IO、维表关联	标注延迟、标签噪声
5. 样本切分	按时间、账户、交易类型划分训练/验证/测试集	Flink 窗口、自定义 Partition	数据泄露、分布偏移
6. 样本存储	写入 HDFS、Hive、ClickHouse 等存储系统	Flink FileSink、JDBC Sink	存储格式不统一
7. 样本归档/淘汰	过期样本标记、冷热数据分离	Flink 定时任务、TTL 策略	归档策略不合理导致存储膨胀

1.4 样本切分的特殊挑战

在交易风控场景中，样本切分面临几个独特挑战，这些挑战直接影响 Flink 实时计算架构的设计：

时间依赖性：交易行为具有明显的时间序列特性，不能简单随机切分，必须按时间窗口划分训练集和测试集。
类别不平衡：欺诈样本通常占比极低（0.1%~1%），切分时需要保证验证集和测试集中保留足够的正样本。
数据漂移：业务策略、用户行为、欺诈手法随时间变化，训练集和线上数据分布可能不一致。
实时性要求：Flink 需要同时处理离线样本生成和实时样本采集，保证样本的时效性。

  Flink 实战要点： 在 Flink 中实现样本切分时，推荐使用 KeyedProcessFunction 结合 ValueState 维护每个账户的样本时间戳，确保切分逻辑在分布式环境下的一致性。同时利用 Flink 的 Watermark 机制处理乱序到达的样本数据。

// Flink 样本切分核心逻辑示例
DataStream<TransactionSample> samples = ...;

DataStream<TransactionSample> partitionedSamples = samples
    .keyBy(sample -> sample.getAccountId())
    .process(new KeyedProcessFunction<String, TransactionSample, TransactionSample>() {
        
        private ValueState<Long> lastTrainTime;
        
        @Override
        public void open(Configuration parameters) {
            lastTrainTime = getRuntimeContext()
                .getState(new ValueStateDescriptor<>("lastTrainTime", Types.LONG));
        }
        
        @Override
        public void processElement(TransactionSample sample, Context ctx, Collector<TransactionSample> out) {
            Long lastTime = lastTrainTime.value();
            if (lastTime == null) {
                // 首次样本，按时间窗口划分
                if (sample.getTimestamp() < TRAIN_CUTOFF) {
                    sample.setPartition("train");
                } else {
                    sample.setPartition("test");
                }
            } else {
                // 后续样本，确保时间顺序
                if (sample.getTimestamp() >= lastTime) {
                    // 正常处理
                    if (sample.getTimestamp() < TRAIN_CUTOFF) {
                        sample.setPartition("train");
                    } else {
                        sample.setPartition("test");
                    }
                } else {
                    // 乱序数据，特殊处理
                    sample.setPartition("late");
                }
            }
            lastTrainTime.update(sample.getTimestamp());
            out.collect(sample);
        }
    });