""" DingTalk Notifier - 钉钉群机器人消息推送 支持功能: - Markdown格式消息 - 交易信号格式化 - 错误重试 - 消息去重 """ import logging import json import time from typing import Dict, Any, Optional from datetime import datetime import hmac import hashlib import base64 import urllib.parse logger = logging.getLogger(__name__) class DingTalkNotifier: """钉钉群机器人通知器""" def __init__( self, webhook_url: Optional[str] = None, secret: Optional[str] = None, enabled: bool = True ): """ 初始化钉钉通知器 Args: webhook_url: 钉钉机器人webhook地址 secret: 钉钉机器人加签密钥（可选，增强安全性） enabled: 是否启用通知 """ self.webhook_url = webhook_url self.secret = secret self.enabled = enabled and webhook_url is not None if self.enabled: logger.info(f"📱 钉钉通知已启用 - Webhook: {webhook_url[:50]}...") else: logger.info("📱 钉钉通知未启用 (未配置webhook_url)") # 统计信息 self.stats = { 'total_sent': 0, 'total_failed': 0, 'last_send_time': None } def _generate_sign(self, timestamp: int) -> str: """ 生成钉钉机器人加签 Args: timestamp: 时间戳（毫秒） Returns: 签名字符串 """ if not self.secret: return "" secret_enc = self.secret.encode('utf-8') string_to_sign = f'{timestamp}\n{self.secret}' string_to_sign_enc = string_to_sign.encode('utf-8') hmac_code = hmac.new( secret_enc, string_to_sign_enc, digestmod=hashlib.sha256 ).digest() sign = urllib.parse.quote_plus(base64.b64encode(hmac_code)) return sign def _build_webhook_url(self) -> str: """ 构建带签名的webhook URL Returns: 完整的webhook URL """ if not self.secret: return self.webhook_url timestamp = int(time.time() * 1000) sign = self._generate_sign(timestamp) return f"{self.webhook_url}×tamp={timestamp}&sign={sign}" def send_markdown( self, title: str, text: str, at_mobiles: Optional[list] = None, at_all: bool = False ) -> bool: """ 发送Markdown格式消息 Args: title: 消息标题 text: Markdown格式文本 at_mobiles: @的手机号列表 at_all: 是否@所有人 Returns: 是否发送成功 """ if not self.enabled: logger.debug("钉钉通知未启用，跳过发送") return False try: import requests url = self._build_webhook_url() payload = { "msgtype": "markdown", "markdown": { "title": title, "text": text }, "at": { "atMobiles": at_mobiles or [], "isAtAll": at_all } } headers = {'Content-Type': 'application/json'} response = requests.post( url, data=json.dumps(payload), headers=headers, timeout=5 ) result = response.json() if result.get('errcode') == 0: self.stats['total_sent'] += 1 self.stats['last_send_time'] = datetime.now().isoformat() logger.info(f"✅ 钉钉消息发送成功: {title}") return True else: self.stats['total_failed'] += 1 logger.error(f"❌ 钉钉消息发送失败: {result.get('errmsg')}") return False except Exception as e: self.stats['total_failed'] += 1 logger.error(f"❌ 钉钉消息发送异常: {e}", exc_info=True) return False def send_signal(self, aggregated_signal: Dict[str, Any]) -> bool: """ 发送交易信号通知 Args: aggregated_signal: 聚合后的交易信号 Returns: 是否发送成功 """ if not self.enabled: return False try: # 格式化信号为Markdown markdown = self._format_signal_markdown(aggregated_signal) # 提取标题 signal_type = aggregated_signal.get('final_signal', 'HOLD') confidence = aggregated_signal.get('final_confidence', 0) title = f"🚨 交易信号: {signal_type} (置信度: {confidence:.0%})" # 发送消息 return self.send_markdown(title, markdown) except Exception as e: logger.error(f"❌ 格式化交易信号失败: {e}", exc_info=True) return False def _format_signal_markdown(self, signal: Dict[str, Any]) -> str: """ 格式化交易信号为Markdown文本（多时间级别版本） 支持两种格式: 1. 新格式: trades数组 (优先) 2. 旧格式: opportunities对象 (向后兼容) Args: signal: 聚合信号 Returns: Markdown格式文本 """ # 信号类型对应的emoji signal_type = signal.get('final_signal', 'HOLD') confidence = signal.get('final_confidence', 0) signal_emoji = { 'BUY': '🟢', 'SELL': '🔴', 'HOLD': '🟡' } emoji = signal_emoji.get(signal_type, '⚪') lines = [] # === 核心信号 === symbol = signal.get('symbol', 'BTC/USDT') lines.append(f"# {emoji} {symbol} {signal_type}") lines.append("") lines.append(f"**综合置信度**: {confidence:.0%} | **时间**: {datetime.now().strftime('%H:%M')}") lines.append("") # === 当前价格 === levels = signal.get('levels', {}) current_price = levels.get('current_price', 0) if current_price > 0: lines.append(f"**当前价格**: ${current_price:,.2f}") lines.append("") # === 多时间级别分析 === lines.append("## 📊 多时间级别分析") lines.append("") # 获取LLM信号 llm_signal = signal.get('llm_signal') or {} # 检测是否为新格式 (trades数组) trades = llm_signal.get('trades', []) if trades and isinstance(trades, list) and len(trades) >= 3: # 新格式: trades数组 trades_by_tf = {t.get('timeframe'): t for t in trades if t.get('timeframe')} # 短期分析 self._add_trade_section( lines, "短期 (5m/15m/1h)", "⚡", trades_by_tf.get('short', {}), signal ) # 中期分析 self._add_trade_section( lines, "中期 (4h/1d)", "📈", trades_by_tf.get('medium', {}), signal ) # 长期分析 self._add_trade_section( lines, "长期 (1d/1w)", "📅", trades_by_tf.get('long', {}), signal ) # 综合分析 analysis = llm_signal.get('analysis', {}) reason = analysis.get('summary', '') or llm_signal.get('reasoning', '') else: # 旧格式: opportunities对象 opportunities = llm_signal.get('opportunities', {}) recommendations = llm_signal.get('recommendations_by_timeframe', {}) # 短期分析 self._add_timeframe_section( lines, "短期 (5m/15m/1h)", "⚡", opportunities.get('short_term_5m_15m_1h', {}), recommendations.get('short_term', ''), signal ) # 中期分析 self._add_timeframe_section( lines, "中期 (4h/1d)", "📈", opportunities.get('medium_term_4h_1d', {}), recommendations.get('medium_term', ''), signal ) # 长期分析 self._add_timeframe_section( lines, "长期 (1d/1w)", "📅", opportunities.get('long_term_1d_1w', {}), recommendations.get('long_term', ''), signal ) reason = llm_signal.get('reasoning', '') or self._get_brief_reason(signal) # === 综合建议 === if reason: lines.append("---") lines.append("## 💡 综合分析") lines.append("") lines.append(f"{reason}") lines.append("") # === 关键价位 === key_levels = llm_signal.get('key_levels', {}) if key_levels: support = key_levels.get('support', []) resistance = key_levels.get('resistance', []) if support or resistance: lines.append("---") lines.append("## 📍 关键价位") lines.append("") if support: support_str = ", ".join([f"${p:,.0f}" for p in support[:3]]) lines.append(f"**支撑**: {support_str}") if resistance: resistance_str = ", ".join([f"${p:,.0f}" for p in resistance[:3]]) lines.append(f"**阻力**: {resistance_str}") lines.append("") # === 页脚 === lines.append("---") lines.append("*仅供参考，不构成投资建议*") return "\n".join(lines) def _add_trade_section( self, lines: list, timeframe_label: str, emoji: str, trade: Dict[str, Any], signal: Dict[str, Any] = None ): """ 添加单个时间级别的交易区块（新格式trades数组） Args: lines: 输出行列表 timeframe_label: 时间级别标签 emoji: emoji图标 trade: 该时间级别的交易信息 signal: 完整信号数据 """ lines.append(f"### {emoji} {timeframe_label}") lines.append("") status = trade.get('status', 'INACTIVE') is_active = status == 'ACTIVE' if is_active: direction = trade.get('direction', 'NONE') entry = trade.get('entry', {}) exit_data = trade.get('exit', {}) position = trade.get('position', {}) risk_reward = trade.get('risk_reward', 0) expected_profit = trade.get('expected_profit_pct', 0) reasoning = trade.get('reasoning', '') # 方向标识 direction_emoji = "🟢" if direction == "LONG" else "🔴" if direction == "SHORT" else "⚪" lines.append(f"{direction_emoji} **方向**: {direction}") lines.append("") # 金字塔入场价格 entry_prices = [] for i in range(1, 5): price = entry.get(f'price_{i}', 0) pct = position.get(f'size_pct_{i}', 0) if price > 0: entry_prices.append(f"${price:,.0f}({pct}%)") if entry_prices: lines.append(f"**入场**: {' → '.join(entry_prices)}") # 止损止盈 stop_loss = exit_data.get('stop_loss', 0) tp1 = exit_data.get('take_profit_1', 0) tp2 = exit_data.get('take_profit_2', 0) tp3 = exit_data.get('take_profit_3', 0) if stop_loss > 0: lines.append(f"**止损**: ${stop_loss:,.0f}") take_profits = [] if tp1 > 0: take_profits.append(f"${tp1:,.0f}") if tp2 > 0: take_profits.append(f"${tp2:,.0f}") if tp3 > 0: take_profits.append(f"${tp3:,.0f}") if take_profits: lines.append(f"**止盈**: {' / '.join(take_profits)}") # 风险回报比和预期盈利 if risk_reward > 0: lines.append(f"**风险回报**: 1:{risk_reward:.1f}") if expected_profit > 0: lines.append(f"**预期盈利**: {expected_profit:.1f}%") lines.append("") # 理由 if reasoning: lines.append(f"💭 {reasoning}") lines.append("") else: # 无交易机会 reasoning = trade.get('reasoning', '') if reasoning: lines.append(f"💭 {reasoning}") else: lines.append("💭 暂无明确交易机会") lines.append("") def _add_timeframe_section( self, lines: list, timeframe_label: str, emoji: str, opportunity: Dict[str, Any], recommendation: str, signal: Dict[str, Any] = None ): """ 添加单个时间级别的分析区块 Args: lines: 输出行列表 timeframe_label: 时间级别标签 emoji: emoji图标 opportunity: 该时间级别的交易机会 recommendation: 该时间级别的操作建议 signal: 完整信号数据（用于获取量化评分等） """ lines.append(f"### {emoji} {timeframe_label}") lines.append("") exists = opportunity.get('exists', False) if exists: direction = opportunity.get('direction', 'UNKNOWN') entry = opportunity.get('entry_price', 0) stop = opportunity.get('stop_loss', 0) tp = opportunity.get('take_profit', 0) reasoning = opportunity.get('reasoning', '') # 方向标识 direction_emoji = "🟢" if direction == "LONG" else "🔴" if direction == "SHORT" else "⚪" lines.append(f"{direction_emoji} **方向**: {direction}") lines.append("") # 价格信息 if entry and stop and tp: lines.append(f"**入场**: ${entry:,.2f}") lines.append(f"**止损**: ${stop:,.2f}") lines.append(f"**止盈**: ${tp:,.2f}") # 风险回报比 risk = abs(entry - stop) reward = abs(tp - entry) rr = reward / risk if risk > 0 else 0 lines.append(f"**风险回报**: 1:{rr:.1f}") lines.append("") # 理由 if reasoning: lines.append(f"💭 {reasoning}") lines.append("") else: # 无交易机会时，显示关键数据支撑 # 尝试从市场分析中获取该周期的数据 if signal: market_analysis = signal.get('market_analysis', {}) # 显示量化评分（如果有） quant_signal = signal.get('quantitative_signal', {}) if quant_signal: composite = quant_signal.get('composite_score', 0) lines.append(f"📊 **量化评分**: {composite:.1f}") scores = quant_signal.get('scores', {}) if scores: lines.append(f"- 趋势: {scores.get('trend', 0):.0f} | 动量: {scores.get('momentum', 0):.0f} | 订单流: {scores.get('orderflow', 0):.0f}") lines.append("") # 显示关键价格和技术指标 if market_analysis: trend = market_analysis.get('trend', {}) momentum = market_analysis.get('momentum', {}) if trend or momentum: lines.append(f"📈 **技术状态**:") if trend: lines.append(f"- 趋势: {trend.get('direction', 'unknown')} ({trend.get('strength', 'weak')})") if momentum: lines.append(f"- RSI: {momentum.get('rsi', 50):.0f} ({momentum.get('rsi_status', '中性')})") lines.append(f"- MACD: {momentum.get('macd_signal', 'unknown')}") lines.append("") # 操作建议 if recommendation: lines.append(f"💭 **操作建议**: {recommendation}") lines.append("") else: lines.append("💭 暂无明确交易机会") lines.append("") def _get_best_trade_plan(self, signal: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ 获取最优交易计划（优先中长线，盈利空间更大） 优先级： 1. 中长线机会 (swing) - 4h/1d/1w级别，盈利空间2%-5%+，适合通知 2. 日内机会 (intraday) - 5m/15m/1h级别，盈利空间0.5%-1% 3. 量化信号价格位 - 回退选项 Returns: { 'entry': float, 'stop_loss': float, 'take_profit': float, 'type': 'swing'|'intraday'|'quant' # 交易类型 } or None """ llm = signal.get('llm_signal') if llm and isinstance(llm, dict): opportunities = llm.get('opportunities', {}) # 优先1：中长线机会（盈利空间大，适合手机通知） swing = opportunities.get('swing', {}) if swing.get('exists'): entry = swing.get('entry_price') stop = swing.get('stop_loss') tp = swing.get('take_profit') if entry and stop and tp: return { 'entry': float(entry), 'stop_loss': float(stop), 'take_profit': float(tp), 'type': 'swing' } # 优先2：日内机会（如果没有中长线机会） intraday = opportunities.get('intraday', {}) if intraday.get('exists'): entry = intraday.get('entry_price') stop = intraday.get('stop_loss') tp = intraday.get('take_profit') if entry and stop and tp: return { 'entry': float(entry), 'stop_loss': float(stop), 'take_profit': float(tp), 'type': 'intraday' } # 回退到量化信号的价格位 levels = signal.get('levels', {}) entry = levels.get('entry') stop = levels.get('stop_loss') tp = levels.get('take_profit_1') if entry and stop and tp: return { 'entry': float(entry), 'stop_loss': float(stop), 'take_profit': float(tp), 'type': 'quant' } return None def _get_brief_reason(self, signal: Dict[str, Any]) -> str: """ 获取简短的信号原因（1-2句话） Returns: 简短原因描述 """ reasons = [] # 优先使用LLM的推理（截取前100字） llm = signal.get('llm_signal') if llm and isinstance(llm, dict): llm_reasoning = llm.get('reasoning', '') if llm_reasoning: # 取第一句话或前100字 brief = llm_reasoning.split('。')[0] + '。' if len(brief) > 100: brief = brief[:100] + '...' return brief # 如果有日内机会的说明 opportunities = llm.get('opportunities', {}) intraday = opportunities.get('intraday', {}) if intraday.get('exists') and intraday.get('reasoning'): brief = intraday['reasoning'] if len(brief) > 100: brief = brief[:100] + '...' return brief # 回退到量化信号的推理 quant = signal.get('quantitative_signal', {}) quant_reasoning = quant.get('reasoning', '') if quant_reasoning: return quant_reasoning # 默认 return signal.get('recommendation', '系统分析建议关注') def send_error(self, error_msg: str, context: Optional[str] = None) -> bool: """ 发送错误通知 Args: error_msg: 错误消息 context: 错误上下文 Returns: 是否发送成功 """ if not self.enabled: return False lines = [] lines.append("# ❌ 系统错误通知") lines.append("") lines.append(f"**时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") lines.append(f"**错误**: {error_msg}") if context: lines.append("") lines.append(f"**上下文**: {context}") lines.append("") lines.append("---") lines.append("*请及时检查系统状态*") markdown = "\n".join(lines) return self.send_markdown("系统错误通知", markdown, at_all=True) def get_stats(self) -> Dict[str, Any]: """获取统计信息""" return { 'enabled': self.enabled, 'total_sent': self.stats['total_sent'], 'total_failed': self.stats['total_failed'], 'success_rate': ( self.stats['total_sent'] / (self.stats['total_sent'] + self.stats['total_failed']) if (self.stats['total_sent'] + self.stats['total_failed']) > 0 else 0 ), 'last_send_time': self.stats['last_send_time'] }