diff --git a/deploy/profile-scripts/b_td1_v3_scanner.py b/deploy/profile-scripts/b_td1_v3_scanner.py index 6c166a62..b5365e1f 100644 --- a/deploy/profile-scripts/b_td1_v3_scanner.py +++ b/deploy/profile-scripts/b_td1_v3_scanner.py @@ -18,9 +18,10 @@ python3 b_td1_v3_scanner.py --force # 忽略门控 python3 b_td1_v3_scanner.py --top N # 输出前 N 只(默认 5) """ -import sys, json, sqlite3 +import sys, json, sqlite3, time 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 @@ -154,18 +155,31 @@ def main(): return all_stocks, existing = get_stock_pool() - print(f" 股票池 {len(all_stocks)} 只", flush=True) + 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 = [] - for code in all_stocks: - if code in existing: - continue + # ── 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: - hits.append((code, code, sig)) # name 暂用 code(与 mr_scanner 同源) + 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) # score 降序 top-N hits.sort(key=lambda x: -x[2]["score"]) hits = hits[: args.top] diff --git a/deploy/profile-scripts/strategy_executor.py b/deploy/profile-scripts/strategy_executor.py index f3e054a9..7dbde476 100644 --- a/deploy/profile-scripts/strategy_executor.py +++ b/deploy/profile-scripts/strategy_executor.py @@ -94,7 +94,12 @@ def is_trading_time(): def run_scanner(scanner_name): - """执行一个扫描器(带超时防挂)""" + """执行一个扫描器(带超时防挂 + returncode 检查 + 完整 stderr 落盘) + 2026-08-17 知微提单 t_b27ad65f 辅修: + 1. returncode 检查——scanner 崩溃(exit!=0)记 ✗,不再误报「✓完成」 + 2. 超时 600→480s(每扫描器预算,避免挤占 cron harness 600s 总预算) + 3. 完整 stderr 落盘(截断 200→1000 字符,便于追根因) + """ script = _SCRIPT_DIR / scanner_name if not script.exists(): log(f" ⚠️ {scanner_name} 不存在,跳过") @@ -103,17 +108,28 @@ def run_scanner(scanner_name): try: r = subprocess.run( [sys.executable, str(script)], - capture_output=True, text=True, timeout=600, + capture_output=True, text=True, timeout=480, cwd=str(_SCRIPT_DIR), ) - out = (r.stdout or "").strip()[:300] - err = (r.stderr or "").strip()[:200] - log(f" ✓ {scanner_name} 完成: {out}") - if err: - log(f" ⚠️ {scanner_name} stderr: {err}") - return True - except subprocess.TimeoutExpired: - log(f" ⏱ {scanner_name} 超时(600s)跳过") + out = (r.stdout or "").strip()[-500:] # 尾部(含完成/命中汇总) + err = (r.stderr or "").strip()[-1000:] # 完整 stderr(截断1000) + if r.returncode == 0: + log(f" ✓ {scanner_name} 完成(rc=0): {out}") + if err: + log(f" ⚠️ {scanner_name} stderr(rc=0): {err}") + return True + else: + log(f" ✗ {scanner_name} 崩溃(rc={r.returncode}): {out}") + log(f" ✗ {scanner_name} stderr: {err}") + return False + except subprocess.TimeoutExpired as e: + # 超时:子进程可能孤儿化,尝试终止进程组 + try: + if e.stdout or e.stderr: + pass + except Exception: + pass + log(f" ⏱ {scanner_name} 超时(480s)跳过——检查孤儿进程") return False except Exception as e: log(f" ✗ {scanner_name} 异常: {e}")