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