Signed-off-by: TIANHE <TIANHE@GMAIL.COM>
This commit is contained in:
TIANHE
2026-03-01 03:42:10 +08:00
parent 7c067fe61c
commit a6ea4d967c
77 changed files with 3368 additions and 193 deletions
+10
View File
@@ -33,6 +33,15 @@ def get_pending_order_worker():
return _pending_order_worker
def start_polymarket_worker():
"""启动Polymarket后台任务"""
try:
from app.services.polymarket_worker import get_polymarket_worker
get_polymarket_worker().start()
except Exception as e:
logger.error(f"Failed to start Polymarket worker: {e}")
def start_portfolio_monitor():
"""Start the portfolio monitor service if enabled.
@@ -257,6 +266,7 @@ def create_app(config_name='default'):
start_pending_order_worker()
start_portfolio_monitor()
start_usdt_order_worker()
start_polymarket_worker()
restore_running_strategies()
return app
@@ -0,0 +1,969 @@
"""
Polymarket预测市场数据源
从Polymarket获取预测市场数据
"""
import time
import requests
import json
from typing import Dict, List, Any, Optional
from datetime import datetime, timedelta
from app.utils.logger import get_logger
from app.utils.db import get_db_connection
logger = get_logger(__name__)
class PolymarketDataSource:
"""Polymarket预测市场数据源"""
def __init__(self):
# Polymarket官方API端点(根据官方文档)
# Gamma API: 市场、事件、标签、搜索等(完全公开,无需认证)
self.gamma_api = "https://gamma-api.polymarket.com"
# Data API: 用户持仓、交易、活动等(完全公开,无需认证)
self.data_api = "https://data-api.polymarket.com"
# CLOB API: 订单簿、价格、交易操作(公开端点无需认证)
self.clob_api = "https://clob.polymarket.com"
self.cache_ttl = 300 # 5分钟缓存
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Accept': 'application/json'
})
def get_trending_markets(self, category: str = None, limit: int = 50) -> List[Dict]:
"""
获取热门预测市场
Args:
category: 类别筛选 (crypto, politics, economics, sports, all)
limit: 返回数量限制
Returns:
预测市场列表
"""
try:
# 先从数据库缓存读取
cached = self._get_cached_markets(category, limit)
if cached:
return cached
# 从真实API获取 - 获取多个分类的数据以确保多样性
all_markets = []
if category and category != "all":
# 如果指定了类别,只获取该类别的数据
markets = self._fetch_markets_from_api(category, limit * 2)
all_markets.extend(markets)
else:
# 如果没有指定类别或指定了"all",获取多个分类的数据
categories_to_fetch = ["crypto", "politics", "economics", "sports"]
for cat in categories_to_fetch:
markets = self._fetch_markets_from_api(cat, limit // len(categories_to_fetch) + 10)
all_markets.extend(markets)
# 去重(按market_id
seen = set()
unique_markets = []
for market in all_markets:
market_id = market.get("market_id")
if market_id and market_id not in seen:
seen.add(market_id)
unique_markets.append(market)
# 按交易量排序
unique_markets.sort(key=lambda x: x.get('volume_24h', 0), reverse=True)
# 保存到数据库缓存
if unique_markets:
self._save_markets_to_db(unique_markets)
return unique_markets[:limit]
# 如果API失败,返回空列表(不再使用示例数据)
logger.warning("Polymarket API unavailable, returning empty list")
return []
except Exception as e:
logger.error(f"Failed to get trending markets: {e}", exc_info=True)
return []
def get_market_details(self, market_id: str) -> Optional[Dict]:
"""获取单个市场详情"""
try:
# 确保market_id是字符串
market_id = str(market_id).strip()
if not market_id:
logger.warning("Empty market_id provided")
return None
# 先从数据库读取
try:
with get_db_connection() as db:
cur = db.cursor()
cur.execute("""
SELECT market_id, question, category, current_probability,
volume_24h, liquidity, end_date_iso, status, outcome_tokens
FROM qd_polymarket_markets
WHERE market_id = %s
""", (market_id,))
row = cur.fetchone()
cur.close()
if row:
# RealDictCursor返回字典,使用键访问
db_market_id = str(row.get('market_id') or market_id)
# 解析outcome_tokens(可能是JSON字符串)
outcome_tokens = {}
outcome_tokens_raw = row.get('outcome_tokens')
if outcome_tokens_raw:
try:
if isinstance(outcome_tokens_raw, str):
outcome_tokens = json.loads(outcome_tokens_raw)
else:
outcome_tokens = outcome_tokens_raw if isinstance(outcome_tokens_raw, dict) else {}
except:
outcome_tokens = {}
return {
"market_id": db_market_id,
"question": row.get('question') or '',
"category": row.get('category') or 'other',
"current_probability": float(row.get('current_probability') or 0),
"volume_24h": float(row.get('volume_24h') or 0),
"liquidity": float(row.get('liquidity') or 0),
"end_date_iso": row.get('end_date_iso'),
"status": row.get('status') or 'active',
"outcome_tokens": outcome_tokens,
"polymarket_url": self._build_polymarket_url(row.get('slug'), db_market_id),
"slug": row.get('slug') if row.get('slug') and not str(row.get('slug', '')).isdigit() else None
}
except Exception as db_error:
logger.warning(f"Database query failed for market {market_id}: {db_error}")
# 继续尝试从API获取
# 如果数据库没有,从API获取
logger.info(f"Market {market_id} not in database, fetching from API")
market = self._fetch_market_from_api(market_id)
if market:
try:
self._save_markets_to_db([market])
except Exception as save_error:
logger.warning(f"Failed to save market to DB: {save_error}")
return market
logger.warning(f"Market {market_id} not found in API")
return None
except Exception as e:
logger.error(f"Failed to get market details for {market_id}: {e}", exc_info=True)
return None
def get_market_history(self, market_id: str, days: int = 30) -> List[Dict]:
"""获取市场历史价格数据"""
# 这里需要实现历史数据获取逻辑
# 暂时返回空列表
return []
def search_markets(self, keyword: str, limit: int = 20, use_cache: bool = True) -> List[Dict]:
"""
搜索相关预测市场
优先从API获取实时数据,数据库仅作为可选缓存
Args:
keyword: 搜索关键词
limit: 返回结果数量限制
use_cache: 是否使用数据库缓存(AI分析时应设为False以获取最新数据)
"""
try:
logger.info(f"Searching Polymarket markets for keyword: '{keyword}' (limit={limit}, use_cache={use_cache})")
# 如果允许使用缓存,先尝试从数据库搜索
if use_cache:
with get_db_connection() as db:
cur = db.cursor()
cur.execute("""
SELECT market_id, question, category, current_probability,
volume_24h, liquidity, end_date_iso, status, slug
FROM qd_polymarket_markets
WHERE question ILIKE %s AND status = 'active'
ORDER BY volume_24h DESC
LIMIT %s
""", (f"%{keyword}%", limit))
rows = cur.fetchall()
cur.close()
if rows:
logger.info(f"Found {len(rows)} markets in database for keyword '{keyword}'")
return [{
"market_id": str(row.get('market_id') or ''),
"question": row.get('question') or '',
"category": row.get('category') or 'other',
"current_probability": float(row.get('current_probability') or 0),
"volume_24h": float(row.get('volume_24h') or 0),
"liquidity": float(row.get('liquidity') or 0),
"end_date_iso": row.get('end_date_iso'),
"status": row.get('status') or 'active',
"polymarket_url": self._build_polymarket_url(row.get('slug'), row.get('market_id') or ''),
"slug": row.get('slug') if row.get('slug') and not str(row.get('slug', '')).isdigit() else None
} for row in rows]
# 直接从Gamma API获取并过滤(AI分析时使用)
logger.info(f"Fetching from API for keyword '{keyword}' (use_cache={use_cache})...")
# 获取更多数据以便有足够的选择空间
all_markets = self._fetch_from_gamma_api(category=None, limit=limit * 5)
logger.info(f"Fetched {len(all_markets)} markets from API, filtering for keyword '{keyword}'...")
# 按关键词过滤(支持多个关键词匹配)
keyword_lower = keyword.lower()
filtered = []
for market in all_markets:
question = market.get("question", "").lower()
# 检查关键词是否在问题中
if keyword_lower in question:
filtered.append(market)
logger.debug(f"Matched market: {market.get('question', '')[:60]}")
if len(filtered) >= limit:
break
logger.info(f"Filtered {len(filtered)} markets matching keyword '{keyword}' from API")
return filtered
except Exception as e:
logger.error(f"Failed to search markets: {e}", exc_info=True)
return []
def _get_cached_markets(self, category: str = None, limit: int = 50) -> Optional[List[Dict]]:
"""从数据库缓存读取市场数据"""
try:
with get_db_connection() as db:
cur = db.cursor()
# 检查缓存是否新鲜(5分钟内)
cutoff_time = datetime.now() - timedelta(seconds=self.cache_ttl)
query = """
SELECT market_id, question, category, current_probability,
volume_24h, liquidity, end_date_iso, status, outcome_tokens
FROM qd_polymarket_markets
WHERE status = 'active' AND updated_at > %s
"""
params = [cutoff_time]
if category:
query += " AND category = %s"
params.append(category)
query += " ORDER BY volume_24h DESC LIMIT %s"
params.append(limit)
cur.execute(query, params)
rows = cur.fetchall()
cur.close()
if rows:
return [{
"market_id": str(row.get('market_id') or ''),
"question": row.get('question') or '',
"category": row.get('category') or 'other',
"current_probability": float(row.get('current_probability') or 0),
"volume_24h": float(row.get('volume_24h') or 0),
"liquidity": float(row.get('liquidity') or 0),
"end_date_iso": row.get('end_date_iso'),
"status": row.get('status') or 'active',
"outcome_tokens": row.get('outcome_tokens') if row.get('outcome_tokens') else {},
"polymarket_url": f"https://polymarket.com/event/{row.get('market_id') or ''}"
} for row in rows]
return None
except Exception as e:
logger.debug(f"Failed to get cached markets: {e}")
return None
def _fetch_markets_from_api(self, category: str = None, limit: int = 50) -> List[Dict]:
"""
从Polymarket Gamma API获取市场数据
使用官方推荐的 /events 端点
"""
try:
# 使用Gamma API的/events端点(官方推荐方式)
markets = self._fetch_from_gamma_api(category, limit)
if markets:
# 按volume_24h降序排序(因为API不支持order参数,需要本地排序)
markets.sort(key=lambda x: x.get('volume_24h', 0), reverse=True)
return markets[:limit] # 返回排序后的前limit个
logger.warning("Gamma API failed to fetch markets")
return []
except Exception as e:
logger.error(f"Failed to fetch markets from API: {e}", exc_info=True)
return []
def _fetch_from_gamma_api(self, category: str = None, limit: int = 50) -> List[Dict]:
"""
使用Gamma API的/events端点获取市场数据(官方推荐方式)
参考: https://docs.polymarket.com/market-data/fetching-markets
"""
try:
# 使用/events端点获取活跃市场(推荐方式)
# 根据官方文档:https://docs.polymarket.com/market-data/fetching-markets
# order参数支持的值:volume_24hr, volume, liquidity, competitive, start_date, end_date
# 但某些端点可能不支持,先尝试不带order参数
url = f"{self.gamma_api}/events"
params = {
"active": "true",
"closed": "false",
"limit": min(limit * 2, 100) # 获取更多数据以便排序和筛选
}
# 尝试添加排序参数(如果API支持)
# 根据文档,可能的排序字段:volume_24hr, volume, liquidity等
# 如果API不支持,会在422错误后移除
# 如果指定了类别,需要通过tag_id筛选
# 注意:需要先获取tag_id,这里先用关键词推断
if category:
# 可以尝试通过搜索或标签来筛选
# 暂时先获取所有,然后在解析时过滤
pass
logger.info(f"Fetching from Gamma API: {url} with params: {params}")
response = self.session.get(url, params=params, timeout=15)
logger.info(f"Gamma API response status: {response.status_code}")
if response.status_code == 200:
try:
data = response.json()
logger.debug(f"Gamma API returned data type: {type(data)}, keys: {list(data.keys()) if isinstance(data, dict) else 'list'}")
# Gamma API返回的可能是列表或包含data字段的对象
if isinstance(data, list):
logger.info(f"Gamma API returned list with {len(data)} items")
markets = self._parse_gamma_events(data, category)
logger.info(f"Parsed {len(markets)} markets from Gamma API")
return markets
elif isinstance(data, dict):
# 可能是 {"data": [...]} 格式
if "data" in data:
events_list = data["data"]
logger.info(f"Gamma API returned dict with 'data' field containing {len(events_list) if isinstance(events_list, list) else 'non-list'} items")
markets = self._parse_gamma_events(events_list, category)
logger.info(f"Parsed {len(markets)} markets from Gamma API")
return markets
# 或者直接是事件对象
elif "id" in data or "slug" in data:
logger.info("Gamma API returned single event object")
markets = self._parse_gamma_events([data], category)
logger.info(f"Parsed {len(markets)} markets from Gamma API")
return markets
else:
logger.warning(f"Gamma API returned dict with unexpected keys: {list(data.keys())}")
logger.debug(f"Full response: {str(data)[:500]}")
logger.warning(f"Gamma API returned unexpected format: {type(data)}")
return []
except json.JSONDecodeError as je:
logger.error(f"Gamma API returned invalid JSON: {je}")
logger.error(f"Response text (first 500 chars): {response.text[:500]}")
return []
# 非200状态码
logger.warning(f"Gamma API returned status {response.status_code}")
logger.warning(f"Response headers: {dict(response.headers)}")
logger.warning(f"Response text (first 500 chars): {response.text[:500]}")
return []
except requests.exceptions.Timeout:
logger.error("Gamma API request timeout after 15 seconds")
return []
except requests.exceptions.ConnectionError as ce:
logger.error(f"Gamma API connection error: {ce}")
return []
except Exception as e:
logger.error(f"Gamma API failed: {e}", exc_info=True)
return []
def _parse_gamma_events(self, events_data: List[Dict], category_filter: str = None) -> List[Dict]:
"""
解析Gamma API返回的事件数据
Gamma API的/events端点返回事件对象,每个事件包含关联的市场数据
根据官方文档,事件对象结构:
- event对象包含markets数组
- 每个market包含clobTokenIds、outcomePrices等字段
"""
parsed = []
if not events_data:
logger.warning("_parse_gamma_events received empty events_data")
return parsed
logger.info(f"Parsing {len(events_data)} events from Gamma API")
# 记录第一个事件的键,用于调试
if events_data:
first_event_keys = list(events_data[0].keys())[:10]
logger.info(f"First event keys: {first_event_keys}")
logger.debug(f"First event sample: {str(events_data[0])[:500]}")
for idx, event in enumerate(events_data):
try:
# Gamma API的事件对象结构
# 事件可能有多个市场(markets字段),或者直接包含市场信息
markets = event.get("markets", [])
# 如果事件没有markets字段,可能事件本身就是市场数据
if not markets:
# 检查是否直接是市场对象(有question或title字段)
if "question" in event or "title" in event or "slug" in event:
markets = [event]
else:
if idx < 3: # 只记录前3个的详细信息
logger.debug(f"Event {idx} has no markets and doesn't look like a market. Keys: {list(event.keys())[:10]}")
continue
if idx < 3: # 只记录前3个的详细信息
logger.debug(f"Processing event {idx} with {len(markets)} markets")
for market_idx, market in enumerate(markets):
# 提取市场基本信息
market_id = market.get("id") or market.get("slug") or event.get("id") or event.get("slug", "")
question = market.get("question") or event.get("question") or market.get("title") or event.get("title", "")
if idx < 3 and market_idx < 2: # 记录前几个市场的详细信息
logger.info(f"Event {idx}, Market {market_idx}: id={market_id}, question={question[:50] if question else 'None'}, event_slug={event.get('slug')}, market_slug={market.get('slug')}, keys={list(market.keys())[:10]}")
if not question:
if idx < 3:
logger.warning(f"Event {idx}, Market {market_idx}: No question found, skipping. Market keys: {list(market.keys())[:10]}")
continue
# 推断类别
inferred_category = self._infer_category(question)
# 如果指定了类别筛选,进行过滤
if category_filter and inferred_category != category_filter:
continue
# 获取概率和outcome数据
current_probability = 50.0
outcome_tokens = {}
# 方法1: 从CLOB API获取实时价格(最准确)
try:
condition_id = market.get("conditionId") or event.get("conditionId")
if condition_id:
prices = self._get_market_prices_from_clob(condition_id)
if prices:
yes_price = prices.get("YES", 0)
no_price = prices.get("NO", 0)
if yes_price > 0:
current_probability = yes_price * 100 if yes_price <= 1 else yes_price
outcome_tokens["YES"] = {"price": yes_price if yes_price <= 1 else yes_price / 100, "volume": 0}
if no_price > 0:
outcome_tokens["NO"] = {"price": no_price if no_price <= 1 else no_price / 100, "volume": 0}
except Exception as e:
logger.debug(f"Failed to get prices from CLOB API: {e}")
# 方法2: 处理outcomePrices字段(可能是JSON字符串)
if current_probability == 50.0:
outcome_prices_str = market.get("outcomePrices") or event.get("outcomePrices")
if outcome_prices_str:
try:
if isinstance(outcome_prices_str, str):
outcome_prices = json.loads(outcome_prices_str)
else:
outcome_prices = outcome_prices_str
# outcomePrices通常是["0.65", "0.35"]格式,对应YES和NO
if isinstance(outcome_prices, list) and len(outcome_prices) >= 2:
yes_price = float(outcome_prices[0]) if outcome_prices[0] else 0
no_price = float(outcome_prices[1]) if outcome_prices[1] else 0
current_probability = yes_price * 100 if yes_price <= 1 else yes_price
outcome_tokens["YES"] = {"price": yes_price if yes_price <= 1 else yes_price / 100, "volume": 0}
outcome_tokens["NO"] = {"price": no_price if no_price <= 1 else no_price / 100, "volume": 0}
except Exception as e:
logger.debug(f"Failed to parse outcomePrices: {e}")
# 从market或event中获取outcomes
# outcomes可能是对象数组、字符串数组,或者需要从其他字段解析
outcomes = market.get("outcomes") or market.get("tokens") or event.get("outcomes") or []
# 处理outcomes数组(可能是对象或字符串)
for outcome in outcomes:
try:
# 如果outcome是字符串,跳过或尝试解析
if isinstance(outcome, str):
# 可能是简单的字符串标识,如"YES"或"NO"
outcome_upper = outcome.upper()
if "YES" in outcome_upper:
if "YES" not in outcome_tokens:
outcome_tokens["YES"] = {"price": 0.5, "volume": 0}
elif "NO" in outcome_upper:
if "NO" not in outcome_tokens:
outcome_tokens["NO"] = {"price": 0.5, "volume": 0}
continue
# outcome是对象
if not isinstance(outcome, dict):
continue
title = str(outcome.get("title") or outcome.get("name", "")).upper()
# 获取价格(可能是price、probability或currentPrice
price = float(outcome.get("price") or outcome.get("probability") or outcome.get("currentPrice") or 0)
if "YES" in title or title == "YES" or outcome.get("outcome") == "Yes":
current_probability = price * 100 if price <= 1 else price
outcome_tokens["YES"] = {
"price": price if price <= 1 else price / 100,
"volume": float(outcome.get("volume", outcome.get("volume24hr", 0)) or 0)
}
elif "NO" in title or title == "NO" or outcome.get("outcome") == "No":
outcome_tokens["NO"] = {
"price": price if price <= 1 else price / 100,
"volume": float(outcome.get("volume", outcome.get("volume24hr", 0)) or 0)
}
except Exception as e:
logger.debug(f"Failed to parse outcome: {e}")
continue
# 如果没有找到outcomes,尝试从其他字段获取概率
if current_probability == 50.0:
# 尝试从market的probability字段获取
prob = market.get("probability") or market.get("yesProbability") or event.get("probability")
if prob:
current_probability = float(prob) * 100 if float(prob) <= 1 else float(prob)
# 获取交易量和流动性
volume_24h = float(
market.get("volume_24hr") or
market.get("volume24hr") or
market.get("volume_24h") or
event.get("volume_24hr") or
event.get("volume24hr") or
0
)
liquidity = float(
market.get("liquidity") or
market.get("totalLiquidity") or
event.get("liquidity") or
0
)
# 解析结束日期
end_date_iso = None
end_date = market.get("endDate") or market.get("end_date") or event.get("endDate") or event.get("end_date")
if end_date:
try:
if isinstance(end_date, (int, float)):
end_date_iso = datetime.fromtimestamp(end_date).isoformat() + "Z"
elif isinstance(end_date, str):
# 尝试解析ISO格式字符串
end_date_iso = end_date
except:
pass
# 获取slug用于构建URL
# 根据Polymarket API文档:slug应该直接从API返回的数据中获取
# URL格式: https://polymarket.com/event/{slug}
# slug是字符串标识符,不是数字ID
slug = None
# 优先从event获取slug(因为event包含markets
if event.get("slug"):
slug_str = str(event.get("slug", "")).strip()
# 如果slug不是纯数字,且包含字母或连字符,则是有效slug
if slug_str and not slug_str.isdigit() and ('-' in slug_str or any(c.isalpha() for c in slug_str)):
slug = slug_str
# 如果event没有有效slug,尝试从market获取
if not slug and market.get("slug"):
slug_str = str(market.get("slug", "")).strip()
if slug_str and not slug_str.isdigit() and ('-' in slug_str or any(c.isalpha() for c in slug_str)):
slug = slug_str
# 如果仍然没有有效slug,尝试通过API查询获取
if not slug and market_id:
try:
# 使用markets端点通过ID查询,获取完整的slug信息
detail_market = self._fetch_market_detail_by_id(market_id)
if detail_market and detail_market.get("slug"):
slug_str = str(detail_market.get("slug", "")).strip()
if slug_str and not slug_str.isdigit() and ('-' in slug_str or any(c.isalpha() for c in slug_str)):
slug = slug_str
except Exception as e:
logger.debug(f"Failed to fetch slug for market {market_id}: {e}")
# 构建URL(使用统一的辅助方法)
polymarket_url = self._build_polymarket_url(slug, market_id)
if not slug:
logger.warning(f"Market {market_id} has no valid slug, using markets endpoint as fallback")
market_data = {
"market_id": market_id,
"question": question,
"category": inferred_category,
"current_probability": round(current_probability, 2),
"volume_24h": volume_24h,
"liquidity": liquidity,
"end_date_iso": end_date_iso,
"status": "active" if market.get("active", event.get("active", True)) else "closed",
"outcome_tokens": outcome_tokens,
"polymarket_url": polymarket_url,
"slug": slug if slug else None # 保存slug(如果不是数字)
}
parsed.append(market_data)
if idx < 3 and market_idx < 2: # 记录成功解析的市场
logger.info(f"Successfully parsed market: {question[:50]}, prob={current_probability:.1f}%, volume={volume_24h}")
except Exception as e:
logger.warning(f"Failed to parse event {idx} (id={event.get('id', event.get('slug', 'unknown'))}): {e}", exc_info=True)
continue
logger.info(f"Successfully parsed {len(parsed)} markets from {len(events_data)} events")
return parsed
def _parse_rest_markets(self, markets_data: List[Dict]) -> List[Dict]:
"""解析REST API返回的市场数据"""
parsed = []
for market in markets_data:
try:
# 提取基本信息
market_id = market.get("id") or market.get("slug") or market.get("market_id", "")
question = market.get("question") or market.get("title", "")
# 计算概率
current_probability = 50.0
outcome_tokens = {}
if "outcomes" in market:
for outcome in market["outcomes"]:
title = str(outcome.get("title", "")).upper()
price = float(outcome.get("price", outcome.get("probability", 0)) or 0)
if "YES" in title or title == "YES":
current_probability = price * 100
outcome_tokens["YES"] = {
"price": price,
"volume": float(outcome.get("volume", 0) or 0)
}
elif "NO" in title or title == "NO":
outcome_tokens["NO"] = {
"price": price,
"volume": float(outcome.get("volume", 0) or 0)
}
volume_24h = float(market.get("volume_24h", market.get("volume", 0)) or 0)
liquidity = float(market.get("liquidity", 0) or 0)
# 推断类别
category = self._infer_category(question)
# 解析结束日期
end_date_iso = market.get("end_date") or market.get("endDate")
if isinstance(end_date_iso, (int, float)):
try:
end_date_iso = datetime.fromtimestamp(end_date_iso).isoformat() + "Z"
except:
end_date_iso = None
# 获取slug用于构建URL
slug = None
slug_str = str(market.get('slug', '')).strip() if market.get('slug') else ''
# 检查slug是否有效(不是数字,且包含字母或连字符)
if slug_str and not slug_str.isdigit() and ('-' in slug_str or any(c.isalpha() for c in slug_str)):
slug = slug_str
else:
# 如果slug无效,尝试通过API查询获取
try:
detail_market = self._fetch_market_detail_by_id(market_id)
if detail_market and detail_market.get("slug"):
slug_str = str(detail_market.get("slug", "")).strip()
if slug_str and not slug_str.isdigit() and ('-' in slug_str or any(c.isalpha() for c in slug_str)):
slug = slug_str
except Exception as e:
logger.debug(f"Failed to fetch slug for market {market_id}: {e}")
# 构建URL(使用统一的辅助方法)
polymarket_url = self._build_polymarket_url(slug, market_id)
if not slug:
logger.warning(f"Market {market_id} has no valid slug, using markets endpoint as fallback")
parsed.append({
"market_id": market_id,
"question": question,
"category": category,
"current_probability": round(current_probability, 2),
"volume_24h": volume_24h,
"liquidity": liquidity,
"end_date_iso": end_date_iso,
"status": "active" if market.get("active", True) else "closed",
"outcome_tokens": outcome_tokens,
"polymarket_url": polymarket_url,
"slug": slug if slug else None
})
except Exception as e:
logger.debug(f"Failed to parse market {market.get('id')}: {e}")
continue
return parsed
def _infer_category(self, question: str) -> str:
"""从问题中推断类别"""
question_lower = question.lower()
# 加密货币关键词
crypto_keywords = ['btc', 'bitcoin', 'eth', 'ethereum', 'sol', 'solana', 'crypto', 'token', 'coin', 'defi', 'nft']
if any(kw in question_lower for kw in crypto_keywords):
return "crypto"
# 政治关键词
politics_keywords = ['election', 'president', 'trump', 'biden', 'senate', 'congress', 'vote', 'political', 'democrat', 'republican']
if any(kw in question_lower for kw in politics_keywords):
return "politics"
# 经济关键词
economics_keywords = ['gdp', 'inflation', 'unemployment', 'fed', 'federal reserve', 'interest rate', 'economic', 'economy', 'recession', 'gdp growth', 'cpi', 'ppi']
if any(kw in question_lower for kw in economics_keywords):
return "economics"
# 体育关键词
sports_keywords = ['nfl', 'nba', 'mlb', 'soccer', 'football', 'basketball', 'baseball', 'championship', 'world cup', 'olympics', 'super bowl', 'stanley cup', 'world series']
if any(kw in question_lower for kw in sports_keywords):
return "sports"
# 科技关键词
tech_keywords = ['ai', 'artificial intelligence', 'chatgpt', 'openai', 'tech', 'technology', 'apple', 'google', 'microsoft', 'meta', 'tesla', 'ipo', 'startup']
if any(kw in question_lower for kw in tech_keywords):
return "tech"
# 金融关键词
finance_keywords = ['stock', 's&p', 'dow', 'nasdaq', 'market cap', 'earnings', 'revenue', 'profit', 'bank', 'banking', 'financial', 'trading']
if any(kw in question_lower for kw in finance_keywords):
return "finance"
# 地缘政治关键词
geopolitics_keywords = ['war', 'conflict', 'russia', 'ukraine', 'china', 'taiwan', 'north korea', 'iran', 'israel', 'palestine', 'middle east', 'nato', 'sanctions']
if any(kw in question_lower for kw in geopolitics_keywords):
return "geopolitics"
# 文化关键词
culture_keywords = ['movie', 'film', 'oscar', 'grammy', 'award', 'celebrity', 'music', 'album', 'tv show', 'series', 'netflix', 'disney']
if any(kw in question_lower for kw in culture_keywords):
return "culture"
# 气候关键词
climate_keywords = ['climate', 'global warming', 'temperature', 'carbon', 'emission', 'renewable', 'solar', 'wind energy', 'paris agreement', 'cop']
if any(kw in question_lower for kw in climate_keywords):
return "climate"
# 娱乐关键词
entertainment_keywords = ['game', 'gaming', 'esports', 'tournament', 'streaming', 'youtube', 'twitch', 'podcast', 'comic', 'anime', 'manga']
if any(kw in question_lower for kw in entertainment_keywords):
return "entertainment"
return "other"
def _build_polymarket_url(self, slug: Optional[str], market_id: str) -> str:
"""
根据slug构建Polymarket URL
参考: https://docs.polymarket.com/market-data/fetching-markets
Args:
slug: 从API或数据库获取的slug(可能是None或数字字符串)
market_id: 市场ID(作为备选)
Returns:
Polymarket URL字符串
"""
if slug:
slug_str = str(slug).strip()
# 检查slug是否有效(不是数字,且包含字母或连字符)
if slug_str and not slug_str.isdigit() and ('-' in slug_str or any(c.isalpha() for c in slug_str)):
import re
slug_clean = re.sub(r'[^a-zA-Z0-9\-]', '-', slug_str)
slug_clean = slug_clean.strip('-')
if slug_clean:
return f"https://polymarket.com/event/{slug_clean}"
# 如果没有有效slug,使用markets端点作为备选
return f"https://polymarket.com/markets/{market_id}"
def _fetch_market_detail_by_id(self, market_id: str) -> Optional[Dict]:
"""
通过market ID从API获取市场详情(用于获取slug)
参考: https://docs.polymarket.com/market-data/fetching-markets
"""
try:
# 方法1: 尝试通过events端点查询(推荐,因为events包含markets
url = f"{self.gamma_api}/events"
params = {"active": "true", "closed": "false", "limit": 100}
response = self.session.get(url, params=params, timeout=10)
if response.status_code == 200:
events = response.json()
if isinstance(events, list):
for event in events:
markets = event.get("markets", [])
if not markets and ("question" in event or "slug" in event):
markets = [event]
for market in markets:
m_id = market.get("id") or market.get("slug") or ""
e_id = event.get("id") or event.get("slug") or ""
# 匹配market_id或event_id
if str(m_id) == str(market_id) or str(e_id) == str(market_id):
# 返回event(因为event包含slug
return event
elif isinstance(events, dict):
if "data" in events:
events_list = events["data"]
for event in events_list:
markets = event.get("markets", [])
if not markets and ("question" in event or "slug" in event):
markets = [event]
for market in markets:
m_id = market.get("id") or market.get("slug") or ""
e_id = event.get("id") or event.get("slug") or ""
if str(m_id) == str(market_id) or str(e_id) == str(market_id):
return event
# 方法2: 尝试通过markets端点查询
url = f"{self.gamma_api}/markets"
params = {"id": market_id, "limit": 1}
response = self.session.get(url, params=params, timeout=10)
if response.status_code == 200:
data = response.json()
if isinstance(data, list) and len(data) > 0:
return data[0]
elif isinstance(data, dict) and "id" in data:
return data
return None
except Exception as e:
logger.debug(f"Failed to fetch market detail by ID {market_id}: {e}")
return None
def _fetch_market_from_api(self, market_id: str) -> Optional[Dict]:
"""
从Gamma API获取单个市场数据
支持通过slug或id查询
"""
try:
# 方法1: 通过slug查询(推荐)
# 如果market_id看起来像slug(包含连字符),使用markets端点
if "-" in market_id or "/" not in market_id:
url = f"{self.gamma_api}/markets"
params = {"slug": market_id}
response = self.session.get(url, params=params, timeout=10)
if response.status_code == 200:
data = response.json()
if isinstance(data, list) and len(data) > 0:
markets = self._parse_gamma_events(data)
if markets:
return markets[0]
elif isinstance(data, dict):
markets = self._parse_gamma_events([data])
if markets:
return markets[0]
# 方法2: 通过events端点搜索
url = f"{self.gamma_api}/events"
params = {
"active": "true",
"closed": "false",
"limit": 100
}
response = self.session.get(url, params=params, timeout=10)
if response.status_code == 200:
data = response.json()
events = data if isinstance(data, list) else (data.get("data", []) if isinstance(data, dict) else [])
# 在返回的事件中查找匹配的市场
for event in events:
markets = event.get("markets", [])
if not markets:
markets = [event]
for market in markets:
m_id = market.get("id") or market.get("slug") or event.get("id") or event.get("slug", "")
if m_id == market_id or market.get("slug") == market_id:
parsed = self._parse_gamma_events([event])
if parsed:
return parsed[0]
return None
except Exception as e:
logger.error(f"Failed to fetch market {market_id}: {e}", exc_info=True)
return None
def _save_markets_to_db(self, markets: List[Dict]):
"""保存市场数据到数据库"""
try:
with get_db_connection() as db:
cur = db.cursor()
for market in markets:
# 获取slug,但如果是数字则不要使用(数字不是有效的slug)
slug = market.get('slug') or None
# 如果slug是数字,说明不是有效的slug,设置为None
if slug and str(slug).isdigit():
slug = None
# 清理slug,只保留字母数字和连字符
import re
if slug:
slug = re.sub(r'[^a-zA-Z0-9\-]', '-', str(slug))
slug = slug.strip('-')
# 如果清理后为空或仍然是数字,设置为None
if not slug or slug.isdigit():
slug = None
cur.execute("""
INSERT INTO qd_polymarket_markets
(market_id, question, category, current_probability, volume_24h,
liquidity, end_date_iso, status, outcome_tokens, slug, updated_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())
ON CONFLICT (market_id) DO UPDATE SET
question = EXCLUDED.question,
category = EXCLUDED.category,
current_probability = EXCLUDED.current_probability,
volume_24h = EXCLUDED.volume_24h,
liquidity = EXCLUDED.liquidity,
end_date_iso = EXCLUDED.end_date_iso,
status = EXCLUDED.status,
outcome_tokens = EXCLUDED.outcome_tokens,
slug = EXCLUDED.slug,
updated_at = NOW()
""", (
market.get('market_id'),
market.get('question'),
market.get('category', 'other'),
market.get('current_probability', 50.0),
market.get('volume_24h', 0),
market.get('liquidity', 0),
market.get('end_date_iso'),
market.get('status', 'active'),
json.dumps(market.get('outcome_tokens', {})),
slug
))
db.commit()
cur.close()
except Exception as e:
logger.error(f"Failed to save markets to DB: {type(e).__name__}: {e}", exc_info=True)
def _get_sample_markets(self, category: str = None, limit: int = 50) -> List[Dict]:
"""
获取示例市场数据(已弃用)
现在应该使用真实的API数据
"""
# 不再返回示例数据,返回空列表
logger.warning("Sample data method called, but real API should be used instead")
return []
+3 -1
View File
@@ -26,6 +26,7 @@ def register_routes(app: Flask):
from app.routes.fast_analysis import fast_analysis_bp
from app.routes.billing import billing_bp
from app.routes.quick_trade import quick_trade_bp
from app.routes.polymarket import polymarket_bp
app.register_blueprint(health_bp)
app.register_blueprint(auth_bp, url_prefix='/api/auth') # Auth routes
@@ -46,4 +47,5 @@ def register_routes(app: Flask):
app.register_blueprint(community_bp, url_prefix='/api/community')
app.register_blueprint(fast_analysis_bp, url_prefix='/api/fast-analysis')
app.register_blueprint(billing_bp, url_prefix='/api/billing')
app.register_blueprint(quick_trade_bp, url_prefix='/api/quick-trade')
app.register_blueprint(quick_trade_bp, url_prefix='/api/quick-trade')
app.register_blueprint(polymarket_bp, url_prefix='/api/polymarket')
@@ -1837,11 +1837,61 @@ def _analyze_opportunities_forex(opportunities: list):
})
def _analyze_opportunities_polymarket(opportunities: list):
"""扫描预测市场机会"""
try:
from app.data_sources.polymarket import PolymarketDataSource
from app.services.polymarket_analyzer import PolymarketAnalyzer
polymarket_source = PolymarketDataSource()
analyzer = PolymarketAnalyzer()
# 获取热门市场
markets = polymarket_source.get_trending_markets(limit=20)
for market in markets:
try:
# AI分析
analysis = analyzer.analyze_market(market['market_id'])
if analysis.get('error'):
continue
# 只添加高分机会
if analysis.get('opportunity_score', 0) > 75:
opportunities.append({
"symbol": market['question'][:50], # 简化显示
"name": market['question'],
"price": market['current_probability'],
"change_24h": 0, # 预测市场没有24h涨跌幅概念
"signal": "prediction_opportunity",
"strength": "strong" if analysis.get('opportunity_score', 0) > 85 else "medium",
"reason": f"AI预测概率{analysis.get('ai_predicted_probability', 0):.1f}%,市场概率{market['current_probability']:.1f}%,差异{analysis.get('divergence', 0):.1f}%",
"impact": "bullish" if analysis.get('recommendation') == "YES" else "bearish",
"market": "PredictionMarket",
"market_id": market['market_id'],
"ai_analysis": {
"predicted_probability": analysis.get('ai_predicted_probability', 0),
"recommendation": analysis.get('recommendation', 'HOLD'),
"confidence_score": analysis.get('confidence_score', 0),
"opportunity_score": analysis.get('opportunity_score', 0)
},
"timestamp": int(time.time())
})
except Exception as e:
logger.debug(f"Failed to analyze polymarket {market.get('market_id')}: {e}")
continue
except Exception as e:
logger.error(f"_analyze_opportunities_polymarket failed: {e}")
@global_market_bp.route("/opportunities", methods=["GET"])
@login_required
def trading_opportunities():
"""
Scan for trading opportunities across Crypto, US Stocks, and Forex.
Note: Prediction Markets are excluded as they have their own dedicated page.
Cached for 1 hour. Pass ?force=true to skip cache.
"""
try:
@@ -1878,6 +1928,9 @@ def trading_opportunities():
except Exception as e:
logger.error(f"Failed to analyze forex opportunities: {e}", exc_info=True)
# Note: Prediction Markets are excluded from trading opportunities radar
# as they have their own dedicated page at /polymarket
# Sort by absolute change descending
opportunities.sort(key=lambda x: abs(x.get("change_24h", 0)), reverse=True)
+2 -2
View File
@@ -185,7 +185,7 @@ def save_indicator():
description = (data.get("description") or "").strip()
publish_to_community = 1 if data.get("publishToCommunity") or data.get("publish_to_community") else 0
pricing_type = (data.get("pricingType") or data.get("pricing_type") or "free").strip() or "free"
vip_free = 1 if (data.get("vipFree") or data.get("vip_free")) else 0
vip_free = bool(data.get("vipFree") or data.get("vip_free"))
try:
price = float(data.get("price") or 0)
except Exception:
@@ -265,7 +265,7 @@ def save_indicator():
UPDATE qd_indicator_codes
SET name = ?, code = ?, description = ?,
publish_to_community = ?, pricing_type = ?, price = ?, preview_image = ?,
vip_free = 0,
vip_free = FALSE,
review_status = NULL, review_note = '', reviewed_at = NULL, reviewed_by = NULL,
updatetime = ?, updated_at = NOW()
WHERE id = ? AND user_id = ? AND (is_buy IS NULL OR is_buy = 0)
+442
View File
@@ -0,0 +1,442 @@
"""
Polymarket预测市场API路由
提供预测市场数据和分析接口(只读,不涉及交易)
"""
from flask import Blueprint, jsonify, request, g
from app.utils.auth import login_required
from app.utils.logger import get_logger
from app.data_sources.polymarket import PolymarketDataSource
from app.utils.db import get_db_connection
import json
from datetime import datetime, timedelta
logger = get_logger(__name__)
polymarket_bp = Blueprint('polymarket', __name__)
# 初始化服务
polymarket_source = PolymarketDataSource()
@polymarket_bp.route("/markets", methods=["GET"])
@login_required
def get_markets():
"""
获取预测市场列表
Query params:
category: crypto/politics/economics/sports (optional)
sort_by: volume_24h/ai_score/probability_change (default: volume_24h)
limit: 数量 (default: 20)
"""
try:
category = request.args.get("category")
sort_by = request.args.get("sort_by", "volume_24h")
limit = int(request.args.get("limit", 20))
# 获取市场列表
markets = polymarket_source.get_trending_markets(category, limit * 2)
logger.info(f"Fetched {len(markets)} markets from PolymarketDataSource (category={category}, limit={limit})")
if not markets:
logger.warning(f"No markets returned from PolymarketDataSource. This may indicate API issues or empty cache.")
return jsonify({
"code": 1,
"msg": "success",
"data": [],
"warning": "No markets available. API may be unavailable or cache is empty."
})
# 从数据库读取缓存的AI分析结果(由后台任务批量分析生成)
# 只读取30分钟内的分析结果
try:
with get_db_connection() as db:
cur = db.cursor()
market_ids = [m.get('market_id') for m in markets if m.get('market_id')]
if market_ids:
# 查询30分钟内的分析结果
from datetime import datetime, timedelta
cache_cutoff = datetime.now() - timedelta(minutes=30)
placeholders = ','.join(['%s'] * len(market_ids))
cur.execute(f"""
SELECT market_id, ai_predicted_probability, market_probability, divergence,
recommendation, confidence_score, opportunity_score, reasoning, key_factors
FROM qd_polymarket_ai_analysis
WHERE market_id IN ({placeholders})
AND user_id IS NULL
AND created_at > %s
ORDER BY opportunity_score DESC
""", market_ids + [cache_cutoff])
rows = cur.fetchall()
cur.close()
# 构建分析结果映射
analysis_map = {}
for row in rows:
market_id = row.get('market_id')
if market_id:
key_factors_raw = row.get('key_factors')
key_factors = []
if key_factors_raw:
try:
if isinstance(key_factors_raw, str):
key_factors = json.loads(key_factors_raw)
else:
key_factors = key_factors_raw if isinstance(key_factors_raw, list) else []
except:
key_factors = []
analysis_map[market_id] = {
'predicted_probability': float(row.get('ai_predicted_probability') or 0),
'recommendation': row.get('recommendation') or 'HOLD',
'confidence_score': float(row.get('confidence_score') or 0),
'opportunity_score': float(row.get('opportunity_score') or 0),
'divergence': float(row.get('divergence') or 0),
'reasoning': row.get('reasoning') or '',
'key_factors': key_factors
}
# 为每个市场添加分析结果
for market in markets:
market_id = market.get('market_id')
if market_id and market_id in analysis_map:
market['ai_analysis'] = analysis_map[market_id]
else:
market['ai_analysis'] = None
else:
# 如果没有市场ID,设置分析为None
for market in markets:
market['ai_analysis'] = None
except Exception as e:
logger.warning(f"Failed to load cached analysis: {e}")
# 出错时,所有市场都没有分析结果
for market in markets:
market['ai_analysis'] = None
# 筛选:优先返回有交易机会的市场,但也包含其他活跃市场
# 重点:不是简单复制数据,而是找到交易机会,但也要保证有足够的数据展示
opportunity_markets = []
other_markets = []
for market in markets:
ai_analysis = market.get('ai_analysis')
volume = market.get('volume_24h', 0) or 0
prob = market.get('current_probability', 50.0) or 50.0
if ai_analysis:
opportunity_score = ai_analysis.get('opportunity_score', 0) or 0
divergence = abs(ai_analysis.get('divergence', 0) or 0)
confidence = ai_analysis.get('confidence_score', 0) or 0
# 筛选条件:机会评分>60 或 (差异>15% 且 置信度>70) 或 (差异>10% 且 置信度>60)
if opportunity_score > 60 or (divergence > 15 and confidence > 70) or (divergence > 10 and confidence > 60):
opportunity_markets.append(market)
elif volume > 5000: # 交易量较大的也作为备选
other_markets.append(market)
else:
# 如果没有AI分析,但交易量大且概率不是50%(说明有明确的市场共识),也包含
if volume > 5000 and abs(prob - 50.0) > 5: # 降低阈值,包含更多市场
other_markets.append(market)
# 合并结果:优先显示机会市场,然后补充其他活跃市场
if opportunity_markets:
# 如果有机会市场,优先显示它们,然后补充其他市场直到达到limit
result_markets = opportunity_markets[:limit]
remaining = limit - len(result_markets)
if remaining > 0 and other_markets:
result_markets.extend(other_markets[:remaining])
opportunity_markets = result_markets
elif other_markets:
# 如果没有机会市场,至少返回高交易量的市场
opportunity_markets = other_markets[:limit]
else:
# 如果都没有,返回原始市场列表的前几个
opportunity_markets = markets[:min(limit, len(markets))]
# 排序和筛选
if sort_by == "ai_score":
# 按AI机会评分排序(高概率/高回报比优先)
opportunity_markets.sort(
key=lambda x: (
x.get('ai_analysis', {}).get('opportunity_score', 0) or 0,
abs(x.get('ai_analysis', {}).get('divergence', 0) or 0), # 差异越大越好
x.get('ai_analysis', {}).get('confidence_score', 0) or 0 # 置信度越高越好
),
reverse=True
)
elif sort_by == "high_probability":
# 高概率机会:AI预测概率 > 市场概率 + 10%
opportunity_markets.sort(
key=lambda x: (
x.get('ai_analysis', {}).get('ai_predicted_probability', 0) or 0,
x.get('ai_analysis', {}).get('confidence_score', 0) or 0
),
reverse=True
)
elif sort_by == "high_return":
# 高回报比机会:AI与市场差异大且置信度高
opportunity_markets.sort(
key=lambda x: (
abs(x.get('ai_analysis', {}).get('divergence', 0) or 0) *
(x.get('ai_analysis', {}).get('confidence_score', 0) or 0) / 100,
x.get('ai_analysis', {}).get('opportunity_score', 0) or 0
),
reverse=True
)
elif sort_by == "probability_change":
# 需要历史数据,暂时按volume排序
opportunity_markets.sort(key=lambda x: x.get('volume_24h', 0), reverse=True)
else:
opportunity_markets.sort(key=lambda x: x.get('volume_24h', 0), reverse=True)
return jsonify({
"code": 1,
"msg": "success",
"data": opportunity_markets[:limit],
"total_opportunities": len(opportunity_markets),
"total_markets": len(markets)
})
except Exception as e:
logger.error(f"get_markets failed: {e}", exc_info=True)
return jsonify({
"code": 0,
"msg": str(e),
"data": None
}), 500
@polymarket_bp.route("/markets/<market_id>", methods=["GET"])
@login_required
def get_market_detail(market_id: str):
"""
获取单个市场详情和AI分析
支持通过market ID或slug查询
"""
try:
# 确保market_id是字符串
market_id = str(market_id).strip()
# 获取市场数据
market = polymarket_source.get_market_details(market_id)
if not market:
return jsonify({
"code": 0,
"msg": "Market not found",
"data": None
}), 404
# 从数据库读取缓存的AI分析结果(30分钟内)
analysis = None
try:
with get_db_connection() as db:
cur = db.cursor()
cache_cutoff = datetime.now() - timedelta(minutes=30)
cur.execute("""
SELECT ai_predicted_probability, market_probability, divergence,
recommendation, confidence_score, opportunity_score,
reasoning, key_factors, related_assets, created_at
FROM qd_polymarket_ai_analysis
WHERE market_id = %s AND user_id IS NULL AND created_at > %s
ORDER BY created_at DESC LIMIT 1
""", (market_id, cache_cutoff))
row = cur.fetchone()
cur.close()
if row:
key_factors_raw = row.get('key_factors')
key_factors = []
if key_factors_raw:
try:
if isinstance(key_factors_raw, str):
key_factors = json.loads(key_factors_raw)
else:
key_factors = key_factors_raw if isinstance(key_factors_raw, list) else []
except:
key_factors = []
analysis = {
"ai_predicted_probability": float(row.get('ai_predicted_probability') or 0),
"market_probability": float(row.get('market_probability') or 0),
"divergence": float(row.get('divergence') or 0),
"recommendation": row.get('recommendation') or 'HOLD',
"confidence_score": float(row.get('confidence_score') or 0),
"opportunity_score": float(row.get('opportunity_score') or 0),
"reasoning": row.get('reasoning') or '',
"key_factors": key_factors,
"related_assets": row.get('related_assets') if row.get('related_assets') else []
}
except Exception as e:
logger.warning(f"Failed to load cached analysis for market {market_id}: {e}")
# 资产交易机会(暂时返回空,可以后续实现)
asset_opportunities = []
return jsonify({
"code": 1,
"msg": "success",
"data": {
"market": market,
"ai_analysis": analysis,
"asset_opportunities": asset_opportunities
}
})
except Exception as e:
logger.error(f"get_market_detail failed: {e}", exc_info=True)
return jsonify({
"code": 0,
"msg": str(e),
"data": None
}), 500
@polymarket_bp.route("/markets/<market_id>/opportunities", methods=["GET"])
@login_required
def get_market_opportunities(market_id: str):
"""获取基于该预测市场的资产交易机会(暂时返回空,后续可扩展)"""
try:
# 暂时返回空列表,可以后续实现基于预测市场的资产推荐
opportunities = []
return jsonify({
"code": 1,
"msg": "success",
"data": opportunities
})
except Exception as e:
logger.error(f"get_market_opportunities failed: {e}", exc_info=True)
return jsonify({
"code": 0,
"msg": str(e),
"data": None
}), 500
@polymarket_bp.route("/recommendations", methods=["GET"])
@login_required
def get_recommendations():
"""
获取AI推荐的高价值预测市场
Query params:
limit: 数量 (default: 10)
"""
try:
limit = int(request.args.get("limit", 10))
# 获取所有活跃市场
all_markets = polymarket_source.get_trending_markets(limit=100)
# 从数据库读取缓存的AI分析结果(30分钟内,按机会评分排序)
recommendations = []
try:
with get_db_connection() as db:
cur = db.cursor()
cache_cutoff = datetime.now() - timedelta(minutes=30)
market_ids = [m.get('market_id') for m in all_markets if m.get('market_id')]
if market_ids:
placeholders = ','.join(['%s'] * len(market_ids))
cur.execute(f"""
SELECT a.market_id, a.opportunity_score, a.recommendation,
a.confidence_score, a.reasoning, a.key_factors,
m.question, m.current_probability, m.volume_24h, m.category
FROM qd_polymarket_ai_analysis a
JOIN qd_polymarket_markets m ON a.market_id = m.market_id
WHERE a.market_id IN ({placeholders})
AND a.user_id IS NULL
AND a.created_at > %s
AND a.opportunity_score > 70
ORDER BY a.opportunity_score DESC
LIMIT %s
""", market_ids + [cache_cutoff, limit])
rows = cur.fetchall()
cur.close()
for row in rows:
key_factors_raw = row.get('key_factors')
key_factors = []
if key_factors_raw:
try:
if isinstance(key_factors_raw, str):
key_factors = json.loads(key_factors_raw)
else:
key_factors = key_factors_raw if isinstance(key_factors_raw, list) else []
except:
key_factors = []
recommendations.append({
"market_id": row.get('market_id'),
"question": row.get('question'),
"current_probability": float(row.get('current_probability') or 0),
"volume_24h": float(row.get('volume_24h') or 0),
"category": row.get('category'),
"ai_analysis": {
"opportunity_score": float(row.get('opportunity_score') or 0),
"recommendation": row.get('recommendation') or 'HOLD',
"confidence_score": float(row.get('confidence_score') or 0),
"reasoning": row.get('reasoning') or '',
"key_factors": key_factors
}
})
except Exception as e:
logger.warning(f"Failed to load recommendations: {e}")
# 如果没有缓存结果,返回空列表(等待后台任务分析)
return jsonify({
"code": 1,
"msg": "success",
"data": recommendations[:limit]
})
except Exception as e:
logger.error(f"get_recommendations failed: {e}", exc_info=True)
return jsonify({
"code": 0,
"msg": str(e),
"data": None
}), 500
@polymarket_bp.route("/search", methods=["GET"])
@login_required
def search_markets():
"""搜索预测市场"""
try:
keyword = request.args.get("q", "").strip()
if not keyword:
return jsonify({
"code": 0,
"msg": "Keyword required",
"data": None
}), 400
limit = int(request.args.get("limit", 20))
markets = polymarket_source.search_markets(keyword, limit)
return jsonify({
"code": 1,
"msg": "success",
"data": markets
})
except Exception as e:
logger.error(f"search_markets failed: {e}", exc_info=True)
return jsonify({
"code": 0,
"msg": str(e),
"data": None
}), 500
@@ -43,10 +43,12 @@ class AnalysisMemory:
self._ensure_table()
def _ensure_table(self):
"""Create memory table if not exists."""
"""Create memory table if not exists, and add missing columns if needed."""
try:
with get_db_connection() as db:
cur = db.cursor()
# 创建表(如果不存在)
cur.execute("""
CREATE TABLE IF NOT EXISTS qd_analysis_memory (
id SERIAL PRIMARY KEY,
@@ -64,6 +66,7 @@ class AnalysisMemory:
risks JSONB,
scores JSONB,
indicators_snapshot JSONB,
raw_result JSONB,
created_at TIMESTAMP DEFAULT NOW(),
validated_at TIMESTAMP,
actual_outcome VARCHAR(20),
@@ -72,20 +75,50 @@ class AnalysisMemory:
user_feedback VARCHAR(20),
feedback_at TIMESTAMP
);
""")
# 检查并添加缺失的列(用于已存在的表)
cur.execute("""
DO $$
BEGIN
-- 添加 user_id 列(如果不存在)
IF NOT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'qd_analysis_memory' AND column_name = 'user_id'
) THEN
ALTER TABLE qd_analysis_memory ADD COLUMN user_id INT;
END IF;
-- 添加 raw_result 列(如果不存在)
IF NOT EXISTS (
SELECT 1 FROM information_schema.columns
WHERE table_name = 'qd_analysis_memory' AND column_name = 'raw_result'
) THEN
ALTER TABLE qd_analysis_memory ADD COLUMN raw_result JSONB;
END IF;
END $$;
""")
# 创建索引
cur.execute("""
CREATE INDEX IF NOT EXISTS idx_analysis_memory_symbol
ON qd_analysis_memory(market, symbol);
CREATE INDEX IF NOT EXISTS idx_analysis_memory_created
ON qd_analysis_memory(created_at DESC);
CREATE INDEX IF NOT EXISTS idx_analysis_memory_validated
ON qd_analysis_memory(validated_at) WHERE validated_at IS NOT NULL;
CREATE INDEX IF NOT EXISTS idx_analysis_memory_user
ON qd_analysis_memory(user_id);
""")
db.commit()
cur.close()
logger.debug("Analysis memory table ensured successfully")
except Exception as e:
logger.warning(f"Memory table creation skipped: {e}")
logger.warning(f"Memory table creation/update skipped: {e}")
def store(self, analysis_result: Dict[str, Any], user_id: int = None) -> Optional[int]:
"""
+231 -55
View File
@@ -26,10 +26,10 @@ class BacktestService:
}
# Multi-timeframe backtest threshold configuration
# 1m backtest: max 1 month (~43,200 candles)
# 1m backtest: max 15 days (~21,600 candles) - reduced for performance
# 5m backtest: max 1 year (~105,120 candles)
MTF_CONFIG = {
'max_1m_days': 30, # Max days for 1-minute backtest
'max_1m_days': 15, # Max days for 1-minute backtest (reduced from 30 for performance)
'max_5m_days': 365, # Max days for 5-minute backtest
'default_exec_tf': '1m', # Default execution timeframe
'fallback_exec_tf': '5m', # Fallback execution timeframe
@@ -79,7 +79,7 @@ class BacktestService:
}
if days_diff <= self.MTF_CONFIG['max_1m_days']:
# Within 1 month: use 1-minute precision
# Within 15 days: use 1-minute precision
estimated_candles = days_diff * 24 * 60
return '1m', {
'enabled': True,
@@ -90,7 +90,7 @@ class BacktestService:
'message': f'Using 1-minute precision backtest (~{estimated_candles:,} candles)'
}
elif days_diff <= self.MTF_CONFIG['max_5m_days']:
# 1 month to 1 year: use 5-minute precision
# 15 days to 1 year: use 5-minute precision
estimated_candles = days_diff * 24 * 12
return '5m', {
'enabled': True,
@@ -98,7 +98,7 @@ class BacktestService:
'days': days_diff,
'estimated_candles': estimated_candles,
'precision': 'medium',
'message': f'Range exceeds 30 days, using 5-minute precision (~{estimated_candles:,} candles)'
'message': f'Range exceeds {self.MTF_CONFIG["max_1m_days"]} days, using 5-minute precision (~{estimated_candles:,} candles)'
}
else:
# Over 1 year: high-precision backtest not supported
@@ -192,9 +192,12 @@ class BacktestService:
'trade_direction': trade_direction
}
signals = self._execute_indicator(indicator_code, df_signal, backtest_params)
logger.info(f"Signals generated: {list(signals.keys()) if isinstance(signals, dict) else type(signals)}")
# 3. Fetch execution timeframe candles (for precise trade simulation)
logger.info(f"Fetching execution timeframe data: {exec_tf} for {market}:{symbol}")
df_exec = self._fetch_kline_data(market, symbol, exec_tf, start_date, end_date)
logger.info(f"Execution timeframe data fetched: {len(df_exec)} candles")
if df_exec.empty:
logger.warning(f"Cannot fetch {exec_tf} candles, falling back to standard backtest")
result = self.run(
@@ -221,7 +224,9 @@ class BacktestService:
logger.info(f"Data fetched: signal_candles={len(df_signal)}, exec_candles={len(df_exec)}")
# 4. Use execution timeframe for precise trade simulation
equity_curve, trades, total_commission = self._simulate_trading_mtf(
try:
logger.info("Starting MTF trading simulation...")
equity_curve, trades, total_commission = self._simulate_trading_mtf(
df_signal=df_signal,
df_exec=df_exec,
signals=signals,
@@ -231,19 +236,38 @@ class BacktestService:
leverage=leverage,
trade_direction=trade_direction,
strategy_config=strategy_config,
signal_timeframe=timeframe,
exec_timeframe=exec_tf
)
signal_timeframe=timeframe,
exec_timeframe=exec_tf
)
logger.info(f"MTF simulation completed: {len(trades)} trades executed")
except Exception as e:
logger.error(f"MTF simulation failed: {str(e)}")
logger.error(traceback.format_exc())
raise
# 5. Calculate metrics
metrics = self._calculate_metrics(equity_curve, trades, initial_capital, timeframe, start_date, end_date, total_commission)
try:
logger.info(f"Calculating metrics: equity_curve_len={len(equity_curve)}, trades_len={len(trades)}, initial_capital={initial_capital}")
metrics = self._calculate_metrics(equity_curve, trades, initial_capital, timeframe, start_date, end_date, total_commission)
logger.info(f"Metrics calculated successfully: {list(metrics.keys())}")
except Exception as e:
logger.error(f"Failed to calculate metrics: {str(e)}")
logger.error(traceback.format_exc())
raise
# 6. Format result
result = self._format_result(metrics, equity_curve, trades)
result['precision_info'] = precision_info
result['execution_timeframe'] = exec_tf
result['signal_candles'] = len(df_signal)
result['execution_candles'] = len(df_exec)
try:
logger.info("Formatting backtest result...")
result = self._format_result(metrics, equity_curve, trades)
result['precision_info'] = precision_info
result['execution_timeframe'] = exec_tf
result['signal_candles'] = len(df_signal)
result['execution_candles'] = len(df_exec)
logger.info("Backtest result formatted successfully")
except Exception as e:
logger.error(f"Failed to format result: {str(e)}")
logger.error(traceback.format_exc())
raise
return result
@@ -267,6 +291,11 @@ class BacktestService:
Simulates trades candle by candle on execution timeframe,
using inferred candle price path to determine trigger order.
"""
try:
logger.info(f"Entering _simulate_trading_mtf: df_signal={len(df_signal)}, df_exec={len(df_exec)}, signals_type={type(signals)}")
except Exception as e:
logger.error(f"Error in _simulate_trading_mtf entry logging: {e}")
equity_curve = []
trades = []
total_commission_paid = 0.0
@@ -342,19 +371,38 @@ class BacktestService:
norm_signals = signals
norm_signals['_both_mode'] = False # Explicit 4-signal mode, not both mode
elif all(k in signals for k in ['buy', 'sell']):
buy = signals['buy'].fillna(False).astype(bool)
sell = signals['sell'].fillna(False).astype(bool)
# Ensure signals have the same index as df_signal
buy_series = signals['buy']
sell_series = signals['sell']
# Reindex to match df_signal.index (fill missing with False)
if not buy_series.index.equals(df_signal.index):
logger.warning(f"Buy signal index mismatch! Signal index: {buy_series.index[:5].tolist()}, df_signal index: {df_signal.index[:5].tolist()}")
buy_series = buy_series.reindex(df_signal.index, fill_value=False)
if not sell_series.index.equals(df_signal.index):
logger.warning(f"Sell signal index mismatch! Signal index: {sell_series.index[:5].tolist()}, df_signal index: {df_signal.index[:5].tolist()}")
sell_series = sell_series.reindex(df_signal.index, fill_value=False)
buy = buy_series.fillna(False).astype(bool)
sell = sell_series.fillna(False).astype(bool)
# Debug: log signal statistics
buy_count = buy.sum()
sell_count = sell.sum()
logger.info(f"Signal statistics: buy={buy_count}, sell={sell_count}, total_candles={len(df_signal)}")
td = str(trade_direction or 'both').lower()
logger.info(f"Trade direction: {td} (original: {trade_direction})")
if td == 'long':
norm_signals = {
'open_long': buy, 'close_long': sell,
'open_short': pd.Series([False] * len(df_signal), index=df_signal.index),
'close_short': pd.Series([False] * len(df_signal), index=df_signal.index),
'open_short': pd.Series([False] * len(df_signal), index=df_signal.index, dtype=bool),
'close_short': pd.Series([False] * len(df_signal), index=df_signal.index, dtype=bool),
}
elif td == 'short':
norm_signals = {
'open_long': pd.Series([False] * len(df_signal), index=df_signal.index),
'close_long': pd.Series([False] * len(df_signal), index=df_signal.index),
'open_long': pd.Series([False] * len(df_signal), index=df_signal.index, dtype=bool),
'close_long': pd.Series([False] * len(df_signal), index=df_signal.index, dtype=bool),
'open_short': sell, 'close_short': buy,
}
else:
@@ -363,13 +411,15 @@ class BacktestService:
# We use special signal types 'enter_long' and 'enter_short' to indicate
# that the signal should auto-close opposing position before opening
norm_signals = {
'open_long': buy, 'close_long': pd.Series([False] * len(df_signal), index=df_signal.index),
'open_short': sell, 'close_short': pd.Series([False] * len(df_signal), index=df_signal.index),
'open_long': buy, 'close_long': pd.Series([False] * len(df_signal), index=df_signal.index, dtype=bool),
'open_short': sell, 'close_short': pd.Series([False] * len(df_signal), index=df_signal.index, dtype=bool),
'_both_mode': True # Flag to indicate both mode for special handling
}
else:
raise ValueError("Invalid signal format")
logger.info("Signal normalization completed, starting signal queue building...")
# Map signals to execution timeframe
# Strategy timeframe seconds (e.g. 1H=3600, 1D=86400)
signal_tf_seconds = self.TIMEFRAME_SECONDS.get(signal_timeframe, 3600)
@@ -379,24 +429,38 @@ class BacktestService:
# Preprocessing: create signal queue sorted by effective time
# Each signal executes at the open of the next execution candle after its candle closes
logger.info("Initializing signal queue...")
signal_queue = [] # [(effective_time, signal_type, signal_bar_time), ...]
# Debug: check signal values
debug_signal_counts = {'open_long': 0, 'close_long': 0, 'open_short': 0, 'close_short': 0}
# Verify all norm_signals have matching index
for sig_type in ['open_long', 'close_long', 'open_short', 'close_short']:
if not norm_signals[sig_type].index.equals(df_signal.index):
logger.error(f"Critical: {sig_type} signal index does not match df_signal.index!")
logger.error(f" Signal index: {norm_signals[sig_type].index[:5].tolist()}")
logger.error(f" df_signal index: {df_signal.index[:5].tolist()}")
# Reindex to fix
norm_signals[sig_type] = norm_signals[sig_type].reindex(df_signal.index, fill_value=False)
logger.warning(f" Fixed by reindexing {sig_type}")
for sig_time in df_signal.index:
# Signal candle end time = start time + period
sig_end = sig_time + timedelta(seconds=signal_tf_seconds)
# Check if this signal candle has signals
# Use .loc[] instead of .get() to be more explicit
# All signals should now have matching index, so we can safely use .loc[]
try:
ol = bool(norm_signals['open_long'].loc[sig_time]) if sig_time in norm_signals['open_long'].index else False
cl = bool(norm_signals['close_long'].loc[sig_time]) if sig_time in norm_signals['close_long'].index else False
os = bool(norm_signals['open_short'].loc[sig_time]) if sig_time in norm_signals['open_short'].index else False
cs = bool(norm_signals['close_short'].loc[sig_time]) if sig_time in norm_signals['close_short'].index else False
ol = bool(norm_signals['open_long'].loc[sig_time])
cl = bool(norm_signals['close_long'].loc[sig_time])
os = bool(norm_signals['open_short'].loc[sig_time])
cs = bool(norm_signals['close_short'].loc[sig_time])
except (KeyError, IndexError) as e:
logger.warning(f"Error accessing signal at {sig_time}: {e}, signal index: {norm_signals['open_long'].index[:5].tolist()}, df_signal index: {df_signal.index[:5].tolist()}")
continue
except Exception as e:
logger.warning(f"Error accessing signal at {sig_time}: {e}")
logger.warning(f"Unexpected error accessing signal at {sig_time}: {e}")
continue
if ol:
@@ -414,6 +478,22 @@ class BacktestService:
logger.info(f"Debug signal counts from queue building: {debug_signal_counts}")
# If no signals found, log detailed diagnostic info
if len(signal_queue) == 0:
logger.warning("No signals found in signal queue! Diagnostic info:")
logger.warning(f" df_signal length: {len(df_signal)}")
logger.warning(f" df_signal index range: {df_signal.index[0]} to {df_signal.index[-1]}")
for sig_type in ['open_long', 'close_long', 'open_short', 'close_short']:
sig_series = norm_signals[sig_type]
true_count = sig_series.sum()
logger.warning(f" {sig_type}: {true_count} True values out of {len(sig_series)}")
if true_count > 0:
true_indices = sig_series[sig_series].index.tolist()[:5]
logger.warning(f" First few True indices: {true_indices}")
# Check if signals might be in wrong format
if 'buy' in signals or 'sell' in signals:
logger.warning(" Original signals had 'buy'/'sell' keys - check if conversion was correct")
# Sort by effective time
signal_queue.sort(key=lambda x: x[0])
signal_queue_idx = 0 # Current signal queue pointer
@@ -422,6 +502,8 @@ class BacktestService:
if signal_queue:
logger.info(f"First signal: {signal_queue[0][1]} @ {signal_queue[0][0]} (from {signal_queue[0][2]})")
logger.info(f"Last signal: {signal_queue[-1][1]} @ {signal_queue[-1][0]} (from {signal_queue[-1][2]})")
else:
logger.error("Signal queue is empty! Backtest will fail. Check indicator code to ensure it generates buy/sell signals.")
# Count signals by type
signal_counts = {}
@@ -429,6 +511,12 @@ class BacktestService:
signal_counts[sig_type] = signal_counts.get(sig_type, 0) + 1
logger.info(f"Signal counts: {signal_counts}")
# Log first few signal details for debugging
if signal_queue:
logger.info(f"First 3 signals details:")
for idx, (sig_time, sig_type, sig_bar_time) in enumerate(signal_queue[:3]):
logger.info(f" Signal {idx+1}: {sig_type} @ effective_time={sig_time}, from_bar={sig_bar_time}")
# Log execution data range
if len(df_exec) > 0:
exec_start = df_exec.index[0]
@@ -443,7 +531,17 @@ class BacktestService:
pending_signal_time = None # Signal effective time
executed_trades_count = 0 # Debug counter
# Progress logging for large datasets
total_exec_candles = len(df_exec)
progress_log_interval = max(1000, total_exec_candles // 10) # Log every 10% or every 1000 candles
logger.info(f"Starting execution loop: {total_exec_candles} candles to process, {len(signal_queue)} signals in queue")
for i, (timestamp, row) in enumerate(df_exec.iterrows()):
# Progress logging
if i > 0 and i % progress_log_interval == 0:
progress_pct = (i / total_exec_candles) * 100
logger.info(f"Execution progress: {i}/{total_exec_candles} ({progress_pct:.1f}%), trades={executed_trades_count}, position={position}")
# 爆仓后直接停止回测,输出结果
if is_liquidated:
break
@@ -500,13 +598,15 @@ class BacktestService:
pending_signal = sig_type
pending_signal_time = sig_effective_time
signal_queue_idx += 1
if executed_trades_count < 5:
logger.info(f"Signal ready: {sig_type} @ {timestamp}, will execute at open price (both_mode={both_mode_active})")
if executed_trades_count < 5 or signal_queue_idx <= 5:
logger.info(f"Signal ready: {sig_type} @ {timestamp} (effective_time={sig_effective_time}, sig_bar_time={sig_bar_time}), "
f"will execute at open price (both_mode={both_mode_active}, position={position})")
break
else:
# Signal doesn't meet execution conditions, skip
if signal_queue_idx < 5:
logger.info(f"Skipping signal #{signal_queue_idx}: {sig_type} (position={position}, can_execute=False)")
if signal_queue_idx < 10:
logger.info(f"Skipping signal #{signal_queue_idx}: {sig_type} @ {sig_effective_time} "
f"(position={position}, can_execute=False, both_mode={both_mode_active})")
signal_queue_idx += 1
continue
else:
@@ -713,6 +813,8 @@ class BacktestService:
# 2. Execute pending signal (at open price)
if pending_signal and path_price == open_:
both_mode_active = norm_signals.get('_both_mode', False)
if executed_trades_count < 10:
logger.info(f"Executing pending signal: {pending_signal} @ {timestamp}, path_price={path_price}, open={open_}, position={position}")
# open_long: In both mode, first close short if any, then open long
if pending_signal == 'open_long' and (position == 0 or (both_mode_active and position < 0)):
@@ -909,9 +1011,25 @@ class BacktestService:
})
# Summary log
logger.info(f"MTF simulation complete: executed_trades={executed_trades_count}, total_trades_recorded={len(trades)}, final_capital={capital:.2f}")
logger.info(f"MTF simulation complete: executed_trades={executed_trades_count}, total_trades_recorded={len(trades)}, final_capital={capital:.2f}, final_position={position}")
if len(trades) == 0:
logger.warning(f"No trades executed! signal_queue_idx={signal_queue_idx}, total_signals={len(signal_queue)}")
if len(signal_queue) == 0:
logger.error(f"No trades executed because signal queue is empty! This usually means:")
logger.error(" 1. Indicator code did not generate any buy/sell signals")
logger.error(" 2. Signal index mismatch between indicator output and df_signal")
logger.error(" 3. All signal values are False")
raise ValueError("No signals generated by indicator code. Please check your indicator code to ensure it sets df['buy'] and/or df['sell'] columns with boolean values.")
else:
logger.error(f"No trades executed despite {len(signal_queue)} signals in queue. signal_queue_idx={signal_queue_idx}")
logger.error(f" Signal queue processed: {signal_queue_idx}/{len(signal_queue)}")
logger.error(f" Final position: {position}, Final capital: {capital:.2f}")
logger.error(" This may indicate:")
logger.error(" 1. Signal timing issues (signal effective time doesn't match execution timeframe)")
logger.error(" 2. Position state conflicts (signals skipped due to position state)")
logger.error(" 3. Capital insufficient for trading")
logger.error(f" First few signals: {signal_queue[:min(5, len(signal_queue))]}")
logger.error(f" Exec data range: {df_exec.index[0]} to {df_exec.index[-1]}")
raise ValueError(f"No trades executed despite {len(signal_queue)} signals. Check signal timing and position state logic.")
return equity_curve, trades, total_commission_paid
@@ -1057,28 +1175,70 @@ class BacktestService:
)
if not kline_data:
logger.warning("No candle data retrieved")
logger.warning(f"No candle data retrieved for {market}:{symbol}, timeframe={timeframe}, limit={limit}, before_time={before_time}")
return pd.DataFrame()
if kline_data:
first_time = datetime.fromtimestamp(kline_data[0]['time'])
last_time = datetime.fromtimestamp(kline_data[-1]['time'])
logger.info(f"Retrieved {len(kline_data)} candles for {market}:{symbol}, timeframe={timeframe}")
# Convert to DataFrame
df = pd.DataFrame(kline_data)
df['time'] = pd.to_datetime(df['time'], unit='s')
df = df.set_index('time')
if len(df) > 0:
pass
# Filter date range
df = df[(df.index >= start_date) & (df.index <= end_date)].copy()
if len(df) > 0:
pass
return df
try:
df = pd.DataFrame(kline_data)
if df.empty:
logger.warning(f"DataFrame is empty after conversion")
return pd.DataFrame()
# Handle time column - could be seconds or milliseconds
if 'time' not in df.columns:
logger.error(f"Missing 'time' column in kline data. Columns: {df.columns.tolist()}")
return pd.DataFrame()
# Try seconds first, if fails try milliseconds
try:
df['time'] = pd.to_datetime(df['time'], unit='s')
except (ValueError, OverflowError):
# If seconds fails, try milliseconds
try:
df['time'] = pd.to_datetime(df['time'], unit='ms')
except (ValueError, OverflowError):
# If both fail, try direct conversion
df['time'] = pd.to_datetime(df['time'])
df = df.set_index('time')
if df.empty:
logger.warning(f"DataFrame is empty after setting time index")
return pd.DataFrame()
# Log data range before filtering
data_start = df.index.min()
data_end = df.index.max()
logger.info(f"Kline data range: {data_start} to {data_end}, requested range: {start_date} to {end_date}")
# Check if requested range is within available data
if data_start > start_date:
logger.warning(f"Requested start date {start_date} is before available data start {data_start}. "
f"Using available start date instead.")
if data_end < end_date:
logger.warning(f"Requested end date {end_date} is after available data end {data_end}. "
f"Using available end date instead. This may affect backtest results.")
# Filter date range (use available data range if requested range is outside)
effective_start = max(start_date, data_start)
effective_end = min(end_date, data_end)
df_filtered = df[(df.index >= effective_start) & (df.index <= effective_end)].copy()
if df_filtered.empty:
logger.error(f"After filtering date range ({effective_start} to {effective_end}), no data remains. "
f"Available data range: {data_start} to {data_end}, requested: {start_date} to {end_date}")
return pd.DataFrame()
logger.info(f"After filtering: {len(df_filtered)} candles remain for backtest (effective range: {effective_start} to {effective_end})")
return df_filtered
except Exception as e:
logger.error(f"Error processing kline data: {str(e)}")
logger.error(traceback.format_exc())
return pd.DataFrame()
def _execute_indicator(self, code: str, df: pd.DataFrame, backtest_params: dict = None):
"""Execute indicator code to get signals.
@@ -1213,9 +1373,25 @@ import pandas as pd
# Position sizing, TP/SL, trailing, etc must be handled by strategy_config / strategy logic.
elif all(col in executed_df.columns for col in ['buy', 'sell']):
# Simple buy/sell signals (recommended for indicator authors)
buy_series = executed_df['buy'].fillna(False).astype(bool)
sell_series = executed_df['sell'].fillna(False).astype(bool)
# Ensure signals have the same index as df
if not buy_series.index.equals(df.index):
logger.warning(f"Buy signal index mismatch in _execute_indicator! Reindexing...")
buy_series = buy_series.reindex(df.index, fill_value=False)
if not sell_series.index.equals(df.index):
logger.warning(f"Sell signal index mismatch in _execute_indicator! Reindexing...")
sell_series = sell_series.reindex(df.index, fill_value=False)
# Debug: log signal statistics
buy_count = buy_series.sum()
sell_count = sell_series.sum()
logger.info(f"Indicator execution: buy signals={buy_count}, sell signals={sell_count}, total_candles={len(df)}")
signals = {
'buy': executed_df['buy'].fillna(False).astype(bool),
'sell': executed_df['sell'].fillna(False).astype(bool)
'buy': buy_series,
'sell': sell_series
}
else:
+198 -69
View File
@@ -46,6 +46,7 @@ class FastAnalysisService:
2. 基本面: 公司信息财务数据
3. 宏观数据: DXYVIXTNX黄金等
4. 情绪数据: 新闻市场情绪
5. 预测市场: 相关预测市场事件新增
"""
return self.data_collector.collect_all(
market=market,
@@ -53,6 +54,7 @@ class FastAnalysisService:
timeframe=timeframe,
include_macro=True,
include_news=True,
include_polymarket=True, # 包含预测市场数据
timeout=30
)
@@ -196,6 +198,19 @@ class FastAnalysisService:
return "\n".join(summaries) if summaries else "No recent news available."
def _format_polymarket_summary(self, polymarket_events: List[Dict], max_items: int = 3) -> str:
"""Format prediction market events into a concise summary for the prompt."""
if not polymarket_events:
return "No related prediction market events found."
summaries = []
for event in polymarket_events[:max_items]:
question = event.get('question', '')
prob = event.get('current_probability', 50.0)
summaries.append(f"- {question[:80]}: Market probability {prob:.1f}%")
return "\n".join(summaries) if summaries else "No related prediction market events found."
# ==================== Memory Layer ====================
def _get_memory_context(self, market: str, symbol: str, current_indicators: Dict) -> str:
@@ -247,6 +262,7 @@ class FastAnalysisService:
fundamental = data.get("fundamental") or {}
company = data.get("company") or {}
news_summary = self._format_news_summary(data.get("news") or [])
polymarket_events = data.get("polymarket") or []
# Language instruction - MUST be enforced strictly
lang_map = {
@@ -349,22 +365,31 @@ You are CONSERVATIVE and OBJECTIVE. Your analysis must be based on DATA, not spe
- Consider geopolitical events and their potential impact
- Evaluate how macro trends affect this specific market/symbol
3. **News & Event Analysis**:
- **CRITICAL**: Pay special attention to GEOPOLITICAL EVENTS (wars, conflicts, military actions, sanctions)
- These events can cause sudden and severe market movements, especially for crypto and global markets
- Identify BREAKING NEWS or major events that could cause sudden moves
- Assess news sentiment and its credibility
- Consider regulatory changes, partnerships, scandals, etc.
- Don't ignore major news just because technical indicators look good
4. **Fundamental Analysis**: Evaluate valuation, growth, competitive position if data available. If data is insufficient, say so.
5. **Risk Assessment**:
- Consider regulatory changes, partnerships, scandals, geopolitical tensions, etc.
- **DO NOT ignore major geopolitical news** (e.g., US-Iran conflict, Russia-Ukraine war) even if technical indicators look good
- Global events like wars can override all technical analysis - treat them as HIGHEST PRIORITY
4. **Prediction Market Analysis**:
- Review related prediction market events and their current probabilities
- Prediction markets reflect collective market wisdom and can indicate future price movements
- If prediction markets show high probability for bullish events (e.g., "BTC reaches $100k"), consider this as a positive signal
- If prediction markets show high probability for bearish events, consider this as a risk factor
- Use prediction market probabilities as a sentiment indicator alongside technical analysis
5. **Fundamental Analysis**: Evaluate valuation, growth, competitive position if data available. If data is insufficient, say so.
6. **Risk Assessment**:
- Explain why the stop loss level is appropriate
- List ALL significant risks (technical, macro, news, fundamental)
- Consider tail risks from unexpected events
6. **Clear Recommendation**: BUY/SELL/HOLD with entry, stop loss (near suggested), take profit (near suggested)
7. **Clear Recommendation**: BUY/SELL/HOLD with entry, stop loss (near suggested), take profit (near suggested)
- **BUY**: For long positions when indicators suggest upside
- **SELL**: For short positions when indicators suggest downside - this is a VALID trading opportunity
- **HOLD**: Only when signals are truly unclear - DO NOT default to HOLD just to be safe
- Your decision should reflect the WEIGHTED importance of ALL factors
- If macro/news factors strongly contradict technical, explain why you prioritize one over the other
7. **Trading Opportunity Recognition**:
8. **Trading Opportunity Recognition**:
- When you see RSI > 60, bearish MACD, downtrend Give SELL signal (short opportunity)
- When you see RSI < 40, bullish MACD, uptrend Give BUY signal (long opportunity)
- Only choose HOLD when signals are genuinely mixed or unclear
@@ -397,14 +422,16 @@ Output ONLY valid JSON (do NOT include word counts or format hints in your actua
- Do NOT make up facts or exaggerate. Base everything on the provided data.
📊 OBJECTIVE SCORING SYSTEM (Reference):
The system will calculate an objective score based on technical indicators, fundamentals, sentiment, and macro factors.
- Score >= +40: Bullish signal BUY recommended
- Score <= -40: Bearish signal SELL recommended
- Score between -40 and +40: Neutral HOLD recommended
The system will calculate an objective score based on technical indicators, fundamentals, sentiment (including geopolitical events), and macro factors.
- Score >= +20: Bullish signal BUY recommended
- Score <= -20: Bearish signal SELL recommended
- Score between -20 and +20: Neutral HOLD recommended (narrow range)
- Score >= +70: Strong bullish Strong BUY signal
- Score <= -70: Strong bearish Strong SELL signal
Your decision should align with this objective score when it's significant (>=40 or <=-40).
When the score is neutral (-40 to +40), you can use your judgment, but still consider giving BUY/SELL if technical indicators are clear."""
- Geopolitical events (wars, conflicts) are heavily weighted in sentiment score and can cause severe negative scores
- Macro factors (VIX, DXY, interest rates) are also heavily weighted
Your decision should align with this objective score when it's significant (>=20 or <=-20).
When the score is neutral (-20 to +20), you can use your judgment, but still consider giving BUY/SELL if technical indicators are clear."""
# Format indicator data for prompt (ensure safe defaults)
rsi_data = indicators.get("rsi") or {}
@@ -439,6 +466,9 @@ When the score is neutral (-40 to +40), you can use your judgment, but still con
📰 MARKET NEWS ({len(data.get('news') or [])} items):
{news_summary}
🎯 PREDICTION MARKETS ({len(polymarket_events)} related events):
{self._format_polymarket_summary(polymarket_events)}
💼 FUNDAMENTALS:
- Company: {company.get('name', data['symbol'])}
- Industry: {company.get('industry', 'N/A')}
@@ -460,10 +490,12 @@ When the score is neutral (-40 to +40), you can use your judgment, but still con
{self._format_earnings_data(fundamental.get('earnings', {}))}
IMPORTANT:
1. Consider the macro environment (especially DXY, VIX, rates, geopolitical events) when making your recommendation.
2. Pay attention to BREAKING NEWS and international events that could cause sudden market moves.
3. For US stocks, analyze financial statements and earnings trends to assess company health.
4. Provide your analysis now. Remember: all prices must be within 10% of ${current_price}."""
1. **CRITICAL**: Check for GEOPOLITICAL EVENTS (wars, conflicts, military actions) in the news section. These events have HIGHEST PRIORITY and can override all technical indicators.
2. Consider the macro environment (especially DXY, VIX, rates, geopolitical events) when making your recommendation.
3. Pay attention to BREAKING NEWS and international events that could cause sudden market moves. Geopolitical tensions (e.g., US-Iran conflict) can cause severe market volatility.
4. For US stocks, analyze financial statements and earnings trends to assess company health.
5. If you see news about wars, conflicts, or major geopolitical events, you MUST mention them in your analysis and adjust your recommendation accordingly.
6. Provide your analysis now. Remember: all prices must be within 10% of ${current_price}."""
return system_prompt, user_prompt
@@ -745,8 +777,8 @@ IMPORTANT:
llm_decision = analysis.get("decision", "HOLD")
if llm_decision != score_based_decision:
score_abs = abs(objective_score['overall_score'])
# 降低阈值,因为现在HOLD区间更小了,±40以上的评分就应该覆盖
if score_abs >= 25: # 如果评分达到±25以上,就覆盖LLM决策(因为阈值是±40
# 降低阈值,因为现在HOLD区间更小了(±20),±15以上的评分就应该覆盖
if score_abs >= 15: # 如果评分达到±15以上,就覆盖LLM决策(因为阈值是±20
logger.warning(f"LLM decision '{llm_decision}' conflicts with score-based decision '{score_based_decision}' (score: {objective_score['overall_score']:.1f}). Overriding to score-based decision.")
analysis["decision"] = score_based_decision
# Adjust confidence based on score strength
@@ -908,24 +940,50 @@ IMPORTANT:
def _has_major_news(self, news_data: List[Dict]) -> bool:
"""
检查是否有重大新闻事件
重大新闻包括监管变化重大合作丑闻重大政策等
重大新闻包括监管变化重大合作丑闻重大政策地缘政治事件
"""
if not news_data:
return False
# 检查新闻标题中的关键词
# 检查新闻标题中的关键词(扩展了地缘政治相关关键词)
major_keywords = [
"regulation", "regulatory", "ban", "approval", "partnership", "merger", "acquisition",
"scandal", "lawsuit", "investigation", "policy", "government", "central bank",
"监管", "禁令", "批准", "合作", "合并", "收购", "丑闻", "诉讼", "调查", "政策", "政府", "央行"
# 监管和政策
"regulation", "regulatory", "ban", "approval", "policy", "government", "central bank",
"监管", "禁令", "批准", "政策", "政府", "央行",
# 商业事件
"partnership", "merger", "acquisition", "scandal", "lawsuit", "investigation",
"合作", "合并", "收购", "丑闻", "诉讼", "调查",
# 地缘政治事件(新增)
"war", "conflict", "military", "attack", "strike", "sanctions", "tension", "crisis",
"geopolitical", "iran", "israel", "russia", "ukraine", "china", "taiwan", "north korea",
"middle east", "gulf", "nato", "united states", "us", "usa", "america",
"战争", "冲突", "军事", "袭击", "打击", "制裁", "紧张", "危机",
"地缘政治", "伊朗", "以色列", "俄罗斯", "乌克兰", "中国", "台湾", "朝鲜",
"中东", "海湾", "北约", "美国"
]
for news in news_data[:5]: # 检查前5条最新新闻
for news in news_data[:10]: # 检查前10条最新新闻(增加检查范围)
title = (news.get("title") or news.get("headline") or "").lower()
summary = (news.get("summary") or "").lower()
sentiment = news.get("sentiment", "neutral")
# 如果有重大关键词且情绪强烈(非中性),认为是重大新闻
if any(keyword in title for keyword in major_keywords) and sentiment != "neutral":
# 检查标题和摘要中是否包含重大关键词
text_to_check = f"{title} {summary}"
# 地缘政治事件通常很严重,即使情绪是中性也要识别
geopolitical_keywords = [
"war", "conflict", "military", "attack", "strike", "geopolitical",
"战争", "冲突", "军事", "袭击", "打击", "地缘政治"
]
# 如果是地缘政治相关,直接认为是重大新闻
if any(keyword in text_to_check for keyword in geopolitical_keywords):
logger.info(f"Detected major geopolitical event in news: {title[:60]}")
return True
# 其他重大关键词且情绪强烈(非中性),认为是重大新闻
if any(keyword in text_to_check for keyword in major_keywords) and sentiment != "neutral":
logger.info(f"Detected major news event: {title[:60]}")
return True
return False
@@ -1061,7 +1119,8 @@ IMPORTANT:
conflicts.append("MACD bearish")
# 均线趋势向下时不应该BUY(除非有重大利好)
if "downtrend" in ma_trend.lower():
# 只有当趋势非常强烈时才认为是冲突(避免过于敏感)
if "strong_downtrend" in ma_trend.lower() or ("downtrend" in ma_trend.lower() and rsi_value > 50):
conflicts.append(f"MA trend: {ma_trend}")
if conflicts:
@@ -1140,12 +1199,13 @@ IMPORTANT:
macro_score = self._calculate_macro_score(macro, data.get("market", ""))
# 5. 综合评分(加权平均)
# 权重:技术40%,基本面25%,情绪20%,宏观15%
# 优化权重:技术35%,基本面20%,情绪25%(包含地缘政治),宏观20%(提高宏观权重)
# 提高情绪和宏观权重,因为地缘政治和宏观经济因素对市场影响更大
overall_score = (
technical_score * 0.40 +
fundamental_score * 0.25 +
sentiment_score * 0.20 +
macro_score * 0.15
technical_score * 0.35 +
fundamental_score * 0.20 +
sentiment_score * 0.25 + # 提高情绪权重,包含地缘政治事件
macro_score * 0.20 # 提高宏观权重
)
return {
@@ -1318,16 +1378,49 @@ IMPORTANT:
return max(-100, min(100, score))
def _calculate_sentiment_score(self, news: List[Dict]) -> float:
"""计算新闻情绪评分 (-100 to +100)"""
"""
计算新闻情绪评分 (-100 to +100)
包含地缘政治事件的特殊处理
"""
if not news:
return 0.0 # 无新闻,中性
positive_count = 0
negative_count = 0
neutral_count = 0
geopolitical_penalty = 0 # 地缘政治事件惩罚分数
geopolitical_count = 0 # 地缘政治事件数量
for item in news[:10]: # 只看前10条
# 地缘政治关键词
geopolitical_keywords = [
"war", "conflict", "military", "attack", "strike", "sanctions",
"geopolitical", "crisis", "tension", "iran", "israel", "russia",
"ukraine", "middle east", "nato", "united states",
"战争", "冲突", "军事", "袭击", "制裁", "地缘政治", "危机"
]
for item in news[:15]: # 检查前15条新闻
title = (item.get("headline") or item.get("title") or "").lower()
summary = (item.get("summary") or "").lower()
text = f"{title} {summary}"
sentiment = item.get("sentiment", "neutral")
is_global_event = item.get("is_global_event", False)
# 检查是否是地缘政治事件
is_geopolitical = is_global_event or any(keyword in text for keyword in geopolitical_keywords)
if is_geopolitical:
geopolitical_count += 1
# 地缘政治事件通常是利空的,给予严重惩罚
if any(kw in text for kw in ["war", "conflict", "attack", "strike", "战争", "冲突", "袭击", "打击"]):
geopolitical_penalty -= 50 # 战争/冲突事件严重利空
elif any(kw in text for kw in ["sanctions", "crisis", "tension", "制裁", "危机", "紧张"]):
geopolitical_penalty -= 30 # 制裁/危机事件利空
else:
geopolitical_penalty -= 20 # 其他地缘政治事件利空
logger.info(f"Detected geopolitical event in sentiment scoring: {title[:60]}, penalty: {geopolitical_penalty}")
# 统计普通新闻情绪
if sentiment == "positive":
positive_count += 1
elif sentiment == "negative":
@@ -1336,66 +1429,97 @@ IMPORTANT:
neutral_count += 1
total = positive_count + negative_count + neutral_count
if total == 0:
return 0.0
# 计算净情绪
net_sentiment = (positive_count - negative_count) / total
# 计算净情绪(普通新闻)
if total > 0:
net_sentiment = (positive_count - negative_count) / total
base_score = net_sentiment * 60 # 基础情绪分数(-60到+60
else:
base_score = 0
# 映射到-100到+100
score = net_sentiment * 100
# 地缘政治事件惩罚(如果有地缘政治事件,直接应用惩罚)
if geopolitical_count > 0:
# 地缘政治事件的影响权重很高,直接叠加惩罚
final_score = base_score + geopolitical_penalty
logger.info(f"Sentiment score: base={base_score:.1f}, geopolitical_penalty={geopolitical_penalty}, final={final_score:.1f}")
else:
final_score = base_score
return max(-100, min(100, score))
return max(-100, min(100, final_score))
def _calculate_macro_score(self, macro: Dict, market: str) -> float:
"""计算宏观环境评分 (-100 to +100)"""
"""
计算宏观环境评分 (-100 to +100)
包含VIXDXY利率等宏观经济指标
"""
if not macro:
return 0.0 # 无宏观数据,中性
score = 0.0
factors = 0
# VIX 评分(恐慌指数)
# VIX 评分(恐慌指数)- 权重提高
vix = macro.get("VIX", {})
vix_value = vix.get("price", 0)
if vix_value > 0:
if vix_value > 30:
vix_score = -30 # 高恐慌利空
if vix_value > 35:
vix_score = -50 # 高恐慌(如战争期间),严重利空
elif vix_value > 30:
vix_score = -40 # 高恐慌,严重利空
elif vix_value > 25:
vix_score = -30 # 较高恐慌,利空
elif vix_value > 20:
vix_score = -15
vix_score = -15 # 中等恐慌,轻微利空
elif vix_value < 12:
vix_score = +20 # 低恐慌,利多
elif vix_value < 15:
vix_score = +15 # 低恐慌,利多
vix_score = +10 # 低恐慌,轻微利多
else:
vix_score = 0
score += vix_score
factors += 1
# DXY 评分(美元指数)
# DXY 评分(美元指数)- 权重提高
dxy = macro.get("DXY", {})
dxy_value = dxy.get("price", 0)
dxy_change = dxy.get("changePercent", 0)
if dxy_value > 0:
# 对于加密货币和商品,强美元通常是利空
if market in ["Crypto", "Forex", "Futures"]:
if dxy_change > 1:
if dxy_change > 2:
dxy_score = -30 # 美元大幅走强,严重利空
elif dxy_change > 1:
dxy_score = -20 # 美元走强,利空
elif dxy_change < -2:
dxy_score = +30 # 美元大幅走弱,利多
elif dxy_change < -1:
dxy_score = +20 # 美元走弱,利多
else:
dxy_score = 0
else:
dxy_score = 0 # 对股票影响较小
# 对股票也有影响,但较小
if dxy_change > 2:
dxy_score = -10
elif dxy_change < -2:
dxy_score = +10
else:
dxy_score = 0
score += dxy_score
factors += 1
# 利率评分(TNX
# 利率评分(TNX- 权重提高
tnx = macro.get("TNX", {})
tnx_change = tnx.get("changePercent", 0)
if tnx_change != 0:
tnx_value = tnx.get("price", 0)
if tnx_change != 0 or tnx_value > 0:
# 利率上升对成长股和加密货币通常是利空
if market in ["Crypto", "USStock"]:
if tnx_change > 2:
tnx_score = -20 # 利率大幅上升,利空
if tnx_change > 3:
tnx_score = -30 # 利率大幅上升,严重利空
elif tnx_change > 2:
tnx_score = -20 # 利率上升,利空
elif tnx_change < -3:
tnx_score = +30 # 利率大幅下降,利多
elif tnx_change < -2:
tnx_score = +20 # 利率下降,利多
else:
@@ -1405,9 +1529,12 @@ IMPORTANT:
score += tnx_score
factors += 1
# 归一化
# 归一化(考虑权重)
if factors > 0:
score = score / factors * 100 / 3 # 最大可能分数是3个因素各30分=90,归一化到100
# 最大可能分数:VIX(-50~+20), DXY(-30~+30), TNX(-30~+30) = 约-110到+80
# 归一化到-100到+100
max_possible = 110 # 最大绝对值
score = score / max_possible * 100
return max(-100, min(100, score))
@@ -1415,24 +1542,26 @@ IMPORTANT:
"""
根据客观评分转换为决策
优化后的阈值缩小HOLD区间使决策更明确
- score >= +40: BUY利多
- score <= -40: SELL利空
- -40 < score < +40: HOLD中性
优化后的阈值大幅缩小HOLD区间使决策更明确
- score >= +20: BUY利多
- score <= -20: SELL利空
- -20 < score < +20: HOLD中性
分级决策可选用于更细粒度的判断
分级决策用于更细粒度的判断
- score >= +70: 强烈BUY
- +40 <= score < +70: BUY
- +10 < score < +40: 弱利多倾向于BUY但可HOLD
- -10 <= score <= +10: 中性HOLD
- -40 < score < -10: 弱利空倾向于SELL但可HOLD
- -70 < score <= -40: SELL
- +40 <= score < +70: 明显BUY
- +20 <= score < +40: BUY
- +10 < score < +20: 弱利多倾向于BUY但可HOLD
- -10 <= score <= +10: 中性HOLD真正的中性区间
- -20 < score < -10: 弱利空倾向于SELL但可HOLD
- -40 < score <= -20: SELL
- -70 < score <= -40: 明显SELL
- score <= -70: 强烈SELL
"""
# 使用±40作为主要阈值,缩小HOLD区间
if score >= 40:
# 使用±20作为主要阈值,大幅缩小HOLD区间
if score >= 20:
return "BUY"
elif score <= -40:
elif score <= -20:
return "SELL"
else:
return "HOLD"
@@ -74,6 +74,7 @@ class MarketDataCollector:
timeframe: str = "1D",
include_macro: bool = True,
include_news: bool = True,
include_polymarket: bool = True, # 新增:是否包含预测市场数据
timeout: int = 30
) -> Dict[str, Any]:
"""
@@ -85,6 +86,7 @@ class MarketDataCollector:
timeframe: K线周期
include_macro: 是否包含宏观数据
include_news: 是否包含新闻
include_polymarket: 是否包含预测市场数据
timeout: 总超时时间()
Returns:
@@ -109,6 +111,8 @@ class MarketDataCollector:
# 情绪
"news": [],
"sentiment": {},
# 预测市场
"polymarket": [],
# 元数据
"_meta": {
"success_items": [],
@@ -181,6 +185,17 @@ class MarketDataCollector:
logger.warning(f"News fetch failed: {e}")
data["_meta"]["failed_items"].append("news")
# === 阶段4: 预测市场数据 (如果需要) ===
if include_polymarket:
try:
polymarket_events = self._get_polymarket_events(symbol, market)
data["polymarket"] = polymarket_events
if polymarket_events:
data["_meta"]["success_items"].append("polymarket")
except Exception as e:
logger.debug(f"Polymarket data fetch failed: {e}")
data["_meta"]["failed_items"].append("polymarket")
# 记录总耗时
data["_meta"]["duration_ms"] = int((time.time() - start_time) * 1000)
logger.info(f"Market data collection completed for {market}:{symbol} in {data['_meta']['duration_ms']}ms")
@@ -1045,6 +1060,13 @@ class MarketDataCollector:
search_news = self._get_news_from_search(market, symbol, company_name)
news_list.extend(search_news)
# === 4) 获取全球重大事件新闻(地缘政治、战争等) ===
# 这些事件会影响所有市场,特别是加密货币
global_events = self._get_global_major_events()
if global_events:
news_list.extend(global_events)
logger.info(f"Added {len(global_events)} global major events to news list")
# 去重(按标题)
seen_titles = set()
unique_news = []
@@ -1105,6 +1127,171 @@ class MarketDataCollector:
logger.debug(f"搜索引擎新闻获取失败: {e}")
return news_list
def _get_global_major_events(self) -> List[Dict]:
"""
获取全球重大事件新闻地缘政治战争重大政策等
这些事件会影响所有市场特别是加密货币
Returns:
全球重大事件新闻列表
"""
news_list = []
try:
from app.services.search import get_search_service
search_service = get_search_service()
if not search_service.is_available:
return news_list
# 搜索全球重大事件(最近24小时)
global_event_queries = [
"war conflict breaking news today",
"geopolitical crisis latest",
"US Iran military news"
]
for query in global_event_queries:
try:
response = search_service.search_with_fallback(
query=query,
max_results=2,
days=1 # 只搜索最近1天的新闻
)
if response.success and response.results:
for result in response.results:
# 检查是否是重大事件(包含关键词)
title_lower = result.title.lower()
snippet_lower = (result.snippet or "").lower()
text = f"{title_lower} {snippet_lower}"
# 重大事件关键词
major_event_keywords = [
"war", "conflict", "military", "attack", "strike", "sanctions",
"geopolitical", "crisis", "tension", "iran", "israel", "russia",
"ukraine", "middle east", "nato", "united states",
"战争", "冲突", "军事", "袭击", "制裁", "地缘政治", "危机"
]
if any(keyword in text for keyword in major_event_keywords):
news_list.append({
"datetime": result.published_date or datetime.now().strftime('%Y-%m-%d %H:%M'),
"headline": result.title,
"summary": result.snippet[:300] if result.snippet else '',
"source": f"全球事件:{result.source}",
"url": result.url,
"sentiment": "negative" if any(kw in text for kw in ["war", "conflict", "attack", "战争", "冲突", "袭击"]) else "neutral",
"is_global_event": True # 标记为全球事件
})
logger.info(f"Found global major event: {result.title[:60]}")
except Exception as e:
logger.debug(f"Failed to search global events with query '{query}': {e}")
continue
# 去重
seen_titles = set()
unique_events = []
for item in news_list:
title = item.get('headline', '')
if title and title not in seen_titles:
seen_titles.add(title)
unique_events.append(item)
return unique_events[:5] # 最多返回5条全球重大事件
except Exception as e:
logger.debug(f"Failed to get global major events: {e}")
return []
def _get_polymarket_events(self, symbol: str, market: str) -> List[Dict]:
"""
获取与资产相关的预测市场事件
直接调用Polymarket API获取实时数据不依赖本地数据库
Args:
symbol: 资产符号
market: 市场类型
Returns:
相关预测市场事件列表
"""
try:
from app.data_sources.polymarket import PolymarketDataSource
polymarket_source = PolymarketDataSource()
# 提取关键词
keywords = self._extract_polymarket_keywords(symbol, market)
logger.info(f"Extracted Polymarket keywords for {symbol}: {keywords}")
# 直接调用API搜索,不使用数据库缓存
# 因为AI分析需要最新的、相关的数据,数据库不可能包含所有市场
related_markets = []
for keyword in keywords:
# 使用use_cache=False强制从API获取
markets = polymarket_source.search_markets(keyword, limit=5, use_cache=False)
logger.info(f"Found {len(markets)} markets for keyword '{keyword}' (from API)")
related_markets.extend(markets)
# 去重
seen = set()
result = []
for market_data in related_markets:
market_id = market_data.get('market_id')
if market_id and market_id not in seen:
seen.add(market_id)
result.append({
"market_id": market_id,
"question": market_data.get('question', ''),
"current_probability": market_data.get('current_probability', 50.0),
"volume_24h": market_data.get('volume_24h', 0),
"liquidity": market_data.get('liquidity', 0),
"category": market_data.get('category', 'other'),
"polymarket_url": market_data.get('polymarket_url', f"https://polymarket.com/event/{market_id}")
})
logger.info(f"Total {len(result)} unique Polymarket events found for {symbol}")
return result
except Exception as e:
logger.debug(f"Failed to get polymarket events for {symbol}: {e}")
return []
def _extract_polymarket_keywords(self, symbol: str, market: str) -> List[str]:
"""提取用于搜索预测市场的关键词"""
keywords = []
# 基础符号
if '/' in symbol:
base = symbol.split('/')[0]
keywords.append(base)
else:
keywords.append(symbol)
# 加密货币全名映射
crypto_names = {
'BTC': ['Bitcoin', 'bitcoin'],
'ETH': ['Ethereum', 'ethereum'],
'SOL': ['Solana', 'solana'],
'BNB': ['Binance', 'binance'],
'XRP': ['Ripple', 'ripple'],
'ADA': ['Cardano', 'cardano'],
'DOGE': ['Dogecoin', 'dogecoin'],
'AVAX': ['Avalanche', 'avalanche'],
'DOT': ['Polkadot', 'polkadot'],
'MATIC': ['Polygon', 'polygon']
}
base_symbol = symbol.split('/')[0] if '/' in symbol else symbol
if base_symbol in crypto_names:
keywords.extend(crypto_names[base_symbol])
# 添加价格相关关键词
if market == 'Crypto':
keywords.extend(['$100k', '$100000', '100k', 'ETF', 'approval'])
return keywords
# 全局实例
@@ -0,0 +1,555 @@
"""
Polymarket预测市场分析器
分析预测市场生成AI预测和交易机会推荐
"""
import json
import re
from typing import Dict, List, Any, Optional
from datetime import datetime
from app.utils.logger import get_logger
from app.utils.db import get_db_connection
from app.services.llm import LLMService
from app.services.market_data_collector import get_market_data_collector
from app.data_sources.polymarket import PolymarketDataSource
logger = get_logger(__name__)
class PolymarketAnalyzer:
"""预测市场AI分析器"""
def __init__(self):
self.llm_service = LLMService()
self.data_collector = get_market_data_collector()
self.polymarket_source = PolymarketDataSource()
def analyze_market(self, market_id: str, user_id: int = None, use_cache: bool = True) -> Dict:
"""
分析单个预测市场
Args:
market_id: 市场ID
user_id: 用户ID可选用于用户特定分析
use_cache: 是否使用缓存的分析结果默认True
Returns:
分析结果字典
"""
try:
# 1. 获取市场数据
market = self.polymarket_source.get_market_details(market_id)
if not market:
return {
"error": "Market not found",
"market_id": market_id
}
# 2. 如果使用缓存,检查是否有缓存的分析结果(30分钟有效)
if use_cache:
cached_analysis = self._get_cached_analysis(market_id, user_id)
if cached_analysis:
cache_minutes = 30 # 缓存30分钟
if self._is_analysis_fresh(cached_analysis, max_age_minutes=cache_minutes):
logger.debug(f"Using cached analysis for market {market_id}")
return cached_analysis
# 3. 收集相关数据
related_news = self._get_related_news(market['question'])
related_assets = self._identify_related_assets(market['question'])
asset_data = self._get_asset_data(related_assets)
# 4. AI分析
ai_result = self._ai_predict_probability(
question=market['question'],
current_market_prob=market['current_probability'],
related_news=related_news,
asset_data=asset_data
)
# 5. 计算机会评分
opportunity_score = self._calculate_opportunity_score(
ai_prob=ai_result['predicted_probability'],
market_prob=market['current_probability'],
confidence=ai_result['confidence']
)
# 6. 生成推荐
recommendation = self._generate_recommendation(
divergence=ai_result['predicted_probability'] - market['current_probability'],
confidence=ai_result['confidence']
)
# 7. 构建分析结果
analysis_result = {
"market_id": market_id,
"ai_predicted_probability": ai_result['predicted_probability'],
"market_probability": market['current_probability'],
"divergence": ai_result['predicted_probability'] - market['current_probability'],
"recommendation": recommendation,
"confidence_score": ai_result['confidence'],
"reasoning": ai_result['reasoning'],
"key_factors": ai_result.get('key_factors', []),
"risk_factors": ai_result.get('risk_factors', []),
"related_assets": related_assets,
"risk_level": self._assess_risk(market, ai_result),
"opportunity_score": opportunity_score
}
# 8. 保存到数据库
self._save_analysis_to_db(analysis_result, user_id)
return analysis_result
except Exception as e:
logger.error(f"Failed to analyze market {market_id}: {e}", exc_info=True)
return {
"error": str(e),
"market_id": market_id
}
def generate_asset_trading_opportunities(self, market_id: str) -> List[Dict]:
"""
基于预测市场生成相关资产的交易机会
Args:
market_id: 预测市场ID
Returns:
资产交易机会列表
"""
try:
# 1. 分析预测市场
market_analysis = self.analyze_market(market_id)
if market_analysis.get('error'):
return []
# 2. 识别相关资产
related_assets = market_analysis.get('related_assets', [])
if not related_assets:
return []
# 3. 对每个资产进行技术分析
opportunities = []
for asset in related_assets:
try:
# 推断市场类型
market_type = self._infer_market(asset)
# 获取资产数据
asset_data = self.data_collector.collect_all(
market=market_type,
symbol=asset,
timeframe="1D",
include_polymarket=False # 避免循环
)
# 技术分析
technical_analysis = self._analyze_technical(asset_data)
# 结合预测市场信号
if market_analysis['recommendation'] == "YES":
# 预测事件发生概率高 → 相关资产可能上涨
signal = "BUY" if technical_analysis.get('trend') == "bullish" else "HOLD"
elif market_analysis['recommendation'] == "NO":
# 预测事件发生概率低 → 相关资产可能下跌
signal = "SELL" if technical_analysis.get('trend') == "bearish" else "HOLD"
else:
signal = "HOLD"
# 计算综合置信度
confidence = (
market_analysis['confidence_score'] * 0.6 +
technical_analysis.get('confidence', 50) * 0.4
)
if signal != "HOLD" and confidence > 60:
opportunities.append({
"asset": asset,
"market": market_type,
"signal": signal,
"confidence": round(confidence, 2),
"reasoning": f"预测市场分析:{market_analysis['reasoning'][:200]}。技术面:{technical_analysis.get('summary', '')[:200]}",
"related_prediction": {
"market_id": market_id,
"question": market_analysis.get('question', ''),
"ai_probability": market_analysis['ai_predicted_probability'],
"market_probability": market_analysis['market_probability']
},
"entry_suggestion": technical_analysis.get('entry_suggestion', {})
})
except Exception as e:
logger.debug(f"Failed to analyze asset {asset} for market {market_id}: {e}")
continue
# 保存机会到数据库
if opportunities:
self._save_opportunities_to_db(market_id, opportunities)
return opportunities
except Exception as e:
logger.error(f"Failed to generate asset opportunities for {market_id}: {e}")
return []
def _ai_predict_probability(self, question: str, current_market_prob: float,
related_news: List, asset_data: Dict) -> Dict:
"""使用AI预测事件概率"""
try:
# 构建prompt
news_text = "\n".join([f"- {n.get('title', '')[:100]}" for n in related_news[:5]])
asset_text = ""
if asset_data:
price_data = asset_data.get('price', {})
indicators = asset_data.get('indicators', {})
if price_data:
asset_text = f"""
相关资产数据
- 当前价格: {price_data.get('current_price', 'N/A')}
- 24h涨跌幅: {price_data.get('change_24h', 0):.2f}%
- RSI: {indicators.get('rsi', {}).get('value', 'N/A')}
- MACD: {indicators.get('macd', {}).get('signal', 'N/A')}
"""
prompt = f"""分析以下预测市场事件,评估其发生的概率:
问题{question}
当前市场概率{current_market_prob}%
相关新闻
{news_text if news_text else "暂无相关新闻"}
{asset_text}
请基于以下维度分析
1. 历史类似事件的成功率
2. 当前新闻和趋势
3. 相关资产价格走势和技术指标
4. 宏观环境因素VIXDXY利率等
5. 市场情绪指标
输出JSON格式
{{
"predicted_probability": 72.5, // 你预测的概率0-100
"confidence": 75.0, // 置信度0-100
"reasoning": "详细分析...",
"key_factors": ["因素1", "因素2"],
"risk_factors": ["风险1", "风险2"]
}}"""
# 调用LLM
messages = [
{
"role": "system",
"content": "你是一个专业的市场分析师,擅长分析预测市场事件。请基于提供的数据,客观评估事件发生的概率。"
},
{
"role": "user",
"content": prompt
}
]
result = self.llm_service.call_llm_api(
messages=messages,
use_json_mode=True,
temperature=0.3
)
# 解析结果
if isinstance(result, str):
result = json.loads(result)
# 验证和规范化
predicted_prob = float(result.get('predicted_probability', current_market_prob))
predicted_prob = max(0, min(100, predicted_prob)) # 限制在0-100
confidence = float(result.get('confidence', 70))
confidence = max(0, min(100, confidence))
return {
'predicted_probability': round(predicted_prob, 2),
'confidence': round(confidence, 2),
'reasoning': result.get('reasoning', ''),
'key_factors': result.get('key_factors', []),
'risk_factors': result.get('risk_factors', [])
}
except Exception as e:
logger.error(f"AI prediction failed: {e}", exc_info=True)
# 返回默认值
return {
'predicted_probability': current_market_prob,
'confidence': 50.0,
'reasoning': f'分析失败: {str(e)}',
'key_factors': [],
'risk_factors': []
}
def _calculate_opportunity_score(self, ai_prob: float, market_prob: float,
confidence: float) -> float:
"""
计算机会评分0-100
逻辑
- AI与市场差异越大机会越好
- 置信度越高机会越好
"""
divergence = abs(ai_prob - market_prob)
# 差异越大,机会越好(最大40分)
divergence_score = min(divergence * 2, 40)
# 置信度越高,机会越好(最大60分)
confidence_score = confidence * 0.6
return round(divergence_score + confidence_score, 2)
def _generate_recommendation(self, divergence: float, confidence: float) -> str:
"""
生成推荐YES/NO/HOLD
逻辑
- AI概率 > 市场概率 + 5% 且置信度 > 60 YES
- AI概率 < 市场概率 - 5% 且置信度 > 60 NO
- 其他 HOLD
"""
if divergence > 5 and confidence > 60:
return "YES"
elif divergence < -5 and confidence > 60:
return "NO"
else:
return "HOLD"
def _assess_risk(self, market: Dict, ai_result: Dict) -> str:
"""评估风险等级"""
confidence = ai_result.get('confidence', 50)
divergence = abs(ai_result.get('predicted_probability', 50) - market.get('current_probability', 50))
if confidence < 50 or divergence > 30:
return "high"
elif confidence < 70 or divergence > 15:
return "medium"
else:
return "low"
def _get_related_news(self, question: str) -> List[Dict]:
"""获取相关问题相关的新闻"""
# 提取关键词
keywords = self._extract_keywords(question)
# 这里可以调用新闻API,暂时返回空列表
# 实际实现时可以调用现有的新闻服务
return []
def _identify_related_assets(self, question: str) -> List[str]:
"""识别问题中提到的相关资产"""
assets = []
# 加密货币关键词映射
crypto_keywords = {
'BTC': ['BTC', 'Bitcoin', 'bitcoin', 'btc'],
'ETH': ['ETH', 'Ethereum', 'ethereum', 'eth'],
'SOL': ['SOL', 'Solana', 'solana', 'sol'],
'BNB': ['BNB', 'Binance', 'binance', 'bnb'],
'XRP': ['XRP', 'Ripple', 'ripple', 'xrp'],
'ADA': ['ADA', 'Cardano', 'cardano', 'ada'],
'DOGE': ['DOGE', 'Dogecoin', 'dogecoin', 'doge'],
'AVAX': ['AVAX', 'Avalanche', 'avalanche', 'avax'],
'DOT': ['DOT', 'Polkadot', 'polkadot', 'dot'],
'MATIC': ['MATIC', 'Polygon', 'polygon', 'matic']
}
question_upper = question.upper()
for symbol, keywords in crypto_keywords.items():
if any(kw in question_upper for kw in keywords):
assets.append(f"{symbol}/USDT")
# 去重
return list(set(assets))
def _get_asset_data(self, assets: List[str]) -> Optional[Dict]:
"""获取资产数据(取第一个资产)"""
if not assets:
return None
try:
asset = assets[0]
market_type = self._infer_market(asset)
return self.data_collector.collect_all(
market=market_type,
symbol=asset,
timeframe="1D"
)
except Exception as e:
logger.debug(f"Failed to get asset data for {assets}: {e}")
return None
def _analyze_technical(self, asset_data: Dict) -> Dict:
"""简单的技术分析"""
if not asset_data:
return {
'trend': 'neutral',
'confidence': 50,
'summary': '数据不足',
'entry_suggestion': {}
}
indicators = asset_data.get('indicators', {})
price_data = asset_data.get('price', {})
# 简单的趋势判断
rsi = indicators.get('rsi', {}).get('value', 50)
macd_signal = indicators.get('macd', {}).get('signal', 'neutral')
trend = 'neutral'
if rsi > 60 and macd_signal == 'bullish':
trend = 'bullish'
elif rsi < 40 and macd_signal == 'bearish':
trend = 'bearish'
confidence = 60 if abs(rsi - 50) > 15 else 50
return {
'trend': trend,
'confidence': confidence,
'summary': f'RSI: {rsi:.1f}, MACD: {macd_signal}',
'entry_suggestion': {}
}
def _infer_market(self, symbol: str) -> str:
"""推断市场类型"""
if '/' in symbol:
return "Crypto"
elif len(symbol) <= 5 and symbol.isupper():
return "USStock"
else:
return "Crypto" # 默认
def _extract_keywords(self, text: str) -> List[str]:
"""提取关键词"""
# 简单的关键词提取
words = re.findall(r'\b[A-Z][a-z]+\b|\b[A-Z]{2,}\b', text)
return [w.lower() for w in words if len(w) > 2]
def _get_cached_analysis(self, market_id: str, user_id: int = None) -> Optional[Dict]:
"""获取缓存的分析结果"""
try:
with get_db_connection() as db:
cur = db.cursor()
query = """
SELECT ai_predicted_probability, market_probability, divergence,
recommendation, confidence_score, opportunity_score,
reasoning, key_factors, related_assets, created_at
FROM qd_polymarket_ai_analysis
WHERE market_id = %s
"""
params = [market_id]
if user_id:
query += " AND user_id = %s"
params.append(user_id)
else:
query += " AND user_id IS NULL"
query += " ORDER BY created_at DESC LIMIT 1"
cur.execute(query, params)
row = cur.fetchone()
cur.close()
if row:
# RealDictCursor返回字典,使用键访问
key_factors_raw = row.get('key_factors')
key_factors = []
if key_factors_raw:
try:
if isinstance(key_factors_raw, str):
key_factors = json.loads(key_factors_raw)
else:
key_factors = key_factors_raw if isinstance(key_factors_raw, list) else []
except:
key_factors = []
return {
"market_id": market_id,
"ai_predicted_probability": float(row.get('ai_predicted_probability') or 0),
"market_probability": float(row.get('market_probability') or 0),
"divergence": float(row.get('divergence') or 0),
"recommendation": row.get('recommendation') or 'HOLD',
"confidence_score": float(row.get('confidence_score') or 0),
"opportunity_score": float(row.get('opportunity_score') or 0),
"reasoning": row.get('reasoning') or '',
"key_factors": key_factors,
"related_assets": row.get('related_assets') if row.get('related_assets') else [],
"created_at": row.get('created_at')
}
except Exception as e:
logger.debug(f"Failed to get cached analysis: {e}")
return None
def _is_analysis_fresh(self, analysis: Dict, max_age_minutes: int = 30) -> bool:
"""检查分析结果是否新鲜"""
created_at = analysis.get('created_at')
if not created_at:
return False
if isinstance(created_at, str):
created_at = datetime.fromisoformat(created_at.replace('Z', '+00:00'))
age = (datetime.now() - created_at.replace(tzinfo=None)).total_seconds() / 60
return age < max_age_minutes
def _save_analysis_to_db(self, analysis: Dict, user_id: int = None):
"""保存分析结果到数据库"""
try:
with get_db_connection() as db:
cur = db.cursor()
cur.execute("""
INSERT INTO qd_polymarket_ai_analysis
(market_id, user_id, ai_predicted_probability, market_probability,
divergence, recommendation, confidence_score, opportunity_score,
reasoning, key_factors, related_assets, created_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())
""", (
analysis['market_id'],
user_id,
analysis['ai_predicted_probability'],
analysis['market_probability'],
analysis['divergence'],
analysis['recommendation'],
analysis['confidence_score'],
analysis['opportunity_score'],
analysis['reasoning'],
json.dumps(analysis.get('key_factors', [])),
analysis.get('related_assets', [])
))
db.commit()
cur.close()
except Exception as e:
logger.error(f"Failed to save analysis to DB: {e}")
def _save_opportunities_to_db(self, market_id: str, opportunities: List[Dict]):
"""保存交易机会到数据库"""
try:
with get_db_connection() as db:
cur = db.cursor()
for opp in opportunities:
cur.execute("""
INSERT INTO qd_polymarket_asset_opportunities
(market_id, asset_symbol, asset_market, signal, confidence,
reasoning, entry_suggestion, created_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, NOW())
""", (
market_id,
opp['asset'],
opp['market'],
opp['signal'],
opp['confidence'],
opp['reasoning'],
json.dumps(opp.get('entry_suggestion', {}))
))
db.commit()
cur.close()
except Exception as e:
logger.error(f"Failed to save opportunities to DB: {e}")
@@ -0,0 +1,246 @@
"""
Polymarket批量分析器
一次性分析多个市场由AI筛选出有交易机会的市场
"""
import json
from typing import List, Dict, Optional
from app.utils.logger import get_logger
from app.utils.db import get_db_connection
from app.services.llm import LLMService
from app.data_sources.polymarket import PolymarketDataSource
logger = get_logger(__name__)
class PolymarketBatchAnalyzer:
"""批量分析预测市场,由AI筛选交易机会"""
def __init__(self):
self.llm_service = LLMService()
self.polymarket_source = PolymarketDataSource()
def batch_analyze_markets(self, markets: List[Dict], max_opportunities: int = 20) -> List[Dict]:
"""
批量分析市场由AI筛选出有交易机会的市场
Args:
markets: 市场列表
max_opportunities: 最多返回多少个交易机会
Returns:
筛选后的市场列表包含AI分析结果
"""
if not markets:
return []
try:
# 1. 构建批量分析的prompt
markets_summary = self._build_markets_summary(markets)
prompt = f"""你是一个专业的预测市场分析师。请分析以下预测市场列表,筛选出最有交易机会的市场。
市场列表
{markets_summary}
请基于以下维度评估每个市场
1. **市场活跃度**交易量流动性是否足够
2. **概率偏差**当前市场概率是否偏离合理预期偏离50%越多机会越大
3. **事件重要性**事件对市场的影响程度
4. **时间窗口**距离结算时间是否合适太近或太远都不好
5. **信息优势**是否有明显的信息不对称或市场误判
请返回JSON格式包含筛选出的市场ID和简要分析
{{
"opportunities": [
{{
"market_id": "市场ID",
"opportunity_score": 85, // 机会评分 0-100
"reason": "为什么这个市场有交易机会(简要说明)",
"recommendation": "YES/NO/HOLD", // 推荐方向
"confidence": 75, // 置信度 0-100
"key_factors": ["因素1", "因素2"] // 关键因素
}}
]
}}
要求
- 只返回最有价值的 {max_opportunities} 个机会
- 机会评分 >= 60 才考虑
- 优先选择高交易量 + 明显概率偏差 + 高置信度
- 简要说明原因不要冗长"""
# 2. 调用LLM进行批量分析
messages = [
{
"role": "system",
"content": "你是一个专业的预测市场分析师,擅长从大量市场中快速识别有价值的交易机会。请客观、理性地分析,只推荐真正有优势的机会。"
},
{
"role": "user",
"content": prompt
}
]
logger.info(f"Batch analyzing {len(markets)} markets, requesting {max_opportunities} opportunities")
result = self.llm_service.call_llm_api(
messages=messages,
use_json_mode=True,
temperature=0.3
)
# 3. 解析结果
if isinstance(result, str):
try:
result = json.loads(result)
except:
logger.error(f"Failed to parse LLM result as JSON: {result[:200]}")
return self._fallback_analysis(markets, max_opportunities)
opportunities = result.get('opportunities', [])
if not opportunities:
logger.warning("LLM returned no opportunities, using fallback")
return self._fallback_analysis(markets, max_opportunities)
# 4. 将AI分析结果合并到市场数据中
opportunities_map = {opp.get('market_id'): opp for opp in opportunities}
analyzed_markets = []
for market in markets:
market_id = market.get('market_id')
if not market_id:
continue
opp = opportunities_map.get(market_id)
if opp:
# 获取AI预测的概率
predicted_prob = float(opp.get('predicted_probability', market.get('current_probability', 50.0)))
market_prob = market.get('current_probability', 50.0)
divergence = predicted_prob - market_prob
# 合并AI分析结果
market['ai_analysis'] = {
'predicted_probability': predicted_prob, # 使用AI预测的概率
'recommendation': opp.get('recommendation', 'HOLD'),
'confidence_score': float(opp.get('confidence', 0)),
'opportunity_score': float(opp.get('opportunity_score', 0)),
'divergence': divergence, # AI预测概率 - 市场概率
'reasoning': opp.get('reason', ''),
'key_factors': opp.get('key_factors', [])
}
analyzed_markets.append(market)
# 5. 按机会评分排序
analyzed_markets.sort(
key=lambda x: x.get('ai_analysis', {}).get('opportunity_score', 0),
reverse=True
)
logger.info(f"Batch analysis completed: {len(analyzed_markets)} opportunities identified")
return analyzed_markets
except Exception as e:
logger.error(f"Batch analysis failed: {e}", exc_info=True)
return self._fallback_analysis(markets, max_opportunities)
def _build_markets_summary(self, markets: List[Dict]) -> str:
"""构建市场摘要,用于批量分析"""
summary_lines = []
for i, market in enumerate(markets[:50], 1): # 限制最多50个,避免prompt过长
market_id = market.get('market_id', '')
question = market.get('question', '')[:100] # 限制长度
prob = market.get('current_probability', 50.0)
volume = market.get('volume_24h', 0)
category = market.get('category', 'other')
summary_lines.append(
f"{i}. ID: {market_id}\n"
f" 问题: {question}\n"
f" 当前概率: {prob:.1f}%\n"
f" 24h交易量: ${volume:,.0f}\n"
f" 分类: {category}"
)
return "\n\n".join(summary_lines)
def _fallback_analysis(self, markets: List[Dict], max_opportunities: int) -> List[Dict]:
"""回退分析:基于简单规则筛选"""
opportunities = []
for market in markets:
prob = market.get('current_probability', 50.0)
volume = market.get('volume_24h', 0)
# 简单规则:交易量大 + 概率偏离50%
if volume > 10000 and abs(prob - 50.0) > 10:
opportunity_score = min(60 + abs(prob - 50.0) * 0.5, 90)
market['ai_analysis'] = {
'predicted_probability': prob,
'recommendation': 'YES' if prob > 50 else 'NO',
'confidence_score': 60.0,
'opportunity_score': opportunity_score,
'divergence': 0,
'reasoning': f'高交易量({volume:,.0f}) + 明显概率偏差({prob:.1f}%)',
'key_factors': ['高交易量', '概率偏差']
}
opportunities.append(market)
# 按机会评分排序
opportunities.sort(
key=lambda x: x.get('ai_analysis', {}).get('opportunity_score', 0),
reverse=True
)
return opportunities[:max_opportunities]
def save_batch_analysis(self, markets: List[Dict]):
"""保存批量分析结果到数据库"""
try:
with get_db_connection() as db:
cur = db.cursor()
for market in markets:
market_id = market.get('market_id')
ai_analysis = market.get('ai_analysis')
if not market_id or not ai_analysis:
continue
try:
# 先删除该市场的旧分析记录(user_id为NULL的通用分析)
cur.execute("""
DELETE FROM qd_polymarket_ai_analysis
WHERE market_id = %s AND user_id IS NULL
""", (market_id,))
# 插入新的分析记录
cur.execute("""
INSERT INTO qd_polymarket_ai_analysis
(market_id, user_id, ai_predicted_probability, market_probability,
divergence, recommendation, confidence_score, opportunity_score,
reasoning, key_factors, related_assets, created_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW())
""", (
market_id,
None, # 通用分析
float(ai_analysis.get('predicted_probability', market.get('current_probability', 50.0))),
market.get('current_probability', 50.0),
float(ai_analysis.get('divergence', 0)),
ai_analysis.get('recommendation', 'HOLD'),
ai_analysis.get('confidence_score', 0),
ai_analysis.get('opportunity_score', 0),
ai_analysis.get('reasoning', ''),
json.dumps(ai_analysis.get('key_factors', [])),
[]
))
except Exception as e:
logger.warning(f"Failed to save analysis for market {market_id}: {e}")
continue
db.commit()
cur.close()
logger.info(f"Saved batch analysis for {len(markets)} markets")
except Exception as e:
logger.error(f"Failed to save batch analysis: {e}", exc_info=True)
@@ -0,0 +1,163 @@
"""
Polymarket后台任务
每30分钟更新一次市场数据并批量分析市场机会
"""
import threading
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional
from app.utils.logger import get_logger
from app.utils.db import get_db_connection
from app.data_sources.polymarket import PolymarketDataSource
from app.services.polymarket_batch_analyzer import PolymarketBatchAnalyzer
logger = get_logger(__name__)
class PolymarketWorker:
"""Polymarket数据更新和分析后台任务"""
def __init__(self, update_interval_minutes: int = 30, analysis_cache_minutes: int = 30):
"""
初始化后台任务
Args:
update_interval_minutes: 市场数据更新间隔分钟
analysis_cache_minutes: AI分析结果缓存时间分钟
"""
self.update_interval_minutes = update_interval_minutes
self.analysis_cache_minutes = analysis_cache_minutes
self._stop_event = threading.Event()
self._thread: Optional[threading.Thread] = None
self._lock = threading.Lock()
self.polymarket_source = PolymarketDataSource()
self.batch_analyzer = PolymarketBatchAnalyzer()
self._last_update_ts = 0.0
def start(self) -> bool:
"""启动后台任务"""
with self._lock:
if self._thread and self._thread.is_alive():
return True
self._stop_event.clear()
self._thread = threading.Thread(target=self._run_loop, name="PolymarketWorker", daemon=True)
self._thread.start()
logger.info(f"PolymarketWorker started (update_interval={self.update_interval_minutes}min, cache={self.analysis_cache_minutes}min)")
return True
def stop(self, timeout_sec: float = 5.0) -> None:
"""停止后台任务"""
with self._lock:
if not self._thread or not self._thread.is_alive():
return
self._stop_event.set()
self._thread.join(timeout=timeout_sec)
if self._thread.is_alive():
logger.warning("PolymarketWorker thread did not stop within timeout")
else:
logger.info("PolymarketWorker stopped")
def _run_loop(self) -> None:
"""主循环"""
logger.info("PolymarketWorker loop started")
# 启动时立即执行一次
self._update_markets_and_analyze()
while not self._stop_event.is_set():
try:
# 等待指定时间间隔
wait_seconds = self.update_interval_minutes * 60
if self._stop_event.wait(wait_seconds):
break # 如果收到停止信号,退出循环
# 执行更新和分析
self._update_markets_and_analyze()
except Exception as e:
logger.error(f"PolymarketWorker loop error: {e}", exc_info=True)
# 出错后等待1分钟再重试
self._stop_event.wait(60)
logger.info("PolymarketWorker loop stopped")
def _update_markets_and_analyze(self) -> None:
"""更新市场数据并分析"""
try:
logger.info("Starting Polymarket data update and analysis...")
start_time = time.time()
# 1. 更新市场数据(从所有主要分类获取)
categories = ["crypto", "politics", "economics", "sports", "tech", "finance", "geopolitics", "culture", "climate", "entertainment"]
all_markets = []
for category in categories:
try:
markets = self.polymarket_source.get_trending_markets(category, limit=50)
all_markets.extend(markets)
logger.info(f"Fetched {len(markets)} markets from category: {category}")
except Exception as e:
logger.warning(f"Failed to fetch markets for category {category}: {e}")
# 去重(按market_id
unique_markets = {}
for market in all_markets:
market_id = market.get('market_id')
if market_id:
unique_markets[market_id] = market
logger.info(f"Total unique markets: {len(unique_markets)}")
# 2. 批量分析市场(一次性分析所有市场,由AI筛选机会)
markets_list = list(unique_markets.values())
logger.info(f"Starting batch analysis for {len(markets_list)} markets...")
# 使用批量分析器,一次性分析所有市场
# AI会筛选出最有交易机会的市场(最多20个)
analyzed_markets = self.batch_analyzer.batch_analyze_markets(
markets_list,
max_opportunities=20
)
# 3. 保存分析结果到数据库
if analyzed_markets:
self.batch_analyzer.save_batch_analysis(analyzed_markets)
analyzed_count = len(analyzed_markets)
else:
analyzed_count = 0
elapsed = time.time() - start_time
logger.info(f"Polymarket update completed: {len(unique_markets)} markets updated, {analyzed_count} opportunities identified in {elapsed:.1f}s")
self._last_update_ts = time.time()
except Exception as e:
logger.error(f"Failed to update markets and analyze: {e}", exc_info=True)
def force_update(self) -> None:
"""强制立即更新(用于手动触发)"""
logger.info("Force update triggered")
self._update_markets_and_analyze()
# 全局单例
_polymarket_worker: Optional[PolymarketWorker] = None
_worker_lock = threading.Lock()
def get_polymarket_worker() -> PolymarketWorker:
"""获取PolymarketWorker单例"""
global _polymarket_worker
with _worker_lock:
if _polymarket_worker is None:
update_interval = int(os.getenv("POLYMARKET_UPDATE_INTERVAL_MIN", "30"))
cache_minutes = int(os.getenv("POLYMARKET_ANALYSIS_CACHE_MIN", "30"))
_polymarket_worker = PolymarketWorker(
update_interval_minutes=update_interval,
analysis_cache_minutes=cache_minutes
)
return _polymarket_worker
# 需要导入os
import os
+10 -3
View File
@@ -167,12 +167,16 @@ class PostgresCursor:
row = self._cursor.fetchone()
if row is None:
return None
return dict(row) if row else None
# RealDictCursor already returns a dict, so return as-is
return row if isinstance(row, dict) else dict(row) if row else None
def fetchall(self) -> List[Dict[str, Any]]:
"""Fetch all rows"""
rows = self._cursor.fetchall()
return [dict(row) for row in rows] if rows else []
if not rows:
return []
# RealDictCursor already returns dicts, so return as-is
return [row if isinstance(row, dict) else dict(row) for row in rows]
def close(self):
"""Close cursor"""
@@ -234,7 +238,10 @@ def get_pg_connection():
conn.rollback()
except Exception:
pass
logger.error(f"PostgreSQL operation error: {e}")
# 记录更详细的错误信息
error_msg = str(e) if e else repr(e)
error_type = type(e).__name__
logger.error(f"PostgreSQL operation error ({error_type}): {error_msg}", exc_info=True)
raise
finally:
if conn:
+10
View File
@@ -16,6 +16,16 @@ def setup_logger():
format=log_format
)
# 过滤 werkzeug 的 INFO 级别日志(减少噪音)
# 只保留 WARNING 及以上级别
werkzeug_logger = logging.getLogger('werkzeug')
werkzeug_logger.setLevel(logging.WARNING)
# 过滤 kline 路由的 INFO 级别日志(减少噪音)
# 只保留 WARNING 及以上级别
kline_logger = logging.getLogger('app.routes.kline')
kline_logger.setLevel(logging.WARNING)
# 创建日志目录
log_dir = 'logs'
if not os.path.exists(log_dir):
+1 -1
View File
@@ -11,7 +11,7 @@ backlog = 2048
workers = multiprocessing.cpu_count() * 2 + 1
worker_class = "sync"
worker_connections = 1000
timeout = 120
timeout = 600 # 10 minutes for long-running backtests
keepalive = 5
# 日志
+62
View File
@@ -832,6 +832,68 @@ CREATE TABLE IF NOT EXISTS qd_quick_trades (
CREATE INDEX IF NOT EXISTS idx_quick_trades_user ON qd_quick_trades(user_id);
CREATE INDEX IF NOT EXISTS idx_quick_trades_created ON qd_quick_trades(created_at DESC);
-- =============================================================================
-- Polymarket Prediction Markets (预测市场数据和分析)
-- =============================================================================
-- 预测市场表(缓存)
CREATE TABLE IF NOT EXISTS qd_polymarket_markets (
id SERIAL PRIMARY KEY,
market_id VARCHAR(255) UNIQUE NOT NULL,
question TEXT,
category VARCHAR(100), -- crypto, politics, economics, sports
current_probability DECIMAL(5,2), -- YES概率(0-100
volume_24h DECIMAL(20,2),
liquidity DECIMAL(20,2),
end_date_iso TIMESTAMP,
status VARCHAR(50), -- active, closed, resolved
outcome_tokens JSONB, -- YES/NO价格和交易量
slug VARCHAR(255), -- Polymarket事件slug,用于构建URL
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_polymarket_category ON qd_polymarket_markets(category);
CREATE INDEX IF NOT EXISTS idx_polymarket_status ON qd_polymarket_markets(status);
CREATE INDEX IF NOT EXISTS idx_polymarket_updated ON qd_polymarket_markets(updated_at DESC);
-- AI分析记录表
CREATE TABLE IF NOT EXISTS qd_polymarket_ai_analysis (
id SERIAL PRIMARY KEY,
market_id VARCHAR(255) NOT NULL,
user_id INTEGER, -- 可选:用户特定的分析
ai_predicted_probability DECIMAL(5,2),
market_probability DECIMAL(5,2),
divergence DECIMAL(5,2), -- AI - 市场
recommendation VARCHAR(20), -- YES/NO/HOLD
confidence_score DECIMAL(5,2),
opportunity_score DECIMAL(5,2),
reasoning TEXT,
key_factors JSONB,
related_assets TEXT[], -- 相关资产列表
created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_polymarket_analysis_market ON qd_polymarket_ai_analysis(market_id);
CREATE INDEX IF NOT EXISTS idx_polymarket_analysis_opportunity ON qd_polymarket_ai_analysis(opportunity_score DESC);
CREATE INDEX IF NOT EXISTS idx_polymarket_analysis_user ON qd_polymarket_ai_analysis(user_id);
-- 资产交易机会表(基于预测市场生成)
CREATE TABLE IF NOT EXISTS qd_polymarket_asset_opportunities (
id SERIAL PRIMARY KEY,
market_id VARCHAR(255) NOT NULL,
asset_symbol VARCHAR(100),
asset_market VARCHAR(50),
signal VARCHAR(20), -- BUY/SELL/HOLD
confidence DECIMAL(5,2),
reasoning TEXT,
entry_suggestion JSONB, -- 入场建议
created_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_polymarket_opp_market ON qd_polymarket_asset_opportunities(market_id);
CREATE INDEX IF NOT EXISTS idx_polymarket_opp_asset ON qd_polymarket_asset_opportunities(asset_symbol, asset_market);
-- =============================================================================
-- Completion Notice
-- =============================================================================
+1 -1
View File
@@ -82,7 +82,7 @@ app = create_app()
def main():
"""启动应用"""
# Keep startup messages ASCII-only and short.
print("QuantDinger Python API v2.0.0")
print("QuantDinger Python API v2.2.2")
# Check demo mode status for debugging
demo_status = os.getenv('IS_DEMO_MODE', 'false').lower()