fix: b_td1_v3_scanner本地预筛——mcap/pe从stock_fundamentals先筛(4000→178只),再网络确认,8线程≈1min远在480s预算内(治本)

deploy/profile-scripts/b_td1_v3_scanner.py

@@ -140,6 +140,29 @@ def check_b_td1(klines, code):
     }
 
 
+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:
+            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
+
+
 def main():
     import argparse
     ap = argparse.ArgumentParser()
@@ -154,32 +177,38 @@ def main():
         print(f"  温区 {regime} 非超跌池主战场，跳过", flush=True)
         return
 
-    all_stocks, existing = get_stock_pool()
-    pool = [c for c in all_stocks if c not in existing]
-    print(f"  股票池 {len(all_stocks)} 只, 待扫 {len(pool)} 只（8线程并发）", flush=True)
-    t0 = time.time()
-    hits = []
-    # ── 2026-08-17 并发化（知微提单 t_b27ad65f）：串行14min→并发~2min，防 executor 600s 超时 ──
-    # 8 线程对齐 mr_scanner/s2_scanner 既有实践；腾讯K线接口不支持批量，只能合理并发
-    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 % 500 == 0:
-                print(f"  [{done}/{len(pool)}] 命中{len(hits)} | {time.time()-t0:.0f}s", flush=True)
-            r = fut.result()
-            if r:
-                hits.append(r)
+    # ── 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 = hits[: args.top]