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

#!/usr/bin/env python3
"""
信号历史跟踪DB：
记录每次确认的信号，用于对比加仓/减仓趋势。

表结构：
- confirmed_signals: 已确认的信号（用户回复Y后记录）
- position_history: 仓位变化历史
"""
import sqlite3
from pathlib import Path
from datetime import datetime

DB_PATH = Path.home() / ".hermes/trading/signal_history.db"

def get_conn():
    DB_PATH.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    return conn

def init_db():
    conn = get_conn()
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS confirmed_signals (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            trader TEXT NOT NULL,
            symbol TEXT NOT NULL,
            side TEXT NOT NULL,
            leverage INTEGER,
            trader_size REAL,
            trader_entry REAL,
            trader_pnl REAL,
            our_contracts REAL,
            our_margin REAL,
            our_entry REAL,
            outcome TEXT DEFAULT 'confirmed',
            raw_text TEXT
        );

        CREATE TABLE IF NOT EXISTS position_history (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            trader TEXT NOT NULL,
            symbol TEXT NOT NULL,
            size REAL NOT NULL,
            entry_price REAL,
            pnl REAL,
            signal_type TEXT
        );

        CREATE INDEX IF NOT EXISTS idx_confirmed_trader_symbol
            ON confirmed_signals(trader, symbol, timestamp);

        CREATE INDEX IF NOT EXISTS idx_history_trader_symbol
            ON position_history(trader, symbol, timestamp);
    """)
    conn.commit()
    return conn

def record_confirmed(trader, symbol, side, leverage, trader_size, trader_entry, trader_pnl, our_contracts, our_margin, our_entry, raw_text=""):
    """记录已确认的信号"""
    conn = init_db()
    conn.execute("""
        INSERT INTO confirmed_signals
        (timestamp, trader, symbol, side, leverage, trader_size, trader_entry, trader_pnl, our_contracts, our_margin, our_entry, raw_text)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, (datetime.now().isoformat(), trader, symbol, side, leverage,
          trader_size, trader_entry, trader_pnl, our_contracts, our_margin, our_entry, raw_text[:2000]))

    conn.execute("""
        INSERT INTO position_history
        (timestamp, trader, symbol, size, entry_price, pnl, signal_type)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    """, (datetime.now().isoformat(), trader, symbol, trader_size, trader_entry, trader_pnl, 'confirmed'))

    conn.commit()
    conn.close()

def record_signal(trader, symbol, side, leverage, trader_size, trader_entry, trader_pnl, raw_text="", outcome="pushed"):
    """记录推送的信号（不管是否确认）"""
    conn = init_db()
    conn.execute("""
        INSERT INTO confirmed_signals
        (timestamp, trader, symbol, side, leverage, trader_size, trader_entry, trader_pnl, outcome, raw_text)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, (datetime.now().isoformat(), trader, symbol, side, leverage,
          trader_size, trader_entry, trader_pnl, outcome, raw_text[:2000]))

    conn.execute("""
        INSERT INTO position_history
        (timestamp, trader, symbol, size, entry_price, pnl, signal_type)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    """, (datetime.now().isoformat(), trader, symbol, trader_size, trader_entry, trader_pnl, outcome))

    conn.commit()
    conn.close()

def get_last_confirmed(trader, symbol):
    """获取上次确认的信号"""
    conn = init_db()
    row = conn.execute("""
        SELECT * FROM confirmed_signals
        WHERE trader = ? AND symbol = ? AND outcome = 'confirmed'
        ORDER BY timestamp DESC LIMIT 1
    """, (trader, symbol)).fetchone()
    conn.close()
    return dict(row) if row else None

def get_last_signal(trader, symbol):
    """获取上次推送的信号（不管是否确认）"""
    conn = init_db()
    row = conn.execute("""
        SELECT * FROM confirmed_signals
        WHERE trader = ? AND symbol = ?
        ORDER BY timestamp DESC LIMIT 1
    """, (trader, symbol)).fetchone()
    conn.close()
    return dict(row) if row else None

def get_position_trend(trader, symbol, limit=5):
    """获取仓位变化趋势"""
    conn = init_db()
    rows = conn.execute("""
        SELECT * FROM position_history
        WHERE trader = ? AND symbol = ?
        ORDER BY timestamp DESC LIMIT ?
    """, (trader, symbol, limit)).fetchall()
    conn.close()
    return [dict(r) for r in rows]

def compare_position(trader, symbol, current_size):
    """对比当前仓位与上次，返回变化描述"""
    last = get_last_signal(trader, symbol)

    if not last:
        return None, "首次出现"

    last_size = last.get('trader_size', 0)
    if not last_size or last_size == 0:
        return None, "上次仓位未知"

    change = current_size - last_size
    change_pct = (change / last_size) * 100

    if abs(change_pct) < 1:
        return last_size, "仓位不变"
    elif change > 0:
        return last_size, f"加仓 +{change_pct:.1f}%"
    else:
        return last_size, f"减仓 {change_pct:.1f}%"

def format_comparison(trader, symbol, current_size):
    """格式化对比信息"""
    last_size, desc = compare_position(trader, symbol, current_size)

    # 智能显示: 整数直接显示, 小数保留 2-3 位
    def _fmt(n):
        if n is None:
            return "?"
        if n == int(n) and abs(n) >= 10:
            return f"{int(n):,}"
        return f"{n:,.2f}"

    if last_size is None:
        return f"• {trader} {symbol}: 首次出现，仓位 {_fmt(current_size)}"

    if "不变" in desc:
        return f"• {trader} {symbol}: 仓位不变 {_fmt(current_size)}"
    elif "加仓" in desc:
        return f"• 📈 {trader} {symbol}: {_fmt(last_size)} → {_fmt(current_size)}（{desc}）"
    elif "减仓" in desc:
        delta_pct = 0
        if last_size and last_size > 0:
            delta_pct = (current_size - last_size) / last_size * 100
        return f"• 📉 {trader} {symbol}: {_fmt(last_size)} → {_fmt(current_size)}（减幅 {delta_pct:+.1f}%）"
    else:
        return f"• {trader} {symbol}: {_fmt(last_size)} → {_fmt(current_size)}（{desc}）"

# ─── 交易员统计 ──────────────────────────────────────────────────────────

def get_trader_stats(trader=None):
    """获取交易员统计数据"""
    conn = init_db()

    if trader:
        rows = conn.execute("""
            SELECT trader, symbol, side, outcome, trader_pnl, timestamp
            FROM confirmed_signals
            WHERE trader = ?
            ORDER BY timestamp DESC
        """, (trader,)).fetchall()
    else:
        rows = conn.execute("""
            SELECT trader, symbol, side, outcome, trader_pnl, timestamp
            FROM confirmed_signals
            ORDER BY trader, timestamp DESC
        """).fetchall()

    conn.close()

    # 按交易员分组
    stats = {}
    for row in rows:
        r = dict(row)
        t = r['trader']
        if t not in stats:
            stats[t] = {
                'trader': t,
                'total': 0,
                'pushed': 0,
                'confirmed': 0,
                'auto_executed': 0,
                'cancelled': 0,
                'wins': 0,
                'losses': 0,
                'total_pnl': 0,
                'trades': [],
            }
        s = stats[t]
        s['total'] += 1
        outcome = r.get('outcome', 'pushed')
        if outcome in s:
            s[outcome] += 1
        pnl = r.get('trader_pnl', 0) or 0
        s['total_pnl'] += pnl
        if pnl > 0:
            s['wins'] += 1
        elif pnl < 0:
            s['losses'] += 1
        s['trades'].append({
            'symbol': r['symbol'],
            'side': r['side'],
            'pnl': pnl,
            'outcome': outcome,
            'time': r['timestamp'],
        })

    # 计算胜率
    for t in stats:
        s = stats[t]
        decided = s['wins'] + s['losses']
        s['win_rate'] = (s['wins'] / decided * 100) if decided > 0 else 0
        s['avg_pnl'] = (s['total_pnl'] / s['total']) if s['total'] > 0 else 0

    return stats

def format_trader_rating(trader):
    """格式化交易员评分（用于推送模板）"""
    stats = get_trader_stats(trader)

    if trader not in stats or stats[trader]['total'] < 2:
        return f"📊 {trader}: 数据不足（信号<2条）"

    s = stats[trader]
    win_rate = s['win_rate']
    total = s['total']
    total_pnl = s['total_pnl']

    # 评分等级
    if win_rate >= 70:
        rating = "⭐⭐⭐⭐⭐ 精准"
    elif win_rate >= 60:
        rating = "⭐⭐⭐⭐ 可靠"
    elif win_rate >= 50:
        rating = "⭐⭐⭐ 一般"
    elif win_rate >= 40:
        rating = "⭐⭐ 谨慎"
    else:
        rating = "⭐ 高风险"

    # 最近3笔
    recent = s['trades'][:3]
    recent_str = " → ".join([
        f"{t['symbol']}{'+' if t['pnl']>0 else ''}{t['pnl']:.0f}"
        for t in recent
    ])

    return f"""📊 {trader} 胜率评级: {rating}
• 胜率: {win_rate:.0f}%（{s['wins']}胜/{s['losses']}负/{total}总）
• 总盈亏: {'+' if total_pnl>0 else ''}{total_pnl:.0f} USDT
• 最近: {recent_str}"""

def get_all_traders_summary():
    """获取所有交易员的汇总表"""
    stats = get_trader_stats()
    if not stats:
        return "暂无交易员数据"

    lines = ["| 交易员 | 胜率 | 总盈亏 | 信号数 |",
             "|--------|------|--------|--------|"]

    for t, s in sorted(stats.items(), key=lambda x: x[1]['win_rate'], reverse=True):
        win_rate = s['win_rate']
        total_pnl = s['total_pnl']
        emoji = "⭐" * min(5, max(1, int(win_rate / 20)))
        lines.append(
            f"| {t} | {emoji} {win_rate:.0f}% | {'+' if total_pnl>0 else ''}{total_pnl:.0f} | {s['total']} |"
        )

    return "\n".join(lines)

# CLI
if __name__ == '__main__':
    import sys
    if len(sys.argv) < 2:
        print("用法:")
        print("  python3 signal_tracker.py compare 麻吉大哥 HYPE 12000")
        print("  python3 signal_tracker.py history 麻吉大哥 HYPE")
        print("  python3 signal_tracker.py record 麻吉大哥 HYPE long 10 12000 70.8 -3500")
        print("  python3 signal_tracker.py rating 麻吉大哥")
        print("  python3 signal_tracker.py summary")
        sys.exit(0)

    cmd = sys.argv[1]

    if cmd == 'compare' and len(sys.argv) >= 5:
        trader = sys.argv[2]
        symbol = sys.argv[3]
        size = float(sys.argv[4])
        print(format_comparison(trader, symbol, size))

    elif cmd == 'history' and len(sys.argv) >= 4:
        trader = sys.argv[2]
        symbol = sys.argv[3]
        trend = get_position_trend(trader, symbol)
        for t in trend:
            print(f"  {t['timestamp'][:16]} | {t['size']:,.0f} | {t.get('pnl', 0):+.0f} | {t['signal_type']}")

    elif cmd == 'record' and len(sys.argv) >= 8:
        trader = sys.argv[2]
        symbol = sys.argv[3]
        side = sys.argv[4]
        leverage = int(sys.argv[5])
        size = float(sys.argv[6])
        entry = float(sys.argv[7])
        pnl = float(sys.argv[8]) if len(sys.argv) > 8 else 0
        record_signal(trader, symbol, side, leverage, size, entry, pnl)
        print(f"✅ 已记录: {trader} {symbol} {side} {leverage}x {size:,.0f} @{entry}")

    elif cmd == 'rating' and len(sys.argv) >= 3:
        trader = sys.argv[2]
        print(format_trader_rating(trader))

    elif cmd == 'summary':
        print(get_all_traders_summary())