1. 规则引擎基础

任务目标

实现一个最小可用的规则执行函数：输入一笔交易数据（dict）和一条规则条件（dict），判断该交易是否命中规则，返回布尔值。规则条件支持字段比较（> >= < <= == !=）与逻辑组合（and / or）。

规则结构约定

• 单条件：{"field": "amount", "op": ">", "value": 1000}
• 组合条件：{"logic": "and", "conditions": [cond1, cond2, ...]}，logic 取 and 或 or。
• 字段缺失时视为不命中（返回 False），避免抛异常中断批量执行。

完整实现

import operator

# 支持的比较运算符映射
_OPS = {
    ">": operator.gt,
    ">=": operator.ge,
    "<": operator.lt,
    "<=": operator.le,
    "==": operator.eq,
    "!=": operator.ne,
}

def _eval_single(txn, cond):
    """执行单条字段比较条件。"""
    field = cond.get("field")
    op = cond.get("op")
    value = cond.get("value")

    if field not in txn:
        return False
    if op not in _OPS:
        raise ValueError(f"不支持的运算符: {op}")

    return _OPS[op](txn[field], value)

def match_rule(txn, rule):
    """判断交易 txn 是否命中规则 rule，返回 bool。"""
    if "logic" in rule:
        logic = rule["logic"]
        conditions = rule.get("conditions", [])
        if logic == "and":
            return all(match_rule(txn, c) for c in conditions)
        if logic == "or":
            return any(match_rule(txn, c) for c in conditions)
        raise ValueError(f"不支持的逻辑: {logic}")

    return _eval_single(txn, rule)

if __name__ == "__main__":
    txn = {"amount": 1500, "country": "CN", "is_new_device": True}

    rule1 = {"field": "amount", "op": ">", "value": 1000}
    rule2 = {
        "logic": "and",
        "conditions": [
            {"field": "amount", "op": ">=", "value": 1000},
            {"field": "country", "op": "!=", "value": "US"},
        ],
    }
    rule3 = {
        "logic": "or",
        "conditions": [
            {"field": "amount", "op": ">", "value": 5000},
            {"field": "is_new_device", "op": "==", "value": True},
        ],
    }

    print(match_rule(txn, rule1))  # True
    print(match_rule(txn, rule2))  # True
    print(match_rule(txn, rule3))  # True

运行与验证

1. 将代码保存为 rule_engine.py，执行 python rule_engine.py。
2. 预期输出三行 True，与代码中注释一致（由给定输入唯一确定）。
3. 可自行修改 txn 中 amount 为 500，则 rule1 应输出 False，用于确认比较逻辑生效。
4. 若规则中字段名不在交易数据里，函数返回 False 而非报错，这是批量风控场景下的容错约定。

关键点说明

组合条件通过递归调用 match_rule 实现，因此 and / or 可以任意嵌套，无需额外代码。运算符用 operator 模块映射，避免写一长串 if/elif，也便于后续扩展。