diff --git a/deploy/profile-scripts/b_td1_v3_scanner.py b/deploy/profile-scripts/b_td1_v3_scanner.py index b5365e1f..d2be2218 100644 --- a/deploy/profile-scripts/b_td1_v3_scanner.py +++ b/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]