#!/usr/bin/env python3 """mr_scanner.py — v_weak 弱市策略实盘扫描器（2026-08-05 从 v_mr 精选版切换） v_weak 是六步法+12维框架定稿的弱市策略（docs/v_mr_strategy.md §19/§21）， 全市场 10y 验证：234笔 / wr 62.4% / avg +7.17% / 8槽组合 ret 178.3% / cagr 12.8% / dd 16.4%。 v_mr 已被证伪（2018 依赖 + 去2018 掷硬币，§12），v_weak 是其替代。 入场条件（v_weak 六条件，与回测严格对齐）： 市场门控（main）：大盘 MA20 下方（trend_down/choppy）+ 大盘 ADX ∈ [25,30) 甜区 个股： 1. bias60 ∈ [-35, -20) : 深超跌但不过深（过深易续跌） 2. RSI14 ≤ 25 : 极度超卖 3. r5f ≤ -3% : 5日急跌（下跌动能确认） 4. dist_lo20 < 5% : 距20日低点 < 5%（下方有支撑承接，§17） 5. 信号日收阳 : 入场重评确认=资金进场（§19） 6. 缩量 vol5/vol20<1.0 : 入场重评确认=抛压衰竭（§19） 出场建议：tp=+30% / sl=-12% / max_hold=40 交易日（RR 2.5:1） 排序：bias60 升序（最深超跌优先，§15 自然序=隐式质量排序） 候选写入 sector='v_mr'（管道槽位名不变，下游兼容） 市场门控（2026-08-02 新增）： - 只读 market_regime 表，regime 为 trend_down 或 choppy 时启用扫描 （v_mr 主战场：下跌趋势 + 震荡市；趋势市让位 v_next4） - trend_up 时跳过（v_next4 追涨主战场，不扫超跌） 数据源：腾讯前复权日K（qfq），datalen=120（覆盖 MA60 + 60日回看 + RSI 收敛） 指标算法：与 backtest_framework.py 完全一致（内联，零偏差） 输出：candidates 表（sector='v_mr'），与 accumulation_scanner 同 UPSERT 模式 用法： python3 mr_scanner.py # 完整扫描（regime 门控） python3 mr_scanner.py --force # 忽略 regime 门控强制扫描 python3 mr_scanner.py --top N # 输出前 N 只（默认 10） """ import sys, json, urllib.request, re, time, sqlite3 from pathlib import Path from datetime import datetime # 2026-08-11 重构：工具函数抽取到公共模块（indicators/market_data） # 原 calc_ma/calc_rsi/calc_atr/calc_obv → indicators.py # 原 fetch_tx_klines/get_stock_pool → market_data.py # 保留本文件的 import 兼容（从公共模块导入同名函数） from indicators import calc_ma, calc_rsi, calc_atr, calc_obv from market_data import fetch_tx_klines, get_stock_pool DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") UA = "Mozilla/5.0" # ── v_weak 参数（2026-08-03 六步法+12维框架定稿，见 docs/v_mr_strategy.md §19/§21）── # 弱市(大盘MA20下) + 大盘ADX∈[25,30]甜区(门控在main) + 个股六条件 WEAK_CFG = { "bias60_min": -35, # bias60 ∈ [-35, -20)：深超跌但不过深（过深易续跌） "bias60_max": -20, "rsi_max": 25, # RSI ≤ 25：极度超卖 "r5f_max": -3, # 5日急跌 ≤ -3%（下跌动能确认） "dist_lo20_max": 5, # 距20日低点 < 5%（下方有支撑承接） "vol_shrink_max": 1.0, # 缩量 vol5/vol20 < 1.0（抛压衰竭，入场重评确认） "require_yang": True, # 信号日收阳（资金进场，入场重评确认） } EXIT_CFG = {"tp_pct": 0.30, "sl_pct": 0.12, "max_hold_days": 20} # 2026-08-17分温区: 均值回归族trend_down 20d最优(v_mr实证) TOP_N = 10 def load_regime(): """读取当前温区。优先平滑温区（regime_gate K=5），回退原始 market_regime。 adx 仍需从原始表补（甜区门控用）。""" # 平滑温区优先 try: from regime_gate import get_current_regime _rg = get_current_regime() if _rg and _rg.get("regime") != "unknown": _base = {"date": _rg.get("date", ""), "regime": _rg.get("regime"), "smoothed": True} # 补 adx/above_ma20（甜区门控需要） try: conn = sqlite3.connect(str(DB_PATH), timeout=5) conn.execute("PRAGMA busy_timeout=30000") # 2026-08-18 整点撞锁等待 row = conn.execute( "SELECT date, above_ma20, adx, regime FROM market_regime " "WHERE market='a' ORDER BY date DESC LIMIT 1").fetchone() conn.close() if row: _base["above_ma20"] = bool(row[1]) _base["adx"] = row[2] except Exception: pass return _base except Exception: pass # 回退原始 try: conn = sqlite3.connect(str(DB_PATH), timeout=5) conn.execute("PRAGMA busy_timeout=30000") # 2026-08-18 整点撞锁等待 row = conn.execute( "SELECT date, above_ma20, adx, regime FROM market_regime " "WHERE market='a' ORDER BY date DESC LIMIT 1").fetchone() conn.close() if row: return {"date": row[0], "above_ma20": bool(row[1]), "adx": row[2], "regime": row[3]} except Exception: pass return None # ── v_mr 筛选（精选版：基线6条件 + 正面4因子 + 剔除4负面）── def check_vmr(klines, mkt_adx=None): """对单只股票做 v_weak 入场筛选（2026-08-05 从 v_mr 精选版切换）。 六条件：bias60∈[-35,-20) + RSI≤25 + 5日急跌≤-3% + 距20日低点<5% + 收阳 + 缩量。 klines 为升序日K。mkt_adx 保留参数兼容（甜区门控在 main）。""" if not klines or len(klines) < 70: return None closes = [k["close"] for k in klines] highs = [k["high"] for k in klines] lows = [k["low"] for k in klines] vols = [k["volume"] for k in klines] i = len(klines) - 1 # 最新一日 close = closes[i] if close <= 0: return None ma60 = calc_ma(closes, 60) ma20 = calc_ma(closes, 20) rsi_all = calc_rsi(closes) rsi = rsi_all[i] if i < len(rsi_all) else None m60 = ma60[i] if not m60 or m60 <= 0 or rsi is None: return None # 1. bias60 ∈ [-35, -20)：深超跌但不过深 bias60 = (close - m60) / m60 * 100 if not (WEAK_CFG["bias60_min"] <= bias60 < WEAK_CFG["bias60_max"]): return None # 2. RSI ≤ 25：极度超卖 if rsi > WEAK_CFG["rsi_max"]: return None # 3. 5日急跌 ≤ -3%（下跌动能确认） prev5 = closes[i - 5] if i >= 5 else 0 r5f = (close - prev5) / prev5 * 100 if prev5 > 0 else 0 if r5f > WEAK_CFG["r5f_max"]: return None # 4. 距20日低点 < 5%（下方有支撑承接，§17 老莫洞察） lo20 = min(lows[max(0, i - 19):i + 1]) dist_lo20 = (close - lo20) / lo20 * 100 if lo20 > 0 else 999 if dist_lo20 >= WEAK_CFG["dist_lo20_max"]: return None # 5. 信号日收阳（入场重评确认：资金进场，§19） if WEAK_CFG["require_yang"] and close <= klines[i]["open"]: return None # 6. 缩量 vol5/vol20 < 1.0（入场重评确认：抛压衰竭，§19） vol5 = sum(vols[max(0, i - 4):i + 1]) / 5 if i >= 4 else 0 vol20 = sum(vols[max(0, i - 19):i + 1]) / 20 if i >= 19 else 0 vol_shrink = vol5 / vol20 if vol20 > 0 else 99 if vol_shrink >= WEAK_CFG["vol_shrink_max"]: return None # 命中 → 出场建议（精选版 exit_cfg） tp_pct = EXIT_CFG["tp_pct"] sl_pct = EXIT_CFG["sl_pct"] target = round(close * (1 + tp_pct), 2) stop = round(close * (1 - sl_pct), 2) return { "price": close, "bias60": round(bias60, 2), "rsi": round(rsi, 2), "r5f": round(r5f, 2), "dist_lo20": round(dist_lo20, 2), "vol_shrink": round(vol_shrink, 3), "mkt_adx": mkt_adx, "target": target, "stop_loss": stop, "date": klines[i]["date"], } def _singleton_guard(max_age_sec, script_tag): """自愈式单例守卫（2026-08-05 进程堆积事故后统一加装）""" import subprocess as _sp, os as _os, sys as _sys my_pid = _os.getpid() try: out = _sp.run(["ps", "-C", "python3", "-o", "pid,etimes,cmd"], capture_output=True, text=True, timeout=10).stdout for line in out.splitlines(): if script_tag not in line: continue parts = line.split(None, 2) if len(parts) < 3: continue try: pid = int(parts[0]); age = int(parts[1]) except ValueError: continue if pid == my_pid: continue if age > max_age_sec: try: _os.kill(pid, 9) print(f"[guard] SIGKILL卡死实例 pid={pid} age={age}s", flush=True) except ProcessLookupError: pass else: print(f"[guard] 已有新鲜实例 pid={pid} age={age}s 在跑, 本实例退出", flush=True) _sys.exit(0) except Exception as _e: print(f"[guard] 守卫异常(放行): {_e}", flush=True) def main(): force = "--force" in sys.argv _singleton_guard(1500, "mr_scanner.py") top_n = TOP_N if "--top" in sys.argv: try: top_n = int(sys.argv[sys.argv.index("--top") + 1]) except (ValueError, IndexError): pass print(f"[MR] {datetime.now().strftime('%H:%M')} v_mr 实盘扫描开始", flush=True) # ── regime 门控 + 大盘 ADX（负面因子用 + 甜区门控 2026-08-03 12维发现）── regime = load_regime() mkt_adx = None if regime: rg = regime["regime"] mkt_adx = regime["adx"] print(f" 市场阶段: {regime['date']} → {rg} (adx={regime['adx']})", flush=True) if rg not in ("trend_down", "choppy") and not force: print(f" ⏭ {rg} 非 v_mr 主战场（trend_down/choppy 才扫），跳过", flush=True) return if rg not in ("trend_down", "choppy"): print(f" ⚠ --force 强制扫描（当前 {rg}）", flush=True) # 12维框架验证(2026-08-03): 弱市深超跌edge的前提是大盘ADX∈[25,30]（甜区） # ADX<20 avg+1.63% / 20-25 -1.24% / 25-30 +4.89% / 30-40 -0.04% / >=40 -0.10% # 非甜区时 v_mr 信号质量显著下降，默认跳过（--force 可强制） if mkt_adx is not None and not (25 <= mkt_adx < 30) and not force: print(f" ⏭ 大盘ADX={mkt_adx:.1f} 非甜区[25,30)（12维验证：强跌/弱跌市超跌信号质量差），跳过", flush=True) return if mkt_adx is not None and not (25 <= mkt_adx < 30): print(f" ⚠ --force 强制扫描（当前ADX={mkt_adx:.1f} 非甜区）", flush=True) else: print(" ⚠ market_regime 不可用，默认执行扫描（v_mr 全市场可用）", flush=True) # ── 幂等检查：当天已扫过 v_mr 则跳过（避免 30 分钟调度重复全扫描）── import sqlite3 as _sq _conn = _sq.connect(str(DB_PATH), timeout=5) try: _today = datetime.now().strftime("%Y-%m-%d") # 幂等只认扫描器自己写的候选(reason以v_weak/v_mr精选开头) # —— watchlist_auto_exit 回归池也写 sector='v_mr'，不能算数(2026-08-05 bug修复) _n = _conn.execute( "SELECT COUNT(*) FROM candidates WHERE sector='v_mr' AND substr(created_at,1,10)=?" " AND (reason LIKE 'v_weak%' OR reason LIKE 'v_mr精选%')", (_today,)).fetchone()[0] except Exception: _n = 0 _conn.close() if _n > 0 and not force: print(f" 已有 {_n} 条今日 v_mr 候选，跳过重复扫描（--force 可强制）", flush=True) return # ── 股票池 ── all_stocks, existing = get_stock_pool() print(f" 股票池: {len(all_stocks)}只A股, 已有策略: {len(existing)}只", flush=True) if not all_stocks: print(" ⚠ stocks 表为空", flush=True) return # 并发拉日K（ThreadPool 8 并发） from concurrent.futures import ThreadPoolExecutor, as_completed pool = [c for c in all_stocks if c not in existing] found = [] done = 0 with ThreadPoolExecutor(max_workers=8) as ex: fut_map = {ex.submit(fetch_sina_klines, c): c for c in pool} for fut in as_completed(fut_map): code = fut_map[fut] done += 1 klines = fut.result() if klines: sig = check_vmr(klines, mkt_adx=mkt_adx) if sig: found.append((code, sig)) if done % 400 == 0: print(f" 已扫描 {done}/{len(pool)}", flush=True) print(f" 命中 v_weak 条件: {len(found)} 只", flush=True) # 排序：同分按偏度，精选因子已强过滤，按 rsi_delta 降序（止跌确认最强优先） found.sort(key=lambda x: x[1]["bias60"]) # 最深超跌优先（§15 自然序） # ── 写 candidates 表（UPSERT，保留计算列）── conn = sqlite3.connect(str(DB_PATH), timeout=5) conn.execute("PRAGMA busy_timeout=30000") # 2026-08-18 整点撞锁等待 inserted = 0 for code, sig in found[:top_n]: name = code try: r = conn.execute("SELECT name FROM stocks WHERE code=?", (code,)).fetchone() if r and r[0]: name = r[0] except Exception: pass price = sig["price"] entry_low = round(price * 0.97, 2) entry_high = round(price * 1.02, 2) sl = sig["stop_loss"] tp = sig["target"] reasons = (f"v_weak(bias60={sig['bias60']}% rsi={sig['rsi']} " f"r5f={sig['r5f']}% dist_lo20={sig['dist_lo20']}% " f"vol={sig['vol_shrink']} 收阳+缩量确认)") # 检查是否已在 candidates 且未 promoted exists = conn.execute( "SELECT code FROM candidates WHERE code=? AND (promoted IS NULL OR promoted=0)", (code,)).fetchone() if exists: continue _mid_v = (entry_low + entry_high) / 2 _rr_v = round((tp - _mid_v) / (_mid_v - sl), 2) if _mid_v > sl > 0 else 0 conn.execute( "INSERT INTO candidates (code, name, sector, reason, " "entry_range, stop_loss, target, rr, source_strategy, created_at) " "VALUES (?,?,?,?,?,?,?,?,?,datetime('now','localtime')) " "ON CONFLICT(code) DO UPDATE SET " "name=excluded.name, sector=excluded.sector, reason=excluded.reason, " "entry_range=excluded.entry_range, stop_loss=excluded.stop_loss, target=excluded.target, rr=excluded.rr, source_strategy=excluded.source_strategy", (code, name, "v_mr", reasons, f"{entry_low}~{entry_high}", sl, tp, _rr_v, "v_mr")) inserted += 1 print(f" 🟢 {code} {name} 价{price} bias60={sig['bias60']}% " f"rsi={sig['rsi']} r5f={sig['r5f']}% {reasons}", flush=True) conn.commit() conn.close() print(f" ✅ 新增 {inserted} 只 v_weak 候选（前 {top_n}）", flush=True) if __name__ == "__main__": main()