""" 加密货币交易智能体 - 主控制器 """ import asyncio from typing import Dict, Any, List, Optional from datetime import datetime, timedelta import pandas as pd from app.utils.logger import logger from app.config import get_settings from app.services.binance_service import binance_service from app.services.feishu_service import get_feishu_service from app.services.telegram_service import get_telegram_service from app.services.paper_trading_service import get_paper_trading_service from app.services.price_monitor_service import get_price_monitor_service from app.crypto_agent.signal_analyzer import SignalAnalyzer from app.crypto_agent.strategy import TrendFollowingStrategy class CryptoAgent: """加密货币交易信号智能体""" def __init__(self): """初始化智能体""" self.settings = get_settings() self.binance = binance_service self.feishu = get_feishu_service() self.telegram = get_telegram_service() self.analyzer = SignalAnalyzer() self.strategy = TrendFollowingStrategy() # 模拟交易服务 self.paper_trading_enabled = self.settings.paper_trading_enabled if self.paper_trading_enabled: self.paper_trading = get_paper_trading_service() self.price_monitor = get_price_monitor_service() # 注册价格回调 self.price_monitor.add_price_callback(self._on_price_update) else: self.paper_trading = None self.price_monitor = None # 状态管理 self.last_signals: Dict[str, Dict[str, Any]] = {} # 上次信号 self.last_trends: Dict[str, str] = {} # 上次趋势 self.signal_cooldown: Dict[str, datetime] = {} # 信号冷却时间 # 配置 self.symbols = self.settings.crypto_symbols.split(',') self.analysis_interval = self.settings.crypto_analysis_interval self.llm_threshold = self.settings.crypto_llm_threshold # 运行状态 self.running = False logger.info(f"加密货币智能体初始化完成，监控交易对: {self.symbols}") if self.paper_trading_enabled: logger.info(f"模拟交易已启用") def _on_price_update(self, symbol: str, price: float): """处理实时价格更新（用于模拟交易）""" if not self.paper_trading: return # 检查是否有订单触发止盈止损 triggered = self.paper_trading.check_price_triggers(symbol, price) for result in triggered: # 异步发送平仓通知 asyncio.create_task(self._notify_order_closed(result)) async def _notify_order_closed(self, result: Dict[str, Any]): """发送订单平仓通知""" is_win = result.get('is_win', False) status = result.get('status', '') # 确定图标和文本 if status == 'closed_tp': emoji = "🎯" status_text = "止盈平仓" elif status == 'closed_sl': emoji = "🛑" status_text = "止损平仓" else: emoji = "📤" status_text = "手动平仓" win_text = "盈利" if is_win else "亏损" side_text = "做多" if result.get('side') == 'long' else "做空" # 构建消息 message = f"""{emoji} 订单{status_text} 交易对: {result.get('symbol')} 方向: {side_text} 入场: ${result.get('entry_price', 0):,.2f} 出场: ${result.get('exit_price', 0):,.2f} {win_text}: {result.get('pnl_percent', 0):+.2f}% (${result.get('pnl_amount', 0):+.2f}) 持仓时间: {result.get('hold_duration', 'N/A')}""" # 发送通知 await self.feishu.send_text(message) await self.telegram.send_message(message) logger.info(f"已发送订单平仓通知: {result.get('order_id')}") def _get_seconds_until_next_5min(self) -> int: """计算距离下一个5分钟整点的秒数""" now = datetime.now() current_minute = now.minute current_second = now.second # 计算下一个5的倍数分钟 minutes_past = current_minute % 5 if minutes_past == 0 and current_second == 0: # 刚好在整点，立即执行 return 0 minutes_to_wait = 5 - minutes_past if minutes_past > 0 else 5 seconds_to_wait = minutes_to_wait * 60 - current_second return seconds_to_wait async def run(self): """主运行循环 - 在5的倍数分钟执行""" self.running = True # 启动横幅 logger.info("\n" + "=" * 60) logger.info("🚀 加密货币交易信号智能体") logger.info("=" * 60) logger.info(f" 监控交易对: {', '.join(self.symbols)}") logger.info(f" 运行模式: 每5分钟整点执行 (:00, :05, :10, ...)") logger.info(f" LLM阈值: {self.llm_threshold * 100:.0f}%") if self.paper_trading_enabled: logger.info(f" 模拟交易: 已启用") logger.info("=" * 60 + "\n") # 启动价格监控（用于模拟交易） if self.paper_trading_enabled and self.price_monitor: for symbol in self.symbols: self.price_monitor.subscribe_symbol(symbol) logger.info(f"已启动 WebSocket 价格监控: {', '.join(self.symbols)}") # 发送启动通知（飞书 + Telegram） await self.feishu.send_text( f"🚀 加密货币智能体已启动\n" f"监控交易对: {', '.join(self.symbols)}\n" f"运行时间: 每5分钟整点 (:00, :05, :10, ...)" ) await self.telegram.send_startup_notification(self.symbols) while self.running: try: # 等待到下一个5分钟整点 wait_seconds = self._get_seconds_until_next_5min() if wait_seconds > 0: next_run = datetime.now() + timedelta(seconds=wait_seconds) logger.info(f"⏳ 等待 {wait_seconds} 秒，下次运行: {next_run.strftime('%H:%M:%S')}") await asyncio.sleep(wait_seconds) # 执行分析 run_time = datetime.now() logger.info("\n" + "=" * 60) logger.info(f"⏰ 定时任务执行 [{run_time.strftime('%Y-%m-%d %H:%M:%S')}]") logger.info("=" * 60) for symbol in self.symbols: await self.analyze_symbol(symbol) logger.info("\n" + "─" * 60) logger.info(f"✅ 本轮分析完成，共分析 {len(self.symbols)} 个交易对") logger.info("─" * 60 + "\n") # 等待几秒确保不会在同一分钟内重复执行 await asyncio.sleep(2) except Exception as e: logger.error(f"❌ 分析循环出错: {e}") await asyncio.sleep(10) # 出错后等待10秒再继续 def stop(self): """停止运行""" self.running = False # 停止价格监控 if self.price_monitor: self.price_monitor.stop() logger.info("加密货币智能体已停止") async def analyze_symbol(self, symbol: str): """ 分析单个交易对 Args: symbol: 交易对，如 'BTCUSDT' """ try: # 分隔线 logger.info(f"\n{'─' * 50}") logger.info(f"📊 {symbol} 分析开始") logger.info(f"{'─' * 50}") # 1. 获取多周期数据 data = self.binance.get_multi_timeframe_data(symbol) if not self._validate_data(data): logger.warning(f"⚠️ {symbol} 数据不完整，跳过分析") return # 获取当前价格 current_price = float(data['5m'].iloc[-1]['close']) price_change_24h = self._calculate_price_change(data['1h']) logger.info(f"💰 当前价格: ${current_price:,.2f} ({price_change_24h})") # 2. 分析趋势（1H + 4H）- 返回详细趋势信息 logger.info(f"\n📈 【趋势分析】") trend = self.analyzer.analyze_trend(data['1h'], data['4h']) trend_direction = trend.get('direction', 'neutral') if isinstance(trend, dict) else trend trend_strength = trend.get('strength', 'unknown') if isinstance(trend, dict) else 'unknown' trend_phase = trend.get('phase', 'unknown') if isinstance(trend, dict) else 'unknown' # 趋势方向图标 trend_icon = {'bullish': '🟢 看涨', 'bearish': '🔴 看跌', 'neutral': '⚪ 震荡'}.get(trend_direction, '❓') phase_text = {'impulse': '主升/主跌浪', 'correction': '回调/反弹', 'oversold': '极度超卖', 'overbought': '极度超买', 'sideways': '横盘'}.get(trend_phase, trend_phase) logger.info(f" 方向: {trend_icon} | 强度: {trend_strength} | 阶段: {phase_text}") # 3. 检查趋势变化（只有之前有记录时才检查，避免启动时误报） if symbol in self.last_trends: last_trend = self.last_trends[symbol] if isinstance(last_trend, dict): last_direction = last_trend.get('direction', 'neutral') else: last_direction = last_trend if last_direction != trend_direction: await self._handle_trend_change(symbol, last_direction, trend_direction, data) self.last_trends[symbol] = trend # 4. 分析进场信号（15M 为主，5M 辅助，传入 1H 数据用于支撑阻力位） logger.info(f"\n🎯 【信号分析】") signal = self.analyzer.analyze_entry_signal(data['5m'], data['15m'], trend, data['1h']) signal['symbol'] = symbol signal['trend'] = trend_direction signal['trend_info'] = trend if isinstance(trend, dict) else {'direction': trend} signal['price'] = current_price signal['timestamp'] = datetime.now() # 输出信号详情 action_icon = {'buy': '🟢 买入', 'sell': '🔴 卖出', 'hold': '⏸️ 观望'}.get(signal['action'], '❓') grade_icon = {'A': '⭐⭐⭐', 'B': '⭐⭐', 'C': '⭐', 'D': ''}.get(signal.get('signal_grade', 'D'), '') signal_type = signal.get('signal_type', 'swing') type_text = '📈短线' if signal_type == 'short_term' else '📊波段' logger.info(f" 信号: {action_icon} | 类型: {type_text} | 置信度: {signal['confidence']}% | 等级: {signal.get('signal_grade', 'D')} {grade_icon}") # 输出触发原因 if signal.get('reasons'): logger.info(f" 原因: {', '.join(signal['reasons'][:5])}") # 最多显示5个原因 # 输出权重详情 weights = signal.get('signal_weights', {}) if weights: logger.info(f" 权重: 买入={weights.get('buy', 0):.1f} | 卖出={weights.get('sell', 0):.1f}") # 输出K线形态 patterns = signal.get('patterns', {}) if patterns.get('bullish_patterns') or patterns.get('bearish_patterns'): all_patterns = patterns.get('bullish_patterns', []) + patterns.get('bearish_patterns', []) logger.info(f" 形态: {', '.join(all_patterns)}") # 输出成交量分析 vol = signal.get('volume_analysis', {}) if vol: vol_icon = {'high': '📈', 'low': '📉', 'normal': '➖'}.get(vol.get('volume_signal', 'normal'), '') confirm_text = '✓ 确认' if vol.get('volume_confirms') else '✗ 未确认' logger.info(f" 成交量: {vol_icon} {vol.get('volume_signal', 'normal')} | {confirm_text}") # 输出支撑阻力位 levels = signal.get('levels', {}) if levels.get('nearest_support') or levels.get('nearest_resistance'): support = f"${levels['nearest_support']:,.2f}" if levels.get('nearest_support') else '-' resistance = f"${levels['nearest_resistance']:,.2f}" if levels.get('nearest_resistance') else '-' logger.info(f" 关键位: 支撑={support} | 阻力={resistance}") # 5. 检查是否需要发送信号 if self._should_send_signal(symbol, signal): logger.info(f"\n📤 【发送通知】") # 6. 计算止损止盈 atr = float(data['15m'].iloc[-1].get('atr', 0)) if atr > 0: sl_tp = self.analyzer.calculate_stop_loss_take_profit( signal['price'], signal['action'], atr ) signal.update(sl_tp) logger.info(f" 止损: ${signal['stop_loss']:,.2f} | 止盈: ${signal['take_profit']:,.2f}") # 7. LLM 深度分析（置信度超过阈值时） if signal['confidence'] >= self.llm_threshold * 100: logger.info(f" 🤖 触发 LLM 深度分析...") llm_result = await self.analyzer.llm_analyze(data, signal, symbol) # 处理 LLM 分析结果 if llm_result.get('parsed'): parsed = llm_result['parsed'] # 新格式使用 signal 而不是 recommendation recommendation = parsed.get('signal', parsed.get('recommendation', {})) # 如果 LLM 建议观望，降低置信度 if recommendation.get('action') == 'wait': signal['confidence'] = min(signal['confidence'], 40) signal['llm_analysis'] = llm_result.get('summary', 'LLM 建议观望') logger.info(f" 🤖 LLM 建议: 观望") else: # 使用 LLM 的止损止盈建议 if recommendation.get('stop_loss'): signal['stop_loss'] = recommendation['stop_loss'] if recommendation.get('targets'): signal['take_profit'] = recommendation['targets'][0] elif recommendation.get('take_profit'): signal['take_profit'] = recommendation['take_profit'] signal['llm_analysis'] = llm_result.get('summary', '') logger.info(f" 🤖 LLM 建议: {recommendation.get('action', 'N/A')}") else: signal['llm_analysis'] = llm_result.get('summary', llm_result.get('raw', '')[:200]) # 8. 发送通知（飞书 + Telegram，置信度仍然足够高时） if signal['confidence'] >= 50: await self.feishu.send_trading_signal(signal) await self.telegram.send_trading_signal(signal) # 9. 更新状态 self.last_signals[symbol] = signal self.signal_cooldown[symbol] = datetime.now() action_text = '买入' if signal['action'] == 'buy' else '卖出' logger.info(f" ✅ 已发送 {action_text} 信号通知（飞书+Telegram）") # 10. 创建模拟订单 if self.paper_trading_enabled and self.paper_trading: if signal.get('signal_grade', 'D') != 'D': order = self.paper_trading.create_order_from_signal(signal) if order: logger.info(f" 📝 已创建模拟订单: {order.order_id}") else: logger.info(f" ⏸️ 置信度不足({signal['confidence']}%)，不发送通知") else: # 输出为什么不发送 if signal['action'] == 'hold': logger.info(f"\n⏸️ 结论: 观望，无交易机会") elif signal['confidence'] < 50: logger.info(f"\n⏸️ 结论: 置信度不足({signal['confidence']}%)，继续观望") else: logger.info(f"\n⏸️ 结论: 信号冷却中，跳过") except Exception as e: logger.error(f"❌ 分析 {symbol} 出错: {e}") import traceback logger.error(traceback.format_exc()) def _calculate_price_change(self, h1_data: pd.DataFrame) -> str: """计算24小时价格变化""" if len(h1_data) < 24: return "N/A" price_now = h1_data.iloc[-1]['close'] price_24h_ago = h1_data.iloc[-24]['close'] change = ((price_now - price_24h_ago) / price_24h_ago) * 100 if change >= 0: return f"+{change:.2f}%" return f"{change:.2f}%" def _validate_data(self, data: Dict[str, pd.DataFrame]) -> bool: """验证数据完整性""" required_intervals = ['5m', '15m', '1h', '4h'] for interval in required_intervals: if interval not in data or data[interval].empty: return False if len(data[interval]) < 20: # 至少需要20条数据 return False return True def _should_send_signal(self, symbol: str, signal: Dict[str, Any]) -> bool: """ 判断是否应该发送信号 Args: symbol: 交易对 signal: 信号数据 Returns: 是否发送 """ # 如果是观望，不发送 if signal['action'] == 'hold': return False # 置信度太低，不发送 if signal['confidence'] < 50: return False # 检查冷却时间（同一交易对30分钟内不重复发送相同方向的信号） if symbol in self.signal_cooldown: cooldown_end = self.signal_cooldown[symbol] + timedelta(minutes=30) if datetime.now() < cooldown_end: # 检查是否是相同方向的信号 if symbol in self.last_signals: if self.last_signals[symbol]['action'] == signal['action']: logger.debug(f"{symbol} 信号冷却中，跳过") return False return True async def _handle_trend_change(self, symbol: str, old_trend: str, new_trend: str, data: Dict[str, pd.DataFrame]): """处理趋势变化""" price = float(data['1h'].iloc[-1]['close']) await self.feishu.send_trend_change(symbol, old_trend, new_trend, price) await self.telegram.send_trend_change(symbol, old_trend, new_trend, price) logger.info(f"{symbol} 趋势变化: {old_trend} -> {new_trend}") async def analyze_once(self, symbol: str) -> Dict[str, Any]: """ 单次分析（用于测试或手动触发） Args: symbol: 交易对 Returns: 分析结果 """ data = self.binance.get_multi_timeframe_data(symbol) if not self._validate_data(data): return {'error': '数据不完整'} trend = self.analyzer.analyze_trend(data['1h'], data['4h']) signal = self.analyzer.analyze_entry_signal(data['5m'], data['15m'], trend) signal['symbol'] = symbol signal['trend'] = trend signal['price'] = float(data['5m'].iloc[-1]['close']) # 计算止损止盈 atr = float(data['15m'].iloc[-1].get('atr', 0)) if atr > 0: sl_tp = self.analyzer.calculate_stop_loss_take_profit( signal['price'], signal['action'], atr ) signal.update(sl_tp) return signal def get_status(self) -> Dict[str, Any]: """获取智能体状态""" return { 'running': self.running, 'symbols': self.symbols, 'analysis_interval': self.analysis_interval, 'last_signals': { symbol: { 'action': sig.get('action'), 'confidence': sig.get('confidence'), 'timestamp': sig.get('timestamp').isoformat() if sig.get('timestamp') else None } for symbol, sig in self.last_signals.items() }, 'last_trends': self.last_trends } # 全局实例 crypto_agent = CryptoAgent()