From 7c067fe61cbd905e4bfbfdd27c8544ea8d6765a7 Mon Sep 17 00:00:00 2001 From: TIANHE Date: Sat, 28 Feb 2026 20:34:33 +0800 Subject: [PATCH] v2.2.1 Signed-off-by: TIANHE --- backend_api_python/app/data_sources/crypto.py | 206 ++++- .../app/services/fast_analysis.py | 830 +++++++++++++++++- .../app/services/live_trading/execution.py | 47 + .../app/services/live_trading/symbols.py | 20 +- backend_api_python/app/services/llm.py | 78 +- .../app/services/market_data_collector.py | 206 ++++- .../app/services/pending_order_worker.py | 26 + 7 files changed, 1346 insertions(+), 67 deletions(-) diff --git a/backend_api_python/app/data_sources/crypto.py b/backend_api_python/app/data_sources/crypto.py index 039bbfa..f7bbf9f 100644 --- a/backend_api_python/app/data_sources/crypto.py +++ b/backend_api_python/app/data_sources/crypto.py @@ -2,7 +2,7 @@ 加密货币数据源 使用 CCXT (Coinbase) 获取数据 """ -from typing import Dict, List, Any, Optional +from typing import Dict, List, Any, Optional, Tuple from datetime import datetime, timedelta import ccxt @@ -21,6 +21,9 @@ class CryptoDataSource(BaseDataSource): # 时间周期映射 TIMEFRAME_MAP = CCXTConfig.TIMEFRAME_MAP + # 常见的报价货币列表（按优先级排序） + COMMON_QUOTES = ['USDT', 'USD', 'BTC', 'ETH', 'BUSD', 'USDC', 'BNB', 'EUR', 'GBP'] + def __init__(self): config = { 'timeout': CCXTConfig.TIMEOUT, @@ -43,27 +46,189 @@ class CryptoDataSource(BaseDataSource): exchange_class = getattr(ccxt, exchange_id) self.exchange = exchange_class(config) + + # 延迟加载 markets（首次使用时加载） + self._markets_loaded = False + self._markets_cache = None + + def _ensure_markets_loaded(self) -> bool: + """确保 markets 已加载（用于符号验证）""" + if self._markets_loaded and self._markets_cache is not None: + return True + + try: + # 某些交易所需要显式加载 markets + if hasattr(self.exchange, 'load_markets'): + self.exchange.load_markets(reload=False) + self._markets_cache = getattr(self.exchange, 'markets', {}) + self._markets_loaded = True + return True + except Exception as e: + logger.debug(f"Failed to load markets for {self.exchange.id}: {e}") + return False + + def _normalize_symbol(self, symbol: str) -> Tuple[str, str]: + """ + 规范化符号格式，返回 (normalized_symbol, base_currency) + + 处理各种输入格式： + - BTC/USDT -> BTC/USDT + - BTCUSDT -> BTC/USDT + - BTC/USDT:USDT -> BTC/USDT + - BTC -> BTC/USDT (默认) + - PI, TRX -> PI/USDT, TRX/USDT + """ + if not symbol: + return '', '' + + sym = symbol.strip() + + # 移除 swap/futures 后缀 + if ':' in sym: + sym = sym.split(':', 1)[0] + + sym = sym.upper() + + # 如果已经有分隔符，直接解析 + if '/' in sym: + parts = sym.split('/', 1) + base = parts[0].strip() + quote = parts[1].strip() if len(parts) > 1 else '' + if base and quote: + return f"{base}/{quote}", base + + # 尝试从常见报价货币中识别 + for quote in self.COMMON_QUOTES: + if sym.endswith(quote) and len(sym) > len(quote): + base = sym[:-len(quote)] + if base: + return f"{base}/{quote}", base + + # 如果无法识别，默认使用 USDT + return f"{sym}/USDT", sym + + def _find_valid_symbol(self, base: str, preferred_quote: str = 'USDT') -> Optional[str]: + """ + 在交易所的 markets 中查找有效的符号 + + Args: + base: 基础货币（如 'PI', 'TRX'） + preferred_quote: 首选的报价货币 + + Returns: + 找到的有效符号，如果找不到则返回 None + """ + if not self._ensure_markets_loaded(): + return None + + markets = self._markets_cache or {} + if not markets: + return None + + # 按优先级尝试不同的报价货币 + quotes_to_try = [preferred_quote] + [q for q in self.COMMON_QUOTES if q != preferred_quote] + + for quote in quotes_to_try: + candidate = f"{base}/{quote}" + if candidate in markets: + market = markets[candidate] + # 检查市场是否活跃 + if market.get('active', True): + return candidate + + return None + + def _normalize_symbol_for_exchange(self, symbol: str) -> str: + """ + 根据交易所特性规范化符号 + + 不同交易所的符号格式要求： + - Binance: BTC/USDT (标准格式) + - OKX: BTC/USDT (标准格式，但某些币种可能不支持) + - Coinbase: BTC/USD (通常使用 USD 而不是 USDT) + - Kraken: XBT/USD (BTC 映射为 XBT) + - Bitfinex: tBTCUST (特殊格式) + """ + normalized, base = self._normalize_symbol(symbol) + + if not normalized or not base: + return symbol + + exchange_id = getattr(self.exchange, 'id', '').lower() + + # 特殊处理：某些交易所的符号映射 + if exchange_id == 'coinbase': + # Coinbase 通常使用 USD 而不是 USDT + if normalized.endswith('/USDT'): + usd_version = normalized.replace('/USDT', '/USD') + if self._ensure_markets_loaded(): + markets = self._markets_cache or {} + if usd_version in markets: + return usd_version + + # 尝试在交易所中查找有效符号 + if self._ensure_markets_loaded(): + valid_symbol = self._find_valid_symbol(base, normalized.split('/')[1] if '/' in normalized else 'USDT') + if valid_symbol: + return valid_symbol + + return normalized def get_ticker(self, symbol: str) -> Dict[str, Any]: """ Get latest ticker for a crypto symbol via CCXT. Accepts common formats: - - BTC/USDT - - BTCUSDT - - BTC/USDT:USDT (swap-style suffix, will be normalized) + - BTC/USDT, BTCUSDT, BTC/USDT:USDT + - PI, TRX (will be normalized and searched across exchanges) + - 自动适配不同交易所的符号格式要求 """ - sym = (symbol or "").strip() - if ":" in sym: - sym = sym.split(":", 1)[0] - sym = sym.upper() - if "/" not in sym: - # Coinbase often uses USD, check if we need to adapt - if sym.endswith("USDT") and len(sym) > 4: - sym = f"{sym[:-4]}/USDT" - elif sym.endswith("USD") and len(sym) > 3: - sym = f"{sym[:-3]}/USD" - return self.exchange.fetch_ticker(sym) + if not symbol or not symbol.strip(): + return {'last': 0, 'symbol': symbol} + + # 规范化符号 + normalized = self._normalize_symbol_for_exchange(symbol) + + if not normalized: + logger.warning(f"Failed to normalize symbol: {symbol}") + return {'last': 0, 'symbol': symbol} + + # 尝试获取 ticker + try: + ticker = self.exchange.fetch_ticker(normalized) + if ticker and isinstance(ticker, dict): + return ticker + except Exception as e: + error_msg = str(e).lower() + is_symbol_error = any(keyword in error_msg for keyword in [ + 'does not have market symbol', + 'symbol not found', + 'invalid symbol', + 'market does not exist', + 'trading pair not found' + ]) + + if is_symbol_error: + # 尝试查找替代符号 + base = normalized.split('/')[0] if '/' in normalized else normalized + if self._ensure_markets_loaded(): + valid_symbol = self._find_valid_symbol(base) + if valid_symbol and valid_symbol != normalized: + try: + logger.debug(f"Trying alternative symbol: {valid_symbol} (original: {symbol}, first attempt: {normalized})") + ticker = self.exchange.fetch_ticker(valid_symbol) + if ticker and isinstance(ticker, dict): + return ticker + except Exception as e2: + logger.debug(f"Alternative symbol {valid_symbol} also failed: {e2}") + + # 如果所有尝试都失败，记录警告并返回默认值 + logger.warning( + f"Symbol '{symbol}' (normalized: {normalized}) not found on {self.exchange.id}. " + f"Error: {str(e)[:100]}" + ) + + return {'last': 0, 'symbol': symbol} def get_kline( self, @@ -78,11 +243,12 @@ class CryptoDataSource(BaseDataSource): try: ccxt_timeframe = self.TIMEFRAME_MAP.get(timeframe, '1d') - # 构建交易对符号 - if not symbol.endswith('USDT') and not symbol.endswith('USD'): - symbol_pair = f'{symbol}/USDT' - else: - symbol_pair = symbol + # 使用统一的符号规范化方法 + symbol_pair = self._normalize_symbol_for_exchange(symbol) + + if not symbol_pair: + logger.warning(f"Failed to normalize symbol for K-line: {symbol}") + return [] # logger.info(f"获取加密货币K线: {symbol_pair}, 周期: {ccxt_timeframe}, 条数: {limit}") diff --git a/backend_api_python/app/services/fast_analysis.py b/backend_api_python/app/services/fast_analysis.py index 1633abb..fae42a3 100644 --- a/backend_api_python/app/services/fast_analysis.py +++ b/backend_api_python/app/services/fast_analysis.py @@ -281,11 +281,53 @@ class FastAnalysisService: else: price_lower_bound = price_upper_bound = entry_range_low = entry_range_high = 0 + # Get technical indicator values for decision constraints + rsi_value = indicators.get("rsi", {}).get("value", 50) + macd_signal = indicators.get("macd", {}).get("signal", "neutral") + ma_trend = indicators.get("moving_averages", {}).get("trend", "sideways") + + # Build decision guidance based on technical indicators + decision_guidance = self._build_decision_guidance(rsi_value, macd_signal, ma_trend, change_24h) + system_prompt = f"""You are QuantDinger's Senior Financial Analyst with 20+ years of experience. -Provide professional, detailed analysis like a Wall Street analyst report. +You are CONSERVATIVE and OBJECTIVE. Your analysis must be based on DATA, not speculation. {lang_instruction} +🎯 CRITICAL DECISION RULES (MUST FOLLOW): +1. **Market Context**: This market supports BOTH long (BUY) and short (SELL) positions. SELL signals are VALID trading opportunities, not just risk warnings. +2. **Multi-Factor Analysis** (IMPORTANT - Consider ALL factors): + - **Technical Indicators** (RSI, MACD, MA trends): Provide baseline direction + - **Macro Environment** (DXY, VIX, interest rates, geopolitical events): Can override technical signals + - **Breaking News & Events**: Major news can cause sudden reversals - pay attention! + - **Fundamental Data**: Valuation, growth, financial health matter for medium/long-term + - **Market Sentiment**: News sentiment, fear/greed index, market mood +3. **Decision Priority** (When factors conflict): + - **Major macro events** (war, policy changes, major economic data) > Technical indicators + - **Breaking news** (regulatory changes, major partnerships, scandals) > Short-term technical + - **Technical indicators** > General news sentiment (when no major events) + - **Fundamental data** > Short-term price movements (for long-term decisions) +4. **Balance Your Decisions** (IMPORTANT - Give SELL signals when appropriate): + - BUY: When technical indicators show oversold (RSI < 40), bullish MACD, uptrend, OR strong macro/fundamental catalyst + - SELL: When technical indicators show overbought (RSI > 60), bearish MACD, downtrend, OR major negative macro/news event + - HOLD: Only when signals are truly mixed or unclear - DO NOT default to HOLD just because you're uncertain + - **Remember**: SELL is a valid trading signal for short positions, not just a warning to avoid buying +5. **Confidence Thresholds**: + - BUY requires confidence >= 60 AND (technical support OR macro/fundamental catalyst) + - SELL requires confidence >= 60 AND (technical support OR negative event) - SELL signals are encouraged when indicators suggest downside + - HOLD only when confidence < 60 AND signals are truly unclear +6. **Identify Trading Opportunities**: + - When RSI > 60, MACD bearish, downtrend: Consider SELL (short position opportunity) + - When RSI < 40, MACD bullish, uptrend: Consider BUY (long position opportunity) + - Do NOT default to HOLD when clear technical signals exist +7. **Consider Macro Impact**: + - Strong USD (DXY ↑) usually negative for crypto/commodities → Consider SELL + - High VIX (>30) indicates fear → Consider SELL or HOLD, avoid BUY + - Rising interest rates usually negative for growth assets → Consider SELL + - Geopolitical tensions can cause sudden volatility → Consider SELL if risk-off sentiment + +{decision_guidance} + 📐 TECHNICAL LEVELS (Pre-calculated from chart data): - Support: ${support} | Resistance: ${resistance} | Pivot: ${pivot} - ATR (14-day): ${atr:.4f} ({volatility.get('pct', 0)}% volatility) @@ -300,22 +342,42 @@ Provide professional, detailed analysis like a Wall Street analyst report. 4. Entry price: ${entry_range_low:.4f} ~ ${entry_range_high:.4f} 5. These levels are based on ATR and support/resistance analysis - use them as reference! -📊 YOUR ANALYSIS MUST INCLUDE: -1. **Technical Analysis**: Interpret the indicators, explain why support/resistance levels matter -2. **Fundamental Analysis**: Evaluate valuation, growth if data available -3. **Sentiment Analysis**: Assess market mood, news impact, macro factors -4. **Risk Assessment**: Explain why the stop loss level is appropriate -5. **Clear Recommendation**: BUY/SELL/HOLD with entry, stop loss (near suggested), take profit (near suggested) +📊 YOUR ANALYSIS MUST INCLUDE (ALL factors are important): +1. **Technical Analysis**: Objectively interpret RSI, MACD, MA, support/resistance. Be honest about conflicting signals. +2. **Macro Environment Analysis**: + - Analyze DXY, VIX, interest rates impact on the asset + - Consider geopolitical events and their potential impact + - Evaluate how macro trends affect this specific market/symbol +3. **News & Event Analysis**: + - 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**: + - 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) + - **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**: + - 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 Output ONLY valid JSON (do NOT include word counts or format hints in your actual response): {{ "decision": "BUY" | "SELL" | "HOLD", "confidence": 0-100, - "summary": "Executive summary in 2-3 sentences", + "summary": "Executive summary in 2-3 sentences - be honest about uncertainty if present", "analysis": {{ - "technical": "Your detailed technical analysis here - interpret RSI, MACD, MA, support/resistance", - "fundamental": "Your fundamental assessment here - valuation, growth, competitive position", - "sentiment": "Your market sentiment analysis here - news impact, macro factors, mood" + "technical": "Your detailed technical analysis here - interpret RSI, MACD, MA, support/resistance objectively", + "fundamental": "Your fundamental assessment here - valuation, growth, competitive position. If data is limited, state that clearly.", + "sentiment": "Your market sentiment analysis here - news impact, macro factors, mood. Don't overreact." }}, "entry_price": number, "stop_loss": number, @@ -329,7 +391,20 @@ Output ONLY valid JSON (do NOT include word counts or format hints in your actua "sentiment_score": 0-100 }} -⚠️ IMPORTANT: The analysis fields should contain your ACTUAL analysis text, NOT the format description above.""" +⚠️ IMPORTANT: +- The analysis fields should contain your ACTUAL analysis text, NOT the format description above. +- Be HONEST and CONSERVATIVE. If you're not confident, choose HOLD with lower confidence. +- 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 +- 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.""" # Format indicator data for prompt (ensure safe defaults) rsi_data = indicators.get("rsi") or {} @@ -372,12 +447,124 @@ Output ONLY valid JSON (do NOT include word counts or format hints in your actua - Market Cap: {fundamental.get('market_cap', 'N/A')} - 52W High/Low: {fundamental.get('52w_high', 'N/A')} / {fundamental.get('52w_low', 'N/A')} - ROE: {fundamental.get('roe', 'N/A')} +- Revenue Growth: {fundamental.get('revenue_growth', 'N/A')} +- Profit Margin: {fundamental.get('profit_margin', 'N/A')} +- Debt to Equity: {fundamental.get('debt_to_equity', 'N/A')} +- Current Ratio: {fundamental.get('current_ratio', 'N/A')} +- Free Cash Flow: {fundamental.get('free_cash_flow', 'N/A')} -IMPORTANT: Consider the macro environment (especially DXY, VIX, rates) when making your recommendation. -Provide your analysis now. Remember: all prices must be within 10% of ${current_price}.""" +📊 FINANCIAL STATEMENTS (Latest Quarter): +{self._format_financial_statements(fundamental.get('financial_statements', {}))} + +📈 EARNINGS DATA: +{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}.""" return system_prompt, user_prompt + def _format_financial_statements(self, statements: Dict[str, Any]) -> str: + """格式化财务报表数据用于提示词""" + if not statements: + return "财务报表数据暂不可用" + + lines = [] + + # 资产负债表 + if 'balance_sheet' in statements: + bs = statements['balance_sheet'] + lines.append("资产负债表 (Balance Sheet):") + if bs.get('total_assets'): + lines.append(f" - 总资产: ${bs['total_assets']:,.0f}") + if bs.get('total_liabilities'): + lines.append(f" - 总负债: ${bs['total_liabilities']:,.0f}") + if bs.get('total_equity'): + lines.append(f" - 股东权益: ${bs['total_equity']:,.0f}") + if bs.get('cash'): + lines.append(f" - 现金: ${bs['cash']:,.0f}") + if bs.get('debt'): + lines.append(f" - 总债务: ${bs['debt']:,.0f}") + if bs.get('current_assets') and bs.get('current_liabilities'): + current_ratio = bs['current_assets'] / bs['current_liabilities'] if bs['current_liabilities'] > 0 else 0 + lines.append(f" - 流动比率: {current_ratio:.2f}") + + # 利润表 + if 'income_statement' in statements: + is_stmt = statements['income_statement'] + lines.append("利润表 (Income Statement):") + if is_stmt.get('total_revenue'): + lines.append(f" - 总收入: ${is_stmt['total_revenue']:,.0f}") + if is_stmt.get('gross_profit'): + lines.append(f" - 毛利润: ${is_stmt['gross_profit']:,.0f}") + if is_stmt.get('operating_income'): + lines.append(f" - 营业利润: ${is_stmt['operating_income']:,.0f}") + if is_stmt.get('net_income'): + lines.append(f" - 净利润: ${is_stmt['net_income']:,.0f}") + if is_stmt.get('eps'): + lines.append(f" - 每股收益: ${is_stmt['eps']:.2f}") + + # 现金流量表 + if 'cash_flow' in statements: + cf = statements['cash_flow'] + lines.append("现金流量表 (Cash Flow):") + if cf.get('operating_cash_flow'): + lines.append(f" - 经营现金流: ${cf['operating_cash_flow']:,.0f}") + if cf.get('free_cash_flow'): + lines.append(f" - 自由现金流: ${cf['free_cash_flow']:,.0f}") + + return "\n".join(lines) if lines else "财务报表数据暂不可用" + + def _format_earnings_data(self, earnings: Dict[str, Any]) -> str: + """格式化盈利数据用于提示词""" + if not earnings: + return "盈利数据暂不可用" + + lines = [] + + # 历史盈利 + if 'history' in earnings and earnings['history']: + lines.append("历史盈利 (Earnings History):") + for i, hist in enumerate(earnings['history'][:4], 1): + date = hist.get('date', 'N/A') + eps_actual = hist.get('eps_actual') + eps_estimate = hist.get('eps_estimate') + surprise = hist.get('surprise') + + if eps_actual is not None: + line = f" {i}. {date}: EPS实际={eps_actual:.2f}" + if eps_estimate is not None: + line += f", 预期={eps_estimate:.2f}" + if surprise is not None: + surprise_str = f"{surprise:+.1f}%" + line += f", 超预期={surprise_str}" + lines.append(line) + + # 未来盈利 + if 'upcoming' in earnings: + upcoming = earnings['upcoming'] + if upcoming.get('next_earnings_date'): + lines.append(f"下次盈利报告: {upcoming['next_earnings_date']}") + if upcoming.get('eps_estimate'): + lines.append(f" - EPS预期: ${upcoming['eps_estimate']:.2f}") + if upcoming.get('revenue_estimate'): + lines.append(f" - 收入预期: ${upcoming['revenue_estimate']:,.0f}") + + # 季度盈利 + if 'quarterly' in earnings: + q = earnings['quarterly'] + if q.get('latest_quarter'): + lines.append(f"最新季度 ({q['latest_quarter']}):") + if q.get('revenue'): + lines.append(f" - 收入: ${q['revenue']:,.0f}") + if q.get('earnings'): + lines.append(f" - 盈利: ${q['earnings']:,.0f}") + + return "\n".join(lines) if lines else "盈利数据暂不可用" + def _format_macro_summary(self, macro: Dict[str, Any], market: str) -> str: """格式化宏观数据摘要""" if not macro: @@ -546,8 +733,50 @@ Provide your analysis now. Remember: all prices must be within 10% of ${current_ llm_time = int((time.time() - llm_start) * 1000) logger.info(f"LLM call completed in {llm_time}ms") - # Phase 4: Validate and constrain output - analysis = self._validate_and_constrain(analysis, current_price) + # Phase 4: Calculate objective score and determine decision based on score + objective_score = self._calculate_objective_score(data, current_price) + logger.info(f"Objective score calculated: {objective_score['overall_score']:.1f} (Technical: {objective_score['technical_score']:.1f}, Fundamental: {objective_score['fundamental_score']:.1f}, Sentiment: {objective_score['sentiment_score']:.1f}, Macro: {objective_score['macro_score']:.1f})") + + # Determine decision based on objective score thresholds + score_based_decision = self._score_to_decision(objective_score['overall_score']) + logger.info(f"Score-based decision: {score_based_decision} (score: {objective_score['overall_score']:.1f})") + + # Override LLM decision with score-based decision if they differ significantly + 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） + 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 + # 评分越高，置信度越高（最高95，最低60） + analysis["confidence"] = min(95, max(60, int(50 + score_abs * 0.45))) + # Update summary to mention score-based decision + original_summary = analysis.get("summary", "") + score_level = "强烈" if score_abs >= 70 else "明显" if score_abs >= 40 else "轻微" + analysis["summary"] = f"{original_summary} [基于客观评分系统：综合评分{objective_score['overall_score']:.1f}分（{score_level}{'利多' if objective_score['overall_score'] > 0 else '利空'}），建议{score_based_decision}]" + else: + logger.info(f"LLM decision '{llm_decision}' differs from score-based '{score_based_decision}' but score is close to neutral ({objective_score['overall_score']:.1f}), keeping LLM decision") + + # Add objective scores to analysis + analysis["objective_score"] = objective_score + analysis["score_based_decision"] = score_based_decision + + # Phase 5: Validate and constrain output (pass indicators for decision validation) + # Check for major news or macro events that could override technical indicators + news_data = data.get("news") or [] + macro_data = data.get("macro") or {} + has_major_news = self._has_major_news(news_data) + has_macro_event = self._has_macro_event(macro_data, data.get("market", "")) + + analysis = self._validate_and_constrain( + analysis, + current_price, + indicators=data.get("indicators"), + has_major_news=has_major_news, + has_macro_event=has_macro_event + ) # Build final result total_time = int((time.time() - start_time) * 1000) @@ -582,6 +811,8 @@ Provide your analysis now. Remember: all prices must be within 10% of ${current_ "sentiment": analysis.get("sentiment_score", 50), "overall": self._calculate_overall_score(analysis), }, + "objective_score": analysis.get("objective_score", {}), + "score_based_decision": analysis.get("score_based_decision", "HOLD"), "market_data": { "current_price": current_price, "change_24h": data["price"].get("changePercent", 0), @@ -607,10 +838,134 @@ Provide your analysis now. Remember: all prices must be within 10% of ${current_ return result - def _validate_and_constrain(self, analysis: Dict, current_price: float) -> Dict: + def _build_decision_guidance(self, rsi_value: float, macd_signal: str, ma_trend: str, change_24h: float) -> str: + """ + 根据技术指标构建决策指导，帮助AI做出更合理的决策。 + 强调SELL信号是有效的做空机会。 + """ + guidance_parts = [] + + # RSI 指导 - 更积极地识别做空机会 + if rsi_value > 70: + guidance_parts.append("🔴 RSI > 70 (超买): 强烈建议SELL做空，避免BUY") + elif rsi_value > 60: + guidance_parts.append("🟠 RSI > 60 (偏超买): 建议SELL做空，谨慎BUY") + elif rsi_value < 30: + guidance_parts.append("🟢 RSI < 30 (超卖): 建议BUY做多，避免SELL") + elif rsi_value < 40: + guidance_parts.append("🟡 RSI < 40 (偏超卖): 可以考虑BUY做多") + else: + guidance_parts.append("⚪ RSI 40-60 (中性): 技术面中性，需要结合其他指标判断") + + # MACD 指导 - 明确做空信号 + if macd_signal == "bullish": + guidance_parts.append("🟢 MACD 看涨: 支持BUY做多") + elif macd_signal == "bearish": + guidance_parts.append("🔴 MACD 看跌: 支持SELL做空，这是有效的做空机会") + else: + guidance_parts.append("⚪ MACD 中性: 无明显方向") + + # MA 趋势指导 - 识别趋势反转机会 + if "uptrend" in ma_trend.lower() or "strong_uptrend" in ma_trend.lower(): + if rsi_value > 60: + guidance_parts.append("⚠️ 均线向上但RSI超买: 可能接近顶部，考虑SELL做空") + else: + guidance_parts.append("🟢 均线趋势向上: 支持BUY做多") + elif "downtrend" in ma_trend.lower() or "strong_downtrend" in ma_trend.lower(): + guidance_parts.append("🔴 均线趋势向下: 这是SELL做空的良好机会，避免BUY") + else: + guidance_parts.append("⚪ 均线横盘: 趋势不明确") + + # 24小时涨跌幅指导 - 识别过度波动 + if change_24h > 5: + guidance_parts.append("🔴 24h涨幅 > 5%: 可能已过度上涨，建议SELL做空或获利了结") + elif change_24h < -5: + guidance_parts.append("🟢 24h跌幅 > 5%: 可能已过度下跌，可以考虑BUY做多") + + # 综合建议 + sell_signals = sum([ + rsi_value > 60, + macd_signal == "bearish", + "downtrend" in ma_trend.lower(), + change_24h > 5 + ]) + buy_signals = sum([ + rsi_value < 40, + macd_signal == "bullish", + "uptrend" in ma_trend.lower(), + change_24h < -5 + ]) + + if sell_signals >= 2: + guidance_parts.append(f"📊 综合判断: {sell_signals}个做空信号，建议考虑SELL") + elif buy_signals >= 2: + guidance_parts.append(f"📊 综合判断: {buy_signals}个做多信号，建议考虑BUY") + else: + guidance_parts.append("📊 综合判断: 信号混合，需要结合宏观和新闻判断") + + return "\n".join(guidance_parts) if guidance_parts else "技术指标数据不足，请谨慎判断" + + 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", + "监管", "禁令", "批准", "合作", "合并", "收购", "丑闻", "诉讼", "调查", "政策", "政府", "央行" + ] + + for news in news_data[:5]: # 只检查前5条最新新闻 + title = (news.get("title") or news.get("headline") or "").lower() + sentiment = news.get("sentiment", "neutral") + + # 如果有重大关键词且情绪强烈（非中性），认为是重大新闻 + if any(keyword in title for keyword in major_keywords) and sentiment != "neutral": + return True + + return False + + def _has_macro_event(self, macro_data: Dict, market: str) -> bool: + """ + 检查是否有重大宏观事件。 + 重大宏观事件包括：VIX异常高、DXY大幅波动、利率政策变化等。 + """ + if not macro_data: + return False + + # 检查VIX（恐慌指数） + if "VIX" in macro_data: + vix = macro_data["VIX"] + vix_value = vix.get("price", 0) + if vix_value > 30: # VIX > 30 表示极度恐慌 + return True + + # 检查DXY大幅波动（>1%） + if "DXY" in macro_data: + dxy = macro_data["DXY"] + change_pct = abs(dxy.get("changePercent", 0)) + if change_pct > 1.0: # 美元指数波动超过1% + return True + + # 检查利率变化（对股票和加密货币影响大） + if "TNX" in macro_data and market in ["USStock", "Crypto"]: + tnx = macro_data["TNX"] + change_pct = abs(tnx.get("changePercent", 0)) + if change_pct > 2.0: # 利率变化超过2% + return True + + return False + + def _validate_and_constrain(self, analysis: Dict, current_price: float, indicators: Dict = None, + has_major_news: bool = False, has_macro_event: bool = False) -> Dict: """ Validate LLM output and constrain prices to reasonable ranges. - This prevents absurd recommendations like "BTC at 95000, buy at 75000". + Also validate decision against technical indicators to prevent absurd recommendations. """ if not current_price or current_price <= 0: return analysis @@ -651,10 +1006,447 @@ Provide your analysis now. Remember: all prices must be within 10% of ${current_ else: analysis["decision"] = decision + # 基于技术指标验证决策合理性（允许宏观/新闻因素覆盖） + if indicators: + analysis = self._validate_decision_against_indicators( + analysis, indicators, confidence, + has_major_news=has_major_news, + has_macro_event=has_macro_event + ) + return analysis + def _validate_decision_against_indicators(self, analysis: Dict, indicators: Dict, confidence: int, + has_major_news: bool = False, has_macro_event: bool = False) -> Dict: + """ + 根据技术指标验证决策的合理性，但允许宏观/新闻因素覆盖技术指标。 + + Args: + analysis: AI分析结果 + indicators: 技术指标数据 + confidence: 置信度 + has_major_news: 是否有重大新闻事件 + has_macro_event: 是否有重大宏观事件 + """ + decision = analysis.get("decision", "HOLD") + rsi_data = indicators.get("rsi", {}) + macd_data = indicators.get("macd", {}) + ma_data = indicators.get("moving_averages", {}) + + rsi_value = rsi_data.get("value", 50) + macd_signal = macd_data.get("signal", "neutral") + ma_trend = ma_data.get("trend", "sideways") + + # 如果置信度太低，强制改为HOLD + if confidence < 60: + if decision != "HOLD": + logger.warning(f"Decision {decision} with low confidence {confidence}, forcing to HOLD") + analysis["decision"] = "HOLD" + analysis["confidence"] = max(confidence, 45) # 降低置信度 + return analysis + + # 如果有重大新闻或宏观事件，允许覆盖技术指标（但记录警告） + allow_override = has_major_news or has_macro_event + + # 检查BUY决策是否与技术指标矛盾 + if decision == "BUY": + conflicts = [] + + # RSI > 70 时不应该BUY（除非有重大利好） + if rsi_value > 70: + conflicts.append(f"RSI {rsi_value:.1f} > 70 (超买)") + + # MACD看跌时不应该BUY（除非有重大利好） + if macd_signal == "bearish": + conflicts.append("MACD bearish") + + # 均线趋势向下时不应该BUY（除非有重大利好） + if "downtrend" in ma_trend.lower(): + conflicts.append(f"MA trend: {ma_trend}") + + if conflicts: + if allow_override: + # 允许覆盖，但降低置信度并添加说明 + logger.info(f"BUY decision conflicts with indicators but major news/macro event allows override: {', '.join(conflicts)}") + analysis["confidence"] = max(confidence - 15, 50) + original_summary = analysis.get("summary", "") + analysis["summary"] = f"{original_summary} [注意：技术指标显示{', '.join(conflicts)}，但重大事件可能改变趋势]" + else: + # 没有重大事件，强制改为HOLD + logger.warning(f"BUY decision conflicts with indicators and no major event: {', '.join(conflicts)}. Forcing to HOLD") + analysis["decision"] = "HOLD" + analysis["confidence"] = max(confidence - 20, 40) + original_summary = analysis.get("summary", "") + analysis["summary"] = f"{original_summary} [注意：技术指标显示{', '.join(conflicts)}，建议观望]" + + # 检查SELL决策是否与技术指标矛盾（放宽限制，因为SELL是有效的做空机会） + elif decision == "SELL": + conflicts = [] + + # 只有在强烈看涨信号时才阻止SELL（放宽条件） + # RSI < 30 且 MACD看涨 且 均线向上时，才认为矛盾 + if rsi_value < 30 and macd_signal == "bullish" and "uptrend" in ma_trend.lower(): + conflicts.append(f"Strong bullish signals (RSI {rsi_value:.1f} < 30, MACD bullish, uptrend)") + # 或者 RSI < 30 且 均线强烈向上 + elif rsi_value < 30 and "strong_uptrend" in ma_trend.lower(): + conflicts.append(f"Very strong uptrend with oversold RSI {rsi_value:.1f}") + + if conflicts: + if allow_override: + # 允许覆盖，但降低置信度并添加说明 + logger.info(f"SELL decision conflicts with strong bullish indicators but major news/macro event allows override: {', '.join(conflicts)}") + analysis["confidence"] = max(confidence - 15, 50) + original_summary = analysis.get("summary", "") + analysis["summary"] = f"{original_summary} [注意：技术指标显示{', '.join(conflicts)}，但重大事件可能改变趋势]" + else: + # 只有在非常强烈的看涨信号时才改为HOLD + logger.warning(f"SELL decision conflicts with very strong bullish indicators: {', '.join(conflicts)}. Forcing to HOLD") + analysis["decision"] = "HOLD" + analysis["confidence"] = max(confidence - 20, 40) + original_summary = analysis.get("summary", "") + analysis["summary"] = f"{original_summary} [注意：技术指标显示{', '.join(conflicts)}，建议观望]" + + return analysis + + def _calculate_objective_score(self, data: Dict[str, Any], current_price: float) -> Dict[str, float]: + """ + 基于客观数据计算量化评分系统 + + 返回一个-100到+100的分数： + - +100: 强烈利多（强烈BUY） + - +70到+100: 强烈利多（强烈BUY） + - +40到+70: 利多（BUY） + - -40到+40: 中性（HOLD） + - -70到-40: 利空（SELL） + - -100到-70: 强烈利空（强烈SELL） + - -100: 强烈利空（强烈SELL） + """ + indicators = data.get("indicators") or {} + fundamental = data.get("fundamental") or {} + news = data.get("news") or [] + macro = data.get("macro") or {} + price_data = data.get("price") or {} + + # 1. 技术指标评分 (-100 to +100) + technical_score = self._calculate_technical_score(indicators, price_data) + + # 2. 基本面评分 (-100 to +100) + fundamental_score = self._calculate_fundamental_score(fundamental, data.get("market", "")) + + # 3. 新闻情绪评分 (-100 to +100) + sentiment_score = self._calculate_sentiment_score(news) + + # 4. 宏观环境评分 (-100 to +100) + macro_score = self._calculate_macro_score(macro, data.get("market", "")) + + # 5. 综合评分（加权平均） + # 权重：技术40%，基本面25%，情绪20%，宏观15% + overall_score = ( + technical_score * 0.40 + + fundamental_score * 0.25 + + sentiment_score * 0.20 + + macro_score * 0.15 + ) + + return { + "technical_score": technical_score, + "fundamental_score": fundamental_score, + "sentiment_score": sentiment_score, + "macro_score": macro_score, + "overall_score": overall_score + } + + def _calculate_technical_score(self, indicators: Dict, price_data: Dict) -> float: + """计算技术指标评分 (-100 to +100)""" + score = 0.0 + weight_sum = 0.0 + + # RSI 评分 (-50 to +50) + rsi_data = indicators.get("rsi", {}) + rsi_value = rsi_data.get("value", 50) + if rsi_value > 0: + if rsi_value > 70: + rsi_score = -50 # 超买，强烈利空 + elif rsi_value > 60: + rsi_score = -30 # 偏超买，利空 + elif rsi_value < 30: + rsi_score = +50 # 超卖，强烈利多 + elif rsi_value < 40: + rsi_score = +30 # 偏超卖，利多 + else: + rsi_score = (50 - rsi_value) * 0.6 # 40-60之间，线性映射 + score += rsi_score * 0.30 + weight_sum += 0.30 + + # MACD 评分 (-40 to +40) + macd_data = indicators.get("macd", {}) + macd_signal = macd_data.get("signal", "neutral") + if macd_signal == "bullish": + macd_score = +40 + elif macd_signal == "bearish": + macd_score = -40 + else: + macd_score = 0 + score += macd_score * 0.25 + weight_sum += 0.25 + + # 均线趋势评分 (-40 to +40) + ma_data = indicators.get("moving_averages", {}) + ma_trend = ma_data.get("trend", "sideways") + if "strong_uptrend" in ma_trend.lower(): + ma_score = +40 + elif "uptrend" in ma_trend.lower(): + ma_score = +25 + elif "strong_downtrend" in ma_trend.lower(): + ma_score = -40 + elif "downtrend" in ma_trend.lower(): + ma_score = -25 + else: + ma_score = 0 + score += ma_score * 0.25 + weight_sum += 0.25 + + # 24小时涨跌幅评分 (-20 to +20) + change_24h = price_data.get("changePercent", 0) + if change_24h > 10: + change_score = -20 # 过度上涨，利空 + elif change_24h > 5: + change_score = -10 + elif change_24h < -10: + change_score = +20 # 过度下跌，利多 + elif change_24h < -5: + change_score = +10 + else: + change_score = change_24h * 2 # 线性映射 + score += change_score * 0.20 + weight_sum += 0.20 + + # 归一化到-100到+100 + if weight_sum > 0: + score = score / weight_sum * 100 + + return max(-100, min(100, score)) + + def _calculate_fundamental_score(self, fundamental: Dict, market: str) -> float: + """计算基本面评分 (-100 to +100)""" + if market != "USStock" or not fundamental: + return 0.0 # 非美股或无基本面数据，返回中性 + + score = 0.0 + factors = 0 + + # PE Ratio 评分 + pe_ratio = fundamental.get("pe_ratio") + if pe_ratio and pe_ratio > 0: + if pe_ratio < 15: + pe_score = +20 # 低PE，利多 + elif pe_ratio < 25: + pe_score = +10 + elif pe_ratio > 50: + pe_score = -20 # 高PE，利空 + elif pe_ratio > 35: + pe_score = -10 + else: + pe_score = 0 + score += pe_score + factors += 1 + + # ROE 评分 + roe = fundamental.get("roe") + if roe: + if roe > 20: + roe_score = +20 # 高ROE，利多 + elif roe > 15: + roe_score = +10 + elif roe < 5: + roe_score = -20 # 低ROE，利空 + elif roe < 10: + roe_score = -10 + else: + roe_score = 0 + score += roe_score + factors += 1 + + # 营收增长评分 + revenue_growth = fundamental.get("revenue_growth") + if revenue_growth: + if revenue_growth > 20: + growth_score = +20 # 高增长，利多 + elif revenue_growth > 10: + growth_score = +10 + elif revenue_growth < -10: + growth_score = -20 # 负增长，利空 + elif revenue_growth < 0: + growth_score = -10 + else: + growth_score = 0 + score += growth_score + factors += 1 + + # 利润率评分 + profit_margin = fundamental.get("profit_margin") + if profit_margin: + if profit_margin > 20: + margin_score = +15 # 高利润率，利多 + elif profit_margin > 10: + margin_score = +7 + elif profit_margin < 0: + margin_score = -15 # 亏损，利空 + elif profit_margin < 5: + margin_score = -7 + else: + margin_score = 0 + score += margin_score + factors += 1 + + # 债务权益比评分 + debt_to_equity = fundamental.get("debt_to_equity") + if debt_to_equity: + if debt_to_equity < 0.5: + debt_score = +10 # 低负债，利多 + elif debt_to_equity > 2.0: + debt_score = -10 # 高负债，利空 + else: + debt_score = 0 + score += debt_score + factors += 1 + + # 归一化（如果有多个因素） + if factors > 0: + score = score / factors * 100 / 4 # 最大可能分数是4个因素各20分=80，归一化到100 + + return max(-100, min(100, score)) + + def _calculate_sentiment_score(self, news: List[Dict]) -> float: + """计算新闻情绪评分 (-100 to +100)""" + if not news: + return 0.0 # 无新闻，中性 + + positive_count = 0 + negative_count = 0 + neutral_count = 0 + + for item in news[:10]: # 只看前10条 + sentiment = item.get("sentiment", "neutral") + if sentiment == "positive": + positive_count += 1 + elif sentiment == "negative": + negative_count += 1 + else: + neutral_count += 1 + + total = positive_count + negative_count + neutral_count + if total == 0: + return 0.0 + + # 计算净情绪 + net_sentiment = (positive_count - negative_count) / total + + # 映射到-100到+100 + score = net_sentiment * 100 + + return max(-100, min(100, score)) + + def _calculate_macro_score(self, macro: Dict, market: str) -> float: + """计算宏观环境评分 (-100 to +100)""" + if not macro: + return 0.0 # 无宏观数据，中性 + + score = 0.0 + factors = 0 + + # VIX 评分（恐慌指数） + vix = macro.get("VIX", {}) + vix_value = vix.get("price", 0) + if vix_value > 0: + if vix_value > 30: + vix_score = -30 # 高恐慌，利空 + elif vix_value > 20: + vix_score = -15 + elif vix_value < 15: + vix_score = +15 # 低恐慌，利多 + else: + vix_score = 0 + score += vix_score + factors += 1 + + # 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: + dxy_score = -20 # 美元走强，利空 + elif dxy_change < -1: + dxy_score = +20 # 美元走弱，利多 + else: + dxy_score = 0 + else: + dxy_score = 0 # 对股票影响较小 + score += dxy_score + factors += 1 + + # 利率评分（TNX） + tnx = macro.get("TNX", {}) + tnx_change = tnx.get("changePercent", 0) + if tnx_change != 0: + # 利率上升对成长股和加密货币通常是利空 + if market in ["Crypto", "USStock"]: + if tnx_change > 2: + tnx_score = -20 # 利率大幅上升，利空 + elif tnx_change < -2: + tnx_score = +20 # 利率下降，利多 + else: + tnx_score = 0 + else: + tnx_score = 0 + score += tnx_score + factors += 1 + + # 归一化 + if factors > 0: + score = score / factors * 100 / 3 # 最大可能分数是3个因素各30分=90，归一化到100 + + return max(-100, min(100, score)) + + def _score_to_decision(self, score: float) -> str: + """ + 根据客观评分转换为决策 + + 优化后的阈值（缩小HOLD区间，使决策更明确）： + - score >= +40: BUY（利多） + - score <= -40: SELL（利空） + - -40 < score < +40: 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 + - score <= -70: 强烈SELL + """ + # 使用±40作为主要阈值，缩小HOLD区间 + if score >= 40: + return "BUY" + elif score <= -40: + return "SELL" + else: + return "HOLD" + def _calculate_overall_score(self, analysis: Dict) -> int: - """Calculate weighted overall score.""" + """Calculate weighted overall score (legacy method, now uses objective score if available).""" + # 优先使用客观评分 + if "objective_score" in analysis: + objective = analysis["objective_score"] + overall = objective.get("overall_score", 50) + # 转换为0-100格式（原系统使用） + return max(0, min(100, int(50 + overall * 0.5))) + + # 降级到LLM评分 tech = analysis.get("technical_score", 50) fund = analysis.get("fundamental_score", 50) sent = analysis.get("sentiment_score", 50) diff --git a/backend_api_python/app/services/live_trading/execution.py b/backend_api_python/app/services/live_trading/execution.py index 6f3227d..fb24a5a 100644 --- a/backend_api_python/app/services/live_trading/execution.py +++ b/backend_api_python/app/services/live_trading/execution.py @@ -36,6 +36,50 @@ IBKRClient = None MT5Client = None +def _normalize_symbol_for_order(symbol: str, market_type: str = "swap") -> str: + """ + 规范化符号格式，确保符号符合交易所要求。 + + 处理各种输入格式： + - BTC/USDT -> BTC/USDT + - BTCUSDT -> BTC/USDT + - BTC/USDT:USDT -> BTC/USDT + - PI, TRX -> PI/USDT, TRX/USDT (默认添加 /USDT) + + Args: + symbol: 原始符号 + market_type: 市场类型 (spot/swap) + + Returns: + 规范化后的符号 + """ + if not symbol: + return symbol + + sym = symbol.strip() + + # 移除 swap/futures 后缀 + if ':' in sym: + sym = sym.split(':', 1)[0] + + sym = sym.upper() + + # 如果已经有分隔符，直接返回（假设格式正确） + if '/' in sym: + return sym + + # 尝试从常见报价货币中识别 + common_quotes = ['USDT', 'USD', 'BTC', 'ETH', 'BUSD', 'USDC'] + for quote in common_quotes: + if sym.endswith(quote) and len(sym) > len(quote): + base = sym[:-len(quote)] + if base: + return f"{base}/{quote}" + + # 如果无法识别，默认使用 USDT + return f"{sym}/USDT" + + def _signal_to_sides(signal_type: str) -> Tuple[str, str, bool]: """ Returns (side, pos_side, reduce_only) @@ -80,6 +124,9 @@ def place_order_from_signal( # Spot does not support short signals in this system. if mt == "spot" and ("short" in (signal_type or "").lower()): raise LiveTradingError("spot market does not support short signals") + + # 规范化符号格式（统一处理裸符号如 PI, TRX 等） + symbol = _normalize_symbol_for_order(symbol, market_type=mt) if isinstance(client, BinanceFuturesClient): return client.place_market_order( diff --git a/backend_api_python/app/services/live_trading/symbols.py b/backend_api_python/app/services/live_trading/symbols.py index 56b9497..ac6c0d2 100644 --- a/backend_api_python/app/services/live_trading/symbols.py +++ b/backend_api_python/app/services/live_trading/symbols.py @@ -14,12 +14,28 @@ from typing import Dict, Tuple def _split_base_quote(symbol: str) -> Tuple[str, str]: + """ + 分割符号为基础货币和报价货币。 + + 处理各种格式： + - BTC/USDT -> (BTC, USDT) + - BTCUSDT -> (BTCUSDT, "") - 需要进一步处理 + - PI, TRX -> (PI, "") - 需要进一步处理 + """ s = (symbol or "").strip() if ":" in s: s = s.split(":", 1)[0] if "/" not in s: - # Already exchange-specific (best-effort) - return s, "" + # 尝试识别报价货币（常见格式：BASEQUOTE） + s_upper = s.upper() + common_quotes = ['USDT', 'USD', 'BTC', 'ETH', 'BUSD', 'USDC', 'BNB'] + for quote in common_quotes: + if s_upper.endswith(quote) and len(s_upper) > len(quote): + base = s_upper[:-len(quote)] + if base: + return base, quote + # 无法识别，返回原符号和空报价 + return s_upper, "" base, quote = s.split("/", 1) return base.strip().upper(), quote.strip().upper() diff --git a/backend_api_python/app/services/llm.py b/backend_api_python/app/services/llm.py index c4911e3..6213a5e 100644 --- a/backend_api_python/app/services/llm.py +++ b/backend_api_python/app/services/llm.py @@ -313,7 +313,7 @@ class LLMService: def call_llm_api(self, messages: list, model: str = None, temperature: float = 0.7, use_fallback: bool = True, provider: LLMProvider = None, - use_json_mode: bool = True) -> str: + use_json_mode: bool = True, try_alternative_providers: bool = True) -> str: """ Call LLM API with the specified or default provider. @@ -324,6 +324,7 @@ class LLMService: use_fallback: Whether to try fallback model on failure provider: Override the service's default provider use_json_mode: Whether to request JSON output format (default True for analysis, False for code generation) + try_alternative_providers: Whether to try alternative providers when current provider fails with 403/402 Returns: Generated text content @@ -347,11 +348,22 @@ class LLMService: api_key = self.get_api_key(p) if not api_key: - raise ValueError(f"API key not configured for provider: {p.value}") + # If no API key for current provider, try to find any available provider + if try_alternative_providers: + for alt_provider in [LLMProvider.DEEPSEEK, LLMProvider.GROK, LLMProvider.OPENAI, LLMProvider.GOOGLE, LLMProvider.OPENROUTER]: + if alt_provider != p and self.get_api_key(alt_provider): + logger.warning(f"No API key for {p.value}, switching to {alt_provider.value}") + p = alt_provider + api_key = self.get_api_key(p) + break + + if not api_key: + raise ValueError(f"API key not configured for provider: {p.value}. Please configure at least one LLM provider API key.") base_url = self.get_base_url(p) # Normalize model name for the provider + original_model = model model = self._normalize_model_for_provider(model, p) config = load_addon_config() @@ -366,6 +378,7 @@ class LLMService: models_to_try.append(fallback) last_error = None + last_status_code = None for current_model in models_to_try: try: @@ -384,13 +397,25 @@ class LLMService: except requests.exceptions.HTTPError as e: error_detail = e.response.text if e.response else str(e) - logger.error(f"{p.value} API HTTP error ({current_model}): {error_detail}") + status_code = e.response.status_code if e.response else None + last_status_code = status_code + + logger.error(f"{p.value} API HTTP error ({current_model}): {status_code} - {error_detail}") last_error = str(e) + # 403/402 errors usually mean API key issue - try alternative provider + if status_code in (402, 403) and try_alternative_providers and current_model == models_to_try[-1]: + # Only try alternative providers after all models in current provider failed + logger.warning(f"{p.value} returned {status_code} (likely API key issue). Trying alternative providers...") + return self._try_alternative_providers( + messages, original_model, temperature, + use_json_mode, excluded_provider=p + ) + # Check for recoverable errors - try fallback model # 402: Payment required, 403: Forbidden (invalid key), 404: Model not found, 429: Rate limit - if e.response and e.response.status_code in (402, 403, 404, 429): - logger.warning(f"{p.value} returned {e.response.status_code} for model {current_model}; trying fallback...") + if status_code in (402, 403, 404, 429): + logger.warning(f"{p.value} returned {status_code} for model {current_model}; trying fallback...") continue if not use_fallback or current_model == models_to_try[-1]: @@ -408,9 +433,50 @@ class LLMService: if current_model == models_to_try[-1]: raise - error_msg = f"All model calls failed. Last error: {last_error}" + error_msg = f"All model calls failed for {p.value}. Last error: {last_error}" + if last_status_code in (402, 403): + error_msg += f"\nStatus {last_status_code} usually means: API key invalid/expired, insufficient balance, or no access to model." + error_msg += f"\nPlease check your {p.value} API key configuration and account balance." + logger.error(error_msg) raise Exception(error_msg) + + def _try_alternative_providers(self, messages: list, model: str, temperature: float, + use_json_mode: bool, excluded_provider: LLMProvider = None) -> str: + """ + Try alternative providers when current provider fails. + + Priority: DeepSeek > Grok > OpenAI > Google > OpenRouter + """ + priority_order = [ + LLMProvider.DEEPSEEK, + LLMProvider.GROK, + LLMProvider.OPENAI, + LLMProvider.GOOGLE, + LLMProvider.OPENROUTER, + ] + + for alt_provider in priority_order: + if alt_provider == excluded_provider: + continue + + api_key = self.get_api_key(alt_provider) + if not api_key: + continue + + logger.info(f"Trying alternative provider: {alt_provider.value}") + try: + return self.call_llm_api( + messages, model, temperature, + use_fallback=True, provider=alt_provider, + use_json_mode=use_json_mode, + try_alternative_providers=False # Prevent infinite recursion + ) + except Exception as e: + logger.warning(f"Alternative provider {alt_provider.value} also failed: {str(e)}") + continue + + raise Exception(f"All LLM providers failed. Please check your API key configurations.") # Legacy method for backward compatibility def call_openrouter_api(self, messages: list, model: str = None, temperature: float = 0.7, use_fallback: bool = True) -> str: diff --git a/backend_api_python/app/services/market_data_collector.py b/backend_api_python/app/services/market_data_collector.py index 3b64f66..5f0b05a 100644 --- a/backend_api_python/app/services/market_data_collector.py +++ b/backend_api_python/app/services/market_data_collector.py @@ -20,6 +20,7 @@ from datetime import datetime, timedelta from concurrent.futures import ThreadPoolExecutor, as_completed, TimeoutError import yfinance as yf +import pandas as pd from app.data_sources import DataSourceFactory from app.services.kline import KlineService @@ -552,10 +553,13 @@ class MarketDataCollector: return None def _get_us_fundamental(self, symbol: str) -> Optional[Dict[str, Any]]: - """美股基本面 - Finnhub + yfinance""" + """ + 美股基本面 - Finnhub + yfinance + 包括：基础财务指标 + 财报数据（资产负债表、利润表、现金流量表） + """ result = {} - # Finnhub + # === 1. 基础财务指标 (Finnhub) === if self._finnhub_client: try: metrics = self._finnhub_client.company_basic_financials(symbol, 'all') @@ -573,31 +577,193 @@ class MarketDataCollector: 'roe': m.get('roeTTM'), 'eps': m.get('epsBasicExclExtraItemsTTM'), 'revenue_growth': m.get('revenueGrowthTTMYoy'), + 'profit_margin': m.get('netProfitMarginTTM'), + 'debt_to_equity': m.get('totalDebtToEquityQuarterly'), + 'current_ratio': m.get('currentRatioQuarterly'), + 'quick_ratio': m.get('quickRatioQuarterly'), }) except Exception as e: logger.debug(f"Finnhub fundamental failed for {symbol}: {e}") - # yfinance 补充 - if not result: - try: - ticker = yf.Ticker(symbol) - info = ticker.info or {} - result.update({ - 'pe_ratio': info.get('trailingPE') or info.get('forwardPE'), - 'pb_ratio': info.get('priceToBook'), - 'market_cap': info.get('marketCap'), - 'dividend_yield': info.get('dividendYield'), - 'beta': info.get('beta'), - '52w_high': info.get('fiftyTwoWeekHigh'), - '52w_low': info.get('fiftyTwoWeekLow'), - 'roe': info.get('returnOnEquity'), - 'eps': info.get('trailingEps'), - }) - except Exception as e: - logger.debug(f"yfinance fundamental failed for {symbol}: {e}") + # === 2. yfinance 补充基础指标 === + try: + ticker = yf.Ticker(symbol) + info = ticker.info or {} + + # 补充缺失的基础指标 + if not result.get('pe_ratio'): + result['pe_ratio'] = info.get('trailingPE') or info.get('forwardPE') + if not result.get('pb_ratio'): + result['pb_ratio'] = info.get('priceToBook') + if not result.get('market_cap'): + result['market_cap'] = info.get('marketCap') + if not result.get('dividend_yield'): + result['dividend_yield'] = info.get('dividendYield') + if not result.get('beta'): + result['beta'] = info.get('beta') + if not result.get('52w_high'): + result['52w_high'] = info.get('fiftyTwoWeekHigh') + if not result.get('52w_low'): + result['52w_low'] = info.get('fiftyTwoWeekLow') + if not result.get('roe'): + result['roe'] = info.get('returnOnEquity') + if not result.get('eps'): + result['eps'] = info.get('trailingEps') + + # 补充更多财务指标 + result.update({ + 'revenue': info.get('totalRevenue'), + 'gross_profit': info.get('grossProfits'), + 'operating_margin': info.get('operatingMargins'), + 'profit_margin': result.get('profit_margin') or info.get('profitMargins'), + 'ebitda': info.get('ebitda'), + 'debt': info.get('totalDebt'), + 'cash': info.get('totalCash'), + 'free_cash_flow': info.get('freeCashflow'), + 'operating_cash_flow': info.get('operatingCashflow'), + 'book_value': info.get('bookValue'), + 'enterprise_value': info.get('enterpriseValue'), + }) + except Exception as e: + logger.debug(f"yfinance fundamental failed for {symbol}: {e}") + + # === 3. 获取财报数据（资产负债表、利润表、现金流量表）=== + financial_statements = self._get_financial_statements(symbol) + if financial_statements: + result['financial_statements'] = financial_statements + + # === 4. 获取盈利报告（Earnings）=== + earnings_data = self._get_earnings_data(symbol) + if earnings_data: + result['earnings'] = earnings_data return result if result else None + def _get_financial_statements(self, symbol: str) -> Optional[Dict[str, Any]]: + """ + 获取财务报表数据（资产负债表、利润表、现金流量表） + + 使用 yfinance 获取，包含最近几个季度的数据 + """ + try: + ticker = yf.Ticker(symbol) + statements = {} + + # 资产负债表 (Balance Sheet) + try: + balance_sheet = ticker.balance_sheet + if balance_sheet is not None and not balance_sheet.empty: + # 获取最近4个季度 + latest_quarters = balance_sheet.columns[:4] if len(balance_sheet.columns) >= 4 else balance_sheet.columns + statements['balance_sheet'] = { + 'latest_date': str(latest_quarters[0]) if len(latest_quarters) > 0 else None, + 'total_assets': float(balance_sheet.loc['Total Assets', latest_quarters[0]]) if 'Total Assets' in balance_sheet.index and len(latest_quarters) > 0 else None, + 'total_liabilities': float(balance_sheet.loc['Total Liab', latest_quarters[0]]) if 'Total Liab' in balance_sheet.index and len(latest_quarters) > 0 else None, + 'total_equity': float(balance_sheet.loc['Stockholders Equity', latest_quarters[0]]) if 'Stockholders Equity' in balance_sheet.index and len(latest_quarters) > 0 else None, + 'cash': float(balance_sheet.loc['Cash', latest_quarters[0]]) if 'Cash' in balance_sheet.index and len(latest_quarters) > 0 else None, + 'debt': float(balance_sheet.loc['Total Debt', latest_quarters[0]]) if 'Total Debt' in balance_sheet.index and len(latest_quarters) > 0 else None, + 'current_assets': float(balance_sheet.loc['Current Assets', latest_quarters[0]]) if 'Current Assets' in balance_sheet.index and len(latest_quarters) > 0 else None, + 'current_liabilities': float(balance_sheet.loc['Current Liabilities', latest_quarters[0]]) if 'Current Liabilities' in balance_sheet.index and len(latest_quarters) > 0 else None, + } + except Exception as e: + logger.debug(f"Balance sheet fetch failed for {symbol}: {e}") + + # 利润表 (Income Statement) + try: + income_stmt = ticker.financials + if income_stmt is not None and not income_stmt.empty: + latest_quarters = income_stmt.columns[:4] if len(income_stmt.columns) >= 4 else income_stmt.columns + statements['income_statement'] = { + 'latest_date': str(latest_quarters[0]) if len(latest_quarters) > 0 else None, + 'total_revenue': float(income_stmt.loc['Total Revenue', latest_quarters[0]]) if 'Total Revenue' in income_stmt.index and len(latest_quarters) > 0 else None, + 'gross_profit': float(income_stmt.loc['Gross Profit', latest_quarters[0]]) if 'Gross Profit' in income_stmt.index and len(latest_quarters) > 0 else None, + 'operating_income': float(income_stmt.loc['Operating Income', latest_quarters[0]]) if 'Operating Income' in income_stmt.index and len(latest_quarters) > 0 else None, + 'net_income': float(income_stmt.loc['Net Income', latest_quarters[0]]) if 'Net Income' in income_stmt.index and len(latest_quarters) > 0 else None, + 'eps': float(income_stmt.loc['Basic EPS', latest_quarters[0]]) if 'Basic EPS' in income_stmt.index and len(latest_quarters) > 0 else None, + } + except Exception as e: + logger.debug(f"Income statement fetch failed for {symbol}: {e}") + + # 现金流量表 (Cash Flow Statement) + try: + cashflow = ticker.cashflow + if cashflow is not None and not cashflow.empty: + latest_quarters = cashflow.columns[:4] if len(cashflow.columns) >= 4 else cashflow.columns + statements['cash_flow'] = { + 'latest_date': str(latest_quarters[0]) if len(latest_quarters) > 0 else None, + 'operating_cash_flow': float(cashflow.loc['Operating Cash Flow', latest_quarters[0]]) if 'Operating Cash Flow' in cashflow.index and len(latest_quarters) > 0 else None, + 'investing_cash_flow': float(cashflow.loc['Capital Expenditure', latest_quarters[0]]) if 'Capital Expenditure' in cashflow.index and len(latest_quarters) > 0 else None, + 'financing_cash_flow': float(cashflow.loc['Financing Cash Flow', latest_quarters[0]]) if 'Financing Cash Flow' in cashflow.index and len(latest_quarters) > 0 else None, + 'free_cash_flow': float(cashflow.loc['Free Cash Flow', latest_quarters[0]]) if 'Free Cash Flow' in cashflow.index and len(latest_quarters) > 0 else None, + } + except Exception as e: + logger.debug(f"Cash flow statement fetch failed for {symbol}: {e}") + + return statements if statements else None + + except Exception as e: + logger.debug(f"Financial statements fetch failed for {symbol}: {e}") + return None + + def _get_earnings_data(self, symbol: str) -> Optional[Dict[str, Any]]: + """ + 获取盈利报告数据（Earnings） + + 包括：历史盈利、盈利预测、盈利日期等 + """ + try: + ticker = yf.Ticker(symbol) + earnings_data = {} + + # 历史盈利数据 + try: + earnings_history = ticker.earnings_history + if earnings_history is not None and not earnings_history.empty: + # 获取最近4个季度 + recent_earnings = earnings_history.head(4) + earnings_data['history'] = [] + for _, row in recent_earnings.iterrows(): + earnings_data['history'].append({ + 'date': str(row.get('Date', '')), + 'eps_actual': float(row.get('EPS Actual', 0)) if row.get('EPS Actual') is not None else None, + 'eps_estimate': float(row.get('EPS Estimate', 0)) if row.get('EPS Estimate') is not None else None, + 'surprise': float(row.get('Surprise(%)', 0)) if row.get('Surprise(%)') is not None else None, + }) + except Exception as e: + logger.debug(f"Earnings history fetch failed for {symbol}: {e}") + + # 盈利日历（未来盈利日期） + try: + earnings_calendar = ticker.calendar + if earnings_calendar is not None and not earnings_calendar.empty: + earnings_data['upcoming'] = { + 'next_earnings_date': str(earnings_calendar.index[0]) if len(earnings_calendar.index) > 0 else None, + 'eps_estimate': float(earnings_calendar.loc[earnings_calendar.index[0], 'Earnings Estimate']) if len(earnings_calendar.index) > 0 and 'Earnings Estimate' in earnings_calendar.columns else None, + 'revenue_estimate': float(earnings_calendar.loc[earnings_calendar.index[0], 'Revenue Estimate']) if len(earnings_calendar.index) > 0 and 'Revenue Estimate' in earnings_calendar.columns else None, + } + except Exception as e: + logger.debug(f"Earnings calendar fetch failed for {symbol}: {e}") + + # 季度盈利数据 + try: + quarterly_earnings = ticker.quarterly_earnings + if quarterly_earnings is not None and not quarterly_earnings.empty: + latest_q = quarterly_earnings.index[0] if len(quarterly_earnings.index) > 0 else None + if latest_q: + earnings_data['quarterly'] = { + 'latest_quarter': str(latest_q), + 'revenue': float(quarterly_earnings.loc[latest_q, 'Revenue']) if 'Revenue' in quarterly_earnings.columns else None, + 'earnings': float(quarterly_earnings.loc[latest_q, 'Earnings']) if 'Earnings' in quarterly_earnings.columns else None, + } + except Exception as e: + logger.debug(f"Quarterly earnings fetch failed for {symbol}: {e}") + + return earnings_data if earnings_data else None + + except Exception as e: + logger.debug(f"Earnings data fetch failed for {symbol}: {e}") + return None + def _get_crypto_info(self, symbol: str) -> Optional[Dict[str, Any]]: """加密货币信息 (固定描述为主)""" # 常见加密货币的描述 diff --git a/backend_api_python/app/services/pending_order_worker.py b/backend_api_python/app/services/pending_order_worker.py index 3e5d4f0..6e15c1c 100644 --- a/backend_api_python/app/services/pending_order_worker.py +++ b/backend_api_python/app/services/pending_order_worker.py @@ -295,6 +295,32 @@ class PendingOrderWorker: except Exception: pass exch_size.setdefault(hb_sym, {"long": 0.0, "short": 0.0})[side] = float(qty_base) + + # Extract entry price from OKX position data + # OKX API returns avgPx (average price) or avgPxEp (average price in equity) for positions + try: + # Try avgPx first (average entry price) + avg_px = p.get("avgPx") + if avg_px: + entry_price = float(avg_px) + else: + # Fallback to avgPxEp (average price in equity) + avg_px_ep = p.get("avgPxEp") + if avg_px_ep: + entry_price = float(avg_px_ep) + else: + # Fallback to last price if available + last_px = p.get("last") + entry_price = float(last_px) if last_px else 0.0 + + if entry_price > 0: + exch_entry_price.setdefault(hb_sym, {"long": 0.0, "short": 0.0})[side] = entry_price + logger.debug(f"[PositionSync] OKX {hb_sym} {side}: entry_price={entry_price} from avgPx={p.get('avgPx')} or avgPxEp={p.get('avgPxEp')}") + else: + logger.warning(f"[PositionSync] OKX {hb_sym} {side}: Could not extract entry price from position data: {p}") + except Exception as e: + logger.warning(f"[PositionSync] Failed to extract entry price for OKX {hb_sym} {side}: {e}") + # Don't set entry_price, will remain 0.0 elif isinstance(client, BitgetMixClient) and market_type == "swap": product_type = str(exchange_config.get("product_type") or exchange_config.get("productType") or "USDT-FUTURES")