#!/usr/bin/env python3 """ 交易信号处理器(no_agent模式): 1. 解析TG信号文本 2. 调advisor脚本获取正确金额 3. 格式化含📐完整模板 4. 推QQ 5. 信号去重/合并 用法: python3 process_signal.py "信号文本" 或: echo "信号文本" | python3 process_signal.py cron模式: 作为no_agent cron job的script使用 """ import sys import os import re import json import subprocess import sqlite3 import hashlib from pathlib import Path from signal_queue import mark_failed # 2026-09-21: 平仓失败时主动 mark_failed from datetime import datetime, timedelta SKILL_DIR = Path.home() / ".hermes/skills/trading/okx-auto-position" ADVISOR = SKILL_DIR / "scripts" / "okx_position_advisor.py" QQ_PUSH = Path.home() / ".hermes/scripts/push_to_qq.sh" SIGNAL_DB = Path.home() / ".hermes/trading/signal_history.db" DEDUP_DB = Path.home() / ".hermes/trading/signal_dedup.db" # Import signal tracker sys.path.insert(0, str(SKILL_DIR / "scripts")) from signal_tracker import format_comparison, record_signal as _tracker_record, record_confirmed, format_trader_rating, get_last_signal, check_signal_lock, list_locks def _locks_status_text(): """所有锁状态 - 附加在每个推送尾部""" try: locks = list_locks() except Exception: return "" if not locks: return "" lines = ["\n\n🔒 **当前所有锁**:"] for l in locks: # list_locks 返回 sqlite Row (索引访问) symbol = l['symbol'] trader = l['trader'] side = l['side'] locked_at = l['locked_at'] # 算锁定时长 from datetime import datetime try: ts = datetime.fromisoformat(str(locked_at).replace('Z', '')) hours = (datetime.now() - ts).total_seconds() / 3600 age = f"{hours:.1f}h" except Exception: age = "?" lines.append(f" • {symbol} → **{trader}** {side} (锁 {age})") return "\n".join(lines) # ─── 解析 ──────────────────────────────────────────────────────────────── def parse_signal(text): """从TG信号文本提取关键字段""" fields = {'_raw': text} # 存原始文本, 供 classify_signal 判定 # 交易员 - 找"【交易员】"标签, fallback "👉 跟单就选 X",再 fallback 第一个非字段名的方括号 m_trader = re.search(r'【交易员】\s*[::]?\s*([^【\n]{1,20})', text) if m_trader: fields['trader'] = m_trader.group(1).strip() else: # Fallback: 👉 跟单就选 X (这是真 trader 来源) m_follow = re.search(r'👉\s*跟单就选\s*(\S+)', text) if m_follow: fields['trader'] = m_follow.group(1).strip() else: # 最后 fallback: 第一个【xx】但跳过字段名 FIELD_NAMES = {'币种', '方向', '杠杆', '仓位大小', '仓位价值', '开仓价', '当前价', '未实现盈亏', '收益额', '持仓量', '强平价', '数量'} m_first = re.search(r'【([^】]{1,20})】', text) if m_first and m_first.group(1) not in FIELD_NAMES: fields['trader'] = m_first.group(1).strip() else: # 2026-08-07: 实在找不到 trader 标签 — 标 '未知' (不要硬编码 'X聚合社区', 会误导) fields['trader'] = '未知' # 字段映射 extractors = { 'symbol': r'【币种】\s*[::]?\s*(\S+)', 'side': r'【方向】\s*[::]?\s*(做多|做空)', # 2026-09-25: 新格式"币种标签内|永续|Nx"也要识别 (熬鹰币安实盘新版) 'leverage':r'(?:【杠杆】\s*[::]?\s*(\d+)|(?:永续|全仓)\s*\|\s*全仓\s*\|\s*(\d+)\s*[xX]?)', 'size': r'【仓位(?:大小)?】\s*[::]?\s*([\d,.]+)', # 兼容 【仓位】和【仓位大小】 'unit': r'【仓位(?:大小)?】\s*[::]?\s*[\d,.]+\s+(BTC|USDT|ETH|SOL|DOGE|张|USD)\b', # 严格匹配已知单位 (必需) 'value': r'【仓位价值】\s*[::]?\s*\$?\s*([\d,.]+)', 'entry': r'【开仓价】\s*[::]?\s*([\d,.]+)', 'current': r'【当前价】\s*[::]?\s*([\d,.]+)', 'pnl': r'【未实现盈亏】\s*[::]?\s*([-\d,.]+)', 'margin': r'【保证金】\s*[::]?\s*\$?\s*([\d,.]+)', } for key, pattern in extractors.items(): m = re.search(pattern, text) if m: v = m.group(1) or m.group(2) or '' if v: fields[key] = v.replace(',', '') if key != 'unit' else v # 清理symbol if 'symbol' in fields: sym_raw = fields['symbol'] # 提取 |Nx 杠杆 (e.g. SKHYUSDT|永续|5x → 5) m_lev = re.search(r'\|(\d+)\s*x?$', sym_raw) if m_lev and 'leverage' not in fields: fields['leverage'] = m_lev.group(1) sym = re.sub(r'\|.*$', '', sym_raw) # 去掉 |永续|10x sym = sym.replace('USDT', '').strip() fields['symbol'] = sym # 方向转英文 if fields.get('side', '').startswith('做多'): fields['side_en'] = 'long' else: fields['side_en'] = 'short' # 信号首次发出时间 (forwarder 加的 ⏱信号时间: 2026-07-08 18:00:00) m = re.search(r'⏱信号时间[::]\s*(\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2})', text) if m: try: fields['signal_time'] = datetime.strptime(m.group(1), '%Y-%m-%d %H:%M:%S') except ValueError: pass return fields SIGNAL_FRESH_MINUTES = 30 # 30 分钟内算新鲜;>=30 算过期(少误跟) def is_signal_stale(fields): """信号是否过期(>= 30 分钟)。 没有时间戳的按"新鲜"处理(不阻断旧信号源)。 """ st = fields.get('signal_time') if not st: return False age = datetime.now() - st return age.total_seconds() >= SIGNAL_FRESH_MINUTES * 60 def format_age_minutes(fields): """信号已发出多久(分钟)。""" st = fields.get('signal_time') if not st: return "?" return int((datetime.now() - st).total_seconds() // 60) # ─── 去重 ──────────────────────────────────────────────────────────────── def init_dedup_db(): conn = sqlite3.connect(str(DEDUP_DB)) conn.execute(""" CREATE TABLE IF NOT EXISTS recent_signals ( id TEXT PRIMARY KEY, symbol TEXT, trader TEXT, timestamp REAL, raw_text TEXT ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS processed ( msg_hash TEXT PRIMARY KEY, processed_at REAL ) """) conn.commit() return conn def is_duplicate(conn, text, symbol, trader): """检查是否重复信号(同交易员同币种2分钟内)""" msg_hash = hashlib.md5(text.encode()).hexdigest() # 检查完全相同的消息 row = conn.execute( "SELECT 1 FROM processed WHERE msg_hash = ?", (msg_hash,) ).fetchone() if row: return True # 检查同交易员同币种2分钟内的信号 cutoff = datetime.now().timestamp() - 120 # 2分钟 row = conn.execute( """SELECT 1 FROM recent_signals WHERE symbol = ? AND trader = ? AND timestamp > ? ORDER BY timestamp DESC LIMIT 1""", (symbol, trader, cutoff) ).fetchone() return row is not None def record_signal(conn, text, symbol, trader): """记录信号用于去重""" msg_hash = hashlib.md5(text.encode()).hexdigest() now = datetime.now().timestamp() conn.execute( "INSERT OR REPLACE INTO processed (msg_hash, processed_at) VALUES (?, ?)", (msg_hash, now) ) conn.execute( "INSERT OR REPLACE INTO recent_signals (id, symbol, trader, timestamp, raw_text) VALUES (?, ?, ?, ?, ?)", (msg_hash, symbol, trader, now, text[:500]) ) # 清理1小时前的记录 cutoff = now - 3600 conn.execute("DELETE FROM recent_signals WHERE timestamp < ?", (cutoff,)) conn.execute("DELETE FROM processed WHERE processed_at < ?", (cutoff,)) conn.commit() # ─── Advisor ────────────────────────────────────────────────────────────── def run_advisor(symbol, side, leverage): """调advisor脚本获取正确数据""" # advisor 内部 fetch_ticker 需要完整 symbol, 不是 base full_symbol = f"{symbol}/USDT:USDT" cmd = [ 'python3', str(ADVISOR), '--symbol', full_symbol, '--side', side, '--leverage', str(leverage), '--json' ] try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=30, cwd=str(ADVISOR.parent) ) if result.returncode == 0: return json.loads(result.stdout) else: return {'error': result.stderr.strip()[:200]} except subprocess.TimeoutExpired: return {'error': 'advisor超时'} except json.JSONDecodeError: return {'error': 'advisor输出非JSON'} except Exception as e: return {'error': str(e)} # ─── 分类 ──────────────────────────────────────────────────────────────── def classify_signal(fields): """判断信号类型:加仓/新开仓/减仓/平仓 2026-09-26: 严格模式 — 文本必须显式包含动作关键词, 否则 unknown 跳过 (避免汇总播报/实时更新被误判为减仓) """ text = fields.get('_raw', '') # 平仓信号 (含 "平仓" 或 "已平仓") if '平仓' in text: return 'close' # 加仓信号 if '加仓' in text or '追仓' in text: return 'add' # 减仓信号 (必须显式含 "减仓" 关键词) if '减仓' in text: return 'reduce' # 新开仓信号 if '新开仓' in text or '注意,大佬开仓了' in text: return 'open' # 调杠杆信号 if '修改了杠杆' in text or '调整了杠杆' in text or '调杠杆' in text: return 'adjust_leverage' # 其他(汇总播报/实时更新/状态推送)统一跳过 return 'unknown' # ─── 格式化 ────────────────────────────────────────────────────────────── def format_message(fields, rec, signal_type): """格式化完整推送消息""" if 'error' in rec: return f"⚠️ advisor错误: {rec['error']}" def _fmt(x, n=4): """格式化数字: 字符串保留原样, 数字 round 到 n 位.""" try: return f"{float(x):.{n}f}" except (ValueError, TypeError): return str(x) symbol = fields.get('symbol', '?') side_cn = fields.get('side', '做多') emoji = '🟩' if fields.get('side_en') == 'long' else '🟥' leverage = fields.get('leverage', '10') trader = fields.get('trader', '?') size = fields.get('size', '?') unit = fields.get('unit', '') # 仓位单位: BTC/USDT/张 value = fields.get('value', '?') entry_price = fields.get('entry', '?') pnl_str = fields.get('pnl', '0') pnl = float(pnl_str.replace('+', '')) if pnl_str else 0 current = rec.get('price', fields.get('current', '?')) pnl_emoji = '🔥' if pnl > 0 else '🔴' pnl_sign = '+' if pnl > 0 else '' # 性价比 cc = rec.get('cost_check', {}) rr = cc.get('rr_ratio', rec.get('rr', 0)) profit = cc.get('profit_amount', rec.get('tp_pnl', 0)) fee = cc.get('fee_cost', 0) fee_pct = cc.get('fee_pct', 0) net = cc.get('net_profit', 0) rating_emoji = cc.get('rating_emoji', '⚠️') rating_text = cc.get('rating_text', '未知') # 信号类型标签 type_labels = { 'open': '新开仓' if not fields.get('_is_add') else 'A类加仓', 'add': '➕ 加仓', 'reduce': 'B类减仓', 'close': '平仓', } type_label = type_labels.get(signal_type, signal_type) # 信号源仓位(只展示,不参与计算) src_info = f"📊 {trader} {size} {unit} {symbol}(价值${value})← 信号源,非你的仓位" # 仓位变化对比 current_size = 0 try: current_size = float(fields.get('size', '0').replace(',', '')) comparison = format_comparison(trader, symbol, current_size) except: comparison = "" # 跟单建议: fetch 真实持仓 + 算比例 our_position = None our_advice = "" try: import ccxt from okx_position_advisor import load_credentials, create_exchange creds = load_credentials() ex = create_exchange(creds) positions = ex.fetch_positions() for p in positions: if symbol in p.get('symbol', '') and p.get('contracts', 0) != 0: our_position = p break if our_position and current_size > 0: our_contracts = float(our_position.get('contracts', 0)) # 算大佬减仓比例 last_signal = get_last_signal(trader, symbol) last_size = (last_signal or {}).get('trader_size', 0) if last_size and last_size > 0: trader_delta_pct = (current_size - last_size) / last_size * 100 if trader_delta_pct < -0.5: # 大佬减仓 > 0.5% # 跟同比例 our_reduce = our_contracts * abs(trader_delta_pct) / 100 # 取整 (BTC min=0.01张, 其他min=1张) if symbol == 'BTC': our_reduce = max(0.01, round(our_reduce, 2)) else: our_reduce = max(1, round(our_reduce)) our_advice = f"\n🎯 跟单建议: 大佬减 {abs(trader_delta_pct):.1f}%, 你跟减 {our_reduce} 张 ({our_contracts} → {our_contracts - our_reduce:.2f})" except Exception as e: our_advice = f"\n⚠️ 跟单计算跳过: {e}" # 交易员评分 try: trader_rating = format_trader_rating(trader) except: trader_rating = "" msg = f"""⚡ 跟单建议 | {symbol} {side_cn} {emoji} {leverage}x({type_label}) {src_info} 入场: ${_fmt(entry_price)} | 当前: ${_fmt(current)} 浮盈: {pnl_sign}{pnl:.0f} {pnl_emoji} 📊 仓位变化 {comparison} {our_advice} {trader_rating} 📐 建议参数 • 仓位: {rec['contracts']}张(保证金{_fmt(rec['margin'])} USDT) • SL: ${_fmt(rec['sl_price'])} → 预亏 -{_fmt(rec.get('sl_pnl', 0))} USDT (保证金-{rec.get('sl_pnl', 0)/max(rec.get('margin', 1), 0.01)*100:.0f}%) • TP: ${_fmt(rec['tp_price'])} → 预盈 +{_fmt(rec.get('tp_pnl', 0))} USDT (保证金+{rec.get('tp_pnl', 0)/max(rec.get('margin', 1), 0.01)*100:.0f}%) • 盈亏比: {rr}:1 {rating_emoji} {rating_text}""" # 如果余额不足,替换跟单方案 if rec.get('contracts', 0) == 0: msg = f"""⚡ 跟单建议 | {symbol} {side_cn} {emoji} {leverage}x({type_label}) {src_info} 入场: ${_fmt(entry_price)} | 当前: ${_fmt(current)} 浮盈: {pnl_sign}{pnl:.0f} {pnl_emoji} ⚠️ 余额不足,无法开仓 • 可用: {_fmt(rec.get('acct_free', 0))} USDT • 需要: ~{_fmt(rec.get('margin', 0))} USDT 💡 建议:等待其他仓位止盈释放保证金""" return msg def format_stale_message(fields, age_minutes): """过期信号提醒(不发 advisor 分析结果,只提示)。""" symbol = fields.get('symbol', '?') side_cn = fields.get('side', '?') emoji = '🟩' if fields.get('side_en') == 'long' else '🟥' leverage = fields.get('leverage', '?') trader = fields.get('trader', '?') entry_price = fields.get('entry', '?') st = fields.get('signal_time') return f"""⏰ 信号已过期 | {symbol} {side_cn} {emoji} {leverage}x {trader} 信号源(仅展示,非你的仓位) 信号首次发出: {st.strftime('%Y-%m-%d %H:%M:%S') if st else '?'} 已过去: {age_minutes} 分钟 (阈值 {SIGNAL_FRESH_MINUTES} 分钟) 入场: ${entry_price} ⚠️ 信号过期,谨慎跟单 • 行情可能已经反转 • 价格/仓位快照与当前不一致 • 如需跟单请用实时数据重新评估""" # ─── 推送 ──────────────────────────────────────────────────────────────── def push_to_qq(message): """推送到QQ""" try: result = subprocess.run( ['bash', str(QQ_PUSH), message], capture_output=True, text=True, timeout=15 ) return result.returncode == 0 except: return False # ─── 平仓 (raw REST, 绕 ccxt load_markets) ──────────────────────────────── def _okx_raw_request(method, path, params=None, body=None, timeout=15): """OKX raw REST 调用 (避 ccxt fetch_balance→load_markets 超时)。 签名规则 (按 ccxt/okx.py sign()): auth = timestamp + method + request_path if GET and query: auth += '?' + urlencode(query) else: auth += json.dumps(body) """ import hmac, hashlib, base64, urllib.parse creds = {} with open(os.path.expanduser('~/.bashrc')) as f: for line in f: line = line.strip() if line.startswith('export OKX_'): k, v = line.replace('export ', '').split('=', 1) creds[k] = v.strip().strip('"').strip("'") for k, v in creds.items(): if '${' not in v: os.environ[k] = v import re for k, v in creds.items(): if '${' in v: os.environ[k] = re.sub(r'\$\{(\w+)\}', lambda m: os.environ.get(m.group(1), ''), v) ts = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%S.') + f"{datetime.utcnow().microsecond // 1000:03d}Z" body_str = json.dumps(body) if body else '' # 构建签名 message auth = ts + method.upper() + path if method.upper() == 'GET': if params: # OKX: query 字符串按字典序排序后用 ? 拼接到 auth sorted_q = '&'.join(f"{k}={urllib.parse.quote_plus(str(v), safe='')}" for k, v in sorted(params.items())) auth += '?' + sorted_q query = '?' + sorted_q else: query = '' else: # POST auth += body_str query = '' sig = base64.b64encode(hmac.new(os.environ['OKX_SECRET'].encode(), auth.encode(), hashlib.sha256).digest()).decode() headers = { 'OK-ACCESS-KEY': os.environ['OKX_API_KEY'], 'OK-ACCESS-SIGN': sig, 'OK-ACCESS-TIMESTAMP': ts, 'OK-ACCESS-PASSPHRASE': os.environ['OKX_PASSPHRASE'], 'Content-Type': 'application/json', } import requests proxies = {'http': 'http://127.0.0.1:7890', 'https': 'http://127.0.0.1:7890'} url = f'https://www.okx.com{path}{query}' if method.upper() == 'GET': r = requests.get(url, headers=headers, proxies=proxies, timeout=timeout) else: r = requests.post(url, data=body_str, headers=headers, proxies=proxies, timeout=timeout) try: return r.json() except Exception: return {'code': '-1', 'msg': f'非JSON响应: {r.text[:200]}'} def close_position_raw(symbol_usdt, signal_side): """用 raw REST 平掉同币种同方向持仓。 Args: symbol_usdt: 'ETH' / 'SKHY' / 'BTC' (base currency) signal_side: 'long' / 'short' Returns: dict: {action, pos_before, pos_after, pnl, message} """ inst_id = f"{symbol_usdt}-USDT-SWAP" # 1000PEPE 归一化为 PEPE (OKX 实际合约名) if symbol_usdt == '1000PEPE': inst_id = 'PEPE-USDT-SWAP' # 1. 查当前持仓 resp = _okx_raw_request('GET', '/api/v5/account/positions', {'instId': inst_id}) if resp.get('code') != '0': return {'action': 'error', 'message': f"查持仓失败: {resp.get('msg')}"} pos_before = None for p in resp.get('data', []): pos_size = float(p.get('pos', '0') or 0) if pos_size > 0: pos_before = { 'pos': pos_size, 'side': 'long', 'avgPx': float(p.get('avgPx', '0') or 0), 'upl': float(p.get('upl', '0') or 0), 'lever': p.get('lever', '?'), 'cTime': int(p.get('cTime', '0') or 0), # 2026-09-19: 持仓时长 } break elif pos_size < 0: pos_before = { 'pos': abs(pos_size), 'side': 'short', 'avgPx': float(p.get('avgPx', '0') or 0), 'upl': float(p.get('upl', '0') or 0), 'lever': p.get('lever', '?'), 'cTime': int(p.get('cTime', '0') or 0), } break if not pos_before: return {'action': 'none', 'message': f'无 {symbol_usdt} 持仓'} # 2. 方向二次校验 if pos_before['side'] != signal_side: return { 'action': 'skip', 'pos_before': pos_before, 'message': f"方向错位: 信号说{signal_side}但你持仓是{pos_before['side']},不动" } # 3. 取最新价做参考 tk = _okx_raw_request('GET', '/api/v5/market/ticker', {'instId': inst_id}) current_px = None if isinstance(tk, dict) and tk.get('code') == '0' and tk.get('data'): try: data0 = tk['data'][0] # 用 dict.get 链避免 LSP 类型推断 (raw JSON 实际是 dict) if isinstance(data0, dict): last_val = data0.get('last') if last_val is not None: current_px = float(last_val) except (KeyError, ValueError, TypeError): pass # 4. 市价全平 reduceOnly close_side = 'sell' if signal_side == 'long' else 'buy' body = { 'instId': inst_id, 'tdMode': 'cross', 'side': close_side, 'posSide': 'net', 'ordType': 'market', 'sz': str(pos_before['pos']), 'reduceOnly': True, } order_resp = _okx_raw_request('POST', '/api/v5/trade/order', body=body) if order_resp.get('code') == '0': return { 'action': 'closed', 'pos_before': pos_before, 'current_px': current_px, 'order_id': order_resp.get('data', [{}])[0].get('ordId'), 'message': '已市价全平', } else: return { 'action': 'error', 'pos_before': pos_before, 'message': f"下单失败: {order_resp.get('msg', order_resp)}" } def adjust_leverage_raw(symbol_usdt, new_leverage): """用 raw REST 直接设置杠杆 (跨币种通用) Args: symbol_usdt: 'ETH' / 'SKHY' / 'BTC' (base currency) new_leverage: int, 目标杠杆倍数 Returns: dict: {ok: bool, error?: str, raw: dict} """ inst_id = f"{symbol_usdt}-USDT-SWAP" if symbol_usdt == '1000PEPE': inst_id = 'PEPE-USDT-SWAP' body = {'instId': inst_id, 'lever': str(new_leverage), 'mgnMode': 'cross'} resp = _okx_raw_request('POST', '/api/v5/account/set-leverage', body=body) if resp.get('code') == '0': return {'ok': True, 'raw': resp} return {'ok': False, 'error': resp.get('msg', 'unknown'), 'raw': resp} def _fetch_close_pnl_details(symbol_usdt, entry_px, contracts, side): """从 OKX bills 取平仓相关的所有账单, 累加 = OKX app 显示值 返回: (exit_px, exit_ts_ms, real_pnl) """ try: from datetime import datetime, timezone, timedelta inst_id = f"{symbol_usdt}-USDT-SWAP" if symbol_usdt == '1000PEPE': inst_id = 'PEPE-USDT-SWAP' # 拉近 7 天的 ZEC 相关账单 (找 type=2 全部分类) since_ms = int((datetime.now() - timedelta(days=7)).timestamp() * 1000) resp = _okx_raw_request('GET', '/api/v5/account/bills', {'instId': inst_id, 'instType': 'SWAP', 'limit': '50', 'begin': str(since_ms)}) if resp.get('code') != '0' or not resp.get('data'): return (None, 0, 0) # 找最新一次"平仓"时间戳: 任何 type=2 (BTC=subType=1, ZEC=多笔 subType=2) # 用"绝对值最大"的那笔作锚点 (平仓盈亏绝对值最大) latest_ts = None latest_pnl = 0.0 latest_fill_px = None max_abs = 0.0 for b in resp['data']: t = b.get('type') if t != '2': continue st = b.get('subType') bal = float(b.get('balChg', 0) or 0) ts = int(b.get('ts', 0) or 0) # 找最大绝对值那笔 (平仓主盈亏, 排除手续费) # 排除 subType=1 的手续费 (绝对值通常 < 1) abs_val = abs(bal) if st == '1' and abs_val < 5.0: # subType=1 手续费一般 < 5 continue if abs_val > max_abs: max_abs = abs_val latest_ts = ts latest_pnl = 0.0 # 累加同一时刻 (latest_ts 前后 30 秒) 的 type=2 全部 subType if latest_ts: for b in resp['data']: ts = int(b.get('ts', 0) or 0) if abs(ts - latest_ts) > 30000: continue t = b.get('type') if t == '2': bal = float(b.get('balChg', 0) or 0) latest_pnl += bal if b.get('fillPx') and latest_fill_px is None: try: latest_fill_px = float(b['fillPx']) except (ValueError, TypeError): pass return (latest_fill_px, latest_ts or 0, latest_pnl) except Exception as e: return (None, 0, 0) def _format_hold_duration(exit_ts_ms, ctime_ms): """格式化持仓时长: 3天5小时 / 5小时22分 / 22分10秒""" from datetime import datetime, timezone if ctime_ms <= 0: return "未知" start = datetime.fromtimestamp(ctime_ms / 1000, tz=timezone.utc) end = datetime.fromtimestamp(exit_ts_ms / 1000, tz=timezone.utc) if exit_ts_ms > 0 else datetime.now(tz=timezone.utc) delta = end - start total_sec = int(delta.total_seconds()) days = total_sec // 86400 hours = (total_sec % 86400) // 3600 minutes = (total_sec % 3600) // 60 if days > 0: return f"{days}天{hours}小时" elif hours > 0: return f"{hours}小时{minutes}分" else: return f"{minutes}分" def format_close_message(fields, close_result): """平仓信号处理结果推送 (精简版:v4.5.2 禁止过度分析)。""" symbol = fields.get('symbol', '?') side_cn = fields.get('side', '?') emoji = '🟩' if fields.get('side_en') == 'long' else '🟥' leverage = fields.get('leverage', '?') trader = fields.get('trader', '?') action = close_result['action'] if action == 'closed': pos = close_result['pos_before'] # 2026-09-19: 平仓推送盈亏 + 持仓时长 (从 OKX 历史订单拿真实 pnl) entry_px = pos['avgPx'] contracts = pos['pos'] exit_px, exit_ts_ms, real_pnl = _fetch_close_pnl_details( symbol_usdt, entry_px, contracts, pos['side']) # 持仓时长 ctime_ms = pos.get('cTime', 0) hold_str = _format_hold_duration(exit_ts_ms, ctime_ms) if exit_px is not None: pnl_sign = '+' if real_pnl >= 0 else '' pnl_line = f"💰 实际盈亏: **{pnl_sign}${real_pnl:.2f}** (OKX 实测, 含手续费/funding)" exit_px_str = f"${exit_px:.2f}" else: pnl_line = "💰 实际盈亏: 拉不到历史订单, 看 OKX app" exit_px_str = "?" return f"""✅ {symbol} {side_cn} {emoji} {leverage}x 已市价全平 | {trader} {pos['side']} {contracts}张 @ {entry_px:.2f} → 平仓价 {exit_px_str} ⏱ 持仓: {hold_str} {pnl_line} 💡 OKX 历史订单实数据""" elif action == 'skip': pos = close_result['pos_before'] return f"""⏭️ {symbol} 平仓信号跳过 | {trader} 你有反向持仓 {pos['side']} {pos['pos']}张 @ {pos['avgPx']:.2f} 💡 实际盈亏请到 OKX app 查看 (扣手续费 + funding)""" elif action == 'none': return f"""ℹ️ {symbol} {side_cn} 平仓信号 | {trader} 无持仓可平""" else: # error return f"""⚠️ {symbol} {side_cn} 平仓信号处理失败 | {trader} {close_result.get('message', '未知错误')} 需手动处理""" # ─── 执行订单 ──────────────────────────────────────────────────────────── def _get_usdt_free(): """实时查 OKX USDT free (失败返回 0).""" try: from okx_position_advisor import load_credentials, create_exchange creds = load_credentials() ex = create_exchange(creds) bal = ex.fetch_balance() # 兼容 ccxt OKX 多账户结构 usdt_info = bal.get('USDT', {}) if isinstance(bal, dict) else {} free = usdt_info.get('free', 0) if isinstance(usdt_info, dict) else 0 return float(free or 0) except Exception as e: print(f"[warn] _get_usdt_free err: {e}", file=__import__('sys').stderr) return 0.0 def execute_order(symbol, side, leverage, rec): """执行开仓订单""" # advisor 内部 fetch_ticker 需要完整 symbol full_symbol = f"{symbol}/USDT:USDT" cmd = [ 'python3', str(ADVISOR), '--symbol', full_symbol, '--side', side, '--leverage', str(leverage), '--execute', '--json', '--rec-json', json.dumps(rec) ] try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=30, cwd=str(ADVISOR.parent) ) if result.returncode == 0: return json.loads(result.stdout) else: return {'error': result.stderr.strip()[:200]} except Exception as e: return {'error': str(e)} def format_execution_result(fields, rec, exec_result): """格式化执行结果""" symbol = fields.get('symbol', '?') side_cn = fields.get('side', '做多') emoji = '🟩' if fields.get('side_en') == 'long' else '🟥' leverage = fields.get('leverage', '10') trader = fields.get('trader', '?') size = fields.get('size', '?') value = fields.get('value', '?') cc = rec.get('cost_check', {}) rr = cc.get('rr_ratio', rec.get('rr', 0)) profit = cc.get('profit_amount', rec.get('tp_pnl', 0)) fee = cc.get('fee_cost', 0) fee_pct = cc.get('fee_pct', 0) net = cc.get('net_profit', 0) rating_emoji = cc.get('rating_emoji', '⚠️') rating_text = cc.get('rating_text', '未知') pos = exec_result.get('position', {}) algo = exec_result.get('algo', {}) # 根据 signal_type 显示动作 (open=新开仓, add=加仓, close=平仓, reduce=减仓) action_labels = { 'open': '新开仓', 'add': '➕ 加仓', 'reduce': '减仓', 'close': '平仓', } action_label = action_labels.get(fields.get('signal_type', 'open'), '自动开仓') msg = f"""✅ {symbol} {side_cn} {emoji} {leverage}x {action_label} 📊 信号源: {trader} {size} {symbol}(价值${value}) 📐 跟单参数 • 盈亏比: {rr}:1 ✅ • 盈利额: +{profit:.2f} USDT ✅ • 手续费: {fee:.2f} USDT ({fee_pct:.1f}%) ✅ • 净盈利: {net:.2f} USDT ✅ • 评级: {rating_emoji} {rating_text} ✅ 执行结果 • 入场: ${pos.get('entry', rec.get('price', '?'))} • 仓位: {pos.get('contracts', rec.get('contracts', '?'))}张 • TP: ${algo.get('tp', rec.get('tp_price', '?'))} • SL: ${algo.get('sl', rec.get('sl_price', '?'))} • 强平: ${pos.get('liq', '?')} ━━━ 当前全部持仓 ━━━ (查询中...)""" # 尝试获取当前全部持仓 try: acct_cmd = ['python3', '-c', f''' import sys sys.path.insert(0, "{ADVISOR.parent}") from okx_position_advisor import load_credentials, create_exchange, get_account_info creds = load_credentials() exchange = create_exchange(creds) info = get_account_info(exchange) print(f"Free: {{info['usdt_free']:.2f}}") for p in info['positions']: print(f" {{p['symbol']}}: {{p['contracts']}}张 UPL={{p['pnl']:.2f}}") '''] acct_result = subprocess.run(acct_cmd, capture_output=True, text=True, timeout=15) if acct_result.returncode == 0: msg = msg.replace("(查询中...)", f"\n```\n{acct_result.stdout.strip()}\n```") except: pass return msg # ─── 主流程 ────────────────────────────────────────────────────────────── def process_signal(text): """处理一条信号""" # 解析 fields = parse_signal(text) fields['_raw'] = text if not fields.get('symbol') or not fields.get('side'): return "⚠️ 无法解析信号" symbol = fields['symbol'] side = fields['side_en'] leverage = fields.get('leverage', '10') trader = fields.get('trader', '未知') # 2026-08-18: 信号源缺失闸 — trader='未知' 时,开/加/减仓信号跳过不跟单 # 平仓信号不阻断(平仓是兜底,即使来源不明也应执行 — 但目前 trader 字段只在 parse_signal 里标'未知', # 解析不到 symbol/side 时早就在上面 return 了;这里分类后再判断更稳) _sig_type_for_gate = classify_signal(fields) # 2026-08-21: trader 白名单闸 — 白名单放配置 whitelist.json (不要硬编码) _WL_PATH = Path(__file__).parent / 'whitelist.json' _TRADER_WHITELIST = {'熬鹰资本'} # fallback 默认值(防止配置文件损坏) try: import json as _json _wl = _json.loads(_WL_PATH.read_text()) if isinstance(_wl, dict) and isinstance(_wl.get('trader_whitelist'), list): _TRADER_WHITELIST = set(_wl['trader_whitelist']) except Exception: pass # 配置文件失败时用 fallback # 2026-09-09: 把 close 也加入白名单闸 (user 原意: 只跟熬鹰资本, 不应被"平仓兜底"绕过) if _sig_type_for_gate in ('open', 'add', 'reduce', 'adjust_leverage', 'close') and trader not in _TRADER_WHITELIST: return f"⏭️ 跳过: trader [{trader}] 不在白名单 | {symbol} {side}" # 2026-09-01: 实时仓位更新是状态推送,跳过不执行 if _sig_type_for_gate == 'update_only': return f"⏭️ 跳过: 实时仓位更新信号(非真实开/加/减/平仓) | {symbol} {side}" if trader == '未知' and _sig_type_for_gate in ('open', 'add', 'reduce'): msg = f"""⏭️ **跳过跟单: 信号源未知** ⏭️ 📊 币种: {symbol} {side} {leverage}x 🚫 信号文本里没有【交易员】/👉 跟单就选 / 任何 trader 标签 ❌ 不自动跟单,避免执行来路不明的开仓 💡 如果这条信号确实要跟, 请手动开仓 + 告诉我 trader 名字, 我会加到名单。 """ push_to_qq(msg) # 仍然记录到 tracker — 留痕, 但 outcome 标 skipped_unknown_source try: _tracker_record(trader='未知', symbol=symbol, side=side, leverage=int(leverage) if leverage else 10, trader_size=float(fields.get('size', '0').replace(',', '')), trader_entry=float(fields.get('entry', '0').replace(',', '')), trader_pnl=float(fields.get('pnl', '0').replace(',', '')), raw_text=text, outcome='skipped_unknown_source') except Exception: pass return f"⏭️ 跳过: 信号源未知 | {symbol} {side}" # 过期检查(30 分钟阈值,由 forwarder 注入的 ⏱信号时间 决定) if is_signal_stale(fields): dedup_conn = init_dedup_db() record_signal(dedup_conn, text, symbol, trader) dedup_conn.close() age = format_age_minutes(fields) msg = format_stale_message(fields, age) push_to_qq(msg) return f"⏰ 信号已过期 ({age}min) | 已推过期提醒" # 分类先于去重(让 close 信号绕过2分钟去重,因为平仓是必须执行的) signal_type = classify_signal(fields) fields['signal_type'] = signal_type # 写回 fields, 给 format_execution_result 用 # 平仓信号走独立通道 — 不看2分钟窗口,只看 raw_text hash 是否完全重复 # 修 2026-07-08 bug: 同币种同交易员的"减仓→平仓"紧跟信号被 dedup 误跳, # 导致平仓规则从未触发,持仓长期不平 if signal_type == 'close': dedup_conn = init_dedup_db() # close 信号只看 raw_text 是否完全相同(text 内含收益额+标记价,天然唯一) msg_hash = hashlib.md5(text.encode()).hexdigest() if dedup_conn.execute("SELECT 1 FROM processed WHERE msg_hash = ?", (msg_hash,)).fetchone(): dedup_conn.close() return "⏭️ 平仓信号完全重复跳过" symbol_usdt = fields['symbol'] signal_side_en = fields.get('side_en', 'long') close_result = close_position_raw(symbol_usdt, signal_side_en) # 2026-08-26: proxy抽风可能让 _okx_raw_request 返回 None, 加 null guard if not close_result: push_to_qq(f"❌ **平仓失败** (proxy 抽风?)\n" f"币种: {symbol_usdt}\n" f"请手动到 OKX 平仓,或等下次信号重试") # 2026-09-21: mark_failed 让重试机制能识别 (不再标 done, 避免再次触发造成预期外亏损) try: import sqlite3 as _sq _c = _sq.connect(os.path.expanduser('~/.hermes/trading/signal_queue.db')) _row = _c.execute("SELECT id FROM queue WHERE raw_text=? ORDER BY id DESC LIMIT 1", (text,)).fetchone() if _row: mark_failed(_row[0], "平仓失败 proxy抽风") _c.close() except Exception: pass return f"❌ 平仓失败: _okx_raw_request 返回 None (proxy 抽风?) | {symbol_usdt}" action = close_result['action'] # 2026-08-06: 无持仓的平仓信号静默 (用户要求 - 不推冗余消息) if action == 'none': # 仍记录去重 + 释放锁 (但 QQ 不推) record_signal(dedup_conn, text, symbol, trader) dedup_conn.close() from signal_tracker import release_lock release_lock(symbol_usdt, trader, auto=True) return f"⏭️ 平仓信号无持仓静默 | {symbol_usdt} {signal_side_en}" msg = format_close_message(fields, close_result) record_signal(dedup_conn, text, symbol, trader) dedup_conn.close() push_to_qq(msg) # 2026-08-06: 平仓后自动释放锁 (用户要求 - 平了肯定解锁) from signal_tracker import release_lock released, lock_msg = release_lock(symbol_usdt, trader, auto=True) if released: msg += f"\n\n🔓 锁已释放: {lock_msg}" push_to_qq(msg) return f"✅ 平仓处理: {close_result['action']} | {symbol_usdt} {signal_side_en} | {lock_msg}" # 调杠杆信号 (2026-08-26: 熬鹰调整杠杆要跟,不走advisor,直接 set-leverage) if signal_type == 'adjust_leverage': dedup_conn = init_dedup_db() if is_duplicate(dedup_conn, text, symbol, trader): dedup_conn.close() return "⏭️ 重复信号,跳过" record_signal(dedup_conn, text, symbol, trader) dedup_conn.close() symbol_usdt = fields['symbol'] new_lev = int(fields.get('leverage') or '10') lev_result = adjust_leverage_raw(symbol_usdt, new_lev) if lev_result and lev_result.get('ok'): msg = f"⚙️ **杠杆已调整** | {symbol_usdt} → {new_lev}x" push_to_qq(msg) return f"✅ 调杠杆: {symbol_usdt} → {new_lev}x" else: err = lev_result.get('error', '未知') return f"❌ 调杠杆失败: {err} | {symbol_usdt}" # 其他信号(open/reduce): 才走2分钟窗口去重 dedup_conn = init_dedup_db() if is_duplicate(dedup_conn, text, symbol, trader): dedup_conn.close() return "⏭️ 重复信号,跳过" # 平仓信号 → 已在上方独立处理,这里不再重复 # 调advisor rec = run_advisor(symbol, side, leverage) if 'error' in rec: record_signal(dedup_conn, text, symbol, trader) dedup_conn.close() return f"⚠️ advisor错误: {rec['error']}" # 性价比检查 cc = rec.get('cost_check', {}) rr = cc.get('rr_ratio', rec.get('rr', 0)) profit = cc.get('profit_amount', rec.get('tp_pnl', 0)) fee_pct = cc.get('fee_pct', 0) # 锁检查: 同币种只跟一个 trader (在 execute 之前检查) side_en = 'long' if 'long' in str(side).lower() or side == '做多' else 'short' allowed, lock_msg = check_signal_lock(symbol, trader, side_en) if not allowed: # 2026-08-06: 锁了也推 QQ (用户要求推送, 不然以为 bug) # 看 lock_msg 提取被锁的 trader + 当前信号方向 msg = f"""🔒 **跟单被锁** 🔒 📊 币种: {symbol} 📈 当前信号: {trader} {side} {leverage}x 🚫 锁信息: {lock_msg} 💡 提示: 同币种只能锁一个 trader. 如要切跟, 用 `/unlock {symbol}` 解锁或告诉我换 trader. """ push_to_qq(msg) return f"🔒 {lock_msg} | 已推 QQ" # 2026-08-26: 去掉"性价比低"判断 — 用户要求所有熬鹰信号都跟,即使 advisor 报低 contracts = rec.get('contracts', 0) auto_execute = contracts > 0 # 性价比低也不再跳过 (用户要求 2026-08-26) # 2026-09-16: hard pre-trade check (CL bug 9/14 教训) # 不可信 advisor rec['contracts'] — 复核单币种上限 try: margin_to_use = float(rec.get('margin') or 0) single_coin_avail = float(rec.get('single_coin_avail') or 0) except (TypeError, ValueError): margin_to_use = 0 single_coin_avail = 0 if margin_to_use > single_coin_avail + 1.0: push_to_qq(f"❌ 拒绝: {symbol} 单币超限 " f"(本次 ${margin_to_use:.2f} > 上限 ${single_coin_avail:.2f})") record_signal(dedup_conn, text, symbol, trader) dedup_conn.close() return f"⏭️ 单币超限跳过 | {symbol} | ${margin_to_use:.2f} > ${single_coin_avail:.2f}" if auto_execute and signal_type == 'open': # 性价比高 + 新开仓 → 自动执行 exec_result = execute_order(symbol, side, leverage, rec) if exec_result and 'error' not in exec_result: msg = format_execution_result(fields, rec, exec_result) _tracker_record(trader=trader, symbol=symbol, side=side, leverage=int(leverage) if leverage else 10, trader_size=float(fields.get('size', '0').replace(',', '')), trader_entry=float(fields.get('entry', '0').replace(',', '')), trader_pnl=float(fields.get('pnl', '0').replace(',', '')), raw_text=text, outcome='auto_executed') else: # 执行失败,降级为确认模式 auto_execute = False msg = format_message(fields, rec, signal_type) _tracker_record(trader=trader, symbol=symbol, side=side, leverage=int(leverage) if leverage else 10, trader_size=float(fields.get('size', '0').replace(',', '')), trader_entry=float(fields.get('entry', '0').replace(',', '')), trader_pnl=float(fields.get('pnl', '0').replace(',', '')), raw_text=text, outcome='pushed') elif signal_type == 'add': # 加仓信号: 走 execute_order 自动加 (advisor 内部会检查余额, 不足返回 0 张) exec_result = execute_order(symbol, side, leverage, rec) if exec_result and 'error' not in exec_result and rec.get('contracts', 0) > 0: msg = format_execution_result(fields, rec, exec_result) _tracker_record(trader=trader, symbol=symbol, side=side, leverage=int(leverage) if leverage else 10, trader_size=float(fields.get('size', '0').replace(',', '')), trader_entry=float(fields.get('entry', '0').replace(',', '')), trader_pnl=float(fields.get('pnl', '0').replace(',', '')), raw_text=text, outcome='auto_executed_add') else: # 余额不足/执行失败 → 推消息不执行 (2026-08-05: 余额不足显式提示) exec_err = (exec_result or {}).get('error', 'unknown') if exec_result else 'no result' # 看是余额不足还是其他错 is_insufficient = '余额不足' in str(exec_err) or 'insufficient' in str(exec_err).lower() # 实时查 USDT free (用 okx_position_advisor helper) usdt_free = _get_usdt_free() if is_insufficient: msg = format_message(fields, rec, signal_type) msg = f"⚠️ **跟单失败: 余额不足** ⚠️\n\n{msg}\n\n💡 你 USDT ${usdt_free:.2f} 不足以开 1 张 {symbol} 3x 杠杆 (约需 ~$46k)\n" outcome = 'pushed_add_insufficient' else: msg = format_message(fields, rec, signal_type) msg = f"⚠️ **跟单失败: 执行错误** ⚠️\n\n{msg}\n\n❌ 错误: {exec_err}\n" outcome = 'auto_executed_failed' _tracker_record(trader=trader, symbol=symbol, side=side, leverage=int(leverage) if leverage else 10, trader_size=float(fields.get('size', '0').replace(',', '')), trader_entry=float(fields.get('entry', '0').replace(',', '')), trader_pnl=float(fields.get('pnl', '0').replace(',', '')), raw_text=text, outcome=outcome) # 2026-08-05: 在 main() 标记 status=failed (signal_queue.db) — 让 cron retry 时知道 _exec_failed = exec_err elif signal_type == 'reduce': # 减仓信号: 锁了同方向 → 按大佬减仓比例, 自动减我们的同向持仓 try: import ccxt as _ccxt from okx_position_advisor import load_credentials, create_exchange creds = load_credentials() ex = create_exchange(creds) positions = ex.fetch_positions() our_pos = next((p for p in positions if symbol in p.get('symbol', '') and p.get('contracts', 0) != 0), None) if not our_pos: # 2026-09-19: 静默 (无持仓减仓信号不推 QQ, 避免刷屏) msg = None outcome = 'silent_no_pos' else: our_contracts = float(our_pos.get('contracts', 0)) # 算大佬减仓比例 last = get_last_signal(trader, symbol) trader_before = (last or {}).get('trader_size', 0) or 0 trader_after = float(fields.get('size', '0').replace(',', '')) if trader_before <= 0: msg = f"⚠️ 减仓信号: 大佬前仓位未知, 跳过 (你有 {our_contracts} 张)" else: delta_pct = (trader_before - trader_after) / trader_before # 2026-09-26: min 减仓比例过滤 (1%以下忽略, 避免蚂蚁式减仓刷屏) _MIN_REDUCE_PCT = 0.01 if delta_pct <= 0: msg = f"⚠️ 减仓信号: 大佬实际是加仓 +{delta_pct*100:.1f}%, 跳过" elif delta_pct < _MIN_REDUCE_PCT: msg = f"⏭️ 减仓幅度过小 ({delta_pct*100:.2f}% < {_MIN_REDUCE_PCT*100}%), 跳过" else: # 按比例减 reduce_amt = our_contracts * delta_pct if symbol == 'BTC': reduce_amt = max(0.01, round(reduce_amt, 2)) else: reduce_amt = max(1, round(reduce_amt)) # 如果减完 < 0.01 张, 全平 if symbol == 'BTC' and our_contracts - reduce_amt < 0.01: reduce_amt = our_contracts # 全平 # 同向减仓: long → sell, short → buy close_side = 'sell' if our_pos.get('side') == 'long' else 'buy' order = ex.create_order( symbol=f'{symbol}/USDT:USDT', type='market', side=close_side, amount=reduce_amt, params={'reduceOnly': True} ) new_contracts = our_contracts - reduce_amt msg = f"""✅ 减仓执行 | {symbol} {our_pos.get('side')} {delta_pct*100:.1f}% 📊 大佬: {trader_before:,.2f} → {trader_after:,.2f} 💼 你的: {our_contracts} → {new_contracts:.2f} 张 🔻 平 {reduce_amt} 张 (市价 {close_side})""" except Exception as e: msg = f"⚠️ 减仓失败: {e}" _tracker_record(trader=trader, symbol=symbol, side=side, leverage=int(leverage) if leverage else 10, trader_size=float(fields.get('size', '0').replace(',', '')), trader_entry=float(fields.get('entry', '0').replace(',', '')), trader_pnl=float(fields.get('pnl', '0').replace(',', '')), raw_text=text, outcome='auto_executed_reduce') else: # 需要确认或减仓信号 msg = format_message(fields, rec, signal_type) _tracker_record(trader=trader, symbol=symbol, side=side, leverage=int(leverage) if leverage else 10, trader_size=float(fields.get('size', '0').replace(',', '')), trader_entry=float(fields.get('entry', '0').replace(',', '')), trader_pnl=float(fields.get('pnl', '0').replace(',', '')), raw_text=text, outcome='pushed') # 记录去重 record_signal(dedup_conn, text, symbol, trader) dedup_conn.close() # 推送 (msg=None 表示静默跳过) if msg is None: return f"🔇 静默跳过 | {symbol} {side} {leverage}x" success = push_to_qq(msg) if success: return f"✅ 已推送 | {symbol} {side} {leverage}x | {rec['contracts']}张 | 评级:{rec.get('cost_check', {}).get('rating_text', '?').replace('性价比', '')}" else: return f"❌ 推送失败" def main(): import sys as _sys if len(_sys.argv) > 1 and _sys.argv[1] == 'unlock': # 2026-08-06: 解锁 CLI (用户要求手动解锁) from signal_tracker import release_lock, list_locks if len(_sys.argv) < 3: locks = list_locks() if not locks: print("无锁") else: print("当前锁:") for l in locks: print(f" {l['symbol']} → {l['trader']} ({l['side']})") return symbol = _sys.argv[2] trader = _sys.argv[3] if len(_sys.argv) > 3 else None released, msg = release_lock(symbol, trader, auto=True) print(msg) push_to_qq(f"🔓 手动解锁: {msg}") return if len(_sys.argv) > 1: text = ' '.join(_sys.argv[1:]) else: text = _sys.stdin.read() if not text.strip(): print("用法: python3 process_signal.py '信号文本'") print(" python3 process_signal.py unlock BTC [trader]") return result = process_signal(text) print(result) # 2026-08-05: 余额不足/执行失败 → exit code != 0 → signal_queue.retry 标 failed if '跟单失败' in str(result): _sys.exit(2) if __name__ == '__main__': main()