fix: b_td1_v3_scanner v4纯SQL——stock_indicators(加工层)+stock_daily联表,零计算零网络(<1s),架构合规

deploy/profile-scripts/b_td1_v3_scanner.py

@@ -1,40 +1,23 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-"""b_td1_v3_scanner.py — B组超跌·原池优选 实盘扫描器（2026-08-17 择优激活落地）
+"""b_td1_v3_scanner.py — B组超跌·原池优选 实盘扫描器（v4 纯 SQL，2026-08-17）
 
-由果及因（老莫指正）：b_td1 原池(5000信号)信号太密集,用 score 每日top5截断 → 信号572/成交316/比1.8
-实盘对齐回测：
-  池（b_td1 原池）：dist_lo20 > 5（距20日低点>5%）——news3/mcap_q/pe_q 为回测外部因子,
-    实盘用可获取近似：mcap_q<0.3（市值分位,从 stock_daily 市值算）pe_q<0.3 暂缺则放宽
-  score（池内超跌评分）：bias60深度 + rsi + sec_ret20 + ret5
-  每日 top5（score 降序）
-  出场建议：tp15% / sl8% / max35日（原版模拟验证参数）
+架构铁律（老莫）：数据使用层不采集、不计算指标——指标是数据加工层的事。
+  采集层 daily_kline_collector → stock_daily（原始K线）
+  加工层 factor_engine(17:05 cron) → stock_indicators（bias60/rsi/dist_lo20/r5f/mcap_q/pe_q 全算好）
+  使用层 本扫描器 → 只 SQL 查询，零计算零网络
 
-数据源：腾讯前复权日K（与 mr_scanner 同源,零偏差）+ 市值分位从 stock_daily
-输出：candidates 表（sector='b_td1_v3'）
-
-用法：
-  python3 b_td1_v3_scanner.py            # 完整扫描
-  python3 b_td1_v3_scanner.py --force    # 忽略门控
-  python3 b_td1_v3_scanner.py --top N    # 输出前 N 只（默认 5）
+stock_indicators 覆盖 3972只(08-14)，close 需联表 stock_daily 取。
 """
-import sys, json, sqlite3, time
+import sys, sqlite3
 from pathlib import Path
-from datetime import datetime
-from concurrent.futures import ThreadPoolExecutor, as_completed
 
 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 = 5
-EXIT_CFG = {"tp_pct": 0.15, "sl_pct": 0.08, "max_hold_days": 35}
 
 
 def load_regime():
-    """当前温区（平滑优先）"""
     try:
         from regime_gate import get_current_regime
         rg = get_current_regime()
@@ -51,120 +34,42 @@ def load_regime():
         return "unknown"
 
 
-def mcap_quantile(code):
-    """市值分位（mcap_q）：用 stock_fundamentals.mcap_total（总市值，亿元）
-    2026-08-17 修复：原用 stock_daily.amount 成交额不可靠（amount=None 导致全返回0.3缺省，
-    而池条件 mcap_q<0.3 会挡掉所有——扫描0候选的根因）"""
-    try:
-        conn = sqlite3.connect(str(DB_PATH), timeout=5)
-        row = conn.execute(
-            "SELECT mcap_total FROM stock_fundamentals WHERE code=? ORDER BY updated_at DESC LIMIT 1",
-            (code,)).fetchone()
+def _sql_scan():
+    """纯 SQL：stock_indicators(加工层指标) JOIN stock_daily(close) 最新完整日，池条件过滤"""
+    conn = sqlite3.connect(str(DB_PATH), timeout=10)
+    day = conn.execute(
+        "SELECT MAX(date) FROM stock_indicators WHERE date < date('now','localtime')").fetchone()[0]
+    if not day:
         conn.close()
-        if not row or not row[0]:
-            return 0.3  # 缺省中值
-        # 全市场市值分位（用基本面最新 mcap_total 全量）
-        conn = sqlite3.connect(str(DB_PATH), timeout=5)
-        rows = conn.execute(
-            "SELECT code, mcap_total FROM stock_fundamentals f WHERE updated_at = "
-            "(SELECT MAX(updated_at) FROM stock_fundamentals f2 WHERE f2.code=f.code)"
-        ).fetchall()
-        conn.close()
-        mcaps = sorted([r[1] for r in rows if r[1] and r[1] > 0])
-        if not mcaps:
-            return 0.3
-        import bisect
-        pos = bisect.bisect_left(mcaps, row[0])
-        return round(pos / max(len(mcaps), 1), 2)
-    except Exception:
-        return 0.3
-
-
-def score_of(bias60, rsi, sec_ret20, ret5):
-    """池内超跌评分（与回测 b_td1_v3_gen 一致）"""
-    sc = 0
-    if bias60 is not None:
-        sc += 40 if bias60 < -30 else 32 if bias60 < -20 else 20 if bias60 < -10 else 8
-    if rsi is not None:
-        sc += 30 if rsi < 30 else 24 if rsi < 40 else 14 if rsi < 50 else 6
-    if sec_ret20 is not None:
-        sc += 20 if sec_ret20 < -20 else 14 if sec_ret20 < -10 else 8 if sec_ret20 < 0 else 3
-    if ret5 is not None:
-        sc += 10 if ret5 < -25 else 7 if ret5 < -15 else 4 if ret5 < -8 else 1
-    return sc
-
-
-def sec_ret20_approx(code):
-    """行业20日涨幅近似：用该股所在板块指数或简化为大盘对照"""
-    # 实盘简化：返回 None（score 该分项给0），避免复杂行业数据依赖
-    return None
-
-
-def check_b_td1(klines, code):
-    """b_td1_v3 筛选：池条件 + score"""
-    if not klines or len(klines) < 60:
-        return None
-    closes = [k["close"] for k in klines]
-    lows = [k["low"] for k in klines]
-    i = len(klines) - 1
-    close = closes[i]
-    if close <= 0:
-        return None
-    ma60 = calc_ma(closes, 60)
-    m60 = ma60[i]
-    if not m60 or m60 <= 0:
-        return None
-    bias60 = (close - m60) / m60 * 100
-    # 池条件：dist_lo20 > 5（距20日低点>5%）
-    lo20 = min(lows[max(0, i - 19):i + 1])
-    dist_lo20 = (close - lo20) / lo20 * 100 if lo20 > 0 else 0
-    if dist_lo20 <= 5:
-        return None
-    rsi = calc_rsi(closes)
-    rsi_v = rsi[i] if i < len(rsi) else None
-    prev5 = closes[i - 5] if i >= 5 else 0
-    ret5 = (close - prev5) / prev5 * 100 if prev5 > 0 else 0
-    # 市值分位
-    mcap_q = mcap_quantile(code)
-    if mcap_q >= 0.3:
-        return None  # 池条件：小市值
-    # score
-    sec20 = sec_ret20_approx(code)
-    sc = score_of(bias60, rsi_v, sec20, ret5)
-    return {
-        "price": close, "bias60": round(bias60, 2), "rsi": round(rsi_v, 2) if rsi_v else None,
-        "ret5": round(ret5, 2), "dist_lo20": round(dist_lo20, 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 _local_prescreen():
-    """本地预筛：stock_fundamentals 小市值+低估值 → 返回候选 code 列表"""
-    import sqlite3 as _sq, bisect
-    _c = _sq.connect(str(DB_PATH), timeout=10)
-    # 全市场 mcap_total / pe 分位
-    _rows = _c.execute(
-        "SELECT code, mcap_total, pe FROM stock_fundamentals f WHERE updated_at = "
-        "(SELECT MAX(updated_at) FROM stock_fundamentals f2 WHERE f2.code=f.code)"
-    ).fetchall()
-    _c.close()
-    _mcaps = sorted(r[1] for r in _rows if r[1] and r[1] > 0)
-    _pes = sorted(r[2] for r in _rows if r[2] and r[2] > 0)
-    _pool = []
-    for _code, _m, _p in _rows:
-        if not _m or _m <= 0 or not _p or _p <= 0:
+        return []
+    rows = conn.execute(
+        "SELECT i.code, sd.close, i.bias60, i.rsi, i.dist_lo20, i.r5f, i.mcap_q, i.pe_q "
+        "FROM stock_indicators i JOIN stock_daily sd ON sd.code=i.code AND sd.date=i.date "
+        "WHERE i.date=? AND i.mcap_q<0.3 AND i.pe_q<0.3 "
+        "AND i.bias60<-20 AND i.rsi<40 AND i.dist_lo20>5", (day,)).fetchall()
+    conn.close()
+    hits = []
+    for code, close, b60, rsi, dist, r5, mq, pq in rows:
+        if not close or close <= 0:
             continue
-        _mq = round(bisect.bisect_left(_mcaps, _m) / max(len(_mcaps), 1), 2)
-        _pq = round(bisect.bisect_left(_pes, _p) / max(len(_pes), 1), 2)
-        if _mq < 0.3 and _pq < 0.3:
-            _pool.append(_code)
-    return _pool
+        sc = 0
+        sc += 40 if b60 < -30 else 32 if b60 < -20 else 20
+        sc += 30 if rsi < 30 else 24 if rsi < 40 else 14
+        if r5 is not None:
+            sc += 10 if r5 < -25 else 7 if r5 < -15 else 4
+        hits.append((code, code, {
+            "price": close, "bias60": round(b60, 2), "rsi": round(rsi, 2),
+            "dist_lo20": round(dist, 2), "ret5": round(r5 or 0, 2),
+            "mcap_q": mq, "score": sc, "target": round(close * 1.15, 2),
+            "stop_loss": round(close * 0.92, 2), "date": day,
+        }))
+    hits.sort(key=lambda x: -x[2]["score"])
+    return hits
 
 
 def main():
     import argparse
+    from datetime import datetime
     ap = argparse.ArgumentParser()
     ap.add_argument("--force", action="store_true")
     ap.add_argument("--top", type=int, default=TOP_N)
@@ -172,47 +77,16 @@ def main():
 
     regime = load_regime()
     print(f"[b_td1_v3] {datetime.now().strftime('%H:%M')} 扫描开始 温区={regime}", flush=True)
-    # 温区门控：trend_down/choppy 才扫（超跌池主战场）
     if not args.force and regime not in ("trend_down", "choppy"):
         print(f"  温区 {regime} 非超跌池主战场，跳过", flush=True)
         return
 
-    # ── 2026-08-17 重构 v2：mcap/pe 本地预筛 → 小池网络确认 ──
-    # 根因：全A 4000只逐只网络 fetch = 14min(串行)/9.5min(8线程) > 480s 预算
-    # 修复：本地 stock_fundamentals 先筛 小市值(mcap_q<0.3)+低估值(pe_q<0.3)
-    #   → 4000只压到数百只 → 只对这数百只网络 fetch（8线程≈1-2min，预算内）
-    pool = _local_prescreen()
-    if not pool:
-        print("  本地预筛(mcap_q<0.3+pe_q<0.3)无命中", flush=True)
-    else:
-        print(f"  本地预筛命中 {len(pool)} 只，网络确认指标（8线程）", flush=True)
-        t0 = time.time()
-        hits = []
-        def _scan(code):
-            try:
-                klines = fetch_tx_klines(code, datalen=120)
-                sig = check_b_td1(klines, code)
-                if sig:
-                    return (code, code, sig)
-            except Exception:
-                pass
-            return None
-        done = 0
-        with ThreadPoolExecutor(max_workers=8) as ex:
-            fut_map = {ex.submit(_scan, c): c for c in pool}
-            for fut in as_completed(fut_map):
-                done += 1
-                if done % 100 == 0:
-                    print(f"  [{done}/{len(pool)}] 命中{len(hits)} | {time.time()-t0:.0f}s", flush=True)
-                r = fut.result()
-                if r:
-                    hits.append(r)
-        print(f"  网络确认后命中 {len(hits)} 只（{time.time()-t0:.0f}s）", flush=True)
-    # score 降序 top-N
-    # score 降序 top-N
-    hits.sort(key=lambda x: -x[2]["score"])
+    hits = _sql_scan()
+    if not hits:
+        print("  SQL扫描无命中（最新完整日无满足池条件的小市值超跌股）", flush=True)
+        return
     hits = hits[: args.top]
-    print(f"  命中 {len(hits)} 只（score降序前{args.top}）", flush=True)
+    print(f"  SQL扫描命中 {len(hits)} 只（score降序前{args.top}）", flush=True)
 
     conn = sqlite3.connect(str(DB_PATH), timeout=10)
     inserted = 0
@@ -228,7 +102,7 @@ def main():
             (code, code, "b_td1_v3", 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)
+        print(f"  🟢 {code} {code} 价{sig['price']} score={sig['score']} {reasons}", flush=True)
     conn.commit()
     conn.close()
     print(f"  ✅ 新增 {inserted} 只 b_td1_v3 候选", flush=True)