4. 基于规则的实时大额交易拦截

任务说明

实现一个流处理函数 process_transaction，对每笔交易判断金额是否超过预设阈值（默认 10000），超过则标记为可疑并输出告警。函数逐笔处理，返回处理结果，便于在实时管道中调用。

核心实现

THRESHOLD = 10000.0  # 大额交易阈值

def process_transaction(tx, threshold=THRESHOLD):
    """
    处理单笔交易，判断是否为大额可疑交易。
    tx: dict，至少包含 'tx_id'、'amount'
    返回: dict，包含原始交易信息 + 'suspicious' 标记 + 可选 'alert'
    """
    amount = float(tx["amount"])
    suspicious = amount > threshold
    result = {
        "tx_id": tx["tx_id"],
        "amount": amount,
        "suspicious": suspicious,
    }
    if suspicious:
        result["alert"] = (
            f"[ALERT] 大额交易 tx_id={tx['tx_id']} "
            f"amount={amount:.2f} 超过阈值 {threshold:.2f}"
        )
    return result

def run_stream(transactions, threshold=THRESHOLD):
    """模拟流式处理：逐笔调用 process_transaction 并输出告警。"""
    for tx in transactions:
        result = process_transaction(tx, threshold)
        if result["suspicious"]:
            print(result["alert"])
        else:
            print(f"[OK] tx_id={result['tx_id']} amount={result['amount']:.2f}")
        yield result

if __name__ == "__main__":
    stream = [
        {"tx_id": "T001", "amount": 500},
        {"tx_id": "T002", "amount": 25000},
        {"tx_id": "T003", "amount": 10000},
        {"tx_id": "T004", "amount": 99999.99},
    ]
    list(run_stream(stream))

运行与验证

直接运行脚本：python rule_engine.py。观察输出：

• T001（500）应输出 [OK]，suspicious=False。
• T002（25000）应输出 [ALERT]，suspicious=True。
• T003（10000）等于阈值，按 > 判定为 [OK]，suspicious=False。
• T004（99999.99）应输出 [ALERT]。

通过条件：每笔交易返回结果中 suspicious 与金额是否严格大于阈值一致；可疑交易均带 alert 字段并打印告警行。

边界与扩展点

• 阈值可通过参数覆盖，便于按业务线配置不同限额。
• 判定使用严格大于，若业务要求“达到即拦截”，将 > 改为 >= 即可。
• 接入真实流时，将 run_stream 的输入替换为消息队列消费迭代器，输出替换为告警通道即可，核心判定逻辑不变。