MoFin/deploy/profile-scripts/signal_ledger.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""signal_ledger.py — 信号溯源系统（2026-08-13 老莫要求）

核心：记录每次股票被推荐的原因（策略/版本/时间/温区），多策略同推标记共振加关注。

表 signal_ledger：
  id, code, name, strategy, version, regime, temp_band, pushed_at, reason,
  resonance_count, resonance_strategies, source_module, updated_at

共振检测：同一股票 24h 内被多个激活策略推 → resonance_count>1，标记"多策略共振"（加关注）

用法：
  from signal_ledger import record_signal, get_resonance
  record_signal(code="300750", name="宁德时代", strategy="v_oversold", version="v_oversold",
                regime="trend_down", temp_band="panic", reason="进买入区+重评买入", source_module="price_monitor")
  res = get_resonance("300750")  # 返回该股票近24h共振信息
"""
import json
import sqlite3
from pathlib import Path
from datetime import datetime, timedelta

DB = "/home/hmo/MoFin/data/mofin.db"
RESONANCE_WINDOW_H = 24  # 共振窗口（小时）


def _conn():
    conn = sqlite3.connect(DB, timeout=30)
    conn.execute("PRAGMA busy_timeout=30000")
    return conn


def init_table():
    conn = _conn()
    conn.execute("""
        CREATE TABLE IF NOT EXISTS signal_ledger (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            code TEXT NOT NULL,
            name TEXT,
            strategy TEXT,
            version TEXT,
            regime TEXT,
            temp_band TEXT,
            pushed_at TIMESTAMP,
            reason TEXT,
            resonance_count INTEGER DEFAULT 1,
            resonance_strategies TEXT,
            source_module TEXT,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_signal_ledger_code ON signal_ledger(code)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_signal_ledger_pushed ON signal_ledger(pushed_at)")
    conn.commit()
    conn.close()


def record_signal(code, name="", strategy="", version="", regime="", temp_band="",
                  reason="", source_module=""):
    """记录一次信号推送，并检测多策略共振"""
    init_table()
    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    conn = _conn()
    try:
        # 共振检测：近 24h 内同一股票被其他策略推过
        cutoff = (datetime.now() - timedelta(hours=RESONANCE_WINDOW_H)).strftime("%Y-%m-%d %H:%M:%S")
        rows = conn.execute(
            """SELECT DISTINCT strategy FROM signal_ledger
               WHERE code=? AND pushed_at >= ? AND strategy != ? AND strategy != ''""",
            (code, cutoff, strategy)
        ).fetchall()
        other_strats = [r[0] for r in rows if r[0]]
        resonance_count = len(other_strats) + (1 if strategy else 0)
        resonance_strategies = ",".join(sorted(set([strategy] + other_strats))) if strategy else ""

        conn.execute(
            """INSERT INTO signal_ledger
               (code, name, strategy, version, regime, temp_band, pushed_at, reason,
                resonance_count, resonance_strategies, source_module)
               VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
            (code, name, strategy, version, regime, temp_band, now, reason,
             resonance_count, resonance_strategies, source_module)
        )
        conn.commit()
        return {"resonance_count": resonance_count, "resonance_strategies": resonance_strategies}
    finally:
        conn.close()


def get_resonance(code, hours=RESONANCE_WINDOW_H):
    """查某股票近 N 小时的共振信息"""
    init_table()
    cutoff = (datetime.now() - timedelta(hours=hours)).strftime("%Y-%m-%d %H:%M:%S")
    conn = _conn()
    try:
        rows = conn.execute(
            """SELECT strategy, version, regime, pushed_at, reason FROM signal_ledger
               WHERE code=? AND pushed_at >= ? ORDER BY pushed_at DESC""",
            (code, cutoff)
        ).fetchall()
        if not rows:
            return None
        strats = sorted({r[0] for r in rows if r[0]})
        return {
            "code": code,
            "count": len(strats),
            "strategies": strats,
            "latest": rows[0][3],
            "signals": [{"strategy": r[0], "version": r[1], "regime": r[2], "at": r[3], "reason": r[4]} for r in rows],
        }
    finally:
        conn.close()


def get_recent_signals(limit=50):
    """查最近推送的信号（供评估）"""
    init_table()
    conn = _conn()
    try:
        rows = conn.execute(
            """SELECT code, name, strategy, regime, temp_band, pushed_at, reason,
                      resonance_count, resonance_strategies, source_module
               FROM signal_ledger ORDER BY pushed_at DESC LIMIT ?""",
            (limit,)
        ).fetchall()
        return [
            {"code": r[0], "name": r[1], "strategy": r[2], "regime": r[3], "temp": r[4],
             "at": r[5], "reason": r[6], "resonance": r[7], "res_strats": r[8], "source": r[9]}
            for r in rows
        ]
    finally:
        conn.close()


if __name__ == "__main__":
    init_table()
    # 自测
    r1 = record_signal("300750", "宁德时代", "v_oversold", "v_oversold", "trend_down", "panic", "进买入区", "test")
    print("单策略:", r1)
    r2 = record_signal("300750", "宁德时代", "v_mr_sel", "v_mr_sel", "trend_down", "panic", "超跌信号", "test")
    print("多策略共振:", r2)
    print("共振查询:", get_resonance("300750"))