4. 规则引擎基础：规则短路执行

任务说明

本案例实现一个最小可用的规则引擎：按优先级顺序执行规则，一旦某条规则命中（返回 True）就立即停止后续规则执行，并返回命中的规则名与决策结果。这是风控场景中"高风险规则优先拦截"的核心机制。

核心实现

规则用字典描述：name 为规则名，check 为接收交易数据的判定函数，decision 为命中后的决策。引擎按列表顺序遍历，命中即短路返回。

from typing import Callable, Any

# 规则定义：name 规则名，check 判定函数，decision 命中后的决策
Rule = dict

def rule_high_amount(tx: dict) -> bool:
    return tx["amount"] >= 10000

def rule_blacklist(tx: dict) -> bool:
    return tx["user"] in {"u_black_01", "u_black_02"}

def rule_foreign_ip(tx: dict) -> bool:
    return tx["ip_country"] != "CN"

RULES: list[Rule] = [
    {"name": "blacklist", "check": rule_blacklist, "decision": "REJECT"},
    {"name": "high_amount", "check": rule_high_amount, "decision": "REVIEW"},
    {"name": "foreign_ip", "check": rule_foreign_ip, "decision": "REVIEW"},
]

def run_rules(tx: dict, rules: list[Rule]) -> dict:
    """按顺序执行规则，命中即短路返回。"""
    for rule in rules:
        if rule["check"](tx):
            return {
                "hit": True,
                "rule": rule["name"],
                "decision": rule["decision"],
            }
    return {"hit": False, "rule": None, "decision": "PASS"}

if __name__ == "__main__":
    # 用例1：黑名单用户，应命中 blacklist 并短路
    tx1 = {"user": "u_black_01", "amount": 200, "ip_country": "CN"}
    # 用例2：非黑名单但大额，应命中 high_amount
    tx2 = {"user": "u_1001", "amount": 50000, "ip_country": "CN"}
    # 用例3：全部未命中，放行
    tx3 = {"user": "u_1002", "amount": 100, "ip_country": "CN"}

    for tx in (tx1, tx2, tx3):
        print(tx["user"], "->", run_rules(tx, RULES))

运行方式

python rule_engine.py

验证方法与通过条件

• 用例1输出 hit=True, rule='blacklist', decision='REJECT'，说明黑名单规则命中后未继续执行后续规则（短路生效）。
• 用例2输出 hit=True, rule='high_amount', decision='REVIEW'，说明前一条未命中时按顺序继续。
• 用例3输出 hit=False, rule=None, decision='PASS'，说明全部未命中时放行。

若需确认短路行为，可在 rule_high_amount 内加一行 print("check high_amount")：用例1执行时不应打印该行，即证明后续规则未被调用。

关键点

• 短路由 return 在循环内直接返回实现，无需额外标志位。
• 规则顺序即优先级，高风险规则应放在列表前面。
• 新增规则只需向 RULES 追加字典，无需改动引擎逻辑。