fix: b_td1_v3_scanner 8线程并发(14min→2min防超时)+executor returncode检查/480s超时/完整stderr(知微提单t_b27ad65f)

This commit is contained in:
xxm
2026-08-17 11:40:30 +08:00
parent b6558bf9da
commit cc45f604f5
2 changed files with 46 additions and 16 deletions
+20 -6
View File
@@ -18,9 +18,10 @@
python3 b_td1_v3_scanner.py --force # 忽略门控 python3 b_td1_v3_scanner.py --force # 忽略门控
python3 b_td1_v3_scanner.py --top N # 输出前 N 只(默认 5) python3 b_td1_v3_scanner.py --top N # 输出前 N 只(默认 5)
""" """
import sys, json, sqlite3 import sys, json, sqlite3, time
from pathlib import Path from pathlib import Path
from datetime import datetime from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
sys.path.insert(0, str(Path(__file__).parent)) sys.path.insert(0, str(Path(__file__).parent))
from indicators import calc_ma, calc_rsi from indicators import calc_ma, calc_rsi
@@ -154,18 +155,31 @@ def main():
return return
all_stocks, existing = get_stock_pool() 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 = [] hits = []
for code in all_stocks: # ── 2026-08-17 并发化(知微提单 t_b27ad65f):串行14min→并发~2min,防 executor 600s 超时 ──
if code in existing: # 8 线程对齐 mr_scanner/s2_scanner 既有实践;腾讯K线接口不支持批量,只能合理并发
continue def _scan(code):
try: try:
klines = fetch_tx_klines(code, datalen=120) klines = fetch_tx_klines(code, datalen=120)
sig = check_b_td1(klines, code) sig = check_b_td1(klines, code)
if sig: if sig:
hits.append((code, code, sig)) # name 暂用 code(与 mr_scanner 同源) return (code, code, sig)
except Exception: except Exception:
pass 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 # score 降序 top-N
hits.sort(key=lambda x: -x[2]["score"]) hits.sort(key=lambda x: -x[2]["score"])
hits = hits[: args.top] hits = hits[: args.top]
+26 -10
View File
@@ -94,7 +94,12 @@ def is_trading_time():
def run_scanner(scanner_name): 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 script = _SCRIPT_DIR / scanner_name
if not script.exists(): if not script.exists():
log(f" ⚠️ {scanner_name} 不存在,跳过") log(f" ⚠️ {scanner_name} 不存在,跳过")
@@ -103,17 +108,28 @@ def run_scanner(scanner_name):
try: try:
r = subprocess.run( r = subprocess.run(
[sys.executable, str(script)], [sys.executable, str(script)],
capture_output=True, text=True, timeout=600, capture_output=True, text=True, timeout=480,
cwd=str(_SCRIPT_DIR), cwd=str(_SCRIPT_DIR),
) )
out = (r.stdout or "").strip()[:300] out = (r.stdout or "").strip()[-500:] # 尾部(含完成/命中汇总)
err = (r.stderr or "").strip()[:200] err = (r.stderr or "").strip()[-1000:] # 完整 stderr(截断1000
log(f"{scanner_name} 完成: {out}") if r.returncode == 0:
if err: log(f"{scanner_name} 完成(rc=0): {out}")
log(f" ⚠️ {scanner_name} stderr: {err}") if err:
return True log(f" ⚠️ {scanner_name} stderr(rc=0): {err}")
except subprocess.TimeoutExpired: return True
log(f"{scanner_name} 超时(600s)跳过") 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 return False
except Exception as e: except Exception as e:
log(f"{scanner_name} 异常: {e}") log(f"{scanner_name} 异常: {e}")