深入理解规则+模型

# -*- coding: utf-8 -*-import randomimport math  class RuleEngine:    """规则引擎：模拟风控规则打分"""     def __init__(self):        # 规则权重表：每条规则对应一个权重        self.rules = [            {"name": "金额超限", "check": lambda t: t["amount"] > 50000, "w": 0.5},            {"name": "新设备", "check": lambda t: t["is_new_device"], "w": 0.3},            {"name": "夜间交易", "check": lambda t: t["hour"] < 6, "w": 0.2},        ]        self.hit_count = 0        self.total_count = 0     def score(self, transaction):        s = 0.0        self.total_count += 1        for rule in self.rules:            if rule["check"](transaction):                s += rule["w"]                self.hit_count += 1        return s  class RiskModel:    """模拟模型：用简单函数代替真实机器学习模型"""     def __init__(self):        self.pred_count = 0     def predict_proba(self, transaction):        # 模拟模型预测欺诈概率（实际中由训练好的模型提供）        self.pred_count += 1        # 特征组合模拟模型学习到的模式        risk = 0.0        if transaction["amount"] > 30000 and transaction["is_new_device"]:            risk += 0.4        if transaction["hour"] < 6 and transaction["amount"] > 20000:            risk += 0.3        if transaction["ip_risk"] > 0.8:            risk += 0.2        return min(risk + 0.1, 0.99)  class FusionSystem:    """融合决策系统"""     def __init__(self, alpha=0.5, rule_th=0.6, model_th=0.7):        self.alpha = alpha        self.beta = 1.0 - alpha        self.rule_th = rule_th        self.model_th = model_th        self.rule_engine = RuleEngine()        self.risk_model = RiskModel()        self.stats = {"reject": 0, "review": 0, "pass": 0}     def decide(self, transaction):        rule_score = self.rule_engine.score(transaction)        model_score = self.risk_model.predict_proba(transaction)         fusion_score = self.alpha * rule_score + self.beta * model_score         # 冲突检测：规则与模型意见不一致        if (rule_score >= self.rule_th) != (model_score >= self.model_th):            decision = "review"  # 冲突时人工复核        elif fusion_score >= 0.6:            decision = "reject"        else:            decision = "pass"         self.stats[decision] += 1        return decision, fusion_score  def generate_transaction():    """生成模拟交易样本"""    return {        "amount": random.uniform(100, 100000),        "is_new_device": random.random() < 0.2,        "hour": random.randint(0, 23),        "ip_risk": random.random(),    }  def main():    fusion = FusionSystem(alpha=0.4)    sample_n = 500     print("=== 融合系统决策演示 ===")    for i in range(sample_n):        txn = generate_transaction()        decision, score = fusion.decide(txn)        if i < 10:            print(f"交易{i+1}: 金额{txn['amount']:8.1f} "                  f"设备{'新' if txn['is_new_device'] else '旧'} "                  f"时段{txn['hour']:2d}时 -> {decision} (融合分{score:.2f})")     print(f"\n=== 决策统计（共{sample_n}笔）===")    for k, v in fusion.stats.items():        print(f"{k}: {v} 笔 ({v / sample_n * 100:.1f}%)")     # 冲突率统计    conflict = fusion.stats["review"]    print(f"冲突率（人工复核比例）: {conflict / sample_n * 100:.1f}%")  if __name__ == "__main__":    main()