149_金融场景论文（论文中附有源码）：风险与欺诈检测 - 分析时序数据的LLM处理技术与实践实现

应用层: 风险监测仪表盘、异常事件管理、风险报告生成
    |
服务层: API网关、权限管理、服务编排
    |
模型层: LLM推理引擎、异常检测模型、知识图谱引擎
    |
特征层: 特征提取、特征工程、特征存储
    |
数据层: 交易数据库、客户数据库、外部数据源

class LLMEnhancedAnomalyDetection:
    def __init__(self, llm_model_path, ml_model_path, config):
        """
        初始化LLM增强的异常检测模型
        
        Args:
            llm_model_path: LLM模型路径
            ml_model_path: 机器学习模型路径
            config: 配置参数
        """
        self.llm = self._load_llm_model(llm_model_path)
        self.ml_model = self._load_ml_model(ml_model_path)
        self.config = config
        self.scaler = self._load_scaler(config['scaler_path'])
    
    def _load_llm_model(self, model_path):
        """加载LLM模型"""
        from transformers import AutoModelForCausalLM, AutoTokenizer
        tokenizer = AutoTokenizer.from_pretrained(model_path)
        model = AutoModelForCausalLM.from_pretrained(model_path)
        return {'model': model, 'tokenizer': tokenizer}
    
    def _load_ml_model(self, model_path):
        """加载机器学习模型"""
        import joblib
        return joblib.load(model_path)
    
    def _load_scaler(self, scaler_path):
        """加载数据缩放器"""
        import joblib
        return joblib.load(scaler_path)
    
    def preprocess_data(self, data):
        """预处理数据"""
        import pandas as pd
        import numpy as np
        
        # 处理时间戳
        df = pd.DataFrame(data)
        df['timestamp'] = pd.to_datetime(df['timestamp'])
        
        # 提取时间特征
        df['hour'] = df['timestamp'].dt.hour
        df['day_of_week'] = df['timestamp'].dt.dayofweek
        df['is_weekend'] = df['day_of_week'].apply(lambda x: 1 if x >= 5 else 0)
        
        # 提取统计特征
        rolling_features = df.groupby('account_id')['amount'].rolling(window=24).agg(
            ['mean', 'std', 'min', 'max']).reset_index()
        df = df.merge(rolling_features, on=['account_id', 'timestamp'], how='left')
        
        # 填充缺失值
        df = df.fillna(0)
        
        # 选择特征列
        feature_cols = ['amount', 'hour', 'day_of_week', 'is_weekend', 
                       'mean', 'std', 'min', 'max']
        
        return df[feature_cols]
    
    def generate_llm_prompt(self, transaction_data):
        """生成LLM提示"""
        prompt = f"""分析以下交易数据是否存在异常：
        交易ID: {transaction_data['transaction_id']}
        账户ID: {transaction_data['account_id']}
        交易金额: {transaction_data['amount']}
        交易时间: {transaction_data['timestamp']}
        交易类型: {transaction_data['transaction_type']}
        交易对手: {transaction_data['counterparty']}
        
        账户近期交易统计:
        平均交易金额: {transaction_data.get('mean', 0)}
        交易金额标准差: {transaction_data.get('std', 0)}
        最大交易金额: {transaction_data.get('max', 0)}
        
        请回答：
        1. 该交易是否存在异常？
        2. 异常的具体表现是什么？
        3. 可能的异常原因是什么？
        4. 风险评分（0-10分）是多少？"""
        return prompt
    
    def predict(self, data):
        """预测异常"""
        # 预处理数据
        features = self.preprocess_data(data)
        scaled_features = self.scaler.transform(features)
        
        # 传统模型预测
        ml_scores = self.ml_model.predict_proba(scaled_features)[:, 1]
        
        # LLM预测
        llm_scores = []
        for i, record in enumerate(data):
            # 准备交易数据
            transaction_data = record.copy()
            transaction_data.update({
                'mean': features.iloc[i]['mean'],
                'std': features.iloc[i]['std'],
                'max': features.iloc[i]['max']
            })
            
            # 生成提示
            prompt = self.generate_llm_prompt(transaction_data)
            
            # LLM推理
            inputs = self.llm['tokenizer'](prompt, return_tensors="pt")
            outputs = self.llm['model'].generate(**inputs, max_new_tokens=200)
            response = self.llm['tokenizer'].decode(outputs[0], skip_special_tokens=True)
            
            # 解析风险评分
            import re
            score_match = re.search(r'风险评分（0-10分）是多少？?\s*(\d+(?:\.\d+)?)', response)
            if score_match:
                llm_score = float(score_match.group(1)) / 10.0  # 归一化到0-1
            else:
                llm_score = 0.5  # 默认中等风险
            
            llm_scores.append(llm_score)
        
        # 融合评分
        combined_scores = [(ml * 0.4 + llm * 0.6) for ml, llm in zip(ml_scores, llm_scores)]
        
        return {
            'anomaly_scores': combined_scores,
            'is_anomaly': [score > self.config['threshold'] for score in combined_scores]
        }

def fine_tune_financial_llm(model_name, dataset_path, output_dir):
    """
    使用QLoRA微调金融领域LLM
    
    Args:
        model_name: 预训练模型名称
        dataset_path: 数据集路径
        output_dir: 输出目录
    """
    import transformers
    import peft
    import datasets
    import torch
    
    # 加载数据集
    dataset = datasets.load_from_disk(dataset_path)
    
    # 配置模型
    model = transformers.AutoModelForCausalLM.from_pretrained(
        model_name,
        load_in_4bit=True,
        torch_dtype=torch.bfloat16,
        device_map="auto"
    )
    
    # 配置tokenizer
    tokenizer = transformers.AutoTokenizer.from_pretrained(model_name)
    
    # 配置QLoRA
    peft_config = peft.LoraConfig(
        lora_alpha=16,
        lora_dropout=0.1,
        r=64,
        bias="none",
        task_type="CAUSAL_LM",
        target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj"]
    )
    
    # 创建PEFT模型
    model = peft.get_peft_model(model, peft_config)
    
    # 配置训练参数
    training_args = transformers.TrainingArguments(
        output_dir=output_dir,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        learning_rate=2e-4,
        logging_steps=100,
        max_steps=1000,
        save_strategy="steps",
        save_steps=500,
        optim="paged_adamw_32bit",
        fp16=True,
        push_to_hub=False
    )
    
    # 创建Trainer
    trainer = transformers.Trainer(
        model=model,
        args=training_args,
        train_dataset=dataset["train"],
        tokenizer=tokenizer,
        data_collator=transformers.DataCollatorForLanguageModeling(
            tokenizer=tokenizer,
            mlm=False
        )
    )
    
    # 开始训练
    trainer.train()
    
    # 保存模型
    model.save_pretrained(output_dir)
    tokenizer.save_pretrained(output_dir)
    
    return model, tokenizer

输入处理模块 → LLM推理模块 → 结果解析模块 → 输出生成模块

class FinancialRiskReasoningEngine:
    def __init__(self, llm_model, llm_tokenizer, config):
        """
        初始化金融风险推理引擎
        
        Args:
            llm_model: LLM模型
            llm_tokenizer: LLM分词器
            config: 配置参数
        """
        self.llm_model = llm_model
        self.llm_tokenizer = llm_tokenizer
        self.config = config
        self.risk_templates = self._load_risk_templates()
    
    def _load_risk_templates(self):
        """加载风险分析模板"""
        return {
            "risk_assessment": """请对以下金融交易进行风险评估：
{transaction_data}

请从以下几个方面进行分析：
1. 交易特征分析
2. 风险因素识别
3. 风险等级评定（低风险/中风险/高风险）
4. 风险缓解建议""",
            
            "fraud_detection": """请分析以下交易是否存在欺诈嫌疑：
{transaction_data}

请回答：
1. 是否存在欺诈嫌疑？
2. 欺诈指标有哪些？
3. 欺诈可能性评分（0-100）
4. 建议采取的措施""",
            
            "compliance_check": """请检查以下交易是否符合合规要求：
{transaction_data}

请检查：
1. KYC/AML合规性
2. 监管要求符合度
3. 合规风险等级
4. 合规改进建议"""
        }
    
    def process_request(self, request_data):
        """
        处理风险分析请求
        
        Args:
            request_data: 请求数据，包含交易信息和分析类型
            
        Returns:
            风险分析结果
        """
        # 提取请求参数
        analysis_type = request_data.get('analysis_type', 'risk_assessment')
        transaction_data = request_data.get('transaction_data', {})
        
        # 生成分析提示
        prompt = self._create_prompt(analysis_type, transaction_data)
        
        # LLM推理
        response = self._generate_llm_response(prompt)
        
        # 解析结果
        parsed_result = self._parse_response(response, analysis_type)
        
        # 增强结果
        enhanced_result = self._enhance_result(parsed_result, transaction_data)
        
        return enhanced_result
    
    def _create_prompt(self, analysis_type, transaction_data):
        """创建分析提示"""
        if analysis_type not in self.risk_templates:
            analysis_type = 'risk_assessment'
        
        # 格式化交易数据
        formatted_data = "\n".join([f"{key}: {value}" for key, value in transaction_data.items()])
        
        return self.risk_templates[analysis_type].format(transaction_data=formatted_data)
    
    def _generate_llm_response(self, prompt):
        """生成LLM响应"""
        inputs = self.llm_tokenizer(prompt, return_tensors="pt")
        outputs = self.llm_model.generate(
            **inputs,
            max_new_tokens=500,
            temperature=0.2,
            top_p=0.95,
            num_beams=1
        )
        
        return self.llm_tokenizer.decode(outputs[0], skip_special_tokens=True)
    
    def _parse_response(self, response, analysis_type):
        """解析LLM响应"""
        import re
        
        parsed_result = {
            'raw_response': response,
            'analysis_type': analysis_type
        }
        
        # 根据分析类型解析结果
        if analysis_type == 'risk_assessment':
            # 提取风险等级
            risk_level_match = re.search(r'风险等级评定\s*[:：]\s*([低中高]风险)', response)
            if risk_level_match:
                parsed_result['risk_level'] = risk_level_match.group(1)
            
            # 提取风险因素
            risk_factors_match = re.search(r'风险因素识别\s*[:：]\s*(.*?)(?=风险等级评定|$)', response, re.DOTALL)
            if risk_factors_match:
                parsed_result['risk_factors'] = risk_factors_match.group(1).strip()
                
        elif analysis_type == 'fraud_detection':
            # 提取欺诈可能性
            fraud_score_match = re.search(r'欺诈可能性评分\s*[:：]\s*(\d+)', response)
            if fraud_score_match:
                parsed_result['fraud_score'] = int(fraud_score_match.group(1))
            
            # 提取欺诈指标
            fraud_indicators_match = re.search(r'欺诈指标有哪些\s*[:：]\s*(.*?)(?=欺诈可能性评分|$)', response, re.DOTALL)
            if fraud_indicators_match:
                parsed_result['fraud_indicators'] = fraud_indicators_match.group(1).strip()
        
        elif analysis_type == 'compliance_check':
            # 提取合规风险等级
            compliance_risk_match = re.search(r'合规风险等级\s*[:：]\s*(.*?)(?=合规改进建议|$)', response, re.DOTALL)
            if compliance_risk_match:
                parsed_result['compliance_risk'] = compliance_risk_match.group(1).strip()
        
        return parsed_result
    
    def explain_risk_assessment(self, assessment_result, transaction_data):
        """
        生成风险评估解释
        
        Args:
            assessment_result: 风险评估结果
            transaction_data: 交易数据
            
        Returns:
            风险评估解释
        """
        # 生成解释提示
        explain_prompt = f"""
        请详细解释以下风险评估结果：

        评估结果：{assessment_result}

        交易数据：
        {"\n".join([f"{key}: {value}" for key, value in transaction_data.items()])}

        请从以下几个方面进行解释：
        1. 为什么给出这个风险等级？
        2. 主要风险点是什么？
        3. 这些风险点如何影响交易安全性？
        4. 如何理解和应用这些风险信息？"""
        
        # 获取解释
        inputs = self.llm_tokenizer(explain_prompt, return_tensors="pt")
        outputs = self.llm_model.generate(
            **inputs,
            max_new_tokens=800,
            temperature=0.2,
            top_p=0.95
        )
        
        explanation = self.llm_tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        return {
            'explanation': explanation,
            'timestamp': datetime.now().isoformat()
        }

class FinancialKnowledgeGraph:
    def __init__(self, neo4j_uri, neo4j_user, neo4j_password):
        """
        初始化金融知识图谱
        
        Args:
            neo4j_uri: Neo4j数据库URI
            neo4j_user: Neo4j用户名
            neo4j_password: Neo4j密码
        """
        from neo4j import GraphDatabase
        self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))
    
    def close(self):
        """关闭数据库连接"""
        self.driver.close()
    
    def create_entity(self, entity_type, entity_id, properties):
        """
        创建实体
        
        Args:
            entity_type: 实体类型
            entity_id: 实体ID
            properties: 实体属性
        """
        with self.driver.session() as session:
            # 构建属性字符串
            props_str = ", ".join([f"{key}: ${key}" for key in properties.keys()])
            
            # 构建查询
            query = f"CREATE (n:{entity_type} {{id: $entity_id, {props_str}}}) RETURN n"
            
            # 执行查询
            parameters = {"entity_id": entity_id, **properties}
            result = session.run(query, parameters)
            
            return result.single()[0]
    
    def create_relation(self, source_type, source_id, target_type, target_id, relation_type, properties=None):
        """
        创建关系
        
        Args:
            source_type: 源实体类型
            source_id: 源实体ID
            target_type: 目标实体类型
            target_id: 目标实体ID
            relation_type: 关系类型
            properties: 关系属性
        """
        with self.driver.session() as session:
            # 构建属性字符串
            props_str = ", ".join([f"{key}: ${key}" for key in (properties or {}).keys()])
            if props_str:
                props_str = ", " + props_str
            
            # 构建查询
            query = f"""
            MATCH (s:{source_type} {{id: $source_id}}), (t:{target_type} {{id: $target_id}})
            CREATE (s)-[r:{relation_type} {{id: $relation_id{props_str}}}]->(t)
            RETURN r
            """
            
            # 执行查询
            import uuid
            relation_id = str(uuid.uuid4())
            parameters = {
                "source_id": source_id,
                "target_id": target_id,
                "relation_id": relation_id,
                **(properties or {})
            }
            result = session.run(query, parameters)
            
            return result.single()[0]
    
    def find_suspicious_patterns(self, query_params):
        """
        查找可疑模式
        
        Args:
            query_params: 查询参数
            
        Returns:
            可疑模式列表
        """
        with self.driver.session() as session:
            # 构建查询
            query = """
            MATCH (p1:Person)-[:OWNS]->(a1:Account)-[t:TRANSACTS_WITH]->(a2:Account)<-[:OWNS]-(p2:Person)
            WHERE t.amount > $threshold AND t.frequency > $frequency
            RETURN p1, a1, t, a2, p2
            LIMIT $limit
            """
            
            # 默认参数
            params = {
                "threshold": 10000,
                "frequency": 3,
                "limit": 100,
                **query_params
            }
            
            # 执行查询
            result = session.run(query, params)
            
            # 处理结果
            patterns = []
            for record in result:
                patterns.append({
                    "person1": dict(record["p1"]),
                    "account1": dict(record["a1"]),
                    "transaction": dict(record["t"]),
                    "account2": dict(record["a2"]),
                    "person2": dict(record["p2"])
                })
            
            return patterns
    
    def get_entity_connections(self, entity_type, entity_id, depth=2):
        """
        获取实体的连接
        
        Args:
            entity_type: 实体类型
            entity_id: 实体ID
            depth: 查询深度
            
        Returns:
            实体连接图
        """
        with self.driver.session() as session:
            # 构建查询
            query = f"""
            MATCH path = (n:{entity_type} {{id: $entity_id}})-[*1..{depth}]-(m)
            RETURN path
            """
            
            # 执行查询
            result = session.run(query, {"entity_id": entity_id})
            
            # 处理结果
            connections = {
                "nodes": [],
                "edges": []
            }
            
            node_ids = set()
            edge_ids = set()
            
            for record in result:
                path = record["path"]
                
                # 提取节点
                for node in path.nodes:
                    node_data = dict(node)
                    node_id = node_data.get("id", str(node.identity))
                    if node_id not in node_ids:
                        connections["nodes"].append({
                            "id": node_id,
                            "labels": list(node.labels),
                            "properties": node_data
                        })
                        node_ids.add(node_id)
                
                # 提取关系
                for rel in path.relationships:
                    start_id = rel.start_node.get("id", str(rel.start_node.identity))
                    end_id = rel.end_node.get("id", str(rel.end_node.identity))
                    edge_id = f"{start_id}_{rel.type}_{end_id}"
                    
                    if edge_id not in edge_ids:
                        connections["edges"].append({
                            "id": edge_id,
                            "type": rel.type,
                            "source": start_id,
                            "target": end_id,
                            "properties": dict(rel)
                        })
                        edge_ids.add(edge_id)
            
            return connections
    
    def detect_fraud_rings(self, min_size=3):
        """
        检测欺诈团伙
        
        Args:
            min_size: 团伙最小规模
            
        Returns:
            欺诈团伙列表
        """
        with self.driver.session() as session:
            # 构建查询，检测密集连接的实体群组
            query = """
            CALL gds.louvain.stream('fraud_graph')
            YIELD nodeId, communityId, intermediateCommunityIds
            WITH communityId, count(*) AS size
            WHERE size >= $min_size
            MATCH (n)
            WHERE gds.util.asNode(gds.louvain.stream('fraud_graph')
                YIELD nodeId, communityId
                WHERE communityId = communityId).nodeId = id(n)
            RETURN communityId, collect(n) AS members
            """
            
            # 执行查询
            result = session.run(query, {"min_size": min_size})
            
            # 处理结果
            fraud_rings = []
            for record in result:
                members = [dict(member) for member in record["members"]]
                fraud_rings.append({
                    "community_id": record["communityId"],
                    "size": len(members),
                    "members": members
                })
            
            return fraud_rings
    
    def get_relevant_knowledge(self, query, limit=10):
        """
        获取相关知识
        
        Args:
            query: 查询文本
            limit: 返回结果数量限制
            
        Returns:
            相关知识列表
        """
        with self.driver.session() as session:
            # 全文搜索查询
            search_query = """
            CALL db.index.fulltext.queryNodes('entity_search', $query, {limit: $limit})
            YIELD node, score
            RETURN node, score
            """
            
            # 执行查询
            result = session.run(search_query, {"query": query, "limit": limit})
            
            # 处理结果
            knowledge = []
            for record in result:
                node = record["node"]
                knowledge.append({
                    "type": list(node.labels)[0],
                    "properties": dict(node),
                    "score": record["score"]
                })
            
            # 获取相关关系
            if knowledge:
                node_ids = [item["properties"].get("id") for item in knowledge if "id" in item["properties"]]
                
                if node_ids:
                    # 查询欺诈案例
                    case_query = """
                    MATCH (c:Case)-[:INVOLVES]->(n)
                    WHERE n.id IN $node_ids
                    RETURN c, n
                    LIMIT $limit
                    """
                    
                    case_result = session.run(case_query, {"node_ids": node_ids, "limit": limit})
                    
                    for record in case_result:
                        knowledge.append({
                            "type": "Case",
                            "properties": dict(record["c"]),
                            "related_to": dict(record["n"])
                        })
                    
                    # 查询客户风险事件
                    event_query = """
                    MATCH (e:RiskEvent)-[:AFFECTS]->(n)
                    WHERE n.id IN $node_ids
                    RETURN e, n
                    LIMIT $limit
                    """
                    
                    event_result = session.run(event_query, {"node_ids": node_ids, "limit": limit})
                    
                    for record in event_result:
                        knowledge.append({
                            "type": "RiskEvent",
                            "properties": dict(record["e"]),
                            "related_to": dict(record["n"])
                        })
            
            return knowledge

class KnowledgeGraphEnhancedLLM:
    def __init__(self, llm_model, llm_tokenizer, kg_client, config):
        """
        初始化知识图谱增强的LLM
        
        Args:
            llm_model: LLM模型
            llm_tokenizer: LLM分词器
            kg_client: 知识图谱客户端
            config: 配置参数
        """
        self.llm_model = llm_model
        self.llm_tokenizer = llm_tokenizer
        self.kg_client = kg_client
        self.config = config
    
    def enhance_with_knowledge(self, query, context=None):
        """
        使用知识图谱增强查询
        
        Args:
            query: 查询文本
            context: 上下文信息
            
        Returns:
            增强后的查询和知识
        """
        # 从查询中提取实体
        entities = self._extract_entities(query)
        
        # 从知识图谱获取相关知识
        relevant_knowledge = []
        for entity in entities:
            knowledge = self.kg_client.get_relevant_knowledge(entity, limit=5)
            relevant_knowledge.extend(knowledge)
        
        # 去重和排序
        unique_knowledge = self._deduplicate_knowledge(relevant_knowledge)
        sorted_knowledge = sorted(unique_knowledge, key=lambda x: x.get('score', 0), reverse=True)
        
        # 构建增强提示
        enhanced_prompt = self._build_enhanced_prompt(query, sorted_knowledge[:10], context)
        
        return {
            "enhanced_prompt": enhanced_prompt,
            "extracted_entities": entities,
            "relevant_knowledge": sorted_knowledge[:10]
        }
    
    def _extract_entities(self, text):
        """
        从文本中提取实体
        
        Args:
            text: 输入文本
            
        Returns:
            实体列表
        """
        # 使用LLM提取实体
        extract_prompt = f"""
        请从以下文本中提取金融相关的实体，包括客户名称、账户ID、交易对手、产品名称等：
        
        {text}
        
        请以JSON格式返回提取的实体列表：
        {"entities": ["实体1", "实体2", ...]}
        """
        
        inputs = self.llm_tokenizer(extract_prompt, return_tensors="pt")
        outputs = self.llm_model.generate(
            **inputs,
            max_new_tokens=200,
            temperature=0.1,
            top_p=0.9
        )
        
        response = self.llm_tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        # 解析实体
        import json
        try:
            # 提取JSON部分
            json_match = re.search(r'\{[^}]*\}', response)
            if json_match:
                entities_data = json.loads(json_match.group())
                return entities_data.get('entities', [])
        except Exception as e:
            print(f"解析实体失败: {e}")
        
        return []
    
    def _deduplicate_knowledge(self, knowledge_list):
        """
        去重知识列表
        
        Args:
            knowledge_list: 知识列表
            
        Returns:
            去重后的知识列表
        """
        seen = set()
        unique = []
        
        for item in knowledge_list:
            # 生成唯一标识符
            if 'id' in item.get('properties', {}):
                key = f"{item['type']}:{item['properties']['id']}"
            else:
                key = f"{item['type']}:{str(item['properties'])}[:50]"
            
            if key not in seen:
                seen.add(key)
                unique.append(item)
        
        return unique
    
    def _build_enhanced_prompt(self, query, knowledge_list, context=None):
        """
        构建增强提示
        
        Args:
            query: 原始查询
            knowledge_list: 相关知识列表
            context: 上下文信息
            
        Returns:
            增强后的提示
        """
        # 构建知识上下文
        knowledge_context = ""
        if knowledge_list:
            knowledge_context += "基于以下相关知识：\n"
            for i, knowledge in enumerate(knowledge_list, 1):
                entity_type = knowledge['type']
                properties = knowledge['properties']
                
                knowledge_context += f"\n[{entity_type}]\n"
                for key, value in properties.items():
                    if key != 'id':
                        knowledge_context += f"  {key}: {value}\n"
        
        # 构建完整提示
        enhanced_prompt = f"""
        请回答以下问题，并考虑提供的相关知识和上下文信息。
        
        问题: {query}
        
        {knowledge_context}
        
        {f"上下文: {context}\n" if context else ""}
        
        请提供详细、准确的回答，并基于提供的知识进行推理。
        """
        
        return enhanced_prompt
    
    def analyze_with_knowledge(self, query, context=None):
        """
        使用知识增强进行分析
        
        Args:
            query: 查询文本
            context: 上下文信息
            
        Returns:
            分析结果
        """
        # 知识增强
        enhancement_result = self.enhance_with_knowledge(query, context)
        
        # LLM推理
        inputs = self.llm_tokenizer(enhancement_result['enhanced_prompt'], return_tensors="pt")
        outputs = self.llm_model.generate(
            **inputs,
            max_new_tokens=1000,
            temperature=0.2,
            top_p=0.95,
            num_beams=1
        )
        
        response = self.llm_tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        # 提取回答部分
        answer_start = response.find("请提供详细、准确的回答")
        if answer_start != -1:
            answer = response[answer_start + len("请提供详细、准确的回答"):].strip()
        else:
            answer = response
        
        return {
            "query": query,
            "answer": answer,
            "used_knowledge": enhancement_result['relevant_knowledge'],
            "extracted_entities": enhancement_result['extracted_entities']
        }

数据采集层: Kafka, Flume
    |
流处理层: Flink, Spark Streaming
    |
风险处理层: LLM模型, 机器学习模型, 规则引擎
    |
警报生成层: 警报服务, 通知服务

class RealTimeRiskMonitor:
    def __init__(self, config):
        """
        初始化实时风险监控系统
        
        Args:
            config: 配置参数
        """
        self.config = config
        self.llm_detector = self._init_llm_detector()
        self.ml_detector = self._init_ml_detector()
        self.rules_engine = self._init_rules_engine()
        self.alert_service = self._init_alert_service()
        self.kafka_consumer = self._init_kafka_consumer()
        self.mongo_client = self._init_mongo_client()
        self.is_running = False
    
    def _init_llm_detector(self):
        """初始化LLM检测器"""
        from transformers import AutoModelForCausalLM, AutoTokenizer
        
        model_name = self.config['llm_model_name']
        tokenizer = AutoTokenizer.from_pretrained(model_name)
        model = AutoModelForCausalLM.from_pretrained(model_name)
        
        return {
            'model': model,
            'tokenizer': tokenizer
        }
    
    def _init_ml_detector(self):
        """初始化机器学习检测器"""
        import joblib
        
        model_path = self.config['ml_model_path']
        return joblib.load(model_path)
    
    def _init_rules_engine(self):
        """初始化规则引擎"""
        rules = self.config['rules']
        
        def evaluate_rules(transaction):
            """评估规则"""
            violations = []
            
            for rule_name, rule_config in rules.items():
                condition = rule_config['condition']
                threshold = rule_config['threshold']
                
                # 简单规则评估
                if condition == 'amount_gt' and transaction['amount'] > threshold:
                    violations.append({
                        'rule': rule_name,
                        'message': f"交易金额超过阈值: {threshold}",
                        'severity': rule_config['severity']
                    })
                elif condition == 'velocity_check':
                    # 这里应该有更复杂的逻辑
                    pass
            
            return violations
        
        return evaluate_rules
    
    def _init_alert_service(self):
        """初始化警报服务"""
        class AlertService:
            def __init__(self, config):
                self.config = config
                self.alert_threshold = config['alert_threshold']
                
            def create_alert(self, alert_data):
                """创建警报"""
                print(f"创建警报: {alert_data}")
                # 这里应该有实际的警报创建逻辑
                return alert_data
        
        return AlertService(self.config)
    
    def _init_kafka_consumer(self):
        """初始化Kafka消费者"""
        from kafka import KafkaConsumer
        import json
        
        consumer = KafkaConsumer(
            self.config['kafka_topic'],
            bootstrap_servers=self.config['kafka_bootstrap_servers'],
            group_id=self.config['kafka_group_id'],
            auto_offset_reset='latest',
            value_deserializer=lambda x: json.loads(x.decode('utf-8'))
        )
        
        return consumer
    
    def _init_mongo_client(self):
        """初始化MongoDB客户端"""
        from pymongo import MongoClient
        
        client = MongoClient(self.config['mongo_uri'])
        return client[self.config['mongo_db']]
    
    def start(self):
        """启动监控系统"""
        self.is_running = True
        print("启动实时风险监控系统...")
        
        try:
            for message in self.kafka_consumer:
                if not self.is_running:
                    break
                
                # 处理消息
                transaction = message.value
                self._process_transaction(transaction)
        except KeyboardInterrupt:
            print("监控系统被用户中断")
        except Exception as e:
            print(f"监控系统异常: {e}")
        finally:
            self.stop()
    
    def stop(self):
        """停止监控系统"""
        self.is_running = False
        print("停止实时风险监控系统...")
        
        # 关闭资源
        if hasattr(self, 'kafka_consumer'):
            self.kafka_consumer.close()
        
        if hasattr(self, 'mongo_client'):
            self.mongo_client.client.close()
    
    def _process_transaction(self, transaction):
        """
        处理交易数据
        
        Args:
            transaction: 交易数据
        """
        try:
            # 1. 规则引擎检测
            rule_violations = self.rules_engine(transaction)
            
            # 2. 机器学习模型检测
            ml_score = self._detect_with_ml(transaction)
            
            # 3. LLM检测
            llm_score = self._detect_with_llm(transaction)
            
            # 4. 融合风险评分
            risk_score = self._combine_scores(rule_violations, ml_score, llm_score)
            
            # 5. 处理结果
            self._process_result(transaction, risk_score, rule_violations)
            
        except Exception as e:
            print(f"处理交易失败: {e}")
    
    def _detect_with_ml(self, transaction):
        """使用机器学习模型检测风险"""
        # 准备特征
        features = self._prepare_features(transaction)
        
        # 预测
        score = self.ml_detector.predict_proba([features])[0][1]
        
        return score
    
    def _detect_with_llm(self, transaction):
        """使用LLM检测风险"""
        # 生成提示
        prompt = self._generate_llm_prompt(transaction)
        
        # 推理
        inputs = self.llm_detector['tokenizer'](prompt, return_tensors="pt")
        outputs = self.llm_detector['model'].generate(
            **inputs,
            max_new_tokens=100,
            temperature=0.2,
            top_p=0.95
        )
        
        response = self.llm_detector['tokenizer'].decode(outputs[0], skip_special_tokens=True)
        
        # 解析风险评分
        import re
        score_match = re.search(r'风险评分\s*[:：]\s*(\d+(?:\.\d+)?)', response)
        
        if score_match:
            score = float(score_match.group(1)) / 10.0  # 假设评分范围是0-10
        else:
            score = 0.5  # 默认中等风险
        
        return score
    
    def _prepare_features(self, transaction):
        """准备特征"""
        # 简化版本，实际应该更复杂
        return [
            transaction.get('amount', 0),
            transaction.get('hour_of_day', 0),
            transaction.get('day_of_week', 0),
            # 更多特征...
        ]
    
    def _generate_llm_prompt(self, transaction):
        """生成LLM提示"""
        return f"""
        请分析以下金融交易是否存在风险：
        
        {"\n".join([f"{key}: {value}" for key, value in transaction.items()])}
        
        请给出风险评分（0-10分）：
        """
    
    def _combine_scores(self, rule_violations, ml_score, llm_score):
        """融合风险评分"""
        # 规则违规权重
        rule_weight = 0.0
        for violation in rule_violations:
            if violation['severity'] == 'high':
                rule_weight += 0.4
            elif violation['severity'] == 'medium':
                rule_weight += 0.2
            else:
                rule_weight += 0.1
        
        rule_weight = min(rule_weight, 0.5)  # 最大权重0.5
        
        # 融合评分
        combined_score = (rule_weight + ml_score * 0.25 + llm_score * 0.25)
        
        return min(combined_score, 1.0)  # 限制在0-1范围
    
    def _process_result(self, transaction, risk_score, rule_violations):
        """处理检测结果"""
        # 存储结果
        from datetime import datetime
        result = {
            'transaction_id': transaction.get('transaction_id'),
            'timestamp': datetime.now().isoformat(),
            'risk_score': risk_score,
            'rule_violations': rule_violations,
            'transaction_data': transaction
        }
        
        self.mongo_client['risk_results'].insert_one(result)
        
        # 生成警报
        if risk_score >= self.config['alert_threshold']:
            self._send_alert(result)
        
        # 高风险交易审核
        if risk_score >= self.config['review_threshold']:
            self._send_for_review(result)
    
    def _send_alert(self, result):
        """发送警报"""
        import uuid
        alert_data = {
            'alert_id': str(uuid.uuid4()),
            'timestamp': result['timestamp'],
            'risk_score': result['risk_score'],
            'transaction_id': result['transaction_id'],
            'alert_type': 'high_risk_transaction',
            'description': f"高风险交易检测，风险评分: {result['risk_score']:.2f}",
            'rule_violations': [v['rule'] for v in result['rule_violations']],
            'status': 'new'
        }
        
        # 创建警报
        self.alert_service.create_alert(alert_data)
        
        # 记录警报
        self.mongo_client['alerts'].insert_one(alert_data)
    
    def _send_for_review(self, result):
        """发送审核"""
        import uuid
        review_data = {
            'review_id': str(uuid.uuid4()),
            'timestamp': result['timestamp'],
            'risk_score': result['risk_score'],
            'transaction_id': result['transaction_id'],
            'status': 'pending',
            'priority': 'high' if result['risk_score'] >= 0.8 else 'medium'
        }
        
        # 记录审核请求
        self.mongo_client['reviews'].insert_one(review_data)
    
    def _send_result(self, result):
        """发送结果"""
        # 这里可以添加发送结果到其他系统的逻辑
        pass
    
    def _log_high_risk(self, result):
        """记录高风险交易"""
        if result['risk_score'] >= 0.9:
            print(f"⚠️  极高风险交易: {result['transaction_id']}, 评分: {result['risk_score']:.2f}")

# 初始化异常检测模型
config = {
    'threshold': 0.7,
    'llm_model_path': 'path/to/llm/model',
    'ml_model_path': 'path/to/ml/model',
    'scaler_path': 'path/to/scaler'
}

anomaly_detector = LLMEnhancedAnomalyDetection(
    config['llm_model_path'],
    config['ml_model_path'],
    config
)

# 初始化监控系统
monitor_config = {
    'kafka_bootstrap_servers': ['localhost:9092'],
    'kafka_topic': 'financial_transactions',
    'kafka_group_id': 'risk_monitor_group',
    'mongo_uri': 'mongodb://localhost:27017',
    'mongo_db': 'risk_management',
    'llm_model_name': 'path/to/llm/model',
    'ml_model_path': 'path/to/ml/model',
    'alert_threshold': 0.7,
    'review_threshold': 0.8,
    'rules': {
        'high_amount_rule': {
            'condition': 'amount_gt',
            'threshold': 10000,
            'severity': 'high'
        },
        'velocity_rule': {
            'condition': 'velocity_check',
            'threshold': 5,
            'severity': 'medium'
        }
    }
}

monitor = RealTimeRiskMonitor(monitor_config)

# 启动监控系统
if __name__ == "__main__":
    try:
        monitor.start()
    except KeyboardInterrupt:
        monitor.stop()
        print("系统已停止")

def select_and_optimize_llm(task_type, budget_constraint=None):
    """
    根据任务类型和预算约束选择并优化LLM模型
    
    Args:
        task_type: 任务类型，如'real_time_monitoring', 'deep_analysis', 'compliance_check'
        budget_constraint: 预算约束，如'low', 'medium', 'high'
        
    Returns:
        优化后的模型和配置
    """
    import torch
    from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
    from peft import LoraConfig, get_peft_model
    
    # 根据任务类型和预算约束选择模型
    model_choices = {
        'real_time_monitoring': {
            'low': 'microsoft/phi-2',
            'medium': 'meta-llama/Llama-2-7b-chat-hf',
            'high': 'meta-llama/Llama-2-13b-chat-hf'
        },
        'deep_analysis': {
            'low': 'meta-llama/Llama-2-7b-chat-hf',
            'medium': 'meta-llama/Llama-2-13b-chat-hf',
            'high': 'meta-llama/Llama-2-70b-chat-hf'
        },
        'compliance_check': {
            'low': 'meta-llama/Llama-2-7b-chat-hf',
            'medium': 'meta-llama/Llama-2-13b-chat-hf',
            'high': 'meta-llama/Llama-2-70b-chat-hf'
        }
    }
    
    # 默认预算约束
    if budget_constraint not in ['low', 'medium', 'high']:
        budget_constraint = 'medium'
    
    # 获取模型名称
    model_name = model_choices[task_type][budget_constraint]
    
    print(f"选择模型: {model_name} 用于任务: {task_type}，预算约束: {budget_constraint}")
    
    # 配置优化选项
    optimize_config = {
        'quantization': True,
        'use_lora': True,
        'device_map': 'auto',
        'torch_dtype': torch.bfloat16
    }
    
    # 加载tokenizer
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    
    # 配置量化
    if optimize_config['quantization']:
        bnb_config = BitsAndBytesConfig(
            load_in_4bit=True,
            bnb_4bit_compute_dtype=optimize_config['torch_dtype'],
            bnb_4bit_use_double_quant=True,
            bnb_4bit_quant_type="nf4"
        )
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            quantization_config=bnb_config,
            device_map=optimize_config['device_map'],
            torch_dtype=optimize_config['torch_dtype']
        )
    else:
        model = AutoModelForCausalLM.from_pretrained(
            model_name,
            device_map=optimize_config['device_map'],
            torch_dtype=optimize_config['torch_dtype']
        )
    
    # 配置LoRA
    if optimize_config['use_lora']:
        lora_config = LoraConfig(
            r=16,
            lora_alpha=32,
            target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
            lora_dropout=0.05,
            bias="none",
            task_type="CAUSAL_LM"
        )
        model = get_peft_model(model, lora_config)
    
    # 打印模型信息
    print(f"模型加载完成。参数总数: {sum(p.numel() for p in model.parameters())/1e9:.2f}B")
    if optimize_config['use_lora']:
        trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
        print(f"可训练参数: {trainable_params/1e6:.2f}M ({trainable_params/sum(p.numel() for p in model.parameters())*100:.2f}%)")
    
    return {
        'model': model,
        'tokenizer': tokenizer,
        'config': optimize_config
    }

# 示例：为实时监控任务选择模型
monitor_model = select_and_optimize_llm('real_time_monitoring', 'medium')

# 示例：为深度分析任务选择模型
analysis_model = select_and_optimize_llm('deep_analysis', 'high')

def finetune_llm_for_financial_risk(model, tokenizer, dataset_path, output_dir):
    """
    对LLM进行金融风险领域适应微调
    
    Args:
        model: 预训练模型
        tokenizer: 分词器
        dataset_path: 数据集路径
        output_dir: 输出目录
    """
    import transformers
    import datasets
    import torch
    import os
    
    # 确保输出目录存在
    os.makedirs(output_dir, exist_ok=True)
    
    # 加载数据集
    dataset = datasets.load_from_disk(dataset_path)
    print(f"数据集加载完成: {len(dataset['train'])}条训练数据, {len(dataset['test'])}条测试数据")
    
    # 预处理数据集
    def preprocess_function(examples):
        # 格式化输入和输出
        inputs = []
        for i in range(len(examples['transaction_data'])):
            # 将交易数据转换为字符串
            transaction_str = "\n".join([f"{k}: {v}" for k, v in examples['transaction_data'][i].items()])
            
            # 构建指令格式
            input_text = f"""### 指令：
分析以下金融交易是否存在风险，并给出风险评分和理由。

### 输入：
{transaction_str}

### 输出：
"""
            inputs.append(input_text)
        
        # 处理目标输出
        targets = [f"{risk_level}\n风险评分：{score}\n理由：{reason}" 
                  for risk_level, score, reason in 
                  zip(examples['risk_level'], examples['risk_score'], examples['analysis_reason'])]
        
        # 组合输入和输出
        combined = [f"{inp}{tgt}" for inp, tgt in zip(inputs, targets)]
        
        # 标记化
        tokenized = tokenizer(combined, padding="max_length", truncation=True, max_length=1024)
        
        # 设置标签（复制输入作为标签，用于因果语言模型训练）
        tokenized["labels"] = tokenized["input_ids"].copy()
        
        return tokenized
    
    # 应用预处理
    tokenized_dataset = dataset.map(preprocess_function, batched=True)
    
    # 配置训练参数
    training_args = transformers.TrainingArguments(
        output_dir=output_dir,
        per_device_train_batch_size=4,
        gradient_accumulation_steps=4,
        learning_rate=2e-4,
        weight_decay=0.01,
        logging_steps=100,
        max_steps=2000,
        save_strategy="steps",
        save_steps=500,
        evaluation_strategy="steps",
        eval_steps=500,
        fp16=True,
        bf16=False,
        optim="paged_adamw_32bit",
        lr_scheduler_type="cosine",
        warmup_steps=100,
        load_best_model_at_end=True,
        metric_for_best_model="eval_loss",
        greater_is_better=False
    )
    
    # 定义评估指标
    def compute_metrics(eval_pred):
        import numpy as np
        logits, labels = eval_pred
        predictions = np.argmax(logits, axis=-1)
        
        # 由于是生成任务，这里简单计算准确率
        accuracy = np.mean(predictions == labels)
        
        return {
            "accuracy": accuracy,
            "eval_loss": 0.0  # 实际应该计算交叉熵损失
        }
    
    # 创建Trainer
    trainer = transformers.Trainer(
        model=model,
        args=training_args,
        train_dataset=tokenized_dataset["train"],
        eval_dataset=tokenized_dataset["test"],
        tokenizer=tokenizer,
        compute_metrics=compute_metrics,
        data_collator=transformers.DataCollatorForLanguageModeling(
            tokenizer=tokenizer,
            mlm=False
        )
    )
    
    # 开始训练
    print("开始微调训练...")
    trainer.train()
    
    # 保存最佳模型
    print("训练完成，保存模型...")
    model.save_pretrained(os.path.join(output_dir, "best_model"))
    tokenizer.save_pretrained(os.path.join(output_dir, "best_model"))
    
    # 评估模型
    print("评估模型...")
    eval_result = trainer.evaluate()
    print(f"评估结果: {eval_result}")
    
    # 保存评估结果
    import json
    with open(os.path.join(output_dir, "evaluation_result.json"), "w") as f:
        json.dump(eval_result, f, indent=2)
    
    return model, eval_result

# 示例：准备微调数据集
def prepare_finetuning_dataset(sample_size=1000):
    """准备微调数据集示例"""
    import random
    import pandas as pd
    from datasets import Dataset, DatasetDict
    
    # 生成示例数据
    transaction_types = ["transfer", "payment", "deposit", "withdrawal", "investment"]
    risk_levels = ["低风险", "中风险", "高风险"]
    
    data = []
    for i in range(sample_size):
        # 生成随机交易数据
        transaction_data = {
            "transaction_id": f"TX{i:010d}",
            "account_id": f"ACC{random.randint(10000, 99999)}",
            "amount": round(random.uniform(10, 100000), 2),
            "currency": "CNY",
            "transaction_type": random.choice(transaction_types),
            "timestamp": pd.Timestamp.now() - pd.Timedelta(days=random.randint(0, 365)),
            "location": random.choice(["北京", "上海", "广州", "深圳", "杭州"]),
            "device_id": f"DEV{random.randint(1000, 9999)}",
            "ip_address": f"{random.randint(1, 255)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(0, 255)}"
        }
        
        # 生成风险评估
        risk_level = random.choice(risk_levels)
        
        # 根据风险等级设置评分和理由
        if risk_level == "高风险":
            risk_score = round(random.uniform(7, 10), 1)
            reasons = [
                "交易金额异常高",
                "交易地点与账户历史不符",
                "短时间内频繁大额交易",
                "交易对手账户存在可疑活动",
                "交易模式与历史行为差异大"
            ]
            analysis_reason = "；".join(random.sample(reasons, 2))
        elif risk_level == "中风险":
            risk_score = round(random.uniform(4, 7), 1)
            reasons = [
                "交易金额略高于历史平均",
                "非常规交易时间",
                "使用新设备进行交易",
                "首次与该交易对手交易",
                "交易类型与账户常用类型不同"
            ]
            analysis_reason = "；".join(random.sample(reasons, 1))
        else:  # 低风险
            risk_score = round(random.uniform(0, 4), 1)
            analysis_reason = "交易符合账户历史行为模式，未发现异常"
        
        data.append({
            "transaction_data": transaction_data,
            "risk_level": risk_level,
            "risk_score": risk_score,
            "analysis_reason": analysis_reason
        })
    
    # 创建数据集
    df = pd.DataFrame(data)
    
    # 分割训练集和测试集
    train_size = int(0.8 * len(df))
    train_df = df[:train_size]
    test_df = df[train_size:]
    
    # 转换为Hugging Face数据集格式
    train_dataset = Dataset.from_pandas(train_df)
    test_dataset = Dataset.from_pandas(test_df)
    
    dataset = DatasetDict({
        "train": train_dataset,
        "test": test_dataset
    })
    
    # 保存数据集
    dataset.save_to_disk("./financial_risk_dataset")
    
    return dataset

# 准备数据集
dataset = prepare_finetuning_dataset(sample_size=500)

# 微调模型
# model, tokenizer = load_pretrained_model("meta-llama/Llama-2-7b-chat-hf")
# finetuned_model, eval_result = finetune_llm_for_financial_risk(
#     model, tokenizer, "./financial_risk_dataset", "./finetuned_models/risk_analysis"
# )

# docker-compose.yml
version: '3.8'

services:
  # 风险分析API服务
  risk-analysis-api:
    build:
      context: ./risk-analysis-api
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - PYTHONUNBUFFERED=1
      - LLM_MODEL_PATH=/models/llm-model
      - MONGO_URI=mongodb://mongo:27017/risk_management
      - KAFKA_BOOTSTRAP_SERVERS=kafka:9092
    volumes:
      - ./models:/models
    depends_on:
      - mongo
      - kafka
    restart: unless-stopped
    deploy:
      resources:
        limits:
          cpus: '4'
          memory: 8G

  # 实时监控服务
  real-time-monitor:
    build:
      context: ./real-time-monitor
      dockerfile: Dockerfile
    environment:
      - PYTHONUNBUFFERED=1
      - KAFKA_BOOTSTRAP_SERVERS=kafka:9092
      - MONGO_URI=mongodb://mongo:27017/risk_management
      - ALERT_SERVICE_URL=http://alert-service:8080
    depends_on:
      - kafka
      - mongo
      - alert-service
    restart: unless-stopped
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 4G

  # 警报服务
  alert-service:
    build:
      context: ./alert-service
      dockerfile: Dockerfile
    ports:
      - "8080:8080"
    environment:
      - SPRING_DATA_MONGODB_URI=mongodb://mongo:27017/risk_management
    depends_on:
      - mongo
    restart: unless-stopped

  # MongoDB数据库
  mongo:
    image: mongo:5.0
    ports:
      - "27017:27017"
    volumes:
      - mongo-data:/data/db
    restart: unless-stopped
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 4G

  # Kafka消息队列
  kafka:
    image: confluentinc/cp-kafka:7.0.0
    ports:
      - "9092:9092"
    environment:
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
    depends_on:
      - zookeeper
    restart: unless-stopped

  # Zookeeper
  zookeeper:
    image: confluentinc/cp-zookeeper:7.0.0
    ports:
      - "2181:2181"
    environment:
      - ZOOKEEPER_CLIENT_PORT=2181
    restart: unless-stopped

  # 前端应用
  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile
    ports:
      - "80:80"
    depends_on:
      - risk-analysis-api
    restart: unless-stopped

volumes:
  mongo-data:

# risk-analysis-api-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: risk-analysis-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: risk-analysis-api
  template:
    metadata:
      labels:
        app: risk-analysis-api
    spec:
      containers:
      - name: risk-analysis-api
        image: your-registry/risk-analysis-api:latest
        ports:
        - containerPort: 8000
        resources:
          limits:
            cpu: "4"
            memory: "8Gi"
          requests:
            cpu: "2"
            memory: "4Gi"
        env:
        - name: PYTHONUNBUFFERED
          value: "1"
        - name: LLM_MODEL_PATH
          value: "/models/llm-model"
        - name: MONGO_URI
          valueFrom:
            secretKeyRef:
              name: mongo-credentials
              key: uri
        - name: KAFKA_BOOTSTRAP_SERVERS
          value: "kafka:9092"
        volumeMounts:
        - name: model-volume
          mountPath: /models
      volumes:
      - name: model-volume
        persistentVolumeClaim:
          claimName: model-storage-pvc
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 100
            podAffinityTerm:
              labelSelector:
                matchExpressions:
                - key: app
                  operator: In
                  values:
                  - risk-analysis-api
              topologyKey: "kubernetes.io/hostname"

# prometheus_monitoring.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# 定义指标
REQUEST_COUNT = Counter('risk_analysis_requests_total', 'Total number of risk analysis requests', ['endpoint', 'method'])
REQUEST_LATENCY = Histogram('risk_analysis_request_latency_seconds', 'Request latency in seconds', ['endpoint'])
RISK_SCORE = Histogram('risk_analysis_score', 'Risk score distribution', buckets=[0, 2, 4, 6, 8, 10])
ANOMALY_COUNT = Counter('anomaly_detection_total', 'Total number of anomalies detected', ['severity'])
MODEL_LOADING_TIME = Gauge('model_loading_time_seconds', 'Time taken to load models')
MEMORY_USAGE = Gauge('memory_usage_bytes', 'Memory usage of the application')

class PrometheusMonitoring:
    def __init__(self, port=8000):
        """
        初始化Prometheus监控
        
        Args:
            port: 指标暴露端口
        """
        self.port = port
        self.started = False
    
    def start(self):
        """启动监控服务器"""
        if not self.started:
            start_http_server(self.port)
            self.started = True
            print(f"Prometheus监控服务器已启动，端口: {self.port}")
    
    def observe_request(self, endpoint, method, latency):
        """
        记录请求信息
        
        Args:
            endpoint: API端点
            method: HTTP方法
            latency: 请求延迟（秒）
        """
        REQUEST_COUNT.labels(endpoint=endpoint, method=method).inc()
        REQUEST_LATENCY.labels(endpoint=endpoint).observe(latency)
    
    def observe_risk_score(self, score):
        """
        记录风险评分
        
        Args:
            score: 风险评分
        """
        RISK_SCORE.observe(score)
    
    def observe_anomaly(self, severity):
        """
        记录异常检测
        
        Args:
            severity: 异常严重程度
        """
        ANOMALY_COUNT.labels(severity=severity).inc()
    
    def set_model_loading_time(self, time_taken):
        """
        设置模型加载时间
        
        Args:
            time_taken: 加载时间（秒）
        """
        MODEL_LOADING_TIME.set(time_taken)
    
    def set_memory_usage(self, usage):
        """
        设置内存使用量
        
        Args:
            usage: 内存使用量（字节）
        """
        MEMORY_USAGE.set(usage)

# 使用示例
if __name__ == "__main__":
    monitor = PrometheusMonitoring(port=8000)
    monitor.start()
    
    # 模拟请求
    for i in range(100):
        start_time = time.time()
        # 模拟处理时间
        time.sleep(0.1)
        latency = time.time() - start_time
        
        monitor.observe_request("/api/risk/analyze", "POST", latency)
        monitor.observe_risk_score(3.5 + (i % 5))
        
        if i % 10 == 0:
            monitor.observe_anomaly("high")
        
        time.sleep(0.5)

# security_utils.py
import hashlib
import hmac
import json
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64

class SecurityUtils:
    def __init__(self, secret_key=None):
        """
        初始化安全工具类
        
        Args:
            secret_key: 密钥，如果为None则生成新密钥
        """
        if secret_key is None:
            self.secret_key = Fernet.generate_key()
        else:
            self.secret_key = secret_key.encode() if isinstance(secret_key, str) else secret_key
        
        self.cipher = Fernet(self.secret_key)
    
    def encrypt_data(self, data):
        """
        加密数据
        
        Args:
            data: 要加密的数据，可以是字符串或字典
            
        Returns:
            加密后的字节字符串
        """
        if isinstance(data, dict):
            data_str = json.dumps(data)
        else:
            data_str = str(data)
        
        return self.cipher.encrypt(data_str.encode())
    
    def decrypt_data(self, encrypted_data):
        """
        解密数据
        
        Args:
            encrypted_data: 加密的数据（字节字符串）
            
        Returns:
            解密后的数据
        """
        decrypted_bytes = self.cipher.decrypt(encrypted_data)
        
        # 尝试解析为JSON，失败则返回字符串
        try:
            return json.loads(decrypted_bytes.decode())
        except json.JSONDecodeError:
            return decrypted_bytes.decode()
    
    @staticmethod
    def generate_hmac(data, key):
        """
        生成HMAC签名
        
        Args:
            data: 要签名的数据
            key: 签名密钥
            
        Returns:
            HMAC签名（十六进制字符串）
        """
        if isinstance(data, dict):
            data_str = json.dumps(data, sort_keys=True)
        else:
            data_str = str(data)
        
        key_bytes = key.encode() if isinstance(key, str) else key
        return hmac.new(key_bytes, data_str.encode(), hashlib.sha256).hexdigest()
    
    @staticmethod
    def verify_hmac(data, key, signature):
        """
        验证HMAC签名
        
        Args:
            data: 要验证的数据
            key: 签名密钥
            signature: 要验证的签名
            
        Returns:
            是否验证通过
        """
        expected_signature = SecurityUtils.generate_hmac(data, key)
        return hmac.compare_digest(expected_signature, signature)
    
    @staticmethod
    def hash_data(data):
        """
        计算数据哈希值
        
        Args:
            data: 要哈希的数据
            
        Returns:
            哈希值（十六进制字符串）
        """
        if isinstance(data, dict):
            data_str = json.dumps(data, sort_keys=True)
        else:
            data_str = str(data)
        
        return hashlib.sha256(data_str.encode()).hexdigest()
    
    @staticmethod
    def generate_key_from_password(password, salt=None):
        """
        从密码生成密钥
        
        Args:
            password: 用户密码
            salt: 盐值，如果为None则生成新盐值
            
        Returns:
            (密钥, 盐值)
        """
        if salt is None:
            salt = base64.urlsafe_b64encode(os.urandom(16))
        else:
            salt = salt.encode() if isinstance(salt, str) else salt
        
        kdf = PBKDF2HMAC(
            algorithm=hashes.SHA256(),
            length=32,
            salt=salt,
            iterations=100000,
        )
        
        key = base64.urlsafe_b64encode(kdf.derive(password.encode()))
        return key, salt
    
    @staticmethod
    def mask_sensitive_data(data, sensitive_fields=None):
        """
        屏蔽敏感数据
        
        Args:
            data: 原始数据（字典或字符串）
            sensitive_fields: 敏感字段列表
            
        Returns:
            屏蔽后的数据
        """
        if sensitive_fields is None:
            sensitive_fields = ['account_id', 'credit_card', 'ssn', 'password']
        
        if isinstance(data, dict):
            masked_data = data.copy()
            for field in sensitive_fields:
                if field in masked_data:
                    value = str(masked_data[field])
                    # 根据字段类型进行不同的屏蔽处理
                    if field == 'credit_card' and len(value) >= 8:
                        masked_data[field] = f"XXXX-XXXX-XXXX-{value[-4:]}"
                    elif len(value) > 6:
                        masked_data[field] = f"{value[:2]}****{value[-2:]}"
                    else:
                        masked_data[field] = "****"
            return masked_data
        elif isinstance(data, str):
            # 对字符串中的敏感信息进行屏蔽
            import re
            masked_str = data
            
            # 屏蔽信用卡号
            masked_str = re.sub(r'(\d{4}-\d{4}-\d{4}-)(\d{4})', r'XXXX-XXXX-XXXX-\2', masked_str)
            
            # 屏蔽身份证号
            masked_str = re.sub(r'(\d{6})(\d{8})(\d{4})', r'\1********\3', masked_str)
            
            return masked_str
        
        return data

# 使用示例
if __name__ == "__main__":
    import os
    
    # 初始化安全工具
    security = SecurityUtils()
    
    # 加密数据
    sensitive_data = {
        "account_id": "ACC123456",
        "amount": 10000,
        "transaction_id": "TX987654"
    }
    
    encrypted = security.encrypt_data(sensitive_data)
    print(f"加密后: {encrypted}")
    
    # 解密数据
    decrypted = security.decrypt_data(encrypted)
    print(f"解密后: {decrypted}")
    
    # 生成HMAC
    signature = security.generate_hmac(sensitive_data, "my_secret_key")
    print(f"HMAC签名: {signature}")
    
    # 验证HMAC
    is_valid = security.verify_hmac(sensitive_data, "my_secret_key", signature)
    print(f"验证结果: {is_valid}")
    
    # 屏蔽敏感数据
    masked_data = security.mask_sensitive_data(sensitive_data)
    print(f"屏蔽后数据: {masked_data}")