Initial commit: Polymarket Whale Watcher
- Add trade monitoring and whale detection system - Add LLM-powered trade analysis - Add market data fetching from Polymarket API - Include example environment configuration - Add automated report generation 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
SII-leiyu
@@ -0,0 +1,12 @@
|
||||
"""Services module."""
|
||||
from .market_fetcher import MarketFetcher
|
||||
from .trade_monitor import TradeMonitor
|
||||
from .anomaly_detector import AnomalyDetector
|
||||
from .llm_analyzer import LLMAnalyzer
|
||||
|
||||
__all__ = [
|
||||
"MarketFetcher",
|
||||
"TradeMonitor",
|
||||
"AnomalyDetector",
|
||||
"LLMAnalyzer",
|
||||
]
|
||||
@@ -0,0 +1,252 @@
|
||||
"""Anomaly detection service - filters and validates whale trades."""
|
||||
import logging
|
||||
from typing import List, Optional
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from src.config import get_settings
|
||||
from src.models.trade import WhaleTrade, TradeActivity
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AnomalyDetector:
|
||||
"""
|
||||
Detects anomalous (whale) trades based on configurable criteria.
|
||||
|
||||
Criteria:
|
||||
- Trade size >= MIN_TRADE_SIZE_USD (default: $10,000)
|
||||
- Trade price between MIN_PRICE and MAX_PRICE (default: 0.2-0.8)
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.settings = get_settings()
|
||||
|
||||
def is_anomalous_trade(self, activity: TradeActivity) -> bool:
|
||||
"""
|
||||
Check if a trade is anomalous based on size and price.
|
||||
|
||||
Args:
|
||||
activity: The trade activity to check
|
||||
|
||||
Returns:
|
||||
True if the trade is anomalous
|
||||
"""
|
||||
# Check trade size
|
||||
if activity.usdc_size < self.settings.min_trade_size_usd:
|
||||
return False
|
||||
|
||||
# Check price range (0.2-0.8 means not too certain either way)
|
||||
if not (self.settings.min_price <= activity.price <= self.settings.max_price):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def get_anomaly_score(self, activity: TradeActivity) -> float:
|
||||
"""
|
||||
Calculate an anomaly score for a trade.
|
||||
|
||||
Higher score = more interesting anomaly.
|
||||
|
||||
Args:
|
||||
activity: The trade activity to score
|
||||
|
||||
Returns:
|
||||
Anomaly score between 0 and 1
|
||||
"""
|
||||
if not self.is_anomalous_trade(activity):
|
||||
return 0.0
|
||||
|
||||
score = 0.0
|
||||
|
||||
# Size component (bigger trades = higher score)
|
||||
# $10k = 0.3, $50k = 0.5, $100k+ = 0.6
|
||||
size_score = min(0.6, 0.3 + (activity.usdc_size - 10000) / 200000)
|
||||
score += size_score
|
||||
|
||||
# Price component (closer to 0.5 = more uncertain = higher score)
|
||||
# Price at 0.5 = 0.4, price at 0.2 or 0.8 = 0.2
|
||||
price_distance_from_50 = abs(activity.price - 0.5)
|
||||
price_score = 0.4 * (1 - price_distance_from_50 / 0.3)
|
||||
score += max(0, price_score)
|
||||
|
||||
return min(1.0, score)
|
||||
|
||||
def filter_whale_trades(
|
||||
self,
|
||||
trades: List[WhaleTrade],
|
||||
min_score: float = 0.5,
|
||||
) -> List[WhaleTrade]:
|
||||
"""
|
||||
Filter whale trades by anomaly score.
|
||||
|
||||
Args:
|
||||
trades: List of whale trades to filter
|
||||
min_score: Minimum anomaly score to include
|
||||
|
||||
Returns:
|
||||
Filtered list of whale trades
|
||||
"""
|
||||
filtered = []
|
||||
for trade in trades:
|
||||
score = self.get_anomaly_score(trade.trade)
|
||||
if score >= min_score:
|
||||
filtered.append(trade)
|
||||
logger.debug(
|
||||
f"Trade passed filter: ${trade.trade.usdc_size:,.2f} "
|
||||
f"@ {trade.trade.price:.4f} (score: {score:.2f})"
|
||||
)
|
||||
|
||||
return filtered
|
||||
|
||||
def analyze_trade_context(self, whale_trade: WhaleTrade) -> dict:
|
||||
"""
|
||||
Analyze the context of a whale trade for LLM input.
|
||||
|
||||
Args:
|
||||
whale_trade: The whale trade to analyze
|
||||
|
||||
Returns:
|
||||
Dictionary with analysis context
|
||||
"""
|
||||
trade = whale_trade.trade
|
||||
|
||||
# Determine trade direction interpretation
|
||||
if trade.side == "BUY":
|
||||
direction_meaning = f"The trader is betting FOR '{trade.outcome}' occurring"
|
||||
else:
|
||||
direction_meaning = f"The trader is betting AGAINST '{trade.outcome}' occurring"
|
||||
|
||||
# Calculate implied probability from price
|
||||
implied_prob = trade.price if trade.side == "BUY" else (1 - trade.price)
|
||||
|
||||
# Assess market state from outcome prices
|
||||
market_state = "uncertain"
|
||||
if whale_trade.market_outcome_prices:
|
||||
max_price = max(whale_trade.market_outcome_prices)
|
||||
if max_price > 0.7:
|
||||
market_state = "leaning towards one outcome"
|
||||
elif max_price < 0.6:
|
||||
market_state = "highly uncertain"
|
||||
|
||||
# Calculate conviction level based on size
|
||||
conviction = "moderate"
|
||||
if trade.usdc_size >= 50000:
|
||||
conviction = "very high"
|
||||
elif trade.usdc_size >= 25000:
|
||||
conviction = "high"
|
||||
|
||||
return {
|
||||
"trade_size_usd": trade.usdc_size,
|
||||
"trade_side": trade.side,
|
||||
"trade_price": trade.price,
|
||||
"trade_outcome": trade.outcome,
|
||||
"direction_meaning": direction_meaning,
|
||||
"implied_probability": implied_prob,
|
||||
"market_state": market_state,
|
||||
"conviction_level": conviction,
|
||||
"anomaly_score": self.get_anomaly_score(trade),
|
||||
"market_question": whale_trade.market_question,
|
||||
"market_outcomes": whale_trade.market_outcomes,
|
||||
"current_prices": whale_trade.market_outcome_prices,
|
||||
}
|
||||
|
||||
def format_for_llm(self, whale_trade: WhaleTrade) -> str:
|
||||
"""
|
||||
Format whale trade data for LLM analysis.
|
||||
|
||||
Args:
|
||||
whale_trade: The whale trade to format
|
||||
|
||||
Returns:
|
||||
Formatted string for LLM input
|
||||
"""
|
||||
context = self.analyze_trade_context(whale_trade)
|
||||
trade = whale_trade.trade
|
||||
|
||||
# Build outcome prices string
|
||||
prices_str = ""
|
||||
for i, (outcome, price) in enumerate(
|
||||
zip(context["market_outcomes"], context["current_prices"])
|
||||
):
|
||||
prices_str += f" - {outcome}: {price:.2%}\n"
|
||||
|
||||
# Build trader ranking info
|
||||
ranking_str = ""
|
||||
if whale_trade.trader_ranking:
|
||||
rank = whale_trade.trader_ranking
|
||||
rank_display = f"#{rank.rank}" if rank.rank else "未上榜"
|
||||
pnl_display = f"${rank.pnl:,.2f}" if rank.pnl else "N/A"
|
||||
vol_display = f"${rank.volume:,.2f}" if rank.volume else "N/A"
|
||||
verified_display = "✅ 已认证" if rank.verified else "未认证"
|
||||
ranking_str = f"""
|
||||
### 交易者排名信息（盈利排行榜）
|
||||
- **排名**: {rank_display} (时间范围: {rank.time_period})
|
||||
- **累计盈亏 (PnL)**: {pnl_display}
|
||||
- **总交易量**: {vol_display}
|
||||
- **用户名**: {rank.user_name or 'Anonymous'}
|
||||
- **认证状态**: {verified_display}
|
||||
"""
|
||||
else:
|
||||
ranking_str = """
|
||||
### 交易者排名信息
|
||||
- 该交易者不在盈利排行榜上（可能是新用户或小额交易者）
|
||||
"""
|
||||
|
||||
# Build trader history info
|
||||
history_str = ""
|
||||
if whale_trade.trader_history:
|
||||
hist = whale_trade.trader_history
|
||||
history_str = f"""
|
||||
### 交易者历史交易记录（重要！）
|
||||
- **近期交易总数**: {hist.total_trades} 笔
|
||||
- **近期交易总额**: ${hist.total_volume:,.2f} USDC
|
||||
- **平均交易金额**: ${hist.avg_trade_size:,.2f} USDC
|
||||
- **大额交易次数** (≥$5000): {hist.large_trades_count} 笔
|
||||
- **活跃市场**: {', '.join(hist.recent_markets[:5]) if hist.recent_markets else 'N/A'}
|
||||
"""
|
||||
# Add recent large trades details
|
||||
if hist.recent_trades:
|
||||
history_str += "\n**近期大额交易明细**:\n"
|
||||
for i, t in enumerate(hist.recent_trades[:5], 1):
|
||||
title = t.get('title', 'N/A')
|
||||
if len(title) > 40:
|
||||
title = title[:40] + "..."
|
||||
history_str += f" {i}. {t.get('side', 'N/A')} ${t.get('usdc_size', 0):,.2f} @ {t.get('price', 0):.4f} - {title}\n"
|
||||
else:
|
||||
history_str = """
|
||||
### 交易者历史交易记录
|
||||
- 无法获取该交易者的历史交易记录
|
||||
"""
|
||||
|
||||
return f"""
|
||||
## 大额交易异常检测报告
|
||||
|
||||
### 交易详情
|
||||
- **交易金额**: ${context['trade_size_usd']:,.2f} USDC
|
||||
- **交易方向**: {context['trade_side']}
|
||||
- **交易价格**: {context['trade_price']:.4f} ({context['trade_price']:.2%})
|
||||
- **交易结果**: {context['trade_outcome']}
|
||||
- **交易时间**: {datetime.fromtimestamp(trade.timestamp).strftime('%Y-%m-%d %H:%M:%S')}
|
||||
- **交易者钱包**: {trade.proxy_wallet or 'Unknown'}
|
||||
- **异常评分**: {context['anomaly_score']:.2f}/1.00
|
||||
|
||||
### 交易解读
|
||||
- **方向含义**: {context['direction_meaning']}
|
||||
- **隐含概率**: 交易者认为结果发生的概率约为 {context['implied_probability']:.2%}
|
||||
- **信心程度**: {context['conviction_level']}
|
||||
{ranking_str}{history_str}
|
||||
### 市场状态
|
||||
- **市场问题**: {context['market_question']}
|
||||
- **市场状态**: {context['market_state']}
|
||||
- **当前赔率**:
|
||||
{prices_str}
|
||||
|
||||
### 分析要点
|
||||
1. 这是一笔 ${context['trade_size_usd']:,.2f} 的大额交易，表明交易者有{context['conviction_level']}的信心
|
||||
2. 交易价格 {context['trade_price']:.4f} 说明市场尚未形成明确共识
|
||||
3. {context['direction_meaning']}
|
||||
4. **请重点分析交易者的排名和历史交易记录，判断其专业性和可信度**
|
||||
5. 这可能暗示交易者掌握了某些市场尚未充分反映的信息
|
||||
|
||||
请分析这笔交易并给出你的交易建议。
|
||||
"""
|
||||
@@ -0,0 +1,299 @@
|
||||
"""LLM analyzer service - analyzes whale trades using AI."""
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from google import genai
|
||||
from google.genai import types
|
||||
|
||||
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.prompts.whale_analyzer import WhaleAnalyzerPrompts
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class LLMAnalyzer:
|
||||
"""
|
||||
Analyzes whale trades using LLM (Google Gemini models).
|
||||
|
||||
Combines trade context with superforecaster methodology to generate
|
||||
comprehensive analysis reports with trading recommendations.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self.settings = get_settings()
|
||||
|
||||
# Configure Gemini API using new client SDK
|
||||
os.environ["GOOGLE_API_KEY"] = self.settings.gemini_api_key
|
||||
self.client = genai.Client()
|
||||
|
||||
self.anomaly_detector = AnomalyDetector()
|
||||
self.prompts = WhaleAnalyzerPrompts()
|
||||
|
||||
def _extract_json_from_response(self, response: str) -> Optional[dict]:
|
||||
"""
|
||||
Extract JSON from LLM response.
|
||||
|
||||
Args:
|
||||
response: The LLM response text
|
||||
|
||||
Returns:
|
||||
Parsed JSON dict or None
|
||||
"""
|
||||
# Try to find JSON in code blocks
|
||||
json_pattern = r"```(?:json)?\s*([\s\S]*?)```"
|
||||
matches = re.findall(json_pattern, response)
|
||||
|
||||
for match in matches:
|
||||
try:
|
||||
return json.loads(match.strip())
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
# Try to find raw JSON
|
||||
try:
|
||||
# Find JSON-like content
|
||||
start = response.find("{")
|
||||
end = response.rfind("}") + 1
|
||||
if start >= 0 and end > start:
|
||||
return json.loads(response[start:end])
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
return None
|
||||
|
||||
def _parse_recommendation(self, json_data: dict) -> TradeRecommendation:
|
||||
"""
|
||||
Parse JSON data into TradeRecommendation.
|
||||
|
||||
Args:
|
||||
json_data: Parsed JSON from LLM
|
||||
|
||||
Returns:
|
||||
TradeRecommendation object
|
||||
"""
|
||||
action_str = json_data.get("action", "HOLD").upper()
|
||||
try:
|
||||
action = TradeAction(action_str)
|
||||
except ValueError:
|
||||
action = TradeAction.HOLD
|
||||
|
||||
confidence = float(json_data.get("confidence", 0.0))
|
||||
# Clamp confidence to valid range
|
||||
confidence = max(0.0, min(1.0, confidence))
|
||||
|
||||
suggested_price = json_data.get("suggested_price")
|
||||
if suggested_price is not None:
|
||||
suggested_price = float(suggested_price)
|
||||
|
||||
suggested_size = float(json_data.get("suggested_size_percent", 0.1))
|
||||
suggested_size = max(0.0, min(1.0, suggested_size))
|
||||
|
||||
# Parse insider trading assessment fields
|
||||
insider_likelihood = float(json_data.get("insider_trading_likelihood", 0.0))
|
||||
insider_likelihood = max(0.0, min(1.0, insider_likelihood))
|
||||
|
||||
credibility_str = json_data.get("trader_credibility", "UNKNOWN").upper()
|
||||
try:
|
||||
trader_credibility = TraderCredibility(credibility_str)
|
||||
except ValueError:
|
||||
trader_credibility = TraderCredibility.UNKNOWN
|
||||
|
||||
insider_evidence = str(json_data.get("insider_evidence", ""))
|
||||
|
||||
return TradeRecommendation(
|
||||
action=action,
|
||||
outcome=str(json_data.get("outcome", "")),
|
||||
confidence=confidence,
|
||||
suggested_price=suggested_price,
|
||||
suggested_size_percent=suggested_size,
|
||||
reasoning=str(json_data.get("reasoning", "")),
|
||||
insider_trading_likelihood=insider_likelihood,
|
||||
trader_credibility=trader_credibility,
|
||||
insider_evidence=insider_evidence,
|
||||
)
|
||||
|
||||
async def analyze_whale_trade(self, whale_trade: WhaleTrade) -> LLMDecision:
|
||||
"""
|
||||
Analyze a whale trade using LLM.
|
||||
|
||||
Args:
|
||||
whale_trade: The whale trade to analyze
|
||||
|
||||
Returns:
|
||||
LLMDecision with analysis and recommendation
|
||||
"""
|
||||
# Format trade context for LLM
|
||||
trade_context = self.anomaly_detector.format_for_llm(whale_trade)
|
||||
|
||||
# Build prompt (Gemini uses single prompt with system instruction)
|
||||
system_prompt = self.prompts.system_prompt()
|
||||
user_prompt = self.prompts.analyze_whale_trade(trade_context)
|
||||
full_prompt = f"{system_prompt}\n\n---\n\n{user_prompt}"
|
||||
|
||||
try:
|
||||
# Call Gemini API with Google Search tool enabled
|
||||
response = self.client.models.generate_content(
|
||||
model=self.settings.llm_model,
|
||||
contents=full_prompt,
|
||||
config=types.GenerateContentConfig(
|
||||
tools=[types.Tool(google_search=types.GoogleSearch())],
|
||||
),
|
||||
)
|
||||
|
||||
analysis_text = response.text
|
||||
logger.debug(f"LLM response: {analysis_text[:500]}...")
|
||||
|
||||
# Extract JSON from response
|
||||
json_data = self._extract_json_from_response(analysis_text)
|
||||
|
||||
if json_data:
|
||||
recommendation = self._parse_recommendation(json_data)
|
||||
else:
|
||||
# Default to HOLD if we can't parse the response
|
||||
logger.warning("Could not parse LLM response as JSON, defaulting to HOLD")
|
||||
recommendation = TradeRecommendation(
|
||||
action=TradeAction.HOLD,
|
||||
outcome="",
|
||||
confidence=0.0,
|
||||
reasoning="Failed to parse LLM response",
|
||||
)
|
||||
|
||||
return LLMDecision(
|
||||
whale_trade_id=whale_trade.id,
|
||||
market_id=whale_trade.market_id,
|
||||
analysis=analysis_text,
|
||||
recommendation=recommendation,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error calling LLM: {e}")
|
||||
# Return a safe default decision
|
||||
return LLMDecision(
|
||||
whale_trade_id=whale_trade.id,
|
||||
market_id=whale_trade.market_id,
|
||||
analysis=f"Error during analysis: {str(e)}",
|
||||
recommendation=TradeRecommendation(
|
||||
action=TradeAction.HOLD,
|
||||
outcome="",
|
||||
confidence=0.0,
|
||||
reasoning=f"Analysis failed: {str(e)}",
|
||||
),
|
||||
)
|
||||
|
||||
def format_full_report(self, whale_trade: WhaleTrade, decision: LLMDecision) -> str:
|
||||
"""
|
||||
Format a complete analysis report with trade info, analysis, and decision.
|
||||
|
||||
Args:
|
||||
whale_trade: The whale trade
|
||||
decision: The LLM decision
|
||||
|
||||
Returns:
|
||||
Formatted report string
|
||||
"""
|
||||
trade = whale_trade.trade
|
||||
rec = decision.recommendation
|
||||
|
||||
# Format outcome prices
|
||||
prices_str = ""
|
||||
if whale_trade.market_outcomes and whale_trade.market_outcome_prices:
|
||||
prices_str = " | ".join([
|
||||
f"{o}: {p:.1%}"
|
||||
for o, p in zip(whale_trade.market_outcomes, whale_trade.market_outcome_prices)
|
||||
])
|
||||
|
||||
# Action emoji and color indicator
|
||||
action_indicator = {
|
||||
TradeAction.BUY: "🟢 BUY",
|
||||
TradeAction.SELL: "🔴 SELL",
|
||||
TradeAction.HOLD: "⚪ HOLD",
|
||||
}
|
||||
|
||||
# Insider trading likelihood indicator
|
||||
insider_likelihood = rec.insider_trading_likelihood
|
||||
if insider_likelihood >= 0.7:
|
||||
insider_indicator = f"🔴 高度可疑 ({insider_likelihood:.0%})"
|
||||
elif insider_likelihood >= 0.4:
|
||||
insider_indicator = f"🟡 中等可能 ({insider_likelihood:.0%})"
|
||||
else:
|
||||
insider_indicator = f"🟢 普通交易 ({insider_likelihood:.0%})"
|
||||
|
||||
# Trader credibility indicator
|
||||
credibility_indicators = {
|
||||
TraderCredibility.HIGH: "🏆 高可信度 (前100名)",
|
||||
TraderCredibility.MEDIUM: "⭐ 中等可信度 (100-500名)",
|
||||
TraderCredibility.LOW: "📉 低可信度 (500名+)",
|
||||
TraderCredibility.UNKNOWN: "❓ 未知 (未上榜)",
|
||||
}
|
||||
credibility_str = credibility_indicators.get(rec.trader_credibility, "❓ 未知")
|
||||
|
||||
# Trader ranking info
|
||||
trader_ranking_str = ""
|
||||
if whale_trade.trader_ranking:
|
||||
tr = whale_trade.trader_ranking
|
||||
rank_str = f"#{tr.rank}" if tr.rank else "未上榜"
|
||||
pnl_str = f"${tr.pnl:,.2f}" if tr.pnl else "N/A"
|
||||
trader_ranking_str = f"| **交易者排名** | {rank_str} (PnL: {pnl_str}) |"
|
||||
|
||||
report = f"""
|
||||
{'='*70}
|
||||
# 🐋 鲸鱼交易分析报告
|
||||
{'='*70}
|
||||
|
||||
**生成时间**: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC
|
||||
|
||||
## 交易摘要
|
||||
|
||||
| 项目 | 详情 |
|
||||
|------|------|
|
||||
| **市场** | {whale_trade.market_question} |
|
||||
| **交易金额** | ${trade.usdc_size:,.2f} USDC |
|
||||
| **交易方向** | {trade.side} |
|
||||
| **交易价格** | {trade.price:.4f} ({trade.price:.1%}) |
|
||||
| **交易结果** | {trade.outcome} |
|
||||
| **当前赔率** | {prices_str} |
|
||||
| **交易时间** | {datetime.fromtimestamp(trade.timestamp).strftime('%Y-%m-%d %H:%M:%S') if trade.timestamp else 'N/A'} |
|
||||
{trader_ranking_str}
|
||||
|
||||
{'='*70}
|
||||
|
||||
{decision.analysis}
|
||||
|
||||
{'='*70}
|
||||
## 🔍 内幕交易评估
|
||||
{'='*70}
|
||||
|
||||
| 项目 | 评估 |
|
||||
|------|------|
|
||||
| **内幕交易可能性** | {insider_indicator} |
|
||||
| **交易者可信度** | {credibility_str} |
|
||||
|
||||
**关键证据**: {rec.insider_evidence or '无明确证据'}
|
||||
|
||||
{'='*70}
|
||||
## 📊 决策摘要
|
||||
{'='*70}
|
||||
|
||||
| 项目 | 建议 |
|
||||
|------|------|
|
||||
| **操作建议** | {action_indicator.get(rec.action, '⚪ HOLD')} |
|
||||
| **目标结果** | {rec.outcome or 'N/A'} |
|
||||
| **信心程度** | {rec.confidence:.1%} |
|
||||
| **建议仓位** | {rec.suggested_size_percent:.1%} |
|
||||
| **建议价格** | {f'{rec.suggested_price:.4f}' if rec.suggested_price else 'Market'} |
|
||||
|
||||
**决策理由**: {rec.reasoning}
|
||||
|
||||
{'='*70}
|
||||
⚠️ 免责声明：本报告由AI生成，仅供参考，不构成投资建议。
|
||||
预测市场具有高风险，请基于自身判断谨慎决策。
|
||||
{'='*70}
|
||||
"""
|
||||
return report
|
||||
@@ -0,0 +1,301 @@
|
||||
"""Market fetching service - fetches trending markets from Polymarket."""
|
||||
import json
|
||||
import logging
|
||||
from typing import List, Optional
|
||||
|
||||
import httpx
|
||||
|
||||
from src.config import get_settings
|
||||
from src.models.market import Market, TrendingMarket
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class MarketFetcher:
|
||||
"""Fetches and manages trending markets from Polymarket Gamma API."""
|
||||
|
||||
# Sports-related keywords to filter out (case-insensitive)
|
||||
SPORTS_KEYWORDS = [
|
||||
# General sports terms
|
||||
"nba", "nfl", "mlb", "nhl", "mls", "ufc", "wwe", "pga", "atp", "wta",
|
||||
"fifa", "uefa", "epl", "premier league", "la liga", "serie a", "bundesliga",
|
||||
"champions league", "world cup", "olympics", "olympic",
|
||||
# Sports names
|
||||
"basketball", "football", "soccer", "baseball", "hockey", "tennis",
|
||||
"golf", "boxing", "mma", "wrestling", "cricket", "rugby", "f1", "formula 1",
|
||||
"nascar", "racing", "motorsport",
|
||||
# Team/game terms
|
||||
"game", "match", "vs", "versus", "playoff", "playoffs", "finals",
|
||||
"championship", "tournament", "season", "super bowl", "world series",
|
||||
# Player/team actions
|
||||
"score", "points", "goals", "touchdowns", "wins", "win against",
|
||||
"beat", "defeat",
|
||||
# Specific sports betting terms
|
||||
"mvp", "rookie", "all-star", "draft", "trade",
|
||||
# Common sports team cities/names patterns
|
||||
"lakers", "celtics", "warriors", "bulls", "heat", "knicks",
|
||||
"yankees", "dodgers", "red sox", "cubs", "mets",
|
||||
"cowboys", "patriots", "chiefs", "eagles", "49ers",
|
||||
"manchester", "barcelona", "real madrid", "liverpool", "chelsea",
|
||||
]
|
||||
|
||||
def __init__(self):
|
||||
self.settings = get_settings()
|
||||
self.gamma_url = "https://gamma-api.polymarket.com"
|
||||
self.markets_endpoint = f"{self.gamma_url}/markets"
|
||||
self.events_endpoint = f"{self.gamma_url}/events"
|
||||
self._client = httpx.Client(timeout=30.0)
|
||||
|
||||
def __del__(self):
|
||||
"""Cleanup HTTP client."""
|
||||
if hasattr(self, "_client"):
|
||||
self._client.close()
|
||||
|
||||
def _is_sports_market(self, market_data: dict) -> bool:
|
||||
"""
|
||||
Check if a market is sports-related.
|
||||
|
||||
Args:
|
||||
market_data: Raw market data from API
|
||||
|
||||
Returns:
|
||||
True if the market is sports-related
|
||||
"""
|
||||
# Check question and description
|
||||
question = (market_data.get("question") or "").lower()
|
||||
description = (market_data.get("description") or "").lower()
|
||||
slug = (market_data.get("slug") or "").lower()
|
||||
|
||||
text_to_check = f"{question} {description} {slug}"
|
||||
|
||||
for keyword in self.SPORTS_KEYWORDS:
|
||||
if keyword in text_to_check:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _parse_market(self, data: dict) -> Optional[Market]:
|
||||
"""Parse raw market data into Market model."""
|
||||
try:
|
||||
# Parse outcome prices (comes as stringified list)
|
||||
outcome_prices = data.get("outcomePrices", [])
|
||||
if isinstance(outcome_prices, str):
|
||||
outcome_prices = json.loads(outcome_prices)
|
||||
outcome_prices = [float(p) for p in outcome_prices]
|
||||
|
||||
# Parse clob token IDs
|
||||
clob_token_ids = data.get("clobTokenIds", [])
|
||||
if isinstance(clob_token_ids, str):
|
||||
clob_token_ids = json.loads(clob_token_ids)
|
||||
|
||||
# Parse outcomes
|
||||
outcomes = data.get("outcomes", [])
|
||||
if isinstance(outcomes, str):
|
||||
outcomes = json.loads(outcomes)
|
||||
|
||||
return Market(
|
||||
id=str(data.get("id", "")),
|
||||
question=data.get("question", ""),
|
||||
condition_id=data.get("conditionId"),
|
||||
slug=data.get("slug"),
|
||||
description=data.get("description"),
|
||||
end_date=data.get("endDate"),
|
||||
outcomes=outcomes,
|
||||
outcome_prices=outcome_prices,
|
||||
clob_token_ids=clob_token_ids,
|
||||
volume=float(data.get("volume", 0) or 0),
|
||||
volume_24hr=float(data.get("volume24hr", 0) or 0),
|
||||
liquidity=float(data.get("liquidity", 0) or 0),
|
||||
active=data.get("active", False),
|
||||
closed=data.get("closed", False),
|
||||
neg_risk=data.get("negRisk", False),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to parse market {data.get('id')}: {e}")
|
||||
return None
|
||||
|
||||
def get_trending_markets(self, limit: Optional[int] = None) -> List[TrendingMarket]:
|
||||
"""
|
||||
Fetch trending markets sorted by 24-hour volume.
|
||||
|
||||
Filters out sports-related markets since they lack fundamental analysis value.
|
||||
|
||||
Args:
|
||||
limit: Maximum number of non-sports markets to return (defaults to settings)
|
||||
|
||||
Returns:
|
||||
List of TrendingMarket objects (excluding sports markets)
|
||||
"""
|
||||
limit = limit or self.settings.trending_markets_limit
|
||||
trending_markets = []
|
||||
offset = 0
|
||||
batch_size = 100 # Fetch more to account for sports filtering
|
||||
max_iterations = 10 # Safety limit to prevent infinite loops
|
||||
|
||||
try:
|
||||
iteration = 0
|
||||
while len(trending_markets) < limit and iteration < max_iterations:
|
||||
iteration += 1
|
||||
|
||||
# Fetch active markets sorted by volume
|
||||
params = {
|
||||
"active": True,
|
||||
"closed": False,
|
||||
"archived": False,
|
||||
"limit": batch_size,
|
||||
"offset": offset,
|
||||
"order": "volume24hr",
|
||||
"ascending": False,
|
||||
"enableOrderBook": True, # Only markets with CLOB enabled
|
||||
}
|
||||
|
||||
response = self._client.get(self.markets_endpoint, params=params)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
if not data:
|
||||
break # No more markets
|
||||
|
||||
sports_count = 0
|
||||
for market_data in data:
|
||||
# Skip sports markets
|
||||
if self._is_sports_market(market_data):
|
||||
sports_count += 1
|
||||
continue
|
||||
|
||||
market = self._parse_market(market_data)
|
||||
if market:
|
||||
trending_market = TrendingMarket(
|
||||
market=market,
|
||||
volume_24hr=market.volume_24hr,
|
||||
liquidity=market.liquidity,
|
||||
rank=len(trending_markets) + 1,
|
||||
)
|
||||
if trending_market.is_valid_for_monitoring:
|
||||
trending_markets.append(trending_market)
|
||||
|
||||
if len(trending_markets) >= limit:
|
||||
break
|
||||
|
||||
logger.debug(
|
||||
f"Batch {iteration}: fetched {len(data)}, "
|
||||
f"filtered {sports_count} sports markets, "
|
||||
f"total non-sports: {len(trending_markets)}"
|
||||
)
|
||||
|
||||
if len(data) < batch_size:
|
||||
break # No more markets available
|
||||
|
||||
offset += batch_size
|
||||
|
||||
logger.info(
|
||||
f"Fetched {len(trending_markets)} trending markets (sports markets filtered out)"
|
||||
)
|
||||
return trending_markets
|
||||
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"HTTP error fetching trending markets: {e}")
|
||||
return trending_markets # Return what we have so far
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching trending markets: {e}")
|
||||
return trending_markets
|
||||
|
||||
def get_market_by_id(self, market_id: str) -> Optional[Market]:
|
||||
"""
|
||||
Fetch a single market by ID.
|
||||
|
||||
Args:
|
||||
market_id: The market ID
|
||||
|
||||
Returns:
|
||||
Market object or None
|
||||
"""
|
||||
try:
|
||||
url = f"{self.markets_endpoint}/{market_id}"
|
||||
response = self._client.get(url)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
return self._parse_market(data)
|
||||
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"HTTP error fetching market {market_id}: {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching market {market_id}: {e}")
|
||||
return None
|
||||
|
||||
def get_market_by_condition_id(self, condition_id: str) -> Optional[Market]:
|
||||
"""
|
||||
Fetch a market by condition ID.
|
||||
|
||||
Args:
|
||||
condition_id: The condition ID
|
||||
|
||||
Returns:
|
||||
Market object or None
|
||||
"""
|
||||
try:
|
||||
params = {"conditionId": condition_id}
|
||||
response = self._client.get(self.markets_endpoint, params=params)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
if data and len(data) > 0:
|
||||
return self._parse_market(data[0])
|
||||
return None
|
||||
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"HTTP error fetching market by condition {condition_id}: {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching market by condition {condition_id}: {e}")
|
||||
return None
|
||||
|
||||
def get_all_current_markets(self, batch_size: int = 100) -> List[Market]:
|
||||
"""
|
||||
Fetch all current active markets (paginated).
|
||||
|
||||
Args:
|
||||
batch_size: Number of markets per request
|
||||
|
||||
Returns:
|
||||
List of all active Market objects
|
||||
"""
|
||||
all_markets = []
|
||||
offset = 0
|
||||
|
||||
while True:
|
||||
try:
|
||||
params = {
|
||||
"active": True,
|
||||
"closed": False,
|
||||
"archived": False,
|
||||
"limit": batch_size,
|
||||
"offset": offset,
|
||||
"enableOrderBook": True,
|
||||
}
|
||||
|
||||
response = self._client.get(self.markets_endpoint, params=params)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
if not data:
|
||||
break
|
||||
|
||||
for market_data in data:
|
||||
market = self._parse_market(market_data)
|
||||
if market:
|
||||
all_markets.append(market)
|
||||
|
||||
if len(data) < batch_size:
|
||||
break
|
||||
|
||||
offset += batch_size
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching markets at offset {offset}: {e}")
|
||||
break
|
||||
|
||||
logger.info(f"Fetched {len(all_markets)} total active markets")
|
||||
return all_markets
|
||||
@@ -0,0 +1,475 @@
|
||||
"""Trade monitoring service - monitors markets for whale trades."""
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Dict, List, Optional, Set, Callable, Awaitable
|
||||
|
||||
import httpx
|
||||
|
||||
from src.config import get_settings
|
||||
from src.models.market import Market, TrendingMarket
|
||||
from src.models.trade import TradeActivity, WhaleTrade, TraderRanking, TraderHistory
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# File to persist processed transaction hashes
|
||||
PROCESSED_TXNS_FILE = Path(__file__).parent.parent.parent / "data" / "processed_transactions.json"
|
||||
|
||||
|
||||
class TradeMonitor:
|
||||
"""
|
||||
Monitors Polymarket markets for large trades.
|
||||
|
||||
Similar to copy-trading-bot's tradeMonitor, but monitors markets instead of users.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
on_whale_detected: Optional[Callable[[WhaleTrade], Awaitable[None]]] = None,
|
||||
):
|
||||
"""
|
||||
Initialize trade monitor.
|
||||
|
||||
Args:
|
||||
on_whale_detected: Async callback when whale trade is detected
|
||||
"""
|
||||
self.settings = get_settings()
|
||||
self.data_api_url = "https://data-api.polymarket.com"
|
||||
self.trades_endpoint = f"{self.data_api_url}/trades"
|
||||
self.leaderboard_endpoint = f"{self.data_api_url}/v1/leaderboard"
|
||||
self._client = httpx.AsyncClient(timeout=30.0)
|
||||
|
||||
# Cache for trader rankings to avoid repeated API calls
|
||||
self._trader_ranking_cache: Dict[str, TraderRanking] = {}
|
||||
|
||||
# Markets being monitored: condition_id -> Market
|
||||
self._monitored_markets: Dict[str, Market] = {}
|
||||
|
||||
# Track processed transactions to avoid duplicates
|
||||
self._processed_txns: Set[str] = set()
|
||||
|
||||
# Load previously processed transactions from file
|
||||
self._load_processed_txns()
|
||||
|
||||
# Callback for whale detection
|
||||
self._on_whale_detected = on_whale_detected
|
||||
|
||||
# Control flag
|
||||
self._running = False
|
||||
|
||||
# Flag to track if initial scan is complete (ignore historical trades)
|
||||
self._initial_scan_complete = False
|
||||
|
||||
def _load_processed_txns(self):
|
||||
"""Load processed transaction hashes from JSON file."""
|
||||
try:
|
||||
if PROCESSED_TXNS_FILE.exists():
|
||||
with open(PROCESSED_TXNS_FILE, "r") as f:
|
||||
data = json.load(f)
|
||||
self._processed_txns = set(data.get("transactions", []))
|
||||
logger.info(f"Loaded {len(self._processed_txns)} processed transactions from file")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to load processed transactions: {e}")
|
||||
self._processed_txns = set()
|
||||
|
||||
def _save_processed_txns(self):
|
||||
"""Save processed transaction hashes to JSON file."""
|
||||
try:
|
||||
# Ensure directory exists
|
||||
PROCESSED_TXNS_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
with open(PROCESSED_TXNS_FILE, "w") as f:
|
||||
json.dump({
|
||||
"transactions": list(self._processed_txns),
|
||||
"count": len(self._processed_txns),
|
||||
"last_updated": datetime.now().isoformat()
|
||||
}, f, indent=2)
|
||||
logger.debug(f"Saved {len(self._processed_txns)} processed transactions to file")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to save processed transactions: {e}")
|
||||
|
||||
async def close(self):
|
||||
"""Cleanup resources."""
|
||||
# Save processed transactions before closing
|
||||
self._save_processed_txns()
|
||||
await self._client.aclose()
|
||||
|
||||
def set_monitored_markets(self, markets: List[TrendingMarket]):
|
||||
"""
|
||||
Update the list of markets to monitor.
|
||||
|
||||
Args:
|
||||
markets: List of trending markets to monitor
|
||||
"""
|
||||
self._monitored_markets = {}
|
||||
for tm in markets:
|
||||
if tm.market.condition_id:
|
||||
self._monitored_markets[tm.market.condition_id] = tm.market
|
||||
|
||||
logger.info(f"Now monitoring {len(self._monitored_markets)} markets")
|
||||
|
||||
async def fetch_market_trades(self, condition_id: str) -> List[TradeActivity]:
|
||||
"""
|
||||
Fetch recent trades for a market using the /trades endpoint.
|
||||
|
||||
This endpoint allows querying by market without requiring a user address.
|
||||
|
||||
Args:
|
||||
condition_id: The market condition ID
|
||||
|
||||
Returns:
|
||||
List of trade activities
|
||||
"""
|
||||
try:
|
||||
# Use /trades endpoint which supports market-based queries
|
||||
# Docs: https://docs.polymarket.com/api-reference/core/get-trades-for-a-user-or-markets
|
||||
params = {
|
||||
"market": condition_id,
|
||||
"limit": 500,
|
||||
}
|
||||
|
||||
response = await self._client.get(self.trades_endpoint, params=params)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
activities = []
|
||||
|
||||
for item in data:
|
||||
try:
|
||||
# Calculate USDC size from price and size
|
||||
size = float(item.get("size", 0) or 0)
|
||||
price = float(item.get("price", 0) or 0)
|
||||
usdc_size = float(item.get("usdcSize", 0) or 0)
|
||||
|
||||
# If usdcSize not provided, calculate it
|
||||
if usdc_size == 0 and size > 0 and price > 0:
|
||||
usdc_size = size * price
|
||||
|
||||
activity = TradeActivity(
|
||||
transaction_hash=item.get("transactionHash", item.get("id", "")),
|
||||
timestamp=item.get("timestamp", 0),
|
||||
condition_id=item.get("conditionId", condition_id),
|
||||
asset=item.get("asset", item.get("tokenId", "")),
|
||||
side=item.get("side", ""),
|
||||
size=size,
|
||||
usdc_size=usdc_size,
|
||||
price=price,
|
||||
outcome=item.get("outcome", ""),
|
||||
outcome_index=int(item.get("outcomeIndex", 0) or 0),
|
||||
title=item.get("title", item.get("marketTitle", "")),
|
||||
slug=item.get("slug", item.get("marketSlug")),
|
||||
event_slug=item.get("eventSlug"),
|
||||
proxy_wallet=item.get("proxyWallet", item.get("maker", item.get("taker"))),
|
||||
name=item.get("name"),
|
||||
)
|
||||
activities.append(activity)
|
||||
except Exception as e:
|
||||
logger.debug(f"Failed to parse trade: {e}")
|
||||
continue
|
||||
|
||||
return activities
|
||||
|
||||
except httpx.HTTPError as e:
|
||||
logger.warning(f"HTTP error fetching trades for {condition_id}: {e}")
|
||||
return []
|
||||
except Exception as e:
|
||||
logger.warning(f"Error fetching trades for {condition_id}: {e}")
|
||||
return []
|
||||
|
||||
async def fetch_trader_ranking(self, wallet_address: str) -> Optional[TraderRanking]:
|
||||
"""
|
||||
Fetch trader ranking from the leaderboard API.
|
||||
|
||||
Args:
|
||||
wallet_address: The trader's wallet address
|
||||
|
||||
Returns:
|
||||
TraderRanking or None if not found/error
|
||||
"""
|
||||
if not wallet_address:
|
||||
return None
|
||||
|
||||
# Check cache first
|
||||
if wallet_address in self._trader_ranking_cache:
|
||||
return self._trader_ranking_cache[wallet_address]
|
||||
|
||||
try:
|
||||
# Query leaderboard for this specific user (ALL time period for overall ranking)
|
||||
params = {
|
||||
"user": wallet_address,
|
||||
"timePeriod": "ALL",
|
||||
"orderBy": "PNL",
|
||||
}
|
||||
|
||||
response = await self._client.get(self.leaderboard_endpoint, params=params)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
|
||||
if data and len(data) > 0:
|
||||
user_data = data[0]
|
||||
ranking = TraderRanking(
|
||||
rank=user_data.get("rank"),
|
||||
pnl=float(user_data.get("pnl", 0) or 0),
|
||||
volume=float(user_data.get("vol", 0) or 0),
|
||||
user_name=user_data.get("userName"),
|
||||
profile_image=user_data.get("profileImage"),
|
||||
verified=bool(user_data.get("verifiedBadge")),
|
||||
time_period="ALL",
|
||||
)
|
||||
# Cache the result
|
||||
self._trader_ranking_cache[wallet_address] = ranking
|
||||
logger.debug(f"Fetched ranking for {wallet_address}: #{ranking.rank}")
|
||||
return ranking
|
||||
|
||||
# User not on leaderboard
|
||||
return None
|
||||
|
||||
except httpx.HTTPError as e:
|
||||
logger.debug(f"HTTP error fetching ranking for {wallet_address}: {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.debug(f"Error fetching ranking for {wallet_address}: {e}")
|
||||
return None
|
||||
|
||||
async def fetch_trader_history(self, wallet_address: str) -> Optional[TraderHistory]:
|
||||
"""
|
||||
Fetch trader's recent trading history.
|
||||
|
||||
Args:
|
||||
wallet_address: The trader's wallet address
|
||||
|
||||
Returns:
|
||||
TraderHistory or None if not found/error
|
||||
"""
|
||||
if not wallet_address:
|
||||
return None
|
||||
|
||||
try:
|
||||
# Fetch recent trades for this user
|
||||
params = {
|
||||
"user": wallet_address,
|
||||
"limit": 100, # Get last 100 trades
|
||||
}
|
||||
|
||||
response = await self._client.get(self.trades_endpoint, params=params)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
|
||||
if not data:
|
||||
return None
|
||||
|
||||
# Calculate statistics
|
||||
total_trades = len(data)
|
||||
total_volume = 0.0
|
||||
large_trades_count = 0
|
||||
recent_markets = set()
|
||||
recent_trades = []
|
||||
|
||||
for trade in data:
|
||||
usdc_size = float(trade.get("usdcSize", 0) or 0)
|
||||
if usdc_size == 0:
|
||||
size = float(trade.get("size", 0) or 0)
|
||||
price = float(trade.get("price", 0) or 0)
|
||||
usdc_size = size * price
|
||||
|
||||
total_volume += usdc_size
|
||||
|
||||
if usdc_size >= 5000:
|
||||
large_trades_count += 1
|
||||
recent_trades.append({
|
||||
"side": trade.get("side", ""),
|
||||
"usdc_size": usdc_size,
|
||||
"price": float(trade.get("price", 0) or 0),
|
||||
"title": trade.get("title", trade.get("marketTitle", "")),
|
||||
"timestamp": trade.get("timestamp", 0),
|
||||
})
|
||||
|
||||
title = trade.get("title", trade.get("marketTitle", ""))
|
||||
if title:
|
||||
recent_markets.add(title[:50])
|
||||
|
||||
avg_trade_size = total_volume / total_trades if total_trades > 0 else 0
|
||||
|
||||
# Sort recent trades by size (largest first)
|
||||
recent_trades.sort(key=lambda x: x["usdc_size"], reverse=True)
|
||||
|
||||
history = TraderHistory(
|
||||
total_trades=total_trades,
|
||||
total_volume=total_volume,
|
||||
avg_trade_size=avg_trade_size,
|
||||
large_trades_count=large_trades_count,
|
||||
recent_markets=list(recent_markets)[:10],
|
||||
recent_trades=recent_trades[:10],
|
||||
)
|
||||
|
||||
logger.debug(f"Fetched history for {wallet_address}: {total_trades} trades, ${total_volume:,.2f} volume")
|
||||
return history
|
||||
|
||||
except httpx.HTTPError as e:
|
||||
logger.debug(f"HTTP error fetching history for {wallet_address}: {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.debug(f"Error fetching history for {wallet_address}: {e}")
|
||||
return None
|
||||
|
||||
def _is_whale_trade(self, activity: TradeActivity) -> bool:
|
||||
"""
|
||||
Check if a trade qualifies as a whale trade.
|
||||
|
||||
Args:
|
||||
activity: The trade activity to check
|
||||
|
||||
Returns:
|
||||
True if this is a whale trade
|
||||
"""
|
||||
return (
|
||||
activity.usdc_size >= self.settings.min_trade_size_usd
|
||||
and self.settings.min_price <= activity.price <= self.settings.max_price
|
||||
)
|
||||
|
||||
async def _check_market(self, condition_id: str, market: Market) -> List[WhaleTrade]:
|
||||
"""
|
||||
Check a single market for whale trades.
|
||||
|
||||
Args:
|
||||
condition_id: Market condition ID
|
||||
market: Market object
|
||||
|
||||
Returns:
|
||||
List of detected whale trades
|
||||
"""
|
||||
whale_trades = []
|
||||
|
||||
activities = await self.fetch_market_trades(condition_id)
|
||||
|
||||
for activity in activities:
|
||||
# Skip if already processed
|
||||
if activity.transaction_hash in self._processed_txns:
|
||||
continue
|
||||
|
||||
# Mark as processed
|
||||
self._processed_txns.add(activity.transaction_hash)
|
||||
|
||||
# Skip during initial scan (only record historical transactions)
|
||||
if not self._initial_scan_complete:
|
||||
continue
|
||||
|
||||
# Check if it's a whale trade
|
||||
if self._is_whale_trade(activity):
|
||||
# Fetch trader ranking and history concurrently
|
||||
trader_ranking, trader_history = await asyncio.gather(
|
||||
self.fetch_trader_ranking(activity.proxy_wallet),
|
||||
self.fetch_trader_history(activity.proxy_wallet),
|
||||
)
|
||||
|
||||
whale_trade = WhaleTrade(
|
||||
id=f"{condition_id}_{activity.transaction_hash}",
|
||||
trade=activity,
|
||||
market_id=market.id,
|
||||
market_question=market.question,
|
||||
market_description=market.description,
|
||||
market_outcomes=market.outcomes,
|
||||
market_outcome_prices=market.outcome_prices,
|
||||
trader_ranking=trader_ranking,
|
||||
trader_history=trader_history,
|
||||
)
|
||||
|
||||
whale_trades.append(whale_trade)
|
||||
|
||||
# Log with ranking info
|
||||
rank_str = f"(排名 #{trader_ranking.rank})" if trader_ranking and trader_ranking.rank else "(未上榜)"
|
||||
logger.info(
|
||||
f"🐋 Whale trade detected! ${activity.usdc_size:,.2f} "
|
||||
f"{activity.side} @ {activity.price:.4f} {rank_str} on '{market.question[:50]}...'"
|
||||
)
|
||||
|
||||
return whale_trades
|
||||
|
||||
async def check_all_markets(self) -> List[WhaleTrade]:
|
||||
"""
|
||||
Check all monitored markets for whale trades.
|
||||
|
||||
Returns:
|
||||
List of all detected whale trades
|
||||
"""
|
||||
all_whale_trades = []
|
||||
|
||||
# Check markets concurrently in batches
|
||||
batch_size = 10
|
||||
items = list(self._monitored_markets.items())
|
||||
|
||||
for i in range(0, len(items), batch_size):
|
||||
batch = items[i : i + batch_size]
|
||||
tasks = [
|
||||
self._check_market(condition_id, market)
|
||||
for condition_id, market in batch
|
||||
]
|
||||
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
for result in results:
|
||||
if isinstance(result, Exception):
|
||||
logger.error(f"Error checking market: {result}")
|
||||
elif result:
|
||||
all_whale_trades.extend(result)
|
||||
|
||||
return all_whale_trades
|
||||
|
||||
async def run(self):
|
||||
"""
|
||||
Start the monitoring loop.
|
||||
|
||||
Continuously monitors markets at the configured interval.
|
||||
First scan records existing transactions without triggering alerts.
|
||||
"""
|
||||
self._running = True
|
||||
logger.info(
|
||||
f"Starting trade monitor (interval: {self.settings.fetch_interval_seconds}s)"
|
||||
)
|
||||
|
||||
# Initial scan - record existing transactions without alerting
|
||||
logger.info("Performing initial scan to record existing transactions...")
|
||||
await self.check_all_markets()
|
||||
self._initial_scan_complete = True
|
||||
self._save_processed_txns() # Save after initial scan
|
||||
logger.info(f"Initial scan complete. Recorded {len(self._processed_txns)} existing transactions. Now monitoring for NEW trades only.")
|
||||
|
||||
save_counter = 0
|
||||
while self._running:
|
||||
try:
|
||||
whale_trades = await self.check_all_markets()
|
||||
|
||||
# Call callback for each whale trade
|
||||
if self._on_whale_detected:
|
||||
for whale_trade in whale_trades:
|
||||
try:
|
||||
await self._on_whale_detected(whale_trade)
|
||||
except Exception as e:
|
||||
logger.error(f"Error in whale callback: {e}")
|
||||
|
||||
# Save processed transactions periodically (every 12 cycles = ~1 minute)
|
||||
save_counter += 1
|
||||
if save_counter >= 12:
|
||||
self._save_processed_txns()
|
||||
save_counter = 0
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Error in monitoring loop: {e}")
|
||||
|
||||
# Wait for next interval
|
||||
await asyncio.sleep(self.settings.fetch_interval_seconds)
|
||||
|
||||
def stop(self):
|
||||
"""Stop the monitoring loop."""
|
||||
self._running = False
|
||||
logger.info("Trade monitor stopping...")
|
||||
|
||||
def clear_processed_transactions(self):
|
||||
"""Clear the processed transactions cache."""
|
||||
count = len(self._processed_txns)
|
||||
self._processed_txns.clear()
|
||||
logger.info(f"Cleared {count} processed transactions from cache")
|
||||
Reference in New Issue
Block a user