3. 商户风险数据模型：风险事件表结构、商户画像字段、批次偏移量定义

在构建批次偏移诊断系统之前，必须统一数据模型。本章定义三个核心数据实体：风险事件表（记录单次风险行为）、商户画像字段（描述商户静态与动态特征）、批次偏移量（衡量风险迁移的量化指标）。三者共同构成Flink实时计算的输入与输出基础。

3.1 风险事件表结构

风险事件表是实时流的事实表，每行代表一次被检测到的风险行为。表结构设计需兼顾写入吞吐与查询效率，以下为标准化定义：

3.2 商户画像字段

商户画像属于维度表，描述商户的静态注册信息与动态行为特征。画像字段每24小时由批处理任务更新一次，并加载到Flink的BroadcastState中供实时流关联。

3.3 批次偏移量定义

批次偏移量是衡量商户风险在不同批次之间变化程度的量化指标。定义如下：

// 批次偏移量计算公式
batch_offset = (current_batch_risk_score - previous_batch_risk_score) 
               / (max_possible_risk_score - min_possible_risk_score)

// 其中：
// current_batch_risk_score = Σ(事件风险等级 × 事件权重) / 批次内事件总数
// previous_batch_risk_score = 上一批次同商户的加权风险得分
// 分母归一化到 [0, 1] 区间

偏移量的取值范围与业务含义：

3.4 数据模型关系图

下图展示风险事件表、商户画像与批次偏移量之间的数据流转与关联关系：

3.5 批次偏移量计算示例

以下Flink SQL示例展示如何基于风险事件表计算商户的批次偏移量：

-- 步骤1：按批次和商户聚合风险得分
CREATE VIEW batch_risk_score AS
SELECT
    merchant_id,
    batch_id,
    AVG(risk_level) AS avg_risk_level,
    COUNT(*) AS event_count
FROM risk_events
GROUP BY merchant_id, batch_id;

-- 步骤2：关联上一批次得分，计算偏移量
SELECT
    curr.merchant_id,
    curr.batch_id AS current_batch,
    prev.batch_id AS previous_batch,
    curr.avg_risk_level AS current_score,
    COALESCE(prev.avg_risk_level, 0) AS previous_score,
    (curr.avg_risk_level - COALESCE(prev.avg_risk_level, 0)) / 3.0 AS batch_offset
FROM batch_risk_score curr
LEFT JOIN batch_risk_score prev
    ON curr.merchant_id = prev.merchant_id
    AND prev.batch_id = (
        SELECT MAX(batch_id) 
        FROM batch_risk_score 
        WHERE merchant_id = curr.merchant_id 
          AND batch_id < curr.batch_id
    );