金融科技 Multi-Agent 落地：智能投顾与风险监控的协作系统

引言

在金融科技（FinTech）快速发展的今天，人工智能技术正在重塑整个行业的面貌。从算法交易到信用评估，从客户服务到风险管理，AI的应用场景越来越广泛。然而，单一的AI系统往往难以处理金融领域复杂、动态、多维度的问题。这就是为什么Multi-Agent系统（多智能体系统）正在成为金融科技领域的新宠。

想象一下，一个由多个专业化AI智能体组成的团队，它们像人类专家一样协作：有的负责分析市场趋势，有的负责评估投资机会，有的专门监控风险，还有的负责与客户沟通。这些智能体不仅各自发挥专长，还能共享信息、协调决策，形成一个强大的协作系统。

在这篇文章中，我们将深入探讨如何在金融科技领域构建和落地Multi-Agent系统，特别关注智能投顾与风险监控这两个核心功能的协作机制。我们将从理论基础、架构设计、代码实现到实际应用，全方位解析这一前沿技术。

核心概念

什么是Multi-Agent系统？

Multi-Agent系统（MAS）是一种由多个自主智能体（Agent）组成的计算系统，这些智能体在同一环境中相互作用、协作或竞争，以实现各自的目标或共同的目标。每个智能体都是一个自治的实体，具有感知环境、做出决策和采取行动的能力。

在金融科技语境下，一个智能体可以是：

• 一个专门分析股票市场趋势的AI模型
• 一个评估客户风险承受能力的分类器
• 一个实时监控市场异常的预警系统
• 一个生成投资组合建议的优化器

金融科技中的智能投顾

智能投顾（Robo-Advisor）是一种基于算法的在线财富管理服务，它使用数学模型和算法来提供投资建议和管理投资组合，而无需人工干预。传统的智能投顾系统通常基于现代投资组合理论（MPT），通过分散投资来优化风险-收益比。

金融风险监控

金融风险监控是指识别、评估和控制金融风险的过程，包括市场风险、信用风险、操作风险等。在数字化时代，风险监控系统需要处理海量数据，实时检测异常模式，并快速响应潜在威胁。

Multi-Agent在金融科技中的价值

为什么我们需要Multi-Agent系统？因为金融问题具有以下特点：

1. 复杂性：涉及多个市场、多种资产类别的相互作用
2. 动态性：市场条件瞬息万变，需要实时响应
3. 不确定性：充满了不可预测的事件和噪音
4. 多目标性：需要在收益、风险、流动性等多个目标间权衡

Multi-Agent系统通过"分而治之"的策略，将复杂问题分解为多个子问题，由专门的智能体来处理，然后通过协作机制整合结果，从而更有效地解决这些问题。

问题背景

传统金融系统的局限性

传统的金融科技系统往往面临以下挑战：

1. 信息孤岛：不同部门（投资、风控、合规等）的数据和系统相互隔离，难以共享和协同
2. 响应滞后：从数据收集、分析到决策执行，存在较长的时间差
3. 专家依赖：过度依赖少数资深专家的经验和判断，难以规模化
4. 刚性架构：系统设计固定，难以适应市场变化和业务需求的快速迭代

智能投顾的现状与挑战

当前的智能投顾系统虽然取得了一定成功，但仍存在明显不足：

1. 同质化严重：大多基于相似的算法和资产配置模型
2. 缺乏个性化：难以真正理解和适应每个投资者的独特需求和情况变化
3. 风险监控不足：在市场剧烈波动时，往往不能及时调整策略
4. 解释性差：许多模型是"黑盒"，难以向投资者解释决策依据

风险监控的痛点

传统的风险监控系统同样面临诸多问题：

1. 误报率高：产生过多虚假警报，导致真正的风险被忽视
2. 覆盖有限：难以全面监控所有可能的风险点和关联关系
3. 反应迟缓：从发现异常到采取行动，往往需要较长时间
4. 缺乏前瞻性：主要关注已知风险模式，难以发现新的、未知的风险

问题描述

我们的目标是构建一个Multi-Agent协作系统，解决以下核心问题：

1. 如何设计能够高效协作的智能体架构，使智能投顾和风险监控能够协同工作？
2. 如何建立智能体之间的通信机制，确保信息及时、准确地共享？
3. 如何实现智能体的任务分配和协调，避免冲突并最大化整体效能？
4. 如何在保持系统整体稳定性的同时，允许各个智能体自主决策和学习？
5. 如何评估和优化系统的整体性能，确保在复杂金融环境中的有效性？

为了更具体地说明问题，让我们考虑一个典型的投资决策场景：

一个客户有100万元可投资资金，希望获得合理回报同时控制风险。系统需要：

1. 分析客户的财务状况、投资目标和风险偏好
2. 评估当前市场状况和各类资产的预期表现
3. 生成多个可能的投资组合方案
4. 评估每个方案的潜在风险
5. 根据客户偏好和市场情况推荐最优方案
6. 持续监控投资组合表现和市场风险，及时调整

在传统系统中，这些步骤可能由不同的模块甚至不同的系统处理，缺乏有效的协调。而在Multi-Agent系统中，我们可以让专门的智能体负责不同步骤，并通过协作机制实现无缝集成。

问题解决

核心架构设计思路

解决上述问题的关键在于设计一个模块化、松耦合但又能紧密协作的Multi-Agent架构。我们的设计基于以下原则：

1. 专业化分工：每个智能体专注于特定领域，发挥专业优势
2. 自主决策：每个智能体在其职责范围内拥有自主决策权
3. 有效通信：建立标准化的通信协议和共享信息机制
4. 动态协调：能够根据环境变化和任务需求动态调整协作方式
5. 学习进化：系统和各个智能体能够从经验中学习，持续优化

Multi-Agent系统的组成要素

我们的金融Multi-Agent系统由以下核心要素组成：

1. 环境层：模拟或连接真实的金融市场环境
2. 智能体层：各种专业化的智能体
3. 通信层：智能体之间的通信基础设施
4. 协调层：负责智能体之间的任务分配和冲突解决
5. 知识层：共享的知识库和本体
6. 接口层：与用户和外部系统的交互界面

智能体分类与职责

在我们的系统中，智能体可以分为以下几类：

1. 客户智能体（Client Agent）：负责理解和代表客户的利益
2. 市场分析智能体（Market Analyst Agent）：分析市场趋势和资产表现
3. 投资组合智能体（Portfolio Agent）：生成和优化投资组合
4. 风险监控智能体（Risk Monitor Agent）：评估和监控风险
5. 执行智能体（Execution Agent）：负责交易执行
6. 协调智能体（Coordinator Agent）：协调整个系统的运行

边界与外延

系统边界

我们的Multi-Agent系统主要关注以下范围：

1. 投资管理：从客户需求分析到投资组合构建和调整
2. 风险监控：市场风险、信用风险、流动性风险等的实时监控
3. 决策支持：为投资决策提供AI辅助，但保留人工最终决策权
4. 前后台集成：连接前台客户服务和后台风险管理

系统不包括：

1. 直接的高频交易（虽然可以支持相关决策）
2. 完全自主的投资决策（需要人工监督）
3. 跨金融机构的协作（目前聚焦于机构内部）

外延应用

虽然我们主要关注智能投顾和风险监控，但Multi-Agent系统的理念可以扩展到更多金融应用场景：

1. 反洗钱（AML）和欺诈检测：多个智能体协作分析交易模式，识别可疑活动
2. 信用评估：综合多源数据，更准确地评估借款人信用状况
3. 保险定价和理赔：智能体协作分析风险因素，优化定价和理赔流程
4. 算法交易：多个专业化交易智能体在不同市场和策略间协作
5. 客户服务：专门的客户服务智能体与其他智能体协作，提供个性化服务

概念结构与核心要素组成

Multi-Agent系统的核心概念模型

让我们首先建立一个Multi-Agent系统的概念模型，帮助理解各个组成部分及其关系。

智能体的核心要素

每个智能体都包含以下核心要素：

1. 感知模块（Perception Module）：负责从环境和其他智能体获取信息
2. 推理引擎（Reasoning Engine）：处理信息并做出决策
3. 知识库（Knowledge Base）：存储智能体的专业知识和经验
4. 通信模块（Communication Module）：负责与其他智能体交互
5. 执行模块（Action Module）：执行决策并影响环境

概念之间的关系

核心概念对比

为了更好地理解各个智能体的特点和关系，让我们通过一个对比表格来分析它们的核心属性：

智能体类型	主要目标	核心能力	信息输入	决策输出	时间尺度	风险偏好
客户智能体	最大化客户满意度	需求分析、偏好建模	客户数据、反馈	需求规格、约束	长期	保守或进取（因客户而异）
市场分析智能体	准确预测市场趋势	数据分析、模式识别	市场数据、新闻	预测报告、评级	中短期	中性
投资组合智能体	优化风险-收益比	资产配置、组合优化	客户需求、市场预测	投资组合建议	中期	平衡
风险监控智能体	最小化风险暴露	风险评估、异常检测	组合数据、市场数据	风险报告、预警	实时	保守
执行智能体	高效执行交易	交易优化、成本控制	投资组合、市场流动性	执行订单、确认	短期	中性
协调智能体	系统整体效能最大化	任务分配、冲突解决	所有智能体状态	协调指令、资源分配	动态	平衡

智能体协作关系

智能体之间的协作关系可以用实体关系图表示：

智能体交互流程

智能体之间的典型交互流程可以用序列图表示：

数学模型

投资组合优化模型

智能投顾的核心是投资组合优化。我们使用现代投资组合理论（MPT）作为基础，并结合Multi-Agent系统的特点进行扩展。

传统的马科维茨模型可以表示为：

$$\begin{array}{rl}\underset{w}{\min }& {\sigma }_p^2=\sum _{i=1}^n\sum _{j=1}^n{w}_i{w}_j{\sigma }_{ij}\\ \text{s.t.}& {\mu }_p=\sum _{i=1}^n{w}_i{\mu }_i\ge {\mu }_{\text{target}}\\ & \sum _{i=1}^n{w}_i=1\\ & w_i\ge 0\forall i\end{array}$$

其中：

• $w_i$ 是资产 $i$ 的权重
• ${\mu }_i$ 是资产 $i$ 的预期收益率
• ${\sigma }_{ij}$ 是资产 $i$ 和 $j$ 的协方差
• ${\mu }_p$ 和 ${\sigma }_p^2$ 分别是投资组合的预期收益率和方差

在Multi-Agent系统中，我们将这个问题扩展为一个分布式优化问题，不同智能体专注于不同的子目标：

$$\begin{array}{rl}\underset{w}{\min }& \lambda \cdot \text{Risk}(w)+(1-\lambda )\cdot (-\text{Return}(w))+\gamma \cdot \text{Liquidity}(w)\\ \text{s.t.}& {\text{Constraints}}_{\text{client}}(w)\\ & {\text{Constraints}}_{\text{regulatory}}(w)\\ & {\text{Constraints}}_{\text{risk}}(w)\end{array}$$

这里：

• $\lambda$ 是风险-收益权衡参数
• $\text{Risk}(w)$ 由风险监控智能体计算
• $\text{Return}(w)$ 由市场分析智能体预测
• $\text{Liquidity}(w)$ 由执行智能体评估
• 各种约束由相应的智能体提出和验证

风险评估模型

风险监控智能体使用多种风险度量方法，包括：

1. 方差/标准差：传统的波动性度量
2. VaR（风险价值）：在给定置信水平下的最大可能损失
3. CVaR（条件风险价值）：损失超过VaR部分的期望值
4. 最大回撤：从峰值到谷底的最大跌幅

其中，CVaR在优化中更受欢迎，因为它是一个凸风险度量，具有良好的数学性质。CVaR的定义为：

$${\text{CVaR}}_{\alpha }(X)=\frac{1}{1-\alpha }{\int }_{\alpha }^1{\text{VaR}}_{\gamma }(X)d\gamma$$

其中 $X$ 表示投资组合的损失分布，$\alpha$ 是置信水平。

在离散情况下，我们可以使用以下线性规划来优化CVaR：

$$\begin{array}{rl}\underset{w,\zeta ,z_i}{\min }& \zeta +\frac{1}{(1-\alpha )S}\sum _{i=1}^S{z}_i\\ \text{s.t.}& z_i\ge f(w,y_i)-\zeta \\ & z_i\ge 0\\ & \sum _{j=1}^n{w}_j=1\\ & w_j\ge 0\end{array}$$

这里：

• $f(w,y_i)$ 是在情景 $y_i$ 下投资组合的损失
• $S$ 是情景数量
• $\zeta$ 是VaR的近似值

智能体协调模型

智能体之间的协调可以用博弈论模型来描述。我们使用具有社会福利函数的合作博弈模型：

$$\underset{a_1,\dots ,a_n}{\max }\sum _{i=1}^n{\omega }_i\cdot u_i(a_1,\dots ,a_n)$$

其中：

• $a_i$ 是智能体 $i$ 的行动
• $u_i$ 是智能体 $i$ 的效用函数
• ${\omega }_i$ 是智能体 $i$ 的权重

在金融应用中，我们通常根据智能体对系统目标的重要性来设置权重，例如风险监控智能体在市场波动期可能获得更高权重。

另一个重要的协调模型是马尔可夫决策过程（MDP）的扩展——分布式部分可观察马尔可夫决策过程（Dec-POMDP）：

$$\underset{{\pi }_1,\dots ,{\pi }_n}{\max }E\left[\sum _{t=0}^T{\gamma }^{t}R(s_t,a_{1t},\dots ,a_{nt})\right]$$

这里：

• ${\pi }_i$ 是智能体 $i$ 的策略
• $s_t$ 是时间 $t$ 的环境状态
• $a_{it}$ 是智能体 $i$ 在时间 $t$ 的行动
• $R$ 是全局奖励函数
• $\gamma$ 是折现因子

算法流程

Multi-Agent协作算法

让我们通过流程图展示Multi-Agent系统的核心协作算法：

智能体协商与决策流程

智能体之间的协商和决策机制是系统的关键，以下是详细的协商流程：

算法源代码

为了更好地理解Multi-Agent系统的实现，让我们用Python编写一个简化但功能完整的示例系统。我们将创建基本的智能体类和它们的协作机制。

import numpy as np
import pandas as pd
from typing import List, Dict, Any, Tuple, Optional
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
import random
import time
from datetime import datetime, timedelta

# 消息类型枚举
class MessageType(Enum):
    REQUEST = "request"
    RESPONSE = "response"
    NOTIFICATION = "notification"
    NEGOTIATION = "negotiation"

# 消息数据类
@dataclass
class Message:
    sender_id: str
    receiver_id: str
    msg_type: MessageType
    content: Dict[str, Any]
    timestamp: datetime = datetime.now()

# 金融资产数据类
@dataclass
class Asset:
    id: str
    name: str
    asset_type: str
    expected_return: float
    risk: float  # 标准差
    price: float
    liquidity: float  # 流动性评分，0-1

# 投资组合数据类
@dataclass
class Portfolio:
    assets: Dict[str, float]  # 资产ID到权重的映射
    expected_return: float = 0.0
    risk: float = 0.0
    sharpe_ratio: float = 0.0

# 客户数据类
@dataclass
class ClientProfile:
    id: str
    risk_tolerance: float  # 0-1，0为极度保守，1为极度激进
    investment_horizon: int  # 投资期限（月）
    investment_goal: str  # 投资目标
    initial_capital: float
    income_requirement: float = 0.0

# 智能体基类
class BaseAgent(ABC):
    def __init__(self, agent_id: str, environment):
        self.agent_id = agent_id
        self.environment = environment
        self.message_queue: List[Message] = []
        self.knowledge_base: Dict[str, Any] = {}
        self.active = True
        
    def send_message(self, receiver_id: str, msg_type: MessageType, content: Dict[str, Any]) -> None:
        """向其他智能体发送消息"""
        message = Message(
            sender_id=self.agent_id,
            receiver_id=receiver_id,
            msg_type=msg_type,
            content=content
        )
        self.environment.route_message(message)
    
    def receive_message(self, message: Message) -> None:
        """接收消息"""
        self.message_queue.append(message)
    
    def process_messages(self) -> List[Message]:
        """处理队列中的消息"""
        processed = []
        while self.message_queue:
            message = self.message_queue.pop(0)
            self._handle_message(message)
            processed.append(message)
        return processed
    
    @abstractmethod
    def _handle_message(self, message: Message) -> None:
        """处理具体消息的方法，由子类实现"""
        pass
    
    @abstractmethod
    def act(self) -> None:
        """智能体的主要行为，由子类实现"""
        pass
    
    def update(self) -> None:
        """智能体的更新循环"""
        if self.active:
            self.process_messages()
            self.act()

# 环境类，负责消息路由和全局状态
class Environment:
    def __init__(self):
        self.agents: Dict[str, BaseAgent] = {}
        self.market_data: Dict[str, Any] = {}
        self.time_step = 0
        
    def register_agent(self, agent: BaseAgent) -> None:
        """注册智能体"""
        self.agents[agent.agent_id] = agent
    
    def route_message(self, message: Message) -> None:
        """路由消息到目标智能体"""
        if message.receiver_id in self.agents:
            self.agents[message.receiver_id].receive_message(message)
        elif message.receiver_id == "broadcast":
            for agent_id, agent in self.agents.items():
                if agent_id != message.sender_id:
                    agent.receive_message(message)
    
    def update_market_data(self, new_data: Dict[str, Any]) -> None:
        """更新市场数据"""
        self.market_data.update(new_data)
    
    def step(self) -> None:
        """模拟一个时间步长"""
        self.time_step += 1
        for agent in self.agents.values():
            agent.update()
    
    def run(self, steps: int) -> None:
        """运行环境多个时间步"""
        for _ in range(steps):
            self.step()
            time.sleep(0.1)  # 模拟实时性

# 客户智能体
class ClientAgent(BaseAgent):
    def __init__(self, agent_id: str, environment, client_profile: ClientProfile):
        super().__init__(agent_id, environment)
        self.client_profile = client_profile
        self.current_portfolio: Optional[Portfolio] = None
        self.satisfaction_score = 0.5  # 初始满意度中等
        
    def _handle_message(self, message: Message) -> None:
        if message.msg_type == MessageType.REQUEST:
            if message.content.get("request_type") == "client_profile":
                # 发送客户资料
                self.send_message(
                    message.sender_id,
                    MessageType.RESPONSE,
                    {"client_profile": self.client_profile.__dict__}
                )
            elif message.content.get("request_type") == "portfolio_feedback":
                # 提供投资组合反馈
                portfolio = message.content.get("portfolio")
                feedback = self._evaluate_portfolio(portfolio)
                self.send_message(
                    message.sender_id,
                    MessageType.RESPONSE,
                    feedback
                )
        elif message.msg_type == MessageType.NOTIFICATION:
            if message.content.get("notification_type") == "portfolio_update":
                # 更新当前投资组合
                portfolio_data = message.content.get("portfolio")
                self.current_portfolio = self._dict_to_portfolio(portfolio_data)
                self.satisfaction_score = self._calculate_satisfaction()
                print(f"[{self.agent_id}] 投资组合已更新，满意度: {self.satisfaction_score:.2f}")
    
    def act(self) -> None:
        # 定期检查投资组合是否符合目标
        if self.current_portfolio and self.environment.time_step % 10 == 0:
            satisfaction = self._calculate_satisfaction()
            if satisfaction < 0.6:  # 满意度低于阈值，请求重新评估
                print(f"[{self.agent_id}] 满意度较低，请求重新评估投资组合")
                self.send_message(
                    "coordinator",
                    MessageType.REQUEST,
                    {"request_type": "portfolio_review"}
                )
    
    def _evaluate_portfolio(self, portfolio_dict: Dict[str, Any]) -> Dict[str, Any]:
        """评估投资组合并提供反馈"""
        portfolio = self._dict_to_portfolio(portfolio_dict)
        
        # 简单评估逻辑
        risk_ok = portfolio.risk <= (self.client_profile.risk_tolerance * 0.2 + 0.05)
        return_ok = portfolio.expected_return >= (self.client_profile.risk_tolerance * 0.15 + 0.03)
        
        feedback = {
            "risk_acceptable": risk_ok,
            "return_acceptable": return_ok,
            "overall_acceptable": risk_ok and return_ok,
            "comments": "满意" if risk_ok and return_ok else "需要调整",
            "adjustment_preferences": {
                "risk_tolerance": self.client_profile.risk_tolerance,
                "priority": "balanced"
            }
        }
        
        return feedback
    
    def _calculate_satisfaction(self) -> float:
        """计算当前满意度"""
        if not self.current_portfolio:
            return 0.5
        
        # 简单的满意度计算
        risk_score = max(0, 1 - self.current_portfolio.risk / (self.client_profile.risk_tolerance * 0.2 + 0.05))
        return_score = min(1, self.current_portfolio.expected_return / (self.client_profile.risk_tolerance * 0.15 + 0.03))
        
        return 0.5 * risk_score + 0.5 * return_score
    
    def _dict_to_portfolio(self, portfolio_dict: Dict[str, Any]) -> Portfolio:
        """将字典转换为投资组合对象"""
        return Portfolio(
            assets=portfolio_dict.get("assets", {}),
            expected_return=portfolio_dict.get("expected_return", 0.0),
            risk=portfolio_dict.get("risk", 0.0),
            sharpe_ratio=portfolio_dict.get("sharpe_ratio", 0.0)
        )

# 市场分析智能体
class MarketAnalystAgent(BaseAgent):
    def __init__(self, agent_id: str, environment):
        super().__init__(agent_id, environment)
        self.assets: Dict[str, Asset] = {}
        self.market_regime = "normal"  # 市场状态: normal, volatile, bullish, bearish
        self.correlation_matrix = None
        
    def _handle_message(self, message: Message) -> None:
        if message.msg_type == MessageType.REQUEST:
            if message.content.get("request_type") == "market_analysis":
                # 返回市场分析
                analysis = self._generate_market_analysis()
                self.send_message(
                    message.sender_id,
                    MessageType.RESPONSE,
                    analysis
                )
            elif message.content.get("request_type") == "asset_data":
                # 返回资产数据
                asset_ids = message.content.get("asset_ids", list(self.assets.keys()))
                asset_data = {aid: self._asset_to_dict(asset) 
                             for aid, asset in self.assets.items() 
                             if aid in asset_ids}
                self.send_message(
                    message.sender_id,
                    MessageType.RESPONSE,
                    {"assets": asset_data, "correlation_matrix": self.correlation_matrix.tolist() if self.correlation_matrix is not None else None}
                )
    
    def act(self) -> None:
        # 每5个时间步更新一次市场分析
        if self.environment.time_step % 5 == 0:
            self._update_market_regime()
            self._update_asset_forecasts()
            
            # 通知相关方市场变化
            self.send_message(
                "broadcast",
                MessageType.NOTIFICATION,
                {
                    "notification_type": "market_update",
                    "market_regime": self.market_regime,
                    "timestamp": datetime.now().isoformat()
                }
            )
    
    def initialize_assets(self, assets: List[Asset]) -> None:
        """初始化资产池"""
        for asset in assets:
            self.assets[asset.id] = asset
        
        # 生成随机相关矩阵（实际应用中应基于历史数据）
        n = len(assets)
        self.correlation_matrix = np.random.randn(n, n)
        self.correlation_matrix = (self.correlation_matrix + self.correlation_matrix.T) / 2  # 对称
        np.fill_diagonal(self.correlation_matrix, 1.0)  # 对角线为1
        
        # 确保正半定
        eigval, eigvec = np.linalg.eigh(self.correlation_matrix)
        eigval[eigval < 0] = 0
        self.correlation_matrix = eigvec.dot(np.diag(eigval)).dot(eigvec.T)
    
    def _update_market_regime(self) -> None:
        """更新市场状态"""
        # 简单的模拟，实际应用中应基于市场指标
        rand_val = random.random()
        if rand_val < 0.6:
            self.market_regime = "normal"
        elif rand_val < 0.8:
            self.market_regime = "volatile"
        elif rand_val < 0.9:
            self.market_regime = "bullish"
        else:
            self.market_regime = "bearish"
        
        print(f"[{self.agent_id}] 市场状态更新为: {self.market_regime}")
    
    def _update_asset_forecasts(self) -> None:
        """更新资产预期收益和风险预测"""
        for asset in self.assets.values():
            # 根据市场状态调整预期
            if self.market_regime == "bullish":
                asset.expected_return *= 1.2
                asset.risk *= 0.9
            elif self.market_regime == "bearish":
                asset.expected_return *= 0.8
                asset.risk *= 1.1
            elif self.market_regime == "volatile":
                asset.risk *= 1.3
            
            # 添加一些随机波动
            asset.expected_return += random.normalvariate(0, 0.01)
            asset.risk += random.normalvariate(0, 0.005)
            
            # 确保值在合理范围内
            asset.expected_return = max(0, min(0.3, asset.expected_return))
            asset.risk = max(0.01, min(0.5, asset.risk))
    
    def _generate_market_analysis(self) -> Dict[str, Any]:
        """生成市场分析报告"""
        return {
            "market_regime": self.market_regime,
            "outlook": {
                "normal": "市场稳定，预期温和回报",
                "volatile": "市场波动加剧，建议谨慎",
                "bullish": "市场趋势向好，预期较高回报",
                "bearish": "市场趋势向下，注意风险控制"
            }[self.market_regime],
            "recommended_asset_classes": self._get_recommended_assets(),
            "timestamp": datetime.now().isoformat()
        }
    
    def _get_recommended_assets(self) -> List[str]:
        """根据市场状态推荐资产类别"""
        recommendations = {
            "normal": ["stock", "bond", "etf"],
            "volatile": ["bond", "cash", "gold"],
            "bullish": ["stock", "commodity"],
            "bearish": ["bond", "cash", "gold"]
        }
        return recommendations.get(self.market_regime, ["stock", "bond"])
    
    def _asset_to_dict(self, asset: Asset) -> Dict[str, Any]:
        """将资产对象转换为字典"""
        return {
            "id": asset.id,
            "name": asset.name,
            "asset_type": asset.asset_type,
            "expected_return": asset.expected_return,
            "risk": asset.risk,
            "price": asset.price,
            "liquidity": asset.liquidity
        }

# 投资组合智能体
class PortfolioAgent(BaseAgent):
    def __init__(self, agent_id: str, environment):
        super().__init__(agent_id, environment)
        self.candidate_portfolios: List[Portfolio] = []
        self.best_portfolio: Optional[Portfolio] = None
        
    def _handle_message(self, message: Message) -> None:
        if message.msg_type == MessageType.REQUEST:
            if message.content.get("request_type") == "portfolio_generation":
                # 生成投资组合
                client_profile_data = message.content.get("client_profile")
                client_profile = self._dict_to_client_profile(client_profile_data)
                
                # 请求资产数据
                self.send_message(
                    "market_analyst",
                    MessageType.REQUEST,
                    {"request_type": "asset_data"}
                )
                
                # 暂时存储客户资料，等待资产数据
                self.knowledge_base["pending_client"] = client_profile
                
            elif message.content.get("request_type") == "portfolio_optimization":
                # 优化现有投资组合
                current_portfolio_data = message.content.get("current_portfolio")
                client_profile_data = message.content.get("client_profile")
                
                portfolio = self._dict_to_portfolio(current_portfolio_data)
                client_profile = self._dict_to_client_profile(client_profile_data)
                
                optimized = self._optimize_portfolio(portfolio, client_profile)
                self.best_portfolio = optimized
                
                # 发送给风险监控评估
                self.send_message(
                    "risk_monitor",
                    MessageType.REQUEST,
                    {
                        "request_type": "risk_assessment",
                        "portfolio": self._portfolio_to_dict(optimized)
                    }
                )
        
        elif message.msg_type == MessageType.RESPONSE:
            if "assets" in message.content and "pending_client" in self.knowledge_base:
                # 收到资产数据，现在生成投资组合
                assets_data = message.content["assets"]
                correlation_matrix = np.array(message.content.get("correlation_matrix"))
                
                # 转换资产数据
                assets = {aid: self._dict_to_asset(adata) for aid, adata in assets_data.items()}
                
                # 获取待处理的客户资料
                client_profile = self.knowledge_base.pop("pending_client")
                
                # 生成投资组合
                self._generate_portfolios(assets, correlation_matrix, client_profile)
                
                # 选择最佳投资组合
                if self.candidate_portfolios:
                    self.best_portfolio = self._select_best_portfolio(self.candidate_portfolios, client_profile)
                    
                    # 发送给风险监控评估
                    self.send_message(
                        "risk_monitor",
                        MessageType.REQUEST,
                        {
                            "request_type": "risk_assessment",
                            "portfolio": self._portfolio_to_dict(self.best_portfolio)
                        }
                    )
        
        elif message.msg_type == MessageType.NOTIFICATION:
            if message.content.get("notification_type") == "market_update":
                # 市场更新，重新评估投资组合
                if self.best_portfolio:
                    print(f"[{self.agent_id}] 收到市场更新，将重新评估投资组合")

    def act(self) -> None:
        # 定期检查是否需要更新投资组合
        if self.best_portfolio and self.environment.time_step % 20 == 0:
            # 请求客户反馈
            self.send_message(
                "client",
                MessageType.REQUEST,
                {
                    "request_type": "portfolio_feedback",
                    "portfolio": self._portfolio_to_dict(self.best_portfolio)
                }
            )
    
    def _generate_portfolios(self, assets: Dict[str, Asset], correlation_matrix: np.ndarray, 
                           client_profile: ClientProfile, num_candidates: int = 10) -> None:
        """生成多个候选投资组合"""
        self.candidate_portfolios = []
        
        asset_ids = list(assets.keys())
        n_assets = len(asset_ids)
        
        # 生成预期收益向量和协方差矩阵
        expected_returns = np.array([assets[aid].expected_return for aid in asset_ids])
        volatilities = np.array([assets[aid].risk for aid in asset_ids])
        
        # 计算协方差矩阵
        cov_matrix = np.outer(volatilities, volatilities) * correlation_matrix
        
        for _ in range(num_candidates):
            # 生成随机权重
            weights = np.random.rand(n_assets)
            weights /= weights.sum()  # 归一化
            
            # 计算投资组合特征
            port_return = np.dot(weights, expected_returns)
            port_risk = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))
            sharpe_ratio = port_return / port_risk if port_risk > 0 else 0
            
            # 创建投资组合对象
            portfolio = Portfolio(
                assets={asset