xxm
2bca726418
- 与price_monitor同模式: 新鲜实例在跑则退出, 卡死超时则SIGKILL接管 - market_watch 600s / 扫描器 1500s 阈值 - 落实2026-08-05重型任务资源约束铁律第4条
443 lines
17 KiB
Python
443 lines
17 KiB
Python
#!/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
|
||
|
||
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": 40}
|
||
|
||
TOP_N = 10
|
||
|
||
|
||
# ── 技术指标（与 backtest_framework.py 完全同算法）──
|
||
|
||
def calc_ma(series, n):
|
||
result = []
|
||
for i in range(len(series)):
|
||
if i < n - 1:
|
||
result.append(None)
|
||
else:
|
||
result.append(sum(series[i - n + 1:i + 1]) / n)
|
||
return result
|
||
|
||
|
||
def calc_rsi(series, n=14):
|
||
deltas = [series[i] - series[i - 1] for i in range(1, len(series))]
|
||
gains = [d if d > 0 else 0 for d in deltas]
|
||
losses = [-d if d < 0 else 0 for d in deltas]
|
||
result = [None] * (n + 1)
|
||
avg_gain = sum(gains[:n]) / n
|
||
avg_loss = sum(losses[:n]) / n
|
||
if avg_loss == 0:
|
||
result.append(100)
|
||
else:
|
||
rs = avg_gain / avg_loss
|
||
result.append(100 - 100 / (1 + rs))
|
||
for i in range(n, len(gains)):
|
||
avg_gain = (avg_gain * (n - 1) + gains[i]) / n
|
||
avg_loss = (avg_loss * (n - 1) + losses[i]) / n
|
||
if avg_loss == 0:
|
||
result.append(100)
|
||
else:
|
||
rs = avg_gain / avg_loss
|
||
result.append(100 - 100 / (1 + rs))
|
||
while len(result) < len(series):
|
||
result.insert(0, None)
|
||
return result[:len(series)]
|
||
|
||
|
||
def calc_atr(klines, n=14):
|
||
"""ATR（Average True Range），与回测 calc_factors 的 atr_pct 同口径"""
|
||
if len(klines) < n + 1:
|
||
return None
|
||
trs = []
|
||
for i in range(1, len(klines)):
|
||
h, l, pc = klines[i]["high"], klines[i]["low"], klines[i - 1]["close"]
|
||
tr = max(h - l, abs(h - pc), abs(l - pc))
|
||
trs.append(tr)
|
||
atr = sum(trs[-n:]) / n
|
||
return atr
|
||
|
||
|
||
def calc_obv(klines):
|
||
"""OBV 能量潮（近20日变化量），资金流向指标"""
|
||
if len(klines) < 21:
|
||
return 0
|
||
obv = 0
|
||
for i in range(1, len(klines)):
|
||
if klines[i]["close"] > klines[i - 1]["close"]:
|
||
obv += klines[i]["volume"] * 100 # 手→股
|
||
elif klines[i]["close"] < klines[i - 1]["close"]:
|
||
obv -= klines[i]["volume"] * 100
|
||
# 近20日 OBV 变化
|
||
obv_now = 0
|
||
for i in range(max(1, len(klines) - 20), len(klines)):
|
||
if klines[i]["close"] > klines[i - 1]["close"]:
|
||
obv_now += klines[i]["volume"] * 100
|
||
elif klines[i]["close"] < klines[i - 1]["close"]:
|
||
obv_now -= klines[i]["volume"] * 100
|
||
return obv_now
|
||
|
||
|
||
# ── 数据获取 ──
|
||
|
||
def fetch_tx_klines(code, datalen=120):
|
||
"""腾讯前复权日K（qfq），与 stock_daily 数据零偏差，返回 [{date,open,close,high,low,volume}]"""
|
||
raw = str(code).strip()
|
||
if raw.startswith(("6", "9")):
|
||
prefix = "sh"
|
||
elif raw.startswith(("0", "3")):
|
||
prefix = "sz"
|
||
else:
|
||
return None
|
||
url = f"http://ifzq.gtimg.cn/appstock/app/fqkline/get?param={prefix}{raw},day,,,{datalen},qfq"
|
||
try:
|
||
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
||
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||
with opener.open(req, timeout=8) as r:
|
||
text = r.read().decode("utf-8", errors="replace").strip()
|
||
data = json.loads(text)
|
||
node = data.get("data", {}).get(f"{prefix}{raw}", {})
|
||
bars = node.get("qfqday") or node.get("day") or []
|
||
if not bars or len(bars) < 70:
|
||
return None
|
||
result = []
|
||
for b in bars:
|
||
if len(b) < 6:
|
||
continue
|
||
result.append({
|
||
"date": b[0][:10],
|
||
"open": float(b[1]),
|
||
"close": float(b[2]),
|
||
"high": float(b[3]),
|
||
"low": float(b[4]),
|
||
"volume": float(b[5]), # 手
|
||
})
|
||
return result
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
# 兼容别名（供外部引用）
|
||
fetch_sina_klines = fetch_tx_klines
|
||
|
||
|
||
def get_stock_pool():
|
||
"""待扫描股票池：stock_daily 的 distinct code（与回测 run_mr_backtest 完全同口径）
|
||
|
||
回测股票池 = SELECT DISTINCT sd.code FROM stock_daily（4266只，含300/688，
|
||
不含301新创业板——数据源未收录）。实盘扫描用同一口径，保证 v_mr
|
||
信号覆盖的股票都是回测验证过的。
|
||
"""
|
||
conn = sqlite3.connect(str(DB_PATH), timeout=5)
|
||
try:
|
||
existing = set()
|
||
for r in conn.execute("SELECT code FROM holding_strategies WHERE status='active'"):
|
||
existing.add(str(r[0]))
|
||
for r in conn.execute("SELECT code FROM holdings WHERE is_active=1"):
|
||
existing.add(str(r[0]))
|
||
# 与回测完全一致：stock_daily 有K线的股票（回测 universe='a' 排除5位港股）
|
||
all_stocks = [str(r[0]) for r in
|
||
conn.execute("SELECT DISTINCT code FROM stock_daily").fetchall()]
|
||
finally:
|
||
conn.close()
|
||
# 只留 A 股（6位数字），排除港股（5位0开头）—— 与回测 is_hk_code 逻辑一致
|
||
a_stocks = [c for c in all_stocks if len(c) == 6 and c.isdigit()]
|
||
return a_stocks, existing
|
||
|
||
|
||
def load_regime():
|
||
"""读取 market_regime 最新状态"""
|
||
try:
|
||
conn = sqlite3.connect(str(DB_PATH), timeout=5)
|
||
row = conn.execute(
|
||
"SELECT date, above_ma20, adx, regime FROM market_regime "
|
||
"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")
|
||
_n = _conn.execute(
|
||
"SELECT COUNT(*) FROM candidates WHERE sector='v_mr' AND substr(created_at,1,10)=?",
|
||
(_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)
|
||
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
|
||
conn.execute(
|
||
"INSERT INTO candidates (code, name, sector, reason, "
|
||
"entry_range, stop_loss, target, 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",
|
||
(code, name, "v_mr", reasons,
|
||
f"{entry_low}~{entry_high}", sl, tp))
|
||
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()
|