AI 对冲基金与多智能体金融决策系统：从 virattt/ai-hedge-fund 看 Multi-Agent 如何重塑投资

"""
多 Agent 金融决策系统 - 简化版
模拟 ai-hedge-fund 的核心架构
需要: pip install openai yfinance
"""

import yfinance as yf
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import Optional
import json


# ============ 数据结构 ============

class Action(Enum):
    BUY = "买入"
    SELL = "卖出"
    HOLD = "持有"


@dataclass
class MarketData:
    ticker: str
    price: float
    pe_ratio: Optional[float]
    market_cap: Optional[float]
    volume: int
    ma_50: Optional[float]
    ma_200: Optional[float]
    rsi_14: Optional[float]
    change_1d: float
    change_30d: float


@dataclass
class AgentOpinion:
    agent_name: str
    action: Action
    confidence: float  # 0.0 ~ 1.0
    reasoning: str
    target_price: Optional[float] = None
    risk_level: str = "中等"  # 低/中等/高


@dataclass
class TradePlan:
    ticker: str
    action: Action
    size_pct: float  # 仓位占比 0~100
    stop_loss: Optional[float]
    take_profit: Optional[float]
    reasoning: str
    agent_opinions: list[AgentOpinion]


# ============ Agent 基类 ============

class FinancialAgent(ABC):
    """金融分析 Agent 抽象基类"""

    def __init__(self, name: str):
        self.name = name

    @abstractmethod
    def analyze(self, data: MarketData) -> AgentOpinion:
        """分析市场数据，给出投资意见"""
        pass


# ============ 基本面分析 Agent ============

class FundamentalAnalyst(FinancialAgent):
    """基本面分析师：基于估值指标给出建议"""

    def __init__(self):
        super().__init__("基本面分析师")

    def analyze(self, data: MarketData) -> AgentOpinion:
        if data.pe_ratio is None:
            return AgentOpinion(
                agent_name=self.name,
                action=Action.HOLD,
                confidence=0.3,
                reasoning="缺少 PE 数据，无法进行基本面分析",
                risk_level="高"
            )

        # 简化的估值逻辑
        pe = data.pe_ratio
        reasoning_parts = []

        if pe < 15:
            action = Action.BUY
            confidence = min(0.85, 0.5 + (15 - pe) / 50)
            reasoning_parts.append(f"PE={pe} 处于低估区间（<15），具有安全边际")
        elif pe < 25:
            action = Action.HOLD
            confidence = 0.5
            reasoning_parts.append(f"PE={pe} 处于合理区间（15-25），估值适中")
        elif pe < 40:
            action = Action.HOLD
            confidence = 0.6
            reasoning_parts.append(f"PE={pe} 偏高但尚可接受（25-40），需关注盈利增速")
        else:
            action = Action.SELL
            confidence = min(0.8, 0.5 + (pe - 40) / 60)
            reasoning_parts.append(f"PE={pe} 严重高估（>40），泡沫风险大")

        # 结合市值判断
        if data.market_cap and data.market_cap > 1e12:
            reasoning_parts.append(f"市值 ${data.market_cap/1e12:.1f}T，大盘股波动较小")

        # 简单目标价估算
        target = None
        if action == Action.BUY and pe < 20:
            target = data.price * (20 / pe)  # 假设合理 PE 为 20
        elif action == Action.SELL and pe > 30:
            target = data.price * (25 / pe)  # 假设合理 PE 为 25

        return AgentOpinion(
            agent_name=self.name,
            action=action,
            confidence=round(confidence, 2),
            reasoning="；".join(reasoning_parts),
            target_price=round(target, 2) if target else None,
            risk_level="低" if action == Action.HOLD else "中等"
        )


# ============ 技术分析 Agent ============

class TechnicalAnalyst(FinancialAgent):
    """技术分析师：基于价格和指标给出建议"""

    def __init__(self):
        super().__init__("技术分析师")

    def analyze(self, data: MarketData) -> AgentOpinion:
        signals = []
        bullish = 0
        bearish = 0

        # 1. 均线分析
        if data.ma_50 and data.ma_200:
            if data.ma_50 > data.ma_200:
                signals.append("50日均线上穿200日均线（黄金交叉），看涨")
                bullish += 2
            else:
                signals.append("50日均线下穿200日均线（死亡交叉），看跌")
                bearish += 2

            if data.price > data.ma_50:
                signals.append(f"当前价 ${data.price:.2f} 在50日均线 ${data.ma_50:.2f} 之上，短期强势")
                bullish += 1
            else:
                signals.append(f"当前价 ${data.price:.2f} 在50日均线 ${data.ma_50:.2f} 之下，短期弱势")
                bearish += 1

        # 2. RSI 分析
        if data.rsi_14:
            if data.rsi_14 < 30:
                signals.append(f"RSI={data.rsi_14:.1f} 进入超卖区，可能反弹")
                bullish += 1
            elif data.rsi_14 > 70:
                signals.append(f"RSI={data.rsi_14:.1f} 进入超买区，可能回调")
                bearish += 1
            else:
                signals.append(f"RSI={data.rsi_14:.1f} 处于中性区间")

        # 3. 动量分析
        if data.change_30d > 0.15:
            signals.append(f"30日涨幅 {data.change_30d*100:.1f}%，短期涨幅过大，注意回调")
            bearish += 1
        elif data.change_30d < -0.15:
            signals.append(f"30日跌幅 {abs(data.change_30d)*100:.1f}%，超跌反弹机会")
            bullish += 1

        # 综合判断
        net_score = bullish - bearish
        if net_score >= 2:
            action = Action.BUY
            confidence = min(0.85, 0.4 + net_score * 0.1)
        elif net_score <= -2:
            action = Action.SELL
            confidence = min(0.8, 0.4 + abs(net_score) * 0.1)
        else:
            action = Action.HOLD
            confidence = 0.5

        return AgentOpinion(
            agent_name=self.name,
            action=action,
            confidence=round(confidence, 2),
            reasoning="；".join(signals) if signals else "技术指标信号不明确",
            risk_level="低" if action == Action.HOLD else "中等"
        )


# ============ 情绪分析 Agent ============

class SentimentAnalyst(FinancialAgent):
    """情绪分析师：模拟基于新闻/社交的情绪分析"""

    def __init__(self, sentiment_score: float = 0.0):
        """
        sentiment_score: -1.0 (极度悲观) ~ +1.0 (极度乐观)
        实际应用中应通过 LLM 分析新闻和社交媒体得到
        """
        super().__init__("情绪分析师")
        self.sentiment_score = sentiment_score

    def analyze(self, data: MarketData) -> AgentOpinion:
        score = self.sentiment_score
        signals = []

        if score > 0.5:
            signals.append(f"市场情绪极度乐观（{score:.2f}），可能存在非理性繁荣")
            action = Action.HOLD
            confidence = 0.7
            risk = "高"
        elif score > 0.2:
            signals.append(f"市场情绪偏正面（{score:.2f}），有利于上涨")
            action = Action.BUY
            confidence = 0.55
            risk = "中等"
        elif score > -0.2:
            signals.append(f"市场情绪中性（{score:.2f}），无明确方向")
            action = Action.HOLD
            confidence = 0.4
            risk = "中等"
        elif score > -0.5:
            signals.append(f"市场情绪偏负面（{score:.2f}），观望为宜")
            action = Action.HOLD
            confidence = 0.6
            risk = "中等"
        else:
            signals.append(f"市场情绪极度悲观（{score:.2f}），可能是逆向投资机会")
            action = Action.BUY
            confidence = 0.5
            risk = "高"

        # 结合价格变动
        if data.change_1d < -0.05 and score < -0.3:
            signals.append("价格大跌 + 情绪极差 = 恐慌性抛售，关注反弹信号")
            confidence = min(confidence + 0.1, 0.8)

        return AgentOpinion(
            agent_name=self.name,
            action=action,
            confidence=round(confidence, 2),
            reasoning="；".join(signals),
            risk_level=risk
        )


# ============ 风险经理 ============

class RiskManager:
    """风险管理：对交易计划进行风险审查"""

    def __init__(self, max_position_pct: float = 20.0, max_risk_level: str = "高"):
        self.max_position_pct = max_position_pct  # 单票最大仓位
        self.max_risk_level = max_risk_level

    def review(self, plan: TradePlan) -> tuple[bool, str]:
        """审查交易计划，返回 (通过, 原因)"""
        # 1. 检查仓位
        if plan.size_pct > self.max_position_pct:
            return False, f"仓位 {plan.size_pct}% 超过上限 {self.max_position_pct}%"

        # 2. 检查止损
        if plan.action == Action.BUY and plan.stop_loss is None:
            return False, "买入操作必须设置止损"

        # 3. 检查风险等级
        high_risk_count = sum(
            1 for op in plan.agent_opinions if op.risk_level == "高"
        )
        if high_risk_count >= 2:
            return False, f"{high_risk_count} 个 Agent 标记高风险，建议降低仓位或放弃"

        # 4. 检查意见一致性
        buy_count = sum(1 for op in plan.agent_opinions if op.action == Action.BUY)
        sell_count = sum(1 for op in plan.agent_opinions if op.action == Action.SELL)
        if buy_count > 0 and sell_count > 0:
            return True, f"意见分歧（{buy_count} 买入 vs {sell_count} 卖出），已通过但需降低仓位"

        return True, "风险审查通过"


# ============ 投资组合经理 ============

class PortfolioManager:
    """投资组合经理：汇总所有 Agent 意见，生成交易计划"""

    def __init__(self):
        self.risk_manager = RiskManager(max_position_pct=15.0)

    def make_decision(
        self, ticker: str, opinions: list[AgentOpinion]
    ) -> TradePlan:
        # 统计意见
        buy_score = sum(
            op.confidence for op in opinions if op.action == Action.BUY
        )
        sell_score = sum(
            op.confidence for op in opinions if op.action == Action.SELL
        )
        hold_score = sum(
            op.confidence for op in opinions if op.action == Action.HOLD
        )

        # 确定最终行动
        scores = {"BUY": buy_score, "SELL": sell_score, "HOLD": hold_score}
        best_action = max(scores, key=scores.get)

        if best_action == "BUY":
            action = Action.BUY
            # 仓位大小基于综合置信度
            total_score = buy_score + sell_score + hold_score
            size = min(15.0, buy_score / max(total_score, 0.01) * 20)
            stop_loss = None  # 后续由风险管理设定
            take_profit = None
        elif best_action == "SELL":
            action = Action.SELL
            size = 100.0  # 清仓
            stop_loss = None
            take_profit = None
        else:
            action = Action.HOLD
            size = 0.0
            stop_loss = None
            take_profit = None

        reasoning = f"综合评分: 买入={buy_score:.2f}, 卖出={sell_score:.2f}, 持有={hold_score:.2f}"

        plan = TradePlan(
            ticker=ticker,
            action=action,
            size_pct=round(size, 1),
            stop_loss=stop_loss,
            take_profit=take_profit,
            reasoning=reasoning,
            agent_opinions=opinions,
        )

        # 风险审查
        passed, reason = self.risk_manager.review(plan)
        if not passed:
            plan.reasoning += f" | 风险审查未通过: {reason}"
            if action == Action.BUY:
                plan.size_pct = round(plan.size_pct * 0.5, 1)  # 减半仓位
                plan.reasoning += " → 已减半仓位"

        return plan


# ============ 运行示例 ============

def run_hedge_fund(ticker: str):
    """运行完整的对冲基金决策流程"""
    print(f"\n{'='*60}")
    print(f"🏦 AI 对冲基金决策报告 - {ticker}")
    print(f"{'='*60}")

    # 1. 获取市场数据
    stock = yf.Ticker(ticker)
    info = stock.info
    hist = stock.history(period="6mo")

    current_price = info.get("currentPrice", 0) or hist["Close"].iloc[-1]

    data = MarketData(
        ticker=ticker,
        price=current_price,
        pe_ratio=info.get("trailingPE"),
        market_cap=info.get("marketCap"),
        volume=info.get("volume", 0) or int(hist["Volume"].iloc[-1]),
        ma_50=hist["Close"].rolling(50).mean().iloc[-1] if len(hist) >= 50 else None,
        ma_200=hist["Close"].rolling(200).mean().iloc[-1] if len(hist) >= 200 else None,
        rsi_14=None,  # 简化版省略 RSI 计算
        change_1d=info.get("regularMarketChangePercent", 0) / 100,
        change_30d=(current_price / hist["Close"].iloc[-min(21, len(hist))] - 1)
    )

    print(f"\n📊 市场数据:")
    print(f"   当前价格: ${data.price:.2f}")
    print(f"   PE 比率: {data.pe_ratio or 'N/A'}")
    print(f"   市值: ${data.market_cap/1e9:.1f}B" if data.market_cap else "   市值: N/A")
    print(f"   30日涨跌: {data.change_30d*100:.1f}%")

    # 2. 各 Agent 独立分析
    print(f"\n🔍 Agent 分析:")

    agents = [
        FundamentalAnalyst(),
        TechnicalAnalyst(),
        SentimentAnalyst(sentiment_score=0.1),  # 假设轻微正面情绪
    ]

    opinions = []
    for agent in agents:
        opinion = agent.analyze(data)
        opinions.append(opinion)
        emoji = {"买入": "🟢", "卖出": "🔴", "持有": "🟡"}[opinion.action.value]
        print(f"   {emoji} {agent.name}: {opinion.action.value} "
              f"(置信度: {opinion.confidence:.0%})")
        print(f"      {opinion.reasoning[:80]}...")

    # 3. 投资组合经理做决策
    print(f"\n💼 投资决策:")
    pm = PortfolioManager()
    plan = pm.make_decision(ticker, opinions)

    action_emoji = {"买入": "🟢 买入", "卖出": "🔴 卖出", "持有": "🟡 持有"}[plan.action.value]
    print(f"   最终决策: {action_emoji}")
    print(f"   仓位建议: {plan.size_pct}%")
    if plan.stop_loss:
        print(f"   止损位: ${plan.stop_loss:.2f}")
    print(f"   决策依据: {plan.reasoning}")

    return plan


if __name__ == "__main__":
    # 示例：分析几只热门股票
    for ticker in ["AAPL", "NVDA", "TSLA"]:
        try:
            run_hedge_fund(ticker)
        except Exception as e:
            print(f"   分析 {ticker} 时出错: {e}")
        print()