AI Agent在金融风控中的案例拆解：从数据感知到策略生成的完整闭环

关键词

• AI Agent
• 金融风控
• 数据感知
• 策略生成
• 闭环系统
• 强化学习
• 多智能体协作

摘要

在金融领域，风险控制一直是核心挑战之一。传统的风控方法往往依赖静态规则和历史数据，难以应对复杂多变的金融环境。本文将深入探讨如何构建一个从数据感知到策略生成的完整AI Agent闭环系统，应用于金融风控场景。我们将通过生活化的比喻解析核心概念，详细介绍技术实现原理，提供完整的代码示例，并通过实际案例展示系统的应用效果。文章还将探讨AI Agent在金融风控领域的未来发展趋势，以及行业可能面临的挑战与机遇。

---

1. 背景介绍

1.1 金融风控的重要性与挑战

在金融世界里，风险控制就像是航海中的罗盘和船舵——没有它们，船只可能会在茫茫大海中迷失方向，甚至触礁沉没。每天，全球金融系统处理着数以万亿计的交易，每一笔交易都可能伴随着潜在的风险：信用卡欺诈、贷款违约、洗钱活动、市场操纵……这些风险不仅会给金融机构带来巨大的经济损失，还可能动摇整个金融体系的稳定性。

传统的金融风控方法主要依赖于专家制定的规则和简单的统计模型。想象一下，这就像是用一张固定的渔网去捕鱼——渔网的大小和形状是提前定好的，只能捕捉符合特定模式的"鱼"（风险）。然而，金融欺诈者和风险制造者就像是聪明的鱼群，他们不断改变自己的行为模式，绕过这张固定的渔网。更糟糕的是，随着金融市场的发展，交易数据呈现出爆炸性增长，传统方法在处理海量、高维、实时数据时显得力不从心。

1.2 AI Agent带来的变革

就在这时，AI Agent技术的出现为金融风控带来了新的希望。AI Agent就像是一位不知疲倦、学习能力超强的智能风控专家，它能够：

1. 持续感知：像雷达一样全方位监控金融数据的变化
2. 自主学习：从历史经验中学习，不断提升风险识别能力
3. 智能决策：根据实时情况做出最优的风控策略
4. 闭环优化：通过反馈不断改进自己的决策能力

这种从数据感知到策略生成再到反馈优化的完整闭环，正是AI Agent在金融风控中最具价值的特点。

1.3 目标读者与核心问题

本文主要面向以下读者：

• 金融科技从业者，希望了解AI Agent在风控中的应用
• AI/ML工程师，对金融领域的实际应用案例感兴趣
• 风控专家，希望了解新技术如何改进传统风控方法
• 技术管理者，需要评估AI Agent技术在业务中的落地价值

我们将围绕以下核心问题展开探讨：

1. AI Agent如何在金融风控中实现数据感知？
2. 如何设计从感知到决策的技术路径？
3. 策略生成的核心机制是什么？
4. 如何构建完整的闭环系统并实现持续优化？
5. 实际应用中会遇到哪些挑战，如何解决？

通过本文的阅读，您将不仅能够理解AI Agent在金融风控中的工作原理，还能掌握实际构建和部署这类系统的方法。

---

2. 核心概念解析

2.1 AI Agent的基本概念：金融世界中的"智能助手"

让我们先用一个生活化的比喻来理解AI Agent。想象一下，您是一位银行行长，您需要一位特别的助手来帮助您管理风险。这位助手不仅要能听懂您的指令，还要能：

• 主动观察：时刻关注银行的各项业务数据，不需要您时刻提醒
• 独立思考：对观察到的情况进行分析，而不仅仅是传递信息
• 主动行动：根据分析结果采取相应措施，而不是等待您的指令
• 学习成长：从每次行动的结果中学习，下次做得更好

这位"超级助手"就是我们要说的AI Agent。在学术上，AI Agent可以定义为：能够感知环境、做出决策并采取行动，以实现特定目标的智能实体。

在金融风控的场景中，AI Agent的"环境"就是整个金融生态系统——包括交易数据、客户信息、市场动态、监管政策等等。它的"行动"可能是批准一笔贷款、拒绝一笔交易、调整风险限额、发出风险警报等等。而它的"目标"则是在控制风险的前提下，最大化金融机构的收益。

2.2 金融风控中的关键概念

在深入探讨AI Agent如何应用于金融风控之前，我们需要先了解一些金融风控领域的关键概念：

2.2.1 风险类型

金融风险可以分为多种类型，就像不同类型的疾病需要不同的治疗方法：

• 信用风险：借款人无法按时偿还债务的风险（类似借款人可能"生病"无法还钱）
• 市场风险：由于市场价格波动导致损失的风险（类似天气变化影响农作物收成）
• 操作风险：由于内部流程、人员或系统问题导致损失的风险（类似工厂生产线故障）
• 流动性风险：无法及时获得足够资金来履行义务的风险（类似公司资金链断裂）
• 欺诈风险：故意欺骗行为导致损失的风险（类似有人伪造支票骗取钱财）

2.2.2 风控流程

传统的金融风控流程通常包括以下几个步骤：

1. 数据收集：收集客户信息、交易数据等
2. 风险评估：分析数据，评估风险水平
3. 决策制定：根据风险评估结果做出决策（批准/拒绝/调整）
4. 监控跟踪：持续监控已发生的交易或贷款
5. 反馈优化：根据结果调整风控策略

这个流程听起来很合理，但在实际应用中却面临着诸多挑战：数据量大、处理速度慢、规则更新不及时、无法应对新型风险等等。而AI Agent的出现，正是为了解决这些问题。

2.3 AI Agent在金融风控中的核心要素

让我们用一个更具体的比喻来理解AI Agent在金融风控中的工作原理。想象一下，AI Agent就像是一位经验丰富的机场安检员：

• 感知器官（传感器）：安检员的眼睛、耳朵，以及X光机、金属探测器等设备——对应AI Agent的数据收集和处理模块
• 大脑（推理引擎）：安检员的知识和经验，判断乘客是否可疑——对应AI Agent的决策模型
• 手脚（执行器）：安检员的行动，如放行、开箱检查、报警——对应AI Agent的执行模块
• 学习机制：安检员从每次检查中总结经验，提高识别能力——对应AI Agent的反馈学习机制

现在，让我们把这个比喻转化为技术术语，看看AI Agent在金融风控中的核心要素：

2.3.1 感知模块（Perception Module）

这个模块负责从各种数据源中收集和处理信息，就像安检员的眼睛和设备。在金融风控中，感知模块需要处理的数据可能包括：

• 交易数据：金额、时间、地点、方式等
• 客户数据：年龄、职业、收入、信用历史等
• 行为数据：登录时间、操作习惯、浏览记录等
• 外部数据：经济指标、新闻事件、监管政策等

2.3.2 认知模块（Cognition Module）

这个模块负责理解和分析感知到的信息，就像安检员的大脑。在金融风控中，认知模块的主要任务包括：

• 风险识别：发现潜在的风险信号
• 风险评估：量化风险的严重程度
• 情境理解：理解当前的市场环境和业务背景

2.3.3 决策模块（Decision-Making Module）

这个模块负责根据认知结果制定行动方案，就像安检员决定如何处理可疑乘客。在金融风控中，决策模块可能需要做出的决策包括：

• 交易决策：批准、拒绝、延迟或调查一笔交易
• 额度决策：调整客户的信用额度
• 定价决策：根据风险水平调整利率或费用
• 预警决策：是否发出风险警报

2.3.4 执行模块（Action Module）

这个模块负责执行决策模块做出的决定，就像安检员采取具体的行动。在金融风控中，执行模块的任务可能包括：

• 自动执行：自动批准或拒绝交易
• 人工介入：将可疑案例转给风控专家处理
• 系统调整：自动调整系统参数或规则
• 通知警报：向相关人员发送通知或警报

2.3.5 学习模块（Learning Module）

这个模块负责从行动结果中学习，优化系统性能，就像安检员从经验中学习。在金融风控中，学习模块的主要功能包括：

• 结果评估：评估决策的效果（如是否正确识别了欺诈）
• 模型更新：根据新数据更新模型参数
• 策略优化：调整决策策略以提高效果
• 知识积累：积累新的风险模式和案例

2.4 概念之间的关系

现在，让我们用表格和图表来更清晰地展示这些核心概念之间的关系。

2.4.1 概念核心属性维度对比

概念模块	主要功能	输入信息	输出信息	关键技术	实时性要求	复杂度
感知模块	数据收集与处理	原始数据	结构化信息	ETL、NLP、特征工程	高	中
认知模块	风险识别与评估	结构化信息	风险评估结果	机器学习、深度学习	中高	高
决策模块	策略制定	风险评估结果	决策指令	强化学习、优化算法	中高	高
执行模块	决策执行	决策指令	执行结果	API、工作流引擎	高	低
学习模块	系统优化	执行结果	模型更新	在线学习、迁移学习	中低	高

2.4.2 概念联系的ER实体关系图

2.4.3 交互关系图

通过这些图表，我们可以更直观地理解AI Agent在金融风控中的工作流程和各个模块之间的关系。接下来，我们将深入探讨技术原理与实现。

---

3. 技术原理与实现

3.1 从数据感知到认知：技术实现路径

在这一节中，我们将深入探讨AI Agent如何实现从原始数据到风险认知的转化。这就像是将杂乱无章的拼图碎片逐步组装成一幅完整的图片，每一步都需要精确的技术处理。

3.1.1 数据感知层：多源异构数据的融合处理

金融风控的数据来源多种多样，数据格式也各不相同。想象一下，您需要从不同的地方收集食材：有些是新鲜的蔬菜（结构化数据），有些是罐装食品（半结构化数据），还有些是需要特殊处理的海鲜（非结构化数据）。数据感知层的任务就是将这些不同来源、不同格式的"食材"清洗、处理并准备好，供后续"烹饪"使用。

核心技术挑战：

1. 多源数据集成：如何将来自不同系统的数据有效整合
2. 实时数据处理：如何在毫秒级时间内处理大量交易数据
3. 数据质量保障：如何处理缺失值、异常值和错误数据
4. 特征工程：如何从原始数据中提取有意义的特征

技术实现方案：

我们可以使用Lambda架构来同时处理批处理和实时数据：

让我们用Python代码来实现一个简化的数据感知模块：

import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
import json
from datetime import datetime

class RiskDataPerception:
    """金融风控数据感知模块"""
    
    def __init__(self):
        self.numeric_features = ['amount', 'age', 'income', 'credit_score', 
                                 'debt_to_income', 'account_age_days']
        self.categorical_features = ['transaction_type', 'location', 'device_type', 
                                     'occupation', 'education_level']
        self.text_features = ['transaction_description', 'customer_notes']
        self.temporal_features = ['transaction_time']
        
        # 初始化预处理器
        self.numeric_transformer = Pipeline(steps=[
            ('imputer', SimpleImputer(strategy='median')),
            ('scaler', StandardScaler())
        ])
        
        self.categorical_transformer = Pipeline(steps=[
            ('imputer', SimpleImputer(strategy='constant', fill_value='missing')),
            ('onehot', OneHotEncoder(handle_unknown='ignore'))
        ])
        
        self.preprocessor = ColumnTransformer(
            transformers=[
                ('num', self.numeric_transformer, self.numeric_features),
                ('cat', self.categorical_transformer, self.categorical_features)
            ])
        
        self.is_fitted = False
    
    def process_raw_data(self, raw_data):
        """
        处理原始数据，将其转换为结构化格式
        
        参数:
            raw_data (dict): 原始数据字典
            
        返回:
            DataFrame: 处理后的结构化数据
        """
        # 解析原始数据
        processed_data = {}
        
        # 处理数值型特征
        for feature in self.numeric_features:
            if feature in raw_data:
                processed_data[feature] = raw_data[feature]
            else:
                processed_data[feature] = np.nan
        
        # 处理类别型特征
        for feature in self.categorical_features:
            if feature in raw_data:
                processed_data[feature] = raw_data[feature]
            else:
                processed_data[feature] = np.nan
        
        # 处理时间特征
        if 'transaction_time' in raw_data:
            transaction_time = datetime.fromisoformat(raw_data['transaction_time'])
            processed_data['hour_of_day'] = transaction_time.hour
            processed_data['day_of_week'] = transaction_time.weekday()
            processed_data['is_weekend'] = 1 if transaction_time.weekday() >= 5 else 0
        
        # 构造额外特征
        if 'amount' in raw_data and 'income' in raw_data and raw_data['income'] > 0:
            processed_data['amount_to_income_ratio'] = raw_data['amount'] / raw_data['income']
        
        return pd.DataFrame([processed_data])
    
    def extract_features(self, structured_data):
        """
        从结构化数据中提取特征
        
        参数:
            structured_data (DataFrame): 结构化数据
            
        返回:
            array: 提取的特征向量
        """
        if not self.is_fitted:
            features = self.preprocessor.fit_transform(structured_data)
            self.is_fitted = True
        else:
            features = self.preprocessor.transform(structured_data)
        
        return features
    
    def perceive(self, raw_data):
        """
        完整的感知流程：处理原始数据并提取特征
        
        参数:
            raw_data (dict): 原始数据字典
            
        返回:
            dict: 包含特征向量和元数据的感知结果
        """
        # 处理原始数据
        structured_data = self.process_raw_data(raw_data)
        
        # 提取特征
        features = self.extract_features(structured_data)
        
        # 构造感知结果
        perception_result = {
            'features': features,
            'structured_data': structured_data.to_dict('records')[0],
            'timestamp': datetime.now().isoformat(),
            'data_quality_score': self._calculate_data_quality_score(raw_data)
        }
        
        return perception_result
    
    def _calculate_data_quality_score(self, raw_data):
        """
        计算数据质量分数（简化版本）
        
        参数:
            raw_data (dict): 原始数据字典
            
        返回:
            float: 数据质量分数 (0-1)
        """
        expected_fields = set(self.numeric_features + self.categorical_features + self.temporal_features)
        present_fields = set(raw_data.keys())
        completeness = len(present_fields.intersection(expected_fields)) / len(expected_fields)
        
        # 这里可以添加更多数据质量检查逻辑
        # 例如：异常值检测、格式验证等
        
        return completeness

这个简化的数据感知模块展示了如何处理原始数据并提取特征。在实际应用中，我们还需要考虑更多因素，如数据流处理、分布式计算、实时特征更新等。

3.1.2 认知层：风险识别与评估

认知层就像是AI Agent的"大脑"，负责理解感知到的数据并识别潜在风险。这就像是一位经验丰富的医生，通过观察患者的症状（感知数据）来诊断疾病（识别风险）。

在金融风控中，认知层需要解决的核心问题包括：

1. 异常检测：识别异常的交易或行为模式
2. 风险分类：将案例分类为不同的风险类型
3. 风险评分：为每个案例计算风险分数
4. 风险归因：识别导致风险的关键因素

让我们用一个具体的例子来实现风险认知模块：

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier, IsolationForest
from sklearn.neural_network import MLPClassifier
from sklearn.calibration import CalibratedClassifierCV
import joblib
from datetime import datetime

class RiskCognition:
    """金融风控认知模块"""
    
    def __init__(self):
        # 初始化异常检测模型
        self.anomaly_detector = IsolationForest(contamination=0.01, random_state=42)
        
        # 初始化风险分类模型
        self.risk_classifier = RandomForestClassifier(
            n_estimators=100,
            max_depth=10,
            random_state=42
        )
        
        # 初始化风险评分模型（使用校准的分类器以获得更可靠的概率）
        self.risk_scorer = CalibratedClassifierCV(
            base_estimator=MLPClassifier(hidden_layer_sizes=(100, 50), max_iter=500, random_state=42),
            cv=5,
            method='sigmoid'
        )
        
        self.is_trained = False
        self.feature_importance = None
        self.risk_categories = ['low_risk', 'medium_risk', 'high_risk', 'critical_risk']
        self.risk_thresholds = [0.3, 0.6, 0.8, 1.0]  # 对应不同风险类别的阈值
    
    def train(self, features, labels, anomaly_labels=None):
        """
        训练认知模型
        
        参数:
            features (array): 特征矩阵
            labels (array): 风险标签（0-低风险，1-高风险）
            anomaly_labels (array, optional): 异常标签（1-异常，0-正常）
        """
        # 训练异常检测模型
        if anomaly_labels is not None:
            # 如果有标记的异常数据，使用它们
            normal_data = features[anomaly_labels == 0]
            self.anomaly_detector.fit(normal_data)
        else:
            # 否则使用无监督方法
            self.anomaly_detector.fit(features)
        
        # 训练风险分类模型
        self.risk_classifier.fit(features, labels)
        
        # 训练风险评分模型
        self.risk_scorer.fit(features, labels)
        
        # 保存特征重要性
        self.feature_importance = self.risk_classifier.feature_importances_
        
        self.is_trained = True
    
    def detect_anomaly(self, features):
        """
        检测异常
        
        参数:
            features (array): 特征向量
            
        返回:
            dict: 异常检测结果
        """
        if not self.is_trained:
            raise ValueError("模型尚未训练，请先调用train方法")
        
        # 预测异常（1表示正常，-1表示异常）
        is_anomaly = self.anomaly_detector.predict(features)[0]
        anomaly_score = self.anomaly_detector.score_samples(features)[0]
        
        # 转换为更直观的格式
        result = {
            'is_anomaly': bool(is_anomaly == -1),
            'anomaly_score': float(anomaly_score),
            'anomaly_confidence': float(1 - abs(anomaly_score))  # 置信度
        }
        
        return result
    
    def classify_risk(self, features):
        """
        风险分类
        
        参数:
            features (array): 特征向量
            
        返回:
            dict: 风险分类结果
        """
        if not self.is_trained:
            raise ValueError("模型尚未训练，请先调用train方法")
        
        # 预测风险类别
        risk_prediction = self.risk_classifier.predict(features)[0]
        risk_probabilities = self.risk_classifier.predict_proba(features)[0]
        
        result = {
            'risk_binary': int(risk_prediction),
            'risk_probabilities': risk_probabilities.tolist(),
            'risk_category': self.risk_categories[min(int(risk_prediction * len(self.risk_categories)), len(self.risk_categories)-1)]
        }
        
        return result
    
    def score_risk(self, features):
        """
        风险评分
        
        参数:
            features (array): 特征向量
            
        返回:
            dict: 风险评分结果
        """
        if not self.is_trained:
            raise ValueError("模型尚未训练，请先调用train方法")
        
        # 计算风险分数
        risk_score = self.risk_scorer.predict_proba(features)[0, 1]  # 正类概率作为风险分数
        
        # 根据分数确定风险等级
        risk_level = 0
        for i, threshold in enumerate(self.risk_thresholds):
            if risk_score <= threshold:
                risk_level = i
                break
        
        result = {
            'risk_score': float(risk_score),
            'risk_level': risk_level,
            'risk_level_name': self.risk_categories[risk_level]
        }
        
        return result
    
    def explain_risk(self, features, feature_names=None):
        """
        风险归因分析 - 识别导致风险的关键因素
        
        参数:
            features (array): 特征向量
            feature_names (list, optional): 特征名称列表
            
        返回:
            dict: 风险归因结果
        """
        if not self.is_trained:
            raise ValueError("模型尚未训练，请先调用train方法")
        
        if feature_names is None:
            feature_names = [f"feature_{i}" for i in range(len(self.feature_importance))]
        
        # 计算每个特征的贡献（简化版本）
        feature_contributions = features[0] * self.feature_importance
        
        # 找出最重要的特征
        top_features_indices = np.argsort(np.abs(feature_contributions))[::-1][:5]
        top_features = []
        
        for idx in top_features_indices:
            contribution = float(feature_contributions[idx])
            top_features.append({
                'feature_name': feature_names[idx],
                'feature_value': float(features[0, idx]),
                'contribution': contribution,
                'importance': float(self.feature_importance[idx])
            })
        
        result = {
            'top_features': top_features,
            'explanation': self._generate_explanation(top_features)
        }
        
        return result
    
    def _generate_explanation(self, top_features):
        """
        生成自然语言解释（简化版本）
        
        参数:
            top_features (list): 最重要的特征列表
            
        返回:
            str: 自然语言解释
        """
        explanations = []
        
        for feature in top_features:
            if feature['contribution'] > 0:
                direction = "增加了"
            else:
                direction = "降低了"
            
            explanation = f"{feature['feature_name']} = {feature['feature_value']:.2f} {direction}风险"
            explanations.append(explanation)
        
        return "主要风险因素: " + "; ".join(explanations)
    
    def cognize(self, perception_result, feature_names=None):
        """
        完整的认知流程
        
        参数:
            perception_result (dict): 感知结果
            feature_names (list, optional): 特征名称列表
            
        返回:
            dict: 认知结果
        """
        features = perception_result['features']
        
        # 执行各项认知任务
        anomaly_result = self.detect_anomaly(features)
        classification_result = self.classify_risk(features)
        scoring_result = self.score_risk(features)
        explanation_result = self.explain_risk(features, feature_names)
        
        # 综合认知结果
        cognition_result = {
            'anomaly_detection': anomaly_result,
            'risk_classification': classification_result,
            'risk_scoring': scoring_result,
            'risk_explanation': explanation_result,
            'timestamp': datetime.now().isoformat(),
            'overall_risk_level': scoring_result['risk_level_name'],
            'requires_attention': scoring_result['risk_level'] >= 2 or anomaly_result['is_anomaly']
        }
        
        return cognition_result
    
    def save_model(self, filepath):
        """保存模型到文件"""
        model_data = {
            'anomaly_detector': self.anomaly_detector,
            'risk_classifier': self.risk_classifier,
            'risk_scorer': self.risk_scorer,
            'feature_importance': self.feature_importance,
            'is_trained': self.is_trained,
            'risk_categories': self.risk_categories,
            'risk_thresholds': self.risk_thresholds
        }
        joblib.dump(model_data, filepath)
    
    def load_model(self, filepath):
        """从文件加载模型"""
        model_data = joblib.load(filepath)
        self.anomaly_detector = model_data['anomaly_detector']
        self.risk_classifier = model_data['risk_classifier']
        self.risk_scorer = model_data['risk_scorer']
        self.feature_importance = model_data['feature_importance']
        self.is_trained = model_data['is_trained']
        self.risk_categories = model_data['risk_categories']
        self.risk_thresholds = model_data['risk_thresholds']

这个风险认知模块实现了异常检测、风险分类、风险评分和风险归因等核心功能。在实际应用中，我们可能需要使用更复杂的模型，如深度学习模型、图神经网络等，来处理更复杂的风控场景。

3.2 决策与策略生成：基于强化学习的动态优化

在认知模块识别并评估风险后，接下来需要做出决策并生成相应的风控策略。这就像是一位国际象棋大师，在分析棋局后决定下一步怎么走。在金融风控中，决策模块需要考虑多个因素：风险大小、潜在收益、监管要求、客户体验等等。

传统的风控决策通常依赖于固定的规则，如"如果风险分数超过0.8，则拒绝交易"。然而，这种静态规则无法适应不断变化的金融环境。AI Agent可以使用强化学习(Reinforcement Learning, RL)来实现动态决策和策略优化。

3.2.1 强化学习在金融风控中的应用

强化学习是一种让智能体通过与环境交互来学习最优策略的方法。在金融风控场景中：

• 智能体(Agent)：我们的风控AI Agent
• 环境(Environment)：金融市场、客户行为、监管环境等
• 状态(State)：当前的风险状况、市场环境、客户信息等
• 行动(Action)：批准交易、拒绝交易、提高利率、降低额度等
• 奖励(Reward)：决策带来的结果（如收益、损失、客户满意度等）

让我们用一个具体的例子来实现基于强化学习的风控决策模块：

import numpy as np
import pandas as pd
from collections import deque
import random
import json
from datetime import datetime

# 导入深度学习库（如果可用）
try:
    import tensorflow as tf
    from tensorflow.keras import layers, models, optimizers
    TF_AVAILABLE = True
except ImportError:
    TF_AVAILABLE = False
    print("TensorFlow不可用，将使用简化的Q-learning方法")

class RiskDecisionEnvironment:
    """风控决策环境"""
    
    def __init__(self, config=None):
        self.config = config or {
            'risk_weight': 0.6,  # 风险权重
            'profit_weight': 0.3,  # 收益权重
            'customer_satisfaction_weight': 0.1,  # 客户满意度权重
            'false_positive_cost': 100,  # 误报成本
            'false_negative_cost': 500,  # 漏报成本
            'true_positive_gain': 200,  # 正确识别风险的收益
            'true_negative_gain': 50,  # 正确识别正常的收益
        }
        
        # 定义可能的行动
        self.actions = {
            0: 'APPROVE',  # 批准
            1: 'REJECT',   # 拒绝
            2: 'REVIEW',   # 人工审核
            3: 'MONITOR',  # 监控
            4: 'CONDITIONAL_APPROVE'  # 有条件批准
        }
        self.action_space = len(self.actions)
        
        # 状态空间维度（风险分数、风险等级、交易金额、客户价值、历史行为等）
        self.state_dim = 10
    
    def reset(self):
        """重置环境到初始状态"""
        # 随机生成一个初始状态（实际应用中应该基于真实数据）
        state = np.random.rand(self.state_dim)
        state[0] = np.random.uniform(0, 1)  # 风险分数 0-1
        state[1] = np.random.randint(0, 4)   # 风险等级 0-3
        state[2] = np.random.uniform(10, 10000)  # 交易金额
        state[3] = np.random.uniform(0, 1)  # 客户价值分数 0-1
        state[4] = np.random.randint(0, 100)  # 客户历史交易次数
        state[5] = np.random.uniform(0, 0.3)  # 客户历史风险率
        state[6] = np.random.uniform(0, 1)  # 当前市场风险指数
        state[7] = np.random.uniform(0, 1)  # 客户忠诚度分数
        state[8] = np.random.randint(0, 2)  # 是否有异常标记
        state[9] = np.random.uniform(0, 1)  # 数据质量分数
        
        return state
    
    def step(self, state, action, true_label=None):
        """
        执行一个行动并观察结果
        
        参数:
            state: 当前状态
            action: 选择的行动
            true_label: 真实标签（1表示真实风险，0表示正常）
            
        返回:
            next_state: 下一个状态
            reward: 奖励
            done: 是否结束
            info: 额外信息
        """
        # 如果没有提供真实标签，根据状态模拟一个
        if true_label is None:
            risk_score = state[0]
            true_label = 1 if risk_score > 0.6 else 0
        
        # 计算奖励
        reward = self._calculate_reward(state, action, true_label)
        
        # 模拟下一个状态（简化版）
        next_state = np.copy(state)
        # 这里可以添加更真实的状态转移逻辑
        
        done = True  # 简化版：每个交易是一个独立的episode
        
        info = {
            'true_label': true_label,
            'action_taken': self.actions[action],
            'risk_score': state[0],
            'transaction_amount': state[2]
        }
        
        return next_state, reward, done, info
    
    def _calculate_reward(self, state, action, true_label):
        """
        计算奖励函数
        
        参数:
            state: 当前状态
            action: 选择的行动
            true_label: 真实标签
            
        返回:
            float: 奖励值
        """
        risk_score = state[0]
        transaction_amount = state[2]
        customer_value = state[3]
        
        # 基础奖励矩阵
        reward_matrix = {
            # (行动, 真实标签): 基础奖励
            (0, 0): self.config['true_negative_gain'],  # 批准正常交易
            (0, 1): -self.config['false_negative_cost'],  # 批准风险交易
            (1, 0): -self.config['false_positive_cost'],  # 拒绝正常交易
            (1, 1): self.config['true_positive_gain'],  # 拒绝风险交易
            (2, 0): 20,  # 人工审核正常交易
            (2, 1): 150,  # 人工审核风险交易
            (3, 0): 30,  # 监控正常交易
            (3, 1): 80,  # 监控风险交易
            (4, 0): 40,  # 有条件批准正常交易
            (4, 1): 100,  # 有条件批准风险交易
        }
        
        # 基础奖励
        base_reward = reward_matrix.get((action, true_label), 0)
        
        # 根据交易金额调整奖励
        amount_factor = min(transaction_amount / 1000, 2)  # 归一化，最高2倍
        
        # 根据客户价值调整奖励（影响客户满意度）
        customer_factor = 1 + customer_value * 0.5  # 高价值客户的决策影响更大
        
        # 计算最终奖励
        if action == 1 and true_label == 0:  # 拒绝正常交易
            # 对高价值客户，误拒的惩罚更大
            reward = base_reward * amount_factor * customer_factor
        elif action == 0 and true_label == 1:  # 批准风险交易
            # 大额交易的漏损惩罚更大
            reward = base_reward * amount_factor
        else:
            reward = base_reward * (1 + (amount_factor - 1) * 0.5)
        
        return reward

class RiskDecisionModule:
    """风控决策模块"""
    
    def __init__(self, use_deep_learning=False):
        self.use_deep_learning = use_deep_learning and TF_AVAILABLE
        
        # 初始化环境
        self.env = RiskDecisionEnvironment()
        
        # Q表或神经网络
        if self.use_deep_learning:
            self._build_dqn()
        else:
            # 使用离散化的Q表
            self.state_bins = self._create_state_bins()
            self.q_table = np.zeros((*[len(bins) for bins in self.state_bins], self.env.action_space))
        
        # 强化学习参数
        self.learning_rate = 0.01
        self.discount_factor = 0.95
        self.exploration_rate = 1.0
        self.exploration_decay = 0.995
        self.exploration_min = 0.01
        
        # 经验回放
        self.memory = deque(maxlen=2000)
        self.batch_size = 32
        
        # 策略参数
        self.is_trained = False
        self.performance_history = []
    
    def _create_state_bins(self):
        """创建状态离散化的分箱"""
        # 为每个状态维度创建分箱
        bins = [
            np.linspace(0, 1, 10),  # 风险分数
            np.arange(0, 5),  # 风险等级
            np.logspace(1, 5, 10),  # 交易金额（对数刻度）
            np.linspace(0, 1, 5),  # 客户价值
            np.arange(0, 101, 10),  # 交易次数
            np.linspace(0, 0.3, 5),  # 历史风险率
            np.linspace(0, 1, 5),  # 市场风险
            np.linspace(0, 1, 5),  # 客户忠诚度
            np.arange(0, 2),  # 异常标记
            np.linspace(0, 1, 5)  # 数据质量
        ]
        return bins
    
    def _discretize_state(self, state):
        """将连续状态离散化"""
        discretized = []
        for i, value in enumerate(state):
            discretized.append(np.digitize(value, self.state_bins[i]) - 1)
        return tuple(discretized)
    
    def _build_dqn(self):
        """构建深度Q网络"""
        model = models.Sequential()
        model.add(layers.Dense(64, activation='relu', input_dim=self.env.state_dim))
        model.add(layers.Dropout(0.2))
        model.add(layers.Dense(32, activation='relu'))
        model.add(layers.Dense(self.env.action_space, activation='linear'))
        model.compile(optimizer=optimizers.Adam(learning_rate=self.learning_rate), loss='mse')
        self.dqn_model = model
        
        # 目标网络
        self.target_model = models.clone_model(model)
        self.target_model.set_weights(model.get_weights())
        self.update_target_frequency = 100
        self.step_count = 0
    
    def remember(self, state, action, reward, next_state, done):
        """存储经验"""
        self.memory.append((state, action, reward, next_state, done))
    
    def select_action(self, state, use_exploration=True):
        """
        选择行动
        
        参数:
            state: 当前状态
            use_exploration: 是否使用探索
            
        返回:
            int: 选择的行动
        """
        if use_exploration and np.random.rand() <= self.exploration_rate:
            # 随机探索
            return random.randrange(self.env.action_space)
        
        if self.use_deep_learning:
            # 使用DQN
            q_values = self.dqn_model.predict(np.expand_dims(state, axis=0), verbose=0)[0]
        else:
            # 使用Q表
            discretized_state = self._discretize_state(state)
            q_values = self.q_table[discretized_state]
        
        return np.argmax(q_values)
    
    def replay(self):
        """经验回放学习"""
        if len(self.memory) < self.batch_size:
            return
        
        # 随机采样一批经验
        batch = random.sample(self.memory, self.batch_size)
        
        if self.use_deep_learning:
            states = np.array([exp[0] for exp in batch])
            next_states = np.array([exp[3] for exp in batch])
            
            # 预测当前状态和下一个状态的Q值
            q_values = self.dqn_model.predict(states, verbose=0)
            next_q_values = self.target_model.predict(next_states, verbose=0)
            
            # 更新Q值
            for i, (state, action, reward, next_state, done) in enumerate(batch):
                if done:
                    q_values[i][action] = reward
                else:
                    q_values[i][action] = reward + self.discount_factor * np.max(next_q_values[i])
            
            # 训练模型
            self.dqn_model.fit(states, q_values, epochs=1, verbose=0)
            
            # 更新目标网络
            self.step_count += 1
            if self.step_count % self.update_target_frequency == 0:
                self.target_model.set_weights(self.dqn_model.get_weights())
        else:
            # 使用Q表学习
            for state, action, reward, next_state, done in batch:
                discretized_state = self._discretize_state(state)
                discretized_next_state = self._discretize_state(next_state)
                
                if done:
                    target = reward
                else:
                    target = reward + self.discount_factor * np.max(self.q_table[discretized_next_state])
                
                # 更新Q表
                self.q_table[discretized_state][action] = (
                    (1 - self.learning_rate) * self.q_table[discretized_state][action] +
                    self.learning_rate * target
                )
        
        # 衰减探索率
        if self.exploration_rate > self.exploration_min:
            self.exploration_rate *= self.exploration_decay
    
    def train(self, num_episodes=1000, verbose=True):
        """
        训练决策模块
        
        参数:
            num_episodes: 训练回合数
            verbose: 是否打印训练过程
        """
        for episode in range(num_episodes):
            state = self.env.reset()
            total_reward = 0
            done = False
            steps = 0
            
            while not done:
                # 选择行动
                action = self.select_action(state)
                
                # 执行行动
                next_state, reward, done, info = self.env.step(state, action)