ljwx-stock/backend/advisor/claude_investment_advisor.py

"""
Claude Investment Advisor Module
基于 DeepSeek 量化分析结果，通过 Claude 生成自然语言投资建议
形成 Claude × DeepSeek 协作的完整闭环
"""

import json
import logging
from typing import Dict, Any, Optional, List
import httpx

logger = logging.getLogger(__name__)


class ClaudeInvestmentAdvisor:
    """Claude 投资顾问服务 (via OpenRouter)"""

    def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None):
        """
        初始化 Claude 投资顾问服务

        Args:
            api_key: OpenRouter API Key
            base_url: OpenRouter Base URL
        """
        import os
        self.api_key = api_key or os.getenv("OPENROUTER_API_KEY")
        self.base_url = base_url or os.getenv("OPENROUTER_BASE_URL", "https://openrouter.ai/api/v1")

        if not self.api_key:
            raise ValueError("OPENROUTER_API_KEY is required")

        self.model = "anthropic/claude-sonnet-4.5"

    def _get_advisor_system_prompt(self) -> str:
        """获取投资顾问系统提示词"""
        return """
你是一名资深的 AI 投资顾问，专门为"Claude × DeepSeek 智能投顾系统"生成投资建议。

系统架构说明：
1. DeepSeek（本地）：执行量化分析，生成客观的交易信号（BUY/SELL/HOLD）
2. Claude（云端）：你的工作是解释这些信号，生成专业的投资建议

核心职责：
- 解释 DeepSeek 生成的量化信号和技术指标
- 基于多维度数据（技术面、基本面概述）生成投资建议
- 指出投资机会和风险点
- 提供明确的操作建议和风险提示

重要原则：
1. 不编造数据 - 所有数字和指标必须来自输入数据
2. 明确风险 - 每个建议都必须附带风险提示
3. 客观中立 - 基于数据而非主观猜测
4. 可操作性 - 建议必须具有实际的操作指导意义

输出格式要求：
- 建议必须是结构化的，易于前端展示
- 用中文提供专业的财经分析
- 清晰区分不同投资风格（激进/平衡/保守）
"""

    async def _call_claude_openrouter(self, user_prompt: str, max_tokens: int = 2048) -> str:
        """
        调用 Claude OpenRouter API

        Args:
            user_prompt: 用户提示
            max_tokens: 最大输出 token 数

        Returns:
            Claude 的响应文本
        """
        try:
            async with httpx.AsyncClient(timeout=60.0) as client:
                response = await client.post(
                    f"{self.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json",
                    },
                    json={
                        "model": self.model,
                        "messages": [
                            {"role": "system", "content": self._get_advisor_system_prompt()},
                            {"role": "user", "content": user_prompt}
                        ],
                        "max_tokens": max_tokens,
                        "temperature": 0.7,
                    }
                )
                response.raise_for_status()
                result = response.json()
                return result["choices"][0]["message"]["content"]

        except Exception as e:
            logger.error(f"Claude OpenRouter API call failed: {e}")
            raise

    def _extract_json_from_response(self, response_text: str) -> Dict[str, Any]:
        """
        从 Claude 响应中提取 JSON

        Args:
            response_text: Claude 的原始响应文本

        Returns:
            解析后的 JSON 字典
        """
        # 尝试直接解析
        try:
            return json.loads(response_text)
        except json.JSONDecodeError:
            pass

        # 尝试从 markdown 代码块中提取
        if "```json" in response_text:
            json_str = response_text.split("```json")[1].split("```")[0]
        elif "```" in response_text:
            json_str = response_text.split("```")[1].split("```")[0]
        else:
            json_str = response_text

        try:
            return json.loads(json_str.strip())
        except json.JSONDecodeError as e:
            logger.error(f"Failed to parse JSON from Claude response: {e}")
            logger.debug(f"Raw response: {response_text}")
            return {
                "error": "JSON parse failed",
                "raw_text": response_text,
                "recommendation": "分析失败"
            }

    async def generate_investment_advice(
        self,
        symbol: str,
        stock_name: str,
        current_price: float,
        deepseek_signal: Dict[str, Any],
        technical_indicators: Dict[str, Any],
        market_context: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        基于 DeepSeek 信号生成投资建议

        Args:
            symbol: 股票符号
            stock_name: 股票名称
            current_price: 当前价格
            deepseek_signal: DeepSeek 生成的交易信号
            technical_indicators: 技术指标数据
            market_context: 市场背景信息

        Returns:
            包含投资建议的结构化数据
        """
        user_prompt = f"""
请基于以下量化分析结果，为投资者生成专业的投资建议。

【股票信息】
- 代码：{symbol}
- 名称：{stock_name}
- 当前价格：${current_price}

【DeepSeek 量化信号】
{json.dumps(deepseek_signal, ensure_ascii=False, indent=2)}

【技术指标】
{json.dumps(technical_indicators, ensure_ascii=False, indent=2)}
"""
        market_context_str = f'【市场背景】\n{json.dumps(market_context, ensure_ascii=False, indent=2)}' if market_context else ''
        user_prompt += f"""
{market_context_str}

请生成包含以下内容的投资建议（输出 JSON 格式）：

1. 总体建议：基于 DeepSeek 信号的一句话总结
2. 投资逻辑：为什么给出这个建议，主要依据是什么
3. 三档操作策略：
   - 激进（高风险）：适合追求高收益的投资者
   - 平衡（中等风险）：适合风险中立的投资者
   - 保守（低风险）：适合保本优先的投资者
4. 入场价格：建议的买入价格（如果有 DeepSeek 建议则引用）
5. 止损价格：建议的止损点位
6. 止盈价格：建议的止盈点位
7. 投资时间期限：短期/中期/长期
8. 关键风险：列举主要风险点
9. 后续关注：需要关注的关键指标或新闻事件

输出严格 JSON 格式（不需要任何说明文字，直接输出 JSON）：
{{
    "symbol": "{symbol}",
    "recommendation": "强烈建议|建议|中性|建议卖出|强烈建议卖出",
    "summary": "一句话总结",
    "logic": "投资逻辑说明",
    "strategies": {{
        "aggressive": {{
            "action": "BUY|SELL|HOLD",
            "reason": "激进策略的理由"
        }},
        "balanced": {{
            "action": "BUY|SELL|HOLD",
            "reason": "平衡策略的理由"
        }},
        "conservative": {{
            "action": "BUY|SELL|HOLD",
            "reason": "保守策略的理由"
        }}
    }},
    "entry_price": <数字或 null>,
    "stop_loss": <数字或 null>,
    "take_profit": <数字或 null>,
    "time_horizon": "short_term|medium_term|long_term",
    "risks": [
        "风险1",
        "风险2",
        "风险3"
    ],
    "follow_ups": [
        "关注事项1",
        "关注事项2"
    ],
    "confidence": 0.0-1.0
}}
"""

        try:
            response_text = await self._call_claude_openrouter(user_prompt, max_tokens=3072)
            result = self._extract_json_from_response(response_text)

            # 确保返回必需字段
            default_result = {
                "symbol": symbol,
                "recommendation": "中性",
                "summary": "无法生成建议",
                "confidence": 0.0
            }
            result = {**default_result, **result}

            return result

        except Exception as e:
            logger.error(f"Failed to generate investment advice for {symbol}: {e}")
            return {
                "symbol": symbol,
                "recommendation": "中性",
                "summary": "生成建议失败，请稍后重试",
                "error": str(e),
                "confidence": 0.0
            }

    async def generate_comparative_analysis(
        self,
        stocks_with_signals: List[Dict[str, Any]],
        market_context: Optional[Dict[str, Any]] = None
    ) -> Dict[str, Any]:
        """
        对比分析多只股票，生成相对强弱排序

        Args:
            stocks_with_signals: 包含股票数据和 DeepSeek 信号的列表
            market_context: 市场背景

        Returns:
            对比分析结果
        """
        market_context_str = f'市场背景：\n{json.dumps(market_context, ensure_ascii=False, indent=2)}' if market_context else ''
        user_prompt = f"""
请对比以下多只股票的投资机会，生成相对排序和建议。

股票数据与信号：
{json.dumps(stocks_with_signals, ensure_ascii=False, indent=2)}

{market_context_str}

请生成对比分析（输出 JSON 格式）：
1. 相对排序：哪只股票现在最值得投资
2. 为什么：核心理由
3. 配置建议：如何配置这些股票的投资比例
4. 风险提示：整体风险提示

JSON 格式：
{{
    "ranking": [
        {{"symbol": "...", "score": 0-100, "reason": "..."}},
        ...
    ],
    "allocation": {{"symbol1": 40%, "symbol2": 35%, "symbol3": 25%}},
    "portfolio_risk": "LOW|MEDIUM|HIGH",
    "overall_advice": "建议内容"
}}
"""

        try:
            response_text = await self._call_claude_openrouter(user_prompt, max_tokens=2048)
            return self._extract_json_from_response(response_text)

        except Exception as e:
            logger.error(f"Comparative analysis failed: {e}")
            return {
                "error": str(e),
                "ranking": [],
                "portfolio_risk": "UNKNOWN"
            }

    async def explain_deepseek_signal(
        self,
        symbol: str,
        deepseek_signal: Dict[str, Any],
        technical_indicators: Dict[str, Any]
    ) -> str:
        """
        用自然语言解释 DeepSeek 信号

        Args:
            symbol: 股票符号
            deepseek_signal: DeepSeek 生成的信号
            technical_indicators: 技术指标

        Returns:
            自然语言解释
        """
        user_prompt = f"""
请用简洁的自然语言解释以下 {symbol} 的量化信号：

DeepSeek 信号：
{json.dumps(deepseek_signal, ensure_ascii=False, indent=2)}

技术指标：
{json.dumps(technical_indicators, ensure_ascii=False, indent=2)}

要求：
1. 解释信号的含义（为什么是这个信号）
2. 说明主要影响因素
3. 提示可能的风险
4. 保持在 150 字以内
"""

        try:
            return await self._call_claude_openrouter(user_prompt, max_tokens=512)

        except Exception as e:
            logger.error(f"Failed to explain DeepSeek signal: {e}")
            return f"信号解释失败：{str(e)}"


# 单例实例
_claude_advisor: Optional[ClaudeInvestmentAdvisor] = None


def get_claude_advisor() -> ClaudeInvestmentAdvisor:
    """获取 Claude 投资顾问单例"""
    global _claude_advisor
    if _claude_advisor is None:
        _claude_advisor = ClaudeInvestmentAdvisor()
    return _claude_advisor