大模型驱动实时风控：流式特征计算、规则自动合成与零误杀灰度闭环

在电商营销、信贷风控与反欺诈（Anti-Fraud）对抗中，最让风控团队精疲力竭的是黑产的**“高频变异与战术对抗”**：
黑产团伙通过云手机农场、动态代理 IP 池和自动化脚本批量刷券或薅羊毛，每当风控专家通过分析日志写出一条硬编码规则（如 ip_count_1h > 10），黑产在 15 分钟内就会调整攻击参数绕过拦截。

传统风控严重受制于**“人工分析-编写规则-审批测试-上线发布”**的漫长周期（通常为 1~3 天），面对分钟级变异的黑产显得笨重且滞后。

大语言模型（LLM）与**流式特征平台（Streaming Feature Store）的结合，正在将风控规则迭代带入“小时级甚至分钟级”响应时代：系统实时捕获异常交易行为，大模型基于特征画像自动提炼高区分度的候选规则 DSL，并通过离线影子回放（Shadow Replay）与渐进式灰度拦截，在确保“零业务误杀（Zero False Positive）”**的前提下构建动态防御网。

本文深入剖析流式特征计算架构、LLM 规则自动合成、AST 安全沙箱引擎以及生产级防误杀治理。

一、双轨架构：流式特征计算与大模型规则闭环

一个高可用、毫秒级判定的实时风控系统必须遵循**“特征与规则解耦、计算与生成异步”**的双轨架构：

+-----------------------------------------------------------------------------------+
| 轨道 1: 毫秒级在线判定主链路 (Online Real-Time Path, < 20ms)                       |
|                                                                                   |
|  [用户交易事件] ----> Flink 流式特征计算 (近1m/1h/24h 滑动窗口统计)                |
|                             |                                                     |
|                             v                                                     |
|                      在线特征库 (Redis / Aerospike)                               |
|                             |                                                     |
|                             v (拉取特征实体)                                      |
|                      高性能规则引擎 (AST 沙箱求值器) ----> 决策输出 (Pass / Block) |
+-----------------------------------------------------------------------------------+

+-----------------------------------------------------------------------------------+
| 轨道 2: 异步近线规则挖掘与生成闭环 (Nearline / Offline LLM Path)                  |
|                                                                                   |
|  [异常交易聚类] ----> 提取高危特征元数据 ----> LLM 候选规则 DSL 合成               |
|                                                       |                           |
|                                                       v                           |
|  [生产上线] <---- 渐进式灰度 (5% -> 20%) <---- 历史影子回测 (计算误杀率 FPR)       |
+-----------------------------------------------------------------------------------+

核心机制：

1. 重活前置到流式特征侧：所有复杂的跨时间窗口聚合（如 user_failed_login_10m、device_ip_distinct_count_1h）均由 Flink 在底层预计算完毕并写入内存特征库，规则引擎判定时只做极简的单条逻辑判断，确保判定耗时严格压在 15ms 以内。
2. 规则生成完全异步化：LLM 的推理耗时（通常 1~3 秒）完全置于近线侧，严禁直接阻塞线上交易请求。

二、防止误杀：大模型规则生成的四级安全防护网

风控领域的第一铁律是：“误杀正常用户（False Positive）的代价，往往数倍于漏过单笔黑产（False Negative）”。

为了确保大模型生成的规则不会造成灾难性误杀，必须设置四道硬核防线：

+-----------------------------------------------------------------------------------+
| 第一道防线: 语法与特征白名单约束                                                 |
| 强制约束 LLM 仅输出标准 AST 逻辑树或 Aviator 表达式，严禁使用非受控内置函数。     |
+-----------------------------------------------------------------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
| 第二道防线: 历史全量影子回放 (Shadow Backtesting)                                |
| 规则生成后，在数仓最近 7 天的数千万真实交易快照上离线重放。                       |
| 判定指标: 误杀率 (FPR) 必须严格 <= 0.01%，召回率 (Recall) 必须 >= 80%。           |
+-----------------------------------------------------------------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
| 第三道防线: 旁路影子模式运行 (Shadow Mode)                                       |
| 规则部署至生产引擎，但不阻断用户，仅在后台记录命中日志并与人工巡检结果对比 24 小时。|
+-----------------------------------------------------------------------------------+
                                          |
                                          v
+-----------------------------------------------------------------------------------+
| 第四道防线: 渐进式处置降级 (Soft Challenge First)                                |
| 初始灰度阶段命中规则时不直接执行 Block，而是弹出人脸识别或滑动验证码（Captcha）。 |
+-----------------------------------------------------------------------------------+

三、生产级特征提取、LLM 规则合成与 AST 沙箱引擎实现

下面使用 Python 实现一套工业级的风控规则闭环引擎。代码包含特征字典定义、大模型生成结构化规则 DSL、基于 Python ast 的无 eval 安全求值器，以及影子回测评估器。

"""
risk_rule_synthesizer.py
生产级大模型风控规则生成、AST 安全沙箱与影子评估引擎
"""

import ast
import json
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional


@dataclass
class UserFeatureSnapshot:
    """在线特征实体 (由 Flink 预计算)"""
    user_id: str
    order_cnt_1m: int = 0
    pay_amount_1m: float = 0.0
    device_switch_cnt_1h: int = 0
    ip_distinct_user_cnt_1h: int = 0
    account_age_days: int = 365
    is_emulator: int = 0


@dataclass
class SynthesizedRule:
    rule_id: str
    rule_name: str
    dsl_expression: str        # 结构化逻辑表达式, 如 "is_emulator == 1 and order_cnt_1m > 5"
    target_risk: str
    rationale: str
    threshold_fpr: float = 0.0001


class SafeASTEvaluator(ast.NodeVisitor):
    """
    基于抽象语法树 (AST) 的安全沙箱求值器，彻底杜绝 eval 注入与任意代码执行
    """
    def __init__(self, context: Dict[str, Any]):
        self.context = context

    def evaluate(self, expr: str) -> bool:
        try:
            tree = ast.parse(expr, mode='eval')
            return bool(self._eval_node(tree.body))
        except Exception as e:
            # 规则执行异常时，安全降级为 False (放行)，绝不阻塞主交易
            return False

    def _eval_node(self, node: ast.AST) -> Any:
        if isinstance(node, ast.Name):
            # 仅允许从特征上下文中取值
            if node.id not in self.context:
                raise ValueError(f"未定义的特征变量: {node.id}")
            return self.context[node.id]

        elif isinstance(node, ast.Constant):
            return node.value

        elif isinstance(node, ast.Compare):
            left_val = self._eval_node(node.left)
            for op, comparator in zip(node.ops, node.comparators):
                right_val = self._eval_node(comparator)
                if isinstance(op, ast.Eq) and not (left_val == right_val): return False
                elif isinstance(op, ast.NotEq) and not (left_val != right_val): return False
                elif isinstance(op, ast.Lt) and not (left_val < right_val): return False
                elif isinstance(op, ast.LtE) and not (left_val <= right_val): return False
                elif isinstance(op, ast.Gt) and not (left_val > right_val): return False
                elif isinstance(op, ast.GtE) and not (left_val >= right_val): return False
                left_val = right_val
            return True

        elif isinstance(node, ast.BoolOp):
            if isinstance(node.op, ast.And):
                return all(self._eval_node(v) for v in node.values)
            elif isinstance(node.op, ast.Or):
                return any(self._eval_node(v) for v in node.values)

        elif isinstance(node, ast.UnaryOp) and isinstance(node.op, ast.Not):
            return not self._eval_node(node.operand)

        raise ValueError(f"安全拦截: 包含未受支持的操作符 {type(node).__name__}")


class RiskRuleSynthesizer:
    """风控规则生成与影子回测引擎"""

    @staticmethod
    def build_prompt_from_fraud_cases(fraud_cases_summary: str, available_features: List[str]) -> str:
        features_str = ", ".join(available_features)
        prompt = f"""
你是一位资深反欺诈风控架构师。请根据以下黑产攻击案例特征，利用可用特征字典，合成高区分度的风控防御规则。

【近期高频欺诈案例画像】:
{fraud_cases_summary}

【可用流式特征白名单】:
{features_str}

【设计硬性指标】:
1. 重点针对模拟器黑产批量下单、短时间频次剧增行为。
2. 尽量使用复合条件（结合账号年限、设备指纹与短周期频次），降低误杀率。
3. 严格输出标准 JSON 格式：
{{
    "rule_name": "拦截模拟器批量高频刷单",
    "dsl_expression": "is_emulator == 1 and order_cnt_1m >= 5 and account_age_days <= 3",
    "target_risk": "营销黄牛薅羊毛",
    "rationale": "模拟器环境结合新注册账号高频下单是典型的脚本行为"
}}
"""
        return prompt

    @staticmethod
    def shadow_backtest(rule: SynthesizedRule, historical_dataset: List[Dict[str, Any]]) -> Dict[str, Any]:
        """
        在历史带标签样本集上进行影子回放，计算查准率、查全率与误杀率
        """
        evaluator_cache = {}
        total_normal = 0
        total_fraud = 0
        false_positives = 0
        true_positives = 0

        for row in historical_dataset:
            is_actual_fraud = row.get("is_fraud", 0) == 1
            features = {k: v for k, v in row.items() if k != "is_fraud"}

            if is_actual_fraud:
                total_fraud += 1
            else:
                total_normal += 1

            evaluator = SafeASTEvaluator(features)
            is_hit = evaluator.evaluate(rule.dsl_expression)

            if is_hit:
                if is_actual_fraud:
                    true_positives += 1
                else:
                    false_positives += 1

        fpr = false_positives / total_normal if total_normal > 0 else 0.0
        recall = true_positives / total_fraud if total_fraud > 0 else 0.0

        return {
            "rule_id": rule.rule_id,
            "expression": rule.dsl_expression,
            "total_samples": len(historical_dataset),
            "false_positive_rate": round(fpr, 6),
            "recall_rate": round(recall, 4),
            "is_approved_for_gray": fpr <= rule.threshold_fpr
        }

生产规则合成与影子回放演练

# 1. 可用特征字典
FEATURE_CATALOG = [
    "order_cnt_1m", "pay_amount_1m", "device_switch_cnt_1h",
    "ip_distinct_user_cnt_1h", "account_age_days", "is_emulator"
]

# 2. 模拟大模型根据黑产日志合成的候选规则
mock_llm_rule = SynthesizedRule(
    rule_id="RULE_ANTI_EMULATOR_001",
    rule_name="拦截模拟器黑产秒级刷单",
    dsl_expression="is_emulator == 1 and order_cnt_1m >= 3 and account_age_days <= 7",
    target_risk="黑产农场刷单",
    rationale="结合新账号、模拟器特征与 1 分钟内的爆发性下单频次",
    threshold_fpr=0.001
)

# 3. 构造历史 10,000 条真实交易样本 (包含 100 条真实黑产攻击与 9900 条正常用户)
mock_dataset = []
# 正常用户 (无模拟器，频次低)
for i in range(9900):
    mock_dataset.append({
        "is_emulator": 0, "order_cnt_1m": 1, "account_age_days": 120, "is_fraud": 0
    })
# 真实黑产
for i in range(100):
    mock_dataset.append({
        "is_emulator": 1, "order_cnt_1m": 5, "account_age_days": 1, "is_fraud": 1
    })

# 4. 执行影子回放与指标计算
report = RiskRuleSynthesizer.shadow_backtest(mock_llm_rule, mock_dataset)
print("=== 影子回测评估报告 ===")
print(f"【规则名称】: {mock_llm_rule.rule_name}")
print(f"【规则表达式】: {report['expression']}")
print(f"【误杀率 (FPR)】: {report['false_positive_rate'] * 100}%")
print(f"【黑产召回率 (Recall)】: {report['recall_rate'] * 100}%")
print(f"【是否准许进入灰度发布】: {'✅ 是' if report['is_approved_for_gray'] else '❌ 否'}")

# 5. 线上实时判定模拟
online_features = {
    "is_emulator": 1,
    "order_cnt_1m": 4,
    "account_age_days": 2
}
evaluator = SafeASTEvaluator(online_features)
hit_result = evaluator.evaluate(mock_llm_rule.dsl_expression)
print(f"\n线上单笔交易判定结果: {'🚨 命中高危规则，触发拦截' if hit_result else '正常放行'}")

四、生产避坑与安全边界

在将大模型应用到实时风控业务时，必须恪守以下三条底线：

1. 绝对禁止在线判定链路直接使用 eval()：
   直接使用 Python 原生 eval() 会带来灾难性的代码执行安全漏洞（RCE），且在异常输入时极易抛出未捕获异常导致线程阻塞。必须使用基于 AST 的白名单解释器或编译型规则引擎（如 Java Aviator / Easy-Rules）。
2. 规则引擎必须具备 10ms 硬熔断机制（Fail-Open）：
   如果某个复杂规则由于正则回溯或死循环导致判定耗时超过 15ms，规则引擎必须立即中断并执行默认放行（Fail-Open），绝不能因风控引擎卡顿拖垮核心支付交易主流程。
3. 黑产对抗样本投毒防范（Anti-Poisoning）：
   黑产可能会故意在短时间内产生大量正常行为诱导大模型提炼出过于宽泛的规则，从而在后期发起真正攻击时利用误杀倒逼平台下线规则。因此，用于大模型生成规则的样本必须经过高确信度的标签清洗，结合业务专家终审，避免数据源被黑产对抗投毒。

通过将 Flink 流式特征工程的高性能与大模型语义规则合成的敏捷性相结合，数据风控团队能够建立起一套“秒级计算、分钟级合成、零误杀发布”的现代化主动防御体系。