MoFin/archive/hermes-dead-tools-20260820/s2v2_scanner.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""s2_panic_v2_scanner.py — S2恐慌买alpha升级 实盘扫描器（2026-08-17 择优激活落地）

由果及因（72941恐慌日信号验证）：恐慌日买强势——小市值(mcap_q<0.4)+高RSI(rsi>=35)+行业抗跌(sec_ret20>=-10)
  → 胜率60.8% vs 基线30.7%；每日top8截断（信号/成交比1.8）
基于原 s2_scanner 改：原=恐慌日买超跌(bias60<-6.8+r5f<-10+dist_lo20>=10)，数据证明无alpha（甚至负alpha）
  → v2 改为恐慌日买强势（alpha组合），评分同 s2_panic_v2_gen

入场：
  市场门控：大盘 RSI14 < 25（极端恐慌日）
  个股 alpha 组合：
    mcap_q < 0.4（小市值）
    rsi >= 35（相对强势）
    sec_ret20 >= -10（行业抗跌）
  score：mcap分(40) + rsi分(30) + sec分(20) + news分(10)
出场建议：tp30% / sl12% / max60日（s2 原出场）
输出：candidates 表（sector='s2_panic_v2'）

用法：
  python3 s2_panic_v2_scanner.py            # 完整扫描（大盘RSI<25门控）
  python3 s2_panic_v2_scanner.py --force    # 忽略门控
  python3 s2_panic_v2_scanner.py --top N    # 输出前 N 只（默认 8）
"""
import sys, sqlite3
from pathlib import Path
from datetime import datetime

sys.path.insert(0, str(Path(__file__).parent))
from indicators import calc_ma, calc_rsi
from market_data import fetch_tx_klines, get_stock_pool

DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
TOP_N = 8
EXIT_CFG = {"tp_pct": 0.30, "sl_pct": 0.12, "max_hold_days": 60}

# 大盘 RSI 门控（与 s2_scanner 一致）
MKT_RSI_MAX = 25


def load_mkt_rsi():
    """大盘 RSI14（stock_daily sh000001）"""
    try:
        conn = sqlite3.connect(str(DB_PATH), timeout=5)
        rows = conn.execute(
            "SELECT date, close FROM stock_daily WHERE code='sh000001' ORDER BY date DESC LIMIT 40").fetchall()
        conn.close()
        if len(rows) < 20:
            return None, None
        rows = list(reversed(rows))
        closes = [r[1] for r in rows]
        rsi = calc_rsi(closes)
        return rows[-1][0], rsi[-1]
    except Exception:
        return None, None


def mcap_quantile(code):
    try:
        conn = sqlite3.connect(str(DB_PATH), timeout=5)
        row = conn.execute(
            "SELECT amount FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 1", (code,)).fetchone()
        if not row or not row[0]:
            conn.close()
            return 0.3
        rows = conn.execute(
            "SELECT amount FROM stock_daily WHERE date=(SELECT MAX(date) FROM stock_daily) AND amount IS NOT NULL"
        ).fetchall()
        conn.close()
        amounts = sorted([r[0] for r in rows if r[0]])
        if not amounts:
            return 0.3
        import bisect
        return round(bisect.bisect_left(amounts, row[0]) / max(len(amounts), 1), 2)
    except Exception:
        return 0.3


def alpha_score(mcap_q, rsi, sec_ret20, news3=0):
    sc = 0
    if mcap_q is not None:
        sc += 40 if mcap_q < 0.2 else 32 if mcap_q < 0.4 else 24 if mcap_q < 0.6 else 16 if mcap_q < 0.8 else 8
    if rsi is not None:
        sc += 30 if rsi >= 45 else 22 if rsi >= 35 else 12 if rsi >= 25 else 6
    if sec_ret20 is not None:
        sc += 20 if sec_ret20 >= 0 else 16 if sec_ret20 >= -10 else 8 if sec_ret20 >= -20 else 3
    if news3:
        sc += 10 if news3 >= 2 else 7 if news3 >= 1 else 2
    return sc


def check_s2v2(klines, code):
    """s2_panic_v2 筛选：恐慌日 + 强势alpha组合"""
    if not klines or len(klines) < 70:
        return None
    closes = [k["close"] for k in klines]
    i = len(klines) - 1
    close = closes[i]
    if close <= 0:
        return None
    rsi = calc_rsi(closes)
    rsi_v = rsi[i] if i < len(rsi) else None
    mcap_q = mcap_quantile(code)
    # alpha 组合：小市值 + 强势 + （行业抗跌实盘近似简化：跳过 sec_ret20 门控）
    if mcap_q >= 0.4:
        return None
    if rsi_v is None or rsi_v < 35:
        return None
    sc = alpha_score(mcap_q, rsi_v, None)
    return {
        "price": close, "rsi": round(rsi_v, 2), "mcap_q": mcap_q, "score": sc,
        "target": round(close * (1 + EXIT_CFG["tp_pct"]), 2),
        "stop_loss": round(close * (1 - EXIT_CFG["sl_pct"]), 2),
        "date": klines[i]["date"],
    }


def main():
    import argparse
    ap = argparse.ArgumentParser()
    ap.add_argument("--force", action="store_true")
    ap.add_argument("--top", type=int, default=TOP_N)
    args = ap.parse_args()

    mkt_date, mkt_rsi = load_mkt_rsi()
    print(f"[s2_panic_v2] {datetime.now().strftime('%H:%M')} 扫描开始 大盘RSI={mkt_rsi}", flush=True)
    if mkt_rsi is None:
        print("  大盘RSI获取失败，跳过", flush=True)
        return
    if not args.force and mkt_rsi >= MKT_RSI_MAX:
        print(f"  大盘RSI={mkt_rsi:.0f} >= {MKT_RSI_MAX}，非恐慌日，跳过", flush=True)
        return

    all_stocks, existing = get_stock_pool()
    print(f"  股票池 {len(all_stocks)} 只", flush=True)
    hits = []
    for code, name in all_stocks:
        try:
            klines = fetch_tx_klines(code, datalen=120)
            sig = check_s2v2(klines, code)
            if sig:
                hits.append((code, name, sig))
        except Exception:
            pass
    hits.sort(key=lambda x: -x[2]["score"])
    hits = hits[: args.top]

    conn = sqlite3.connect(str(DB_PATH), timeout=10)
    inserted = 0
    for code, name, sig in hits:
        reasons = (f"rsi={sig['rsi']} mcap_q={sig['mcap_q']} score={sig['score']}")
        conn.execute(
            "INSERT INTO candidates (code, name, sector, reason, entry_range, stop_loss, target, created_at) "
            "VALUES (?,?,?,?,?,?,?,datetime('now','localtime')) "
            "ON CONFLICT(code) DO UPDATE SET "
            "name=excluded.name, sector=excluded.sector, reason=excluded.reason, "
            "entry_range=excluded.entry_range, stop_loss=excluded.stop_loss, target=excluded.target",
            (code, name, "s2_panic_v2", reasons,
             f"{sig['price']*0.98:.2f}~{sig['price']:.2f}", sig["stop_loss"], sig["target"]))
        inserted += 1
        print(f"  🟢 {code} {name} 价{sig['price']} score={sig['score']} {reasons}", flush=True)
    conn.commit()
    conn.close()
    print(f"  ✅ 新增 {inserted} 只 s2_panic_v2 候选", flush=True)


if __name__ == "__main__":
    main()