#!/usr/bin/env python3 """market_regime.py — 大盘市场阶段判断（实盘版） 解决实盘"无法判断当前是什么市"的断层： 回测用 _load_index_ctx('sh000001') 算 above_ma20/ma20_slope/roc/adx 做市场过滤 （v_next4 要求 above_ma20=True + adx>=20 才出手，v_mr 用 mkt_mode='any'）， 但实盘 strategy_lifecycle 只有当日涨跌幅 mood（±1%/1.5%→仓位乘数 0.8~1.1）， 没有 MA20+ADX 的市场阶段判断 —— 本模块把回测算法原样搬到实盘。 算法与 /home/hmo/MoFin/strategy_lab.py 的 _load_index_ctx 完全一致： - above_ma20 : 收盘价 > MA20 - ma20_slope : MA20 相对 5 日前 MA20 的变化率(%) - roc : 10 日 Rate of Change(%) - adx : 简易趋势强度 (DI+ - DI-)/(DI+ + DI-) 归一化，14 日窗口 （calc_trend_strength，backtest_framework.py 87 行） regime 分类（与回测 v_next4/v_mr 分工对齐）： - trend_up : above_ma20=True 且 adx>=20 → 趋势市，v_next4 主战场（追涨有效） - choppy : adx<20 → 震荡市，v_mr 主战场（超跌反弹） - trend_down : above_ma20=False 且 adx>=20 → 下跌趋势，v_mr 主战场（深超跌） 用法： python3 market_regime.py # 计算并写 market_regime 表（供定时调度） python3 market_regime.py --print # 只打印当前市场状态 作为库: from market_regime import compute_regime, load_market_regime 写表: market_regime((date, market) PK, above_ma20, ma20_slope, roc, adx, regime, close, created_at) market='a'(A股 sh000001) / 'hk'(港股 hkHSI)，同一交易日两个市场各一条（2026-08-14 阶段3 双市场） 数据源: mofin.db stock_daily 的 sh000001/hkHSI（每日收盘后更新， 盘中用最新可得日线，未收盘日不计入最终判断，adx 用真实历史）。 """ import sys import json import sqlite3 from pathlib import Path from datetime import datetime # ── 路径注入：可被 deploy/profile-scripts 下脚本直接 import ── _SCRIPT_DIR = Path(__file__).resolve().parent _MOFIN_ROOT = _SCRIPT_DIR.parent.parent # deploy/profile-scripts → MoFin for _p in (str(_SCRIPT_DIR), str(_MOFIN_ROOT)): if _p not in sys.path: sys.path.insert(0, _p) DB_PATH = Path(_MOFIN_ROOT) / "data" / "mofin.db" INDEX_CODE = "sh000001" # 上证指数（A股） INDEX_CODE_HK = "hkHSI" # 恒生指数（港股） # ADX 阈值与回测 v_next4 的 mkt_adx_min=20 对齐 ADX_TREND_MIN = 20.0 def calc_trend_strength(highs, lows, closes, n=14): """简易趋势强度（替代 ADX）: (DI+ - DI-) / (DI+ + DI-) 归一化 与 backtest_framework.calc_trend_strength 完全同算法（内联，避免跨目录依赖）""" tr = [] for i in range(len(highs)): if i == 0: tr.append(highs[i] - lows[i]) else: tr.append(max(highs[i] - lows[i], abs(highs[i] - closes[i - 1]), abs(lows[i] - closes[i - 1]))) up = [highs[i] - highs[i - 1] for i in range(1, len(highs))] down = [lows[i - 1] - lows[i] for i in range(1, len(lows))] di_plus_raw = [0.0] * len(up) di_minus_raw = [0.0] * len(up) for i in range(len(up)): if up[i] > down[i] and up[i] > 0: di_plus_raw[i] = up[i] if down[i] > up[i] and down[i] > 0: di_minus_raw[i] = down[i] tr_val = tr[i + 1] if (i + 1) < len(tr) else tr[-1] if tr_val > 0: di_plus_raw[i] = di_plus_raw[i] / tr_val * 100 di_minus_raw[i] = di_minus_raw[i] / tr_val * 100 result = [] for i in range(len(di_plus_raw)): if i < n - 1: result.append(None) else: avg_plus = sum(di_plus_raw[i - n + 1:i + 1]) / n avg_minus = sum(di_minus_raw[i - n + 1:i + 1]) / n if avg_plus + avg_minus > 0: dx = abs(avg_plus - avg_minus) / (avg_plus + avg_minus) * 100 else: dx = 0 result.append(dx) return [None] * (len(highs) - len(result)) + result 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_roc(series, n=10): result = [] for i in range(len(series)): if i < n: result.append(None) else: result.append((series[i] - series[i - n]) / series[i - n] * 100 if series[i - n] != 0 else 0) return result def compute_regime(index_code=INDEX_CODE, db_path=None, lookback_days=120): """计算大盘市场状态（最近一个完整交易日）。 返回 dict： {date, close, above_ma20, ma20_slope, roc, adx, regime, computed_at} 数据不足(<30行)或取数失败时返回 None。 """ db = db_path or DB_PATH conn = sqlite3.connect(str(db), timeout=5) try: rows = conn.execute( "SELECT date, close, high, low FROM stock_daily " "WHERE code=? ORDER BY date DESC LIMIT ?", (index_code, lookback_days)).fetchall() finally: conn.close() if not rows or len(rows) < 30: return None # 升序处理（与回测 prepare_bars 一致） rows = list(reversed(rows)) dates = [r[0] for r in rows] closes = [r[1] for r in rows] highs = [r[2] for r in rows] lows = [r[3] for r in rows] ma20 = calc_ma(closes, 20) trend = calc_trend_strength(highs, lows, closes) roc = calc_roc(closes) i = len(closes) - 1 # 最新一日 close = closes[i] m20 = ma20[i] above_ma20 = (close > m20) if m20 else None # ma20_slope: 相对 5 日前 MA20 的变化率 slope = None if i >= 5 and ma20[i - 5] and ma20[i - 5] > 0 and m20: slope = round((m20 - ma20[i - 5]) / ma20[i - 5] * 100, 3) adx = trend[i] if i < len(trend) else None roc_v = roc[i] if i < len(roc) else None # regime 分类 if above_ma20 is True and adx is not None and adx >= ADX_TREND_MIN: regime = "trend_up" elif adx is not None and adx < ADX_TREND_MIN: regime = "choppy" else: regime = "trend_down" return { "date": dates[i], "close": close, "above_ma20": above_ma20, "ma20_slope": slope, "roc": round(roc_v, 3) if roc_v is not None else None, "adx": round(adx, 2) if adx is not None else None, "regime": regime, "computed_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } def save_regime(regime, market='a', db_path=None): """写入 market_regime 表（按 (date, market) 去重，同日同市场覆盖）""" if not regime: return False db = db_path or DB_PATH conn = sqlite3.connect(str(db), timeout=5) try: conn.execute(""" CREATE TABLE IF NOT EXISTS market_regime ( date TEXT NOT NULL, market TEXT NOT NULL DEFAULT 'a', above_ma20 INTEGER, ma20_slope REAL, roc REAL, adx REAL, regime TEXT, close REAL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (date, market) ) """) conn.execute(""" INSERT OR REPLACE INTO market_regime (date, market, above_ma20, ma20_slope, roc, adx, regime, close, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) """, ( regime["date"], market, 1 if regime["above_ma20"] else 0, regime.get("ma20_slope"), regime.get("roc"), regime.get("adx"), regime["regime"], regime.get("close"), )) conn.commit() finally: conn.close() return True def load_market_regime(db_path=None, market='a'): """读取指定市场最新 market_regime（供 strategy_lifecycle 等消费方调用）。 market='a' 默认 → 与改造前行为完全一致（A 股温区）""" db = db_path or DB_PATH conn = sqlite3.connect(str(db), timeout=5) try: row = conn.execute( "SELECT date, above_ma20, ma20_slope, roc, adx, regime, close " "FROM market_regime WHERE market=? ORDER BY date DESC LIMIT 1", (market,)).fetchone() finally: conn.close() if not row: return None return { "date": row[0], "above_ma20": bool(row[1]), "ma20_slope": row[2], "roc": row[3], "adx": row[4], "regime": row[5], "close": row[6], } REGIME_DESC = { "trend_up": "趋势市（大盘MA20上方+ADX≥20）：v_next4 追涨主战场", "choppy": "震荡市（ADX<20）：v_mr 超跌反弹主战场", "trend_down": "下跌趋势（大盘MA20下方+ADX≥20）：v_mr 深超跌主战场", } def main(): # 双市场：A股(sh000001) + 港股(hkHSI)，各自独立温区 results = {} for market, index_code in (("a", INDEX_CODE), ("hk", INDEX_CODE_HK)): r = compute_regime(index_code=index_code) if not r: print(f"[market_regime] {index_code} 数据不足，跳过 {market} 市场", flush=True) continue results[market] = (index_code, r) if "--print" in sys.argv: for market, (idx, regime) in results.items(): print(json.dumps(regime, ensure_ascii=False, indent=2)) print(f"[{market}] {idx} 判断: {REGIME_DESC.get(regime['regime'], '')}") return 0 if results else 1 ok_any = False for market, (idx, regime) in results.items(): ok = save_regime(regime, market=market) if ok: ok_any = True print(f"[market_regime] {market} {regime['date']} → {regime['regime']} " f"(above_ma20={regime['above_ma20']} adx={regime['adx']} " f"slope={regime['ma20_slope']} roc={regime['roc']})") else: print(f"[market_regime] {market} 写入失败") if not ok_any: print("[market_regime] 两个市场均写入失败/数据不足") return 1 return 0 if __name__ == "__main__": sys.exit(main())