""" LLM决策层 — Claude API对接（M7阶段） 对应提示词八的system prompt和user prompt模板: 1. 信号确认 — 综合分析信号质量 2. 月度复盘 — 市场阶段判断+参数建议 3. 异常市场 — 是否暂停交易 4. 新币评估 — 是否纳入监控 LLM的输出是"建议"而非"指令"——系统可以采纳也可以忽略 """ import os import json import logging from typing import Optional, Dict logger = logging.getLogger(__name__) # System prompt（发给Claude API的系统提示 — 提示词八原文） SYSTEM_PROMPT = """你是浪浪AI交易系统的高级分析师。你的角色是基于浪浪SOP， 结合当前市场数据和历史交易记忆，给出交易建议。 你的决策原则: 1. 永远以浪浪SOP为基础，不发明新规则 2. 你的输出是"建议"不是"指令"，系统可以采纳也可以忽略 3. 在不确定时，建议是"不做"（宁可错过，不可做错） 4. 用数据说话，每个建议附上统计依据 5. 特别关注风险（最大可能亏损是多少） 你可以访问的数据: - 当前12个币种的K线图（1H/4H/日线） - 交易记忆库（所有历史交易的完整记录） - 当前活跃持仓列表 - 近3个月的信号统计（各类型胜率/盈亏比） - BTC大盘趋势指标 输出格式（JSON）: { "action": "建议做多/建议做空/建议观望/建议减仓", "confidence": 0.0-1.0, "reasoning": "分析逻辑", "risk_warning": "风险提醒", "position_suggestion": "仓位建议", "key_levels": {"stop_loss": xxx, "take_profit": xxx} }""" class LLMAdvisor: """ Claude API对接器 用法: advisor = LLMAdvisor(api_key) # 信号确认 result = advisor.confirm_signal(signal_data, market_context, memory_stats) # 月度复盘 result = advisor.monthly_review(review_data) # 异常市场判断 result = advisor.assess_anomaly(market_data) """ def __init__( self, api_key: Optional[str] = None, model: str = "claude-sonnet-4-20250514", ): self.api_key = api_key or os.environ.get('ANTHROPIC_API_KEY', '') self.model = model self._enabled = bool(self.api_key) if not self._enabled: logger.warning("LLM Advisor: ANTHROPIC_API_KEY not set, disabled") @property def enabled(self) -> bool: return self._enabled def _call_api(self, user_message: str, max_tokens: int = 2000) -> Optional[str]: """调用Claude API""" if not self._enabled: return None try: import requests resp = requests.post( "https://api.anthropic.com/v1/messages", headers={ "x-api-key": self.api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", }, json={ "model": self.model, "max_tokens": max_tokens, "system": SYSTEM_PROMPT, "messages": [ {"role": "user", "content": user_message} ], }, timeout=60, ) if resp.status_code != 200: logger.error(f"Claude API error {resp.status_code}: {resp.text[:200]}") return None data = resp.json() text = data.get('content', [{}])[0].get('text', '') return text except Exception as e: logger.error(f"Claude API call failed: {e}") return None def _parse_json_response(self, text: str) -> Optional[dict]: """从LLM响应中提取JSON""" if not text: return None # 尝试直接解析 try: return json.loads(text) except json.JSONDecodeError: pass # 尝试提取JSON块 import re json_match = re.search(r'\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}', text, re.DOTALL) if json_match: try: return json.loads(json_match.group()) except json.JSONDecodeError: pass # 返回原始文本 return {'raw_response': text} # ==================== 信号确认 ==================== def confirm_signal( self, signal: dict, market_context: dict, memory_stats: dict, ) -> Optional[dict]: """ 信号确认场景（提示词八 User Prompt — 信号确认） Args: signal: 信号详情 market_context: 市场环境（BTC价格、趋势等） memory_stats: 历史统计（近30笔胜率等） Returns: LLM建议 dict or None """ prompt = f"""当前时间: {signal.get('entry_time', 'now')} 触发信号: {signal.get('coin', '')} {signal.get('direction', '')} ({signal.get('signal_type', '')}) 市场环境: - BTC价格: ${market_context.get('btc_price', 0):,.0f} 60日趋势: {market_context.get('btc_trend', 0):+.1%} - {signal.get('coin', '')}价格: ${signal.get('entry_price', 0):,.2f} 60日趋势: {signal.get('coin_trend_60d', 0):+.1%} 信号详情: - 盒子区间: ${signal.get('box_low', 0):,.2f} ~ ${signal.get('box_high', 0):,.2f} (宽度{signal.get('box_width_pct', 0):.1%}) - 突破量比: {signal.get('volume_ratio', 0):.1f}x - 止损位: ${signal.get('stop_loss', 0):,.2f} (距离{signal.get('sl_distance', 0):.1%}) - 建议仓位: {signal.get('position_pct', 0):.0%} 历史参考: - 该币种近30笔{signal.get('signal_type','')}信号: 胜率{memory_stats.get('win_rate', 0):.0f}% 平均盈利{memory_stats.get('avg_win', 0):.1f}% 请分析这个信号的质量，给出你的建议。用JSON格式回复。""" text = self._call_api(prompt) return self._parse_json_response(text) # ==================== 月度复盘 ==================== def monthly_review(self, review_data: dict) -> Optional[dict]: """ 月度复盘场景（提示词八 User Prompt — 月度复盘） Args: review_data: MonthlyReview.to_dict() 的输出 Returns: LLM分析 dict or None """ # 格式化统计数据 summary = review_data.get('summary', {}) signal_stats = json.dumps(review_data.get('signal_types', {}), indent=2, ensure_ascii=False) coin_stats = json.dumps(review_data.get('coins', {}), indent=2, ensure_ascii=False) suggestions = json.dumps(review_data.get('suggestions', []), indent=2, ensure_ascii=False) prompt = f"""当前时间: {review_data.get('month', '')} 复盘周期: 近3个月 交易摘要: - 交易: {summary.get('n', 0)}笔 盈{summary.get('wins', 0)}亏{summary.get('losses', 0)} - 净盈亏: ${summary.get('total_pnl', 0):+,.0f} - 胜率: {summary.get('win_rate', 0):.1f}% PF: {summary.get('pf', 0):.2f} 各信号类型表现（近3个月）: {signal_stats} 各币种表现（近3个月）: {coin_stats} 系统参数建议: {suggestions} 请给出: 1. 当前市场阶段判断 2. 哪些信号类型需要调整权重 3. 哪些参数建议微调 4. 下个月的交易策略建议 5. 风险提醒 用JSON格式回复。""" text = self._call_api(prompt, max_tokens=3000) return self._parse_json_response(text) # ==================== 异常市场 ==================== def assess_anomaly(self, market_data: dict) -> Optional[dict]: """ 异常市场评估 触发条件: 单日波动>8% Args: market_data: 市场数据（波动率、价格变动等） Returns: LLM建议 dict or None """ prompt = f"""当前时间: {market_data.get('time', 'now')} 异常市场告警: - BTC 24H变动: {market_data.get('btc_24h_change', 0):+.1%} - 当前波动率: {market_data.get('volatility', 0):.1%} 活跃持仓: {market_data.get('active_positions', 0)}个 当前总保证金使用: {market_data.get('margin_usage', 0):.0%} 请评估: 1. 是否应该暂停新开仓 2. 是否应该对现有持仓减仓 3. 当前风险等级（1-10） 4. 建议操作 用JSON格式回复。""" text = self._call_api(prompt, max_tokens=1500) return self._parse_json_response(text)