""" 信号格式化工具 用于格式化交易信号通知,支持: - 飞书卡片格式 - Telegram 文本格式 - 支持加密货币、美股、港股 """ from typing import Dict, Any class SignalFormatter: """信号格式化工具""" @staticmethod def format_signal_message(signal: Dict[str, Any], symbol: str, agent_type: str = 'crypto', stock_name: str = '') -> str: """ 格式化信号消息(用于 Telegram 通知) Args: signal: 信号数据 symbol: 交易对 agent_type: 智能体类型 (crypto/stock) stock_name: 股票名称(可选,从基本面数据获取) Returns: 格式化的消息文本 """ type_map = { 'short_term': '短线', 'medium_term': '中线', 'long_term': '长线' } action_map = { 'buy': '做多', 'sell': '做空' } # 兼容 timeframe 和 type 字段 signal_type_key = 'timeframe' if 'timeframe' in signal else 'type' signal_type = type_map.get(signal.get(signal_type_key), signal.get(signal_type_key)) action = action_map.get(signal['action'], signal['action']) grade = signal.get('grade', 'C') confidence = signal.get('confidence', 0) entry_type = signal.get('entry_type', 'market') # 等级图标 grade_icon = {'A': '⭐⭐⭐', 'B': '⭐⭐', 'C': '⭐', 'D': ''}.get(grade, '') # 方向图标 action_icon = '🟢' if signal['action'] == 'buy' else '🔴' # 入场类型 entry_type_text = '现价入场' if entry_type == 'market' else '挂单等待' entry_type_icon = '⚡' if entry_type == 'market' else '⏳' # 仓位大小 position_size = signal.get('position_size', 'light') position_map = {'heavy': '重仓', 'medium': '中仓', 'light': '轻仓'} position_icon = {'heavy': '🔥', 'medium': '📊', 'light': '🌱'}.get(position_size, '🌱') position_text = position_map.get(position_size, '轻仓') # 计算风险收益比 entry = signal.get('entry_price', 0) sl = signal.get('stop_loss', 0) tp = signal.get('take_profit', 0) sl_percent = ((sl - entry) / entry * 100) if entry else 0 tp_percent = ((tp - entry) / entry * 100) if entry else 0 # 识别市场类型 if agent_type == 'crypto': market_tag = '[加密货币] ' elif symbol.endswith('.HK'): market_tag = '[港股] ' else: market_tag = '[美股] ' # 构建标题(带股票名称和市场类型) symbol_display = f"{stock_name}({symbol})" if stock_name else symbol message = f"""📊 {market_tag}{symbol_display} {signal_type}信号 {action_icon} **方向**: {action} {entry_type_icon} **入场**: {entry_type_text} {position_icon} **仓位**: {position_text} ⭐ **等级**: {grade} {grade_icon} 📈 **置信度**: {confidence}% 💰 **入场价**: ${entry:,.2f} 🛑 **止损价**: ${sl:,.2f} ({sl_percent:+.1f}%) 🎯 **止盈价**: ${tp:,.2f} ({tp_percent:+.1f}%) 📝 **分析理由**: {signal.get('reason', '无')} ⚠️ **风险提示**: {signal.get('risk_warning', '请注意风险控制')}""" return message @staticmethod def format_feishu_card(signal: Dict[str, Any], symbol: str, agent_type: str = 'crypto', stock_name: str = '') -> Dict[str, Any]: """ 格式化飞书卡片消息 Args: signal: 信号数据 symbol: 交易对 agent_type: 智能体类型 (crypto/stock) stock_name: 股票名称(可选,从基本面数据获取) Returns: 包含 title, content, color 的字典 """ type_map = { 'short_term': '短线', 'medium_term': '中线', 'long_term': '长线' } action_map = { 'buy': '做多', 'sell': '做空' } # 兼容 timeframe 和 type 字段 signal_type_key = 'timeframe' if 'timeframe' in signal else 'type' signal_type = type_map.get(signal.get(signal_type_key), signal.get(signal_type_key)) action = action_map.get(signal['action'], signal['action']) action_icon = '🟢' if signal['action'] == 'buy' else '🔴' grade = signal.get('grade', 'C') confidence = signal.get('confidence', 0) entry_type = signal.get('entry_type', 'market') # 等级图标 grade_icon = {'A': '⭐⭐⭐', 'B': '⭐⭐', 'C': '⭐', 'D': ''}.get(grade, '') # 入场类型 entry_type_text = '现价入场' if entry_type == 'market' else '挂单等待' entry_type_icon = '⚡' if entry_type == 'market' else '⏳' # 仓位大小 position_size = signal.get('position_size', 'light') position_map = {'heavy': '重仓', 'medium': '中仓', 'light': '轻仓'} position_icon = {'heavy': '🔥', 'medium': '📊', 'light': '🌱'}.get(position_size, '🌱') position_text = position_map.get(position_size, '轻仓') # 标题和颜色 - 区分加密货币/美股/港股 is_market_order = entry_type == 'market' market_badge = '【现价】' if is_market_order else '' # 识别市场类型 if agent_type == 'crypto': market_tag = '[加密货币] ' elif symbol.endswith('.HK'): market_tag = '[港股] ' else: market_tag = '[美股] ' # 构建带名称的股票显示 symbol_display = f"{stock_name}({symbol})" if stock_name else symbol if signal['action'] == 'buy': title = f"🟢 {market_tag}{symbol_display} {signal_type}做多信号 {market_badge}" color = "green" else: title = f"🔴 {market_tag}{symbol_display} {signal_type}做空信号 {market_badge}" color = "red" # 计算风险收益比 entry = signal.get('entry_price', 0) sl = signal.get('stop_loss', 0) tp = signal.get('take_profit', 0) sl_percent = ((sl - entry) / entry * 100) if entry else 0 tp_percent = ((tp - entry) / entry * 100) if entry else 0 # 构建内容 content_lines = [ f"{action_icon} **操作**: {action}", f"{entry_type_icon} **入场方式**: {entry_type_text}", f"{position_icon} **仓位**: {position_text} | 📈 信心度: **{confidence}%**", f"⭐ **等级**: {grade} {grade_icon}", f"", f"💰 **入场价**: ${entry:,.2f}", f"🛑 **止损价**: ${sl:,.2f} ({sl_percent:+.1f}%)", f"🎯 **止盈价**: ${tp:,.2f} ({tp_percent:+.1f}%)", f"", f"📝 **分析理由**:", f"{signal.get('reason', '无')}", ] # 添加关键因素(如果有) key_factors = signal.get('key_factors') if key_factors and isinstance(key_factors, list): content_lines.append("") content_lines.append("**关键因素**:") for factor in key_factors[:5]: content_lines.append(f"- {factor}") # 添加风险提示(如果有) risk_warning = signal.get('risk_warning') if risk_warning: content_lines.append("") content_lines.append(f"⚠️ **风险提示**:") content_lines.append(risk_warning) content = "\n".join(content_lines) return { 'title': title, 'content': content, 'color': color } # 全局单例 _signal_formatter = SignalFormatter() def get_signal_formatter() -> SignalFormatter: """获取信号格式化工具单例""" return _signal_formatter