Hermes-Skills/okx-auto-position/scripts/tg_signal_monitor.py

#!/usr/bin/env python3
"""
TG信号监听器（no_agent模式）：
从Telegram forwarder数据库读取新信号→调advisor脚本→格式化含📐→推QQ

用法: python3 tg_signal_monitor.py
配合cron: */1 * * * * python3 ~/.hermes/skills/trading/okx-auto-position/scripts/tg_signal_monitor.py
"""

import sqlite3
import json
import subprocess
import sys
import os
import re
from pathlib import Path
from datetime import datetime

# Paths
FORWARDER_DB = "/tmp/forward.db"  # TG forwarder DB (docker cp出来)
STATE_FILE = Path.home() / ".hermes/trading/.signal_monitor_state"
SKILL_DIR = Path.home() / ".hermes/skills/trading/okx-auto-position"
ADVISOR_SCRIPT = SKILL_DIR / "scripts" / "okx_position_advisor.py"
FORMAT_SCRIPT = SKILL_DIR / "scripts" / "format_signal.py"
QQ_PUSH = Path.home() / ".hermes/scripts/push_to_qq.sh"
SIGNAL_HISTORY_DB = Path.home() / ".hermes/trading/signal_history.db"

def get_last_msg_id():
    """Read last processed message ID"""
    if STATE_FILE.exists():
        return int(STATE_FILE.read_text().strip())
    return 0

def save_last_msg_id(msg_id):
    """Save last processed message ID"""
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(str(msg_id))

def parse_signal(text):
    """Parse TG signal text, extract key fields"""
    # Extract trader name
    trader_match = re.search(r'【([^】]{1,20})】', text)
    trader = trader_match.group(1) if trader_match else "未知"
    
    # Extract fields
    fields = {}
    patterns = {
        'symbol': r'【币种】\s*[:：]?\s*(\S+)',
        'side': r'【方向】\s*[:：]?\s*(做多|做空)',
        'leverage': r'【杠杆】\s*[:：]?\s*(\d+)',
        'size': r'【仓位大小】\s*[:：]?\s*([\d,.]+)',
        'value': r'【仓位价值】\s*[:：]?\s*\$?\s*([\d,.]+)',
        'entry': r'【开仓价】\s*[:：]?\s*([\d,.]+)',
        'current': r'【当前价】\s*[:：]?\s*([\d,.]+)',
        'pnl': r'【未实现盈亏】\s*[:：]?\s*([-\d,.]+)',
    }
    
    for key, pattern in patterns.items():
        match = re.search(pattern, text)
        if match:
            fields[key] = match.group(1).replace(',', '')
    
    return trader, fields

def classify_signal(fields, current_positions):
    """Classify as A(加仓) or C(新开仓)"""
    symbol = fields.get('symbol', '').replace('USDT', '').replace('/USDT', '').strip()
    for pos in current_positions:
        if symbol.upper() in pos['symbol'].upper():
            return 'A'  # 加仓
    return 'C'  # 新开仓

def run_format_script(fields, trader, signal_type):
    """Run format_signal.py and return the formatted message"""
    symbol = fields.get('symbol', '').replace('USDT', '').replace('/USDT', '').strip()
    side = 'long' if fields.get('side', '').startswith('做多') else 'short'
    leverage = fields.get('leverage', '10')
    size = fields.get('size', '0')
    value = fields.get('value', '$0')
    entry = fields.get('entry', '0')
    pnl = fields.get('pnl', '0')
    
    if not value.startswith('$'):
        value = f'${value}'
    
    cmd = [
        'python3', str(FORMAT_SCRIPT),
        '--symbol', symbol,
        '--side', side,
        '--leverage', leverage,
        '--trader', trader,
        '--trader-pos', f'{size} {symbol}',
        '--trader-value', value,
        '--trader-entry', entry,
        '--trader-pnl', pnl,
        '--signal-type', signal_type,
    ]
    
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        if result.returncode == 0:
            return result.stdout.strip()
        else:
            return f"⚠️ format_signal.py 错误: {result.stderr.strip()}"
    except subprocess.TimeoutExpired:
        return "⚠️ format_signal.py 超时"
    except Exception as e:
        return f"⚠️ 执行错误: {e}"

def push_to_qq(message):
    """Push message to QQ via push_to_qq.sh"""
    try:
        result = subprocess.run(
            ['bash', str(QQ_PUSH), message],
            capture_output=True, text=True, timeout=15
        )
        return result.returncode == 0
    except:
        return False

def log_to_db(trader, symbol, side, leverage, raw_text, outcome='pushed'):
    """Log signal to history database"""
    try:
        conn = sqlite3.connect(str(SIGNAL_HISTORY_DB))
        conn.execute("""
            INSERT INTO signals (timestamp, trader, symbol, side, leverage, raw_text, outcome)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        """, (datetime.now().isoformat(), trader, symbol, side, leverage, raw_text, outcome))
        conn.commit()
        conn.close()
    except:
        pass

def main():
    # Check if forwarder DB exists
    if not Path(FORWARDER_DB).exists():
        # Try to copy from docker
        try:
            subprocess.run(
                ['docker', 'cp', 'telegram-forwarder:/app/db/forward.db', FORWARDER_DB],
                capture_output=True, timeout=10
            )
        except:
            print("❌ 无法获取forwarder DB")
            return
    
    last_id = get_last_msg_id()
    
    try:
        conn = sqlite3.connect(FORWARDER_DB)
        conn.row_factory = sqlite3.Row
        
        # Get new messages from the forwarder
        cursor = conn.execute("""
            SELECT id, message_text, created_at 
            FROM forwarded_messages 
            WHERE id > ? AND chat_id = '-1003966251111'
            ORDER BY id ASC
            LIMIT 10
        """, (last_id,))
        
        messages = cursor.fetchall()
        conn.close()
        
        if not messages:
            return  # No new messages, silent exit
        
        for msg in messages:
            text = msg['message_text'] or ''
            msg_id = msg['id']
            
            # Skip non-signal messages
            if '【币种】' not in text and '【方向】' not in text:
                save_last_msg_id(msg_id)
                continue
            
            # Parse signal
            trader, fields = parse_signal(text)
            
            if not fields.get('symbol') or not fields.get('side'):
                save_last_msg_id(msg_id)
                continue
            
            # Classify (simplified - always treat as new for now)
            signal_type = 'C'
            
            # Run format_signal.py
            message = run_format_script(fields, trader, signal_type)
            
            if message and '⚠️' not in message:
                # Push to QQ
                success = push_to_qq(message)
                
                # Log to DB
                symbol = fields.get('symbol', '').replace('USDT', '').strip()
                side = 'long' if fields.get('side', '').startswith('做多') else 'short'
                log_to_db(trader, symbol, side, fields.get('leverage', '10'), text, 
                         'pushed' if success else 'push_failed')
            
            save_last_msg_id(msg_id)
    
    except sqlite3.OperationalError as e:
        print(f"❌ DB错误: {e}")
    except Exception as e:
        print(f"❌ 错误: {e}")

if __name__ == '__main__':
    main()