diff --git a/src/main.py b/src/main.py index 0618921..8e44e6e 100644 --- a/src/main.py +++ b/src/main.py @@ -105,7 +105,11 @@ class WhaleWatcher: decision = await self.llm_analyzer.analyze_whale_trade(whale_trade) # Print the full report (includes analysis + decision summary) - full_report = self.llm_analyzer.format_full_report(whale_trade, decision) + full_report = self.llm_analyzer.format_full_report( + whale_trade, + decision, + historical_report_count=self.llm_analyzer.last_historical_report_count, + ) print(full_report) # Save report to file diff --git a/src/prompts/whale_analyzer.py b/src/prompts/whale_analyzer.py index 0c1e8c6..47b4a57 100644 --- a/src/prompts/whale_analyzer.py +++ b/src/prompts/whale_analyzer.py @@ -69,22 +69,32 @@ class WhaleAnalyzerPrompts: **重要原则**: - **务必使用 Google Search!** 不要仅依赖你的历史知识 - **重视交易者历史记录!** 这是判断交易者专业性的关键依据 +- **如果有历史报告,务必结合历史报告进行综合分析!** 这能帮助你了解该市场的交易模式 - 关注过去24-72小时的最新动态 - 如果搜索不到支持信息,内幕交易可能性应该降低 - 信心不足时建议观望(HOLD)""" @staticmethod - def analyze_whale_trade(trade_context: str) -> str: + def analyze_whale_trade(trade_context: str, historical_context: str = "") -> str: """ Get the prompt for analyzing a whale trade. Args: trade_context: Formatted trade context from AnomalyDetector + historical_context: Formatted historical reports context (optional) Returns: Complete prompt for LLM """ + history_section = "" + if historical_context: + history_section = f""" +{historical_context} + +--- +""" return f"""{trade_context} +{history_section} --- @@ -142,6 +152,27 @@ class WhaleAnalyzerPrompts: --- +## 第三点五步:历史报告综合分析(如有历史报告) + +如果上文提供了历史报告,请进行以下分析: + +### 3.5.1 交易方向对比 +- 历史报告中的交易方向(BUY/SELL)与当前交易是否一致? +- 如果方向一致,这可能表明多个交易者对同一结果有信心 +- 如果方向相反,需要分析原因(时间变化、新信息、不同交易者的判断) + +### 3.5.2 历史内幕交易评估 +- 历史报告对内幕交易的判断如何? +- 如果历史报告也认为是内幕交易,这增强了当前交易的可信度 +- 结合历史报告的证据和当前搜索结果进行综合判断 + +### 3.5.3 趋势演变分析 +- 该市场的交易模式是否有变化? +- 价格从历史报告到现在有何变动? +- 鲸鱼交易的频率和规模是否在增加? + +--- + ## 第四步:内幕交易判定 ### 4.1 内幕交易可能性评估 diff --git a/src/services/anomaly_detector.py b/src/services/anomaly_detector.py index 06ef579..377cbc5 100644 --- a/src/services/anomaly_detector.py +++ b/src/services/anomaly_detector.py @@ -235,8 +235,9 @@ class AnomalyDetector: - **隐含概率**: 交易者认为结果发生的概率约为 {context['implied_probability']:.2%} - **信心程度**: {context['conviction_level']} {ranking_str}{history_str} -### 市场状态 +### 市场信息 - **市场问题**: {context['market_question']} +- **市场描述**: {whale_trade.market_description or 'N/A'} - **市场状态**: {context['market_state']} - **当前赔率**: {prices_str} diff --git a/src/services/llm_analyzer.py b/src/services/llm_analyzer.py index 42d6807..724c9a6 100644 --- a/src/services/llm_analyzer.py +++ b/src/services/llm_analyzer.py @@ -13,6 +13,7 @@ from src.config import get_settings from src.models.trade import WhaleTrade from src.models.decision import LLMDecision, TradeRecommendation, TradeAction, TraderCredibility from src.services.anomaly_detector import AnomalyDetector +from src.services.report_history import ReportHistoryService from src.prompts.whale_analyzer import WhaleAnalyzerPrompts logger = logging.getLogger(__name__) @@ -35,6 +36,15 @@ class LLMAnalyzer: self.anomaly_detector = AnomalyDetector() self.prompts = WhaleAnalyzerPrompts() + self.report_history = ReportHistoryService() + + # Track the number of historical reports used in the last analysis + self._last_historical_report_count = 0 + + @property + def last_historical_report_count(self) -> int: + """Get the number of historical reports used in the last analysis.""" + return self._last_historical_report_count def _extract_json_from_response(self, response: str) -> Optional[dict]: """ @@ -132,9 +142,21 @@ class LLMAnalyzer: # Format trade context for LLM trade_context = self.anomaly_detector.format_for_llm(whale_trade) + # Find and format historical reports for the same market + historical_context = "" + historical_reports = self.report_history.find_historical_reports( + whale_trade.market_question, + similarity_threshold=0.5, + max_reports=5, + ) + self._last_historical_report_count = len(historical_reports) + if historical_reports: + historical_context = self.report_history.format_historical_context(historical_reports) + logger.info(f"Found {len(historical_reports)} historical reports for market: {whale_trade.market_question}") + # Build prompt (Gemini uses single prompt with system instruction) system_prompt = self.prompts.system_prompt() - user_prompt = self.prompts.analyze_whale_trade(trade_context) + user_prompt = self.prompts.analyze_whale_trade(trade_context, historical_context) full_prompt = f"{system_prompt}\n\n---\n\n{user_prompt}" try: @@ -187,13 +209,19 @@ class LLMAnalyzer: ), ) - def format_full_report(self, whale_trade: WhaleTrade, decision: LLMDecision) -> str: + def format_full_report( + self, + whale_trade: WhaleTrade, + decision: LLMDecision, + historical_report_count: int = 0, + ) -> str: """ Format a complete analysis report with trade info, analysis, and decision. Args: whale_trade: The whale trade decision: The LLM decision + historical_report_count: Number of historical reports used in analysis Returns: Formatted report string @@ -242,12 +270,17 @@ class LLMAnalyzer: pnl_str = f"${tr.pnl:,.2f}" if tr.pnl else "N/A" trader_ranking_str = f"| **交易者排名** | {rank_str} (PnL: {pnl_str}) |" + # Historical reports info + historical_info = "" + if historical_report_count > 0: + historical_info = f"\n**参考历史报告**: {historical_report_count} 份 (已综合分析)" + report = f""" {'='*70} # 🐋 鲸鱼交易分析报告 {'='*70} -**生成时间**: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC +**生成时间**: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC{historical_info} ## 交易摘要 diff --git a/src/services/report_history.py b/src/services/report_history.py new file mode 100644 index 0000000..b4193e7 --- /dev/null +++ b/src/services/report_history.py @@ -0,0 +1,257 @@ +"""Report history service - finds and summarizes historical reports for the same market.""" +import os +import re +from pathlib import Path +from typing import List, Optional +from dataclasses import dataclass +from datetime import datetime + + +@dataclass +class HistoricalReport: + """Represents a historical report for the same market.""" + filepath: str + filename: str + timestamp: datetime + side: str + amount_usd: float + market_name: str + content: str + + @property + def summary(self) -> str: + """Extract key decision info from the report.""" + # Try to extract the JSON decision section + json_pattern = r'```json\s*([\s\S]*?)```' + matches = re.findall(json_pattern, self.content) + + decision_info = "" + for match in matches: + try: + import json + data = json.loads(match.strip()) + action = data.get('action', 'N/A') + confidence = data.get('confidence', 'N/A') + insider_likelihood = data.get('insider_trading_likelihood', 'N/A') + reasoning = data.get('reasoning', 'N/A') + + if isinstance(confidence, (int, float)): + confidence = f"{confidence:.0%}" + if isinstance(insider_likelihood, (int, float)): + insider_likelihood = f"{insider_likelihood:.0%}" + + decision_info = f""" +- **操作建议**: {action} +- **信心程度**: {confidence} +- **内幕交易可能性**: {insider_likelihood} +- **决策理由**: {reasoning}""" + break + except (json.JSONDecodeError, KeyError): + continue + + return f"""**报告时间**: {self.timestamp.strftime('%Y-%m-%d %H:%M:%S')} +**交易方向**: {self.side} +**交易金额**: ${self.amount_usd:,.2f} USDC +{decision_info}""" + + +class ReportHistoryService: + """Service for finding and managing historical reports.""" + + def __init__(self, reports_dir: Optional[Path] = None): + """ + Initialize the report history service. + + Args: + reports_dir: Path to the reports directory. Defaults to project's reports dir. + """ + if reports_dir is None: + self.reports_dir = Path(__file__).parent.parent.parent / "reports" + else: + self.reports_dir = reports_dir + + def _parse_filename(self, filename: str) -> Optional[dict]: + """ + Parse a report filename to extract metadata. + + Filename format: {timestamp}_{side}_{amount}USD_{market_name}.md + Example: 20260105_150505_BUY_7520USD_Trump_out_as_President_before_2027.md + + Args: + filename: The filename to parse + + Returns: + Dictionary with parsed metadata or None if parsing fails + """ + if not filename.endswith('.md'): + return None + + # Pattern: timestamp_side_amountUSD_market_name.md + pattern = r'^(\d{8}_\d{6})_(BUY|SELL)_(\d+)USD_(.+)\.md$' + match = re.match(pattern, filename) + + if not match: + return None + + timestamp_str, side, amount_str, market_name = match.groups() + + try: + timestamp = datetime.strptime(timestamp_str, '%Y%m%d_%H%M%S') + amount = float(amount_str) + except ValueError: + return None + + return { + 'timestamp': timestamp, + 'side': side, + 'amount_usd': amount, + 'market_name': market_name, + } + + def _sanitize_market_name(self, market_question: str, max_length: int = 50) -> str: + """ + Sanitize market question for matching with filenames. + + Args: + market_question: The market question to sanitize + max_length: Maximum length of the sanitized name + + Returns: + Sanitized market name + """ + # Remove special characters, keep alphanumeric and spaces + sanitized = re.sub(r'[^\w\s-]', '', market_question) + # Replace spaces with underscores + sanitized = re.sub(r'\s+', '_', sanitized) + # Truncate if too long + return sanitized[:max_length] + + def _calculate_similarity(self, name1: str, name2: str) -> float: + """ + Calculate similarity between two market names. + + Uses a simple word overlap method for fuzzy matching. + + Args: + name1: First market name (sanitized) + name2: Second market name (from filename) + + Returns: + Similarity score between 0 and 1 + """ + # Convert to lowercase and split into words + words1 = set(name1.lower().replace('_', ' ').split()) + words2 = set(name2.lower().replace('_', ' ').split()) + + # Remove common stop words + stop_words = {'the', 'a', 'an', 'is', 'are', 'will', 'by', 'to', 'of', 'in', 'on', 'for'} + words1 = words1 - stop_words + words2 = words2 - stop_words + + if not words1 or not words2: + return 0.0 + + # Calculate Jaccard similarity + intersection = len(words1 & words2) + union = len(words1 | words2) + + return intersection / union if union > 0 else 0.0 + + def find_historical_reports( + self, + market_question: str, + similarity_threshold: float = 0.5, + max_reports: int = 5, + ) -> List[HistoricalReport]: + """ + Find historical reports for the same or similar market. + + Args: + market_question: The market question to search for + similarity_threshold: Minimum similarity score to include a report + max_reports: Maximum number of reports to return + + Returns: + List of HistoricalReport objects, sorted by timestamp (newest first) + """ + if not self.reports_dir.exists(): + return [] + + sanitized_question = self._sanitize_market_name(market_question) + matching_reports = [] + + for filename in os.listdir(self.reports_dir): + metadata = self._parse_filename(filename) + if metadata is None: + continue + + # Calculate similarity between market names + similarity = self._calculate_similarity( + sanitized_question, + metadata['market_name'] + ) + + if similarity >= similarity_threshold: + filepath = self.reports_dir / filename + try: + with open(filepath, 'r', encoding='utf-8') as f: + content = f.read() + except Exception: + continue + + report = HistoricalReport( + filepath=str(filepath), + filename=filename, + timestamp=metadata['timestamp'], + side=metadata['side'], + amount_usd=metadata['amount_usd'], + market_name=metadata['market_name'], + content=content, + ) + matching_reports.append((similarity, report)) + + # Sort by similarity (descending) then by timestamp (descending) + matching_reports.sort(key=lambda x: (x[0], x[1].timestamp), reverse=True) + + # Return only the reports (without similarity scores) + return [report for _, report in matching_reports[:max_reports]] + + def format_historical_context( + self, + reports: List[HistoricalReport], + ) -> str: + """ + Format historical reports into a context string for LLM. + + Args: + reports: List of historical reports + + Returns: + Formatted string for LLM context + """ + if not reports: + return "" + + context = f""" +### 历史报告分析 (共 {len(reports)} 份历史报告) + +**重要**: 该市场之前已经生成过分析报告,请结合历史报告进行综合分析。 + +""" + for i, report in enumerate(reports, 1): + context += f""" +--- +#### 历史报告 {i} +{report.summary} +--- +""" + + context += """ +**综合分析要点**: +1. 对比历史报告中的交易方向和当前交易方向,分析是否有趋势变化 +2. 对比历史的内幕交易可能性评估,判断该市场是否持续有异常交易 +3. 如果多份报告都指向同一方向,这可能加强信号的可信度 +4. 如果报告方向相反,需要分析原因并给出更审慎的判断 +5. 考虑时间因素:越近期的报告越有参考价值 +""" + return context