feat: 事项五落地——策略组合+市场测温+动态切换+失效预警(一次性完成) 市场测温四态(bull_trend/bull_osc/bear/neutral)+策略路由权重+龙头识别框架+三振出局; v_weak近90日胜率22.7%红牌; bull_osc胜率60%最有效(当前状态); bear占111/250天导致近1年不佳
This commit is contained in:
xxm
@@ -0,0 +1,103 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""leader_scanner.py — MoFin 龙头识别策略（bull_trend 适用，2026-08-13 落地）
|
||||
|
||||
核心逻辑（12维方法论 + 龙头识别低波动优化版理念）：
|
||||
- 适用状态：bull_trend（above_ma20 + rsi>55 + adx>20）
|
||||
- 入场：龙头股回调到 MA20 附近（趋势中的健康回调，不追高）
|
||||
- 12维条件（复用 stock_indicators 已有字段，无拍脑袋）：
|
||||
1. 个股 close > MA20（趋势向上）
|
||||
2. dist_ma20 ∈ [-3%, +2%]（回调到 MA20 附近，不追高）
|
||||
3. 行业强势（sector_above_ma20=1 或 sector_ret20>0，需 sector 数据）
|
||||
4. 市值/流动性过滤（mcap_q > 0.3，龙头非小盘）
|
||||
5. RSI ∈ [45, 65]（强势但未超买）
|
||||
6. 量能配合（vol_ratio > 0.8，非极度缩量）
|
||||
- 出场：跌破 MA20 或达到上方压力位
|
||||
- RR：压力位（前高/筹码阻力）/ 支撑位（MA20）计算
|
||||
|
||||
注意：这是初版框架，需 MoFin 引擎验证达标（年化≥13.2%）才正式上线
|
||||
"""
|
||||
import sqlite3
|
||||
import json
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
DB = "/home/hmo/MoFin/data/mofin.db"
|
||||
OUT = Path("/home/hmo/MoFin/data/leader_signals.json")
|
||||
|
||||
def load_market_state():
|
||||
p = Path("/home/hmo/MoFin/data/market_state.json")
|
||||
if p.exists():
|
||||
return json.loads(p.read_text(encoding="utf-8"))
|
||||
return {"state": "neutral"}
|
||||
|
||||
def scan_leaders():
|
||||
"""扫描龙头股回调买点（bull_trend 状态）"""
|
||||
ms = load_market_state()
|
||||
state = ms.get("state", "neutral")
|
||||
if state != "bull_trend":
|
||||
print(f"当前状态 {state}，非 bull_trend，不扫描龙头（避免追高）")
|
||||
return []
|
||||
|
||||
c = sqlite3.connect(DB)
|
||||
# 最新交易日
|
||||
row = c.execute("SELECT MAX(date) FROM stock_indicators").fetchone()
|
||||
if not row or not row[0]:
|
||||
c.close()
|
||||
return []
|
||||
latest = row[0]
|
||||
print(f"扫描日期: {latest}")
|
||||
|
||||
# 龙头条件（12维）
|
||||
rows = c.execute(
|
||||
"""SELECT code, ma20, rsi, dist_ma20, mcap_q, pe_q, vol_ratio, trend_aligned
|
||||
FROM stock_indicators
|
||||
WHERE date=? AND ma20 IS NOT NULL AND rsi IS NOT NULL""",
|
||||
(latest,)
|
||||
).fetchall()
|
||||
c.close()
|
||||
|
||||
signals = []
|
||||
for r in rows:
|
||||
code, ma20, rsi, dist_ma20, mcap_q, pe_q, vol_ratio, trend_aligned = r
|
||||
# 条件1: 趋势向上（close > MA20 → dist_ma20 > 0，或接近）
|
||||
if dist_ma20 is None or dist_ma20 < -3 or dist_ma20 > 2:
|
||||
continue
|
||||
# 条件2: RSI 强势未超买
|
||||
if rsi < 45 or rsi > 65:
|
||||
continue
|
||||
# 条件3: 市值/流动性（龙头非小盘，mcap_q > 0.3）
|
||||
if mcap_q is not None and mcap_q < 0.3:
|
||||
continue
|
||||
# 条件4: 量能配合
|
||||
if vol_ratio is not None and vol_ratio < 0.8:
|
||||
continue
|
||||
# 条件5: 趋势共振（trend_aligned=1）
|
||||
if trend_aligned != 1:
|
||||
continue
|
||||
signals.append({
|
||||
"code": code, "ma20": ma20, "rsi": rsi,
|
||||
"dist_ma20": dist_ma20, "mcap_q": mcap_q, "pe_q": pe_q,
|
||||
"vol_ratio": vol_ratio, "entry_reason": "龙头回调MA20",
|
||||
})
|
||||
|
||||
# 按 dist_ma20 排序（最接近 MA20 的优先）
|
||||
signals.sort(key=lambda x: abs(x["dist_ma20"]))
|
||||
print(f"龙头信号: {len(signals)} 只")
|
||||
for s in signals[:5]:
|
||||
print(f" {s['code']}: dist_ma20={s['dist_ma20']:.1f}% rsi={s['rsi']:.1f} mcap_q={s['mcap_q']}")
|
||||
return signals
|
||||
|
||||
def main():
|
||||
signals = scan_leaders()
|
||||
out = {
|
||||
"state": load_market_state().get("state", "neutral"),
|
||||
"signals": signals,
|
||||
"count": len(signals),
|
||||
"updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
}
|
||||
OUT.write_text(json.dumps(out, ensure_ascii=False, indent=1), encoding="utf-8")
|
||||
print(f"leader_signals.json 写入")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,176 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""market_thermometer.py — MoFin 市场测温模块（2026-08-13 一次性落地）
|
||||
|
||||
核心理念（市场周期测温文/霍华德·马克斯）：周期像钟摆无法预测，但可测温——
|
||||
不是预测拐点，是判断当前摆到哪（牛/熊/震荡/结构性），据此调整策略攻守。
|
||||
|
||||
四态判定（用 market_indicators 已有字段，无拍脑袋）：
|
||||
- bull_trend: above_ma20=1 + rsi>55 + adx>20（MA20上方+强势+强趋势）→ 龙头/趋势策略
|
||||
- bull_osc: above_ma20=1 + rsi>50（MA20上方+偏强+弱趋势）→ 震荡偏强，均衡配置
|
||||
- bear: above_ma20=0 + rsi<45（MA20下方+弱势）→ 弱市均值回复策略（v_weak/v_oversold）
|
||||
- neutral: 其他（中性震荡）→ 观望/轻仓
|
||||
|
||||
输出：market_state.json（当前状态+历史分态统计），供策略动态切换
|
||||
"""
|
||||
import sqlite3
|
||||
import json
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
DB_PATH = "/home/hmo/MoFin/data/mofin.db"
|
||||
OUT_PATH = "/home/hmo/MoFin/data/market_state.json"
|
||||
|
||||
# ── 回填 market_indicators 历史（从 stock_daily 计算上证指数指标）──
|
||||
def backfill_market_indicators(days=500):
|
||||
"""从 stock_daily 计算上证指数（000001.SH）的 mkt_* 指标，回填到 market_indicators"""
|
||||
c = sqlite3.connect(DB_PATH)
|
||||
# 上证指数代码（MoFin 约定：000001 = 上证指数）
|
||||
rows = c.execute(
|
||||
"SELECT date, open, high, low, close, volume FROM stock_daily WHERE code='000001' ORDER BY date DESC LIMIT ?",
|
||||
(days,)
|
||||
).fetchall()
|
||||
if not rows:
|
||||
print("stock_daily 无 000001 数据")
|
||||
return 0
|
||||
|
||||
bars = list(reversed(rows)) # 时间正序
|
||||
n = len(bars)
|
||||
if n < 60:
|
||||
print(f"数据不足 {n} 条，无法计算")
|
||||
return 0
|
||||
|
||||
# 计算 MA20 / RSI14 / ADX14 / 60日高点回撤
|
||||
closes = [b[4] for b in bars]
|
||||
highs = [b[2] for b in bars]
|
||||
lows = [b[3] for b in bars]
|
||||
dates = [b[0] for b in bars]
|
||||
|
||||
# MA20
|
||||
ma20 = [None] * n
|
||||
for i in range(19, n):
|
||||
ma20[i] = sum(closes[i-19:i+1]) / 20
|
||||
|
||||
# RSI14
|
||||
rsi = [None] * n
|
||||
gains, losses = [], []
|
||||
for i in range(1, n):
|
||||
ch = closes[i] - closes[i-1]
|
||||
gains.append(max(ch, 0))
|
||||
losses.append(max(-ch, 0))
|
||||
if i >= 14:
|
||||
avg_g = sum(gains[i-14:i]) / 14
|
||||
avg_l = sum(losses[i-14:i]) / 14
|
||||
rs = avg_g / avg_l if avg_l > 0 else 100
|
||||
rsi[i] = 100 - 100 / (1 + rs)
|
||||
|
||||
# ADX14（简化：用 DMI 近似，实际 MoFin 有 indicators.calc_adx，这里用简化版）
|
||||
adx = [None] * n
|
||||
tr_list = [0.0]
|
||||
pdm, ndm = [0.0], [0.0]
|
||||
for i in range(1, n):
|
||||
h, l, pc = highs[i], lows[i], closes[i-1]
|
||||
tr = max(h - l, abs(h - pc), abs(l - pc))
|
||||
tr_list.append(tr)
|
||||
up_move = highs[i] - highs[i-1]
|
||||
down_move = lows[i-1] - lows[i]
|
||||
pdm.append(up_move if up_move > down_move and up_move > 0 else 0)
|
||||
ndm.append(down_move if down_move > up_move and down_move > 0 else 0)
|
||||
for i in range(14, n):
|
||||
atr = sum(tr_list[i-14:i]) / 14
|
||||
pdi = 100 * sum(pdm[i-14:i]) / 14 / atr if atr > 0 else 0
|
||||
ndi = 100 * sum(ndm[i-14:i]) / 14 / atr if atr > 0 else 0
|
||||
dx = 100 * abs(pdi - ndi) / (pdi + ndi) if (pdi + ndi) > 0 else 0
|
||||
adx[i] = dx # 简化：用 DX 近似 ADX（平滑需更多数据，足够测温）
|
||||
|
||||
# 60日高点回撤
|
||||
dd60 = [None] * n
|
||||
for i in range(59, n):
|
||||
hi60 = max(closes[i-59:i+1])
|
||||
dd60[i] = (closes[i] - hi60) / hi60 * 100 if hi60 > 0 else 0
|
||||
|
||||
# 写入 market_indicators（UPSERT）
|
||||
written = 0
|
||||
for i in range(14, n):
|
||||
if ma20[i] is None or rsi[i] is None:
|
||||
continue
|
||||
above = 1 if closes[i] > ma20[i] else 0
|
||||
# 近20日涨跌（roc）
|
||||
roc = ((closes[i] - closes[i-20]) / closes[i-20] * 100) if i >= 20 and closes[i-20] > 0 else 0
|
||||
# 近20日斜率（简化）
|
||||
slope = (ma20[i] - ma20[i-5]) / ma20[i-5] * 100 if i >= 5 and ma20[i-5] else 0
|
||||
c.execute(
|
||||
"""INSERT OR REPLACE INTO market_indicators
|
||||
(date, mkt_rsi, mkt_dd60, mkt_adx, mkt_above_ma20, mkt_down_days, mkt_slope, mkt_roc, updated_at)
|
||||
VALUES (?,?,?,?,?,?,?,?,?)""",
|
||||
(dates[i], rsi[i], dd60[i], adx[i], above, 0, slope, roc, datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
|
||||
)
|
||||
written += 1
|
||||
c.commit()
|
||||
c.close()
|
||||
print(f"回填 market_indicators {written} 条（最新: {dates[-1]}）")
|
||||
return written
|
||||
|
||||
# ── 四态判定 ──
|
||||
def classify_state(mkt_row):
|
||||
"""判定市场状态。输入：market_indicators 行 dict"""
|
||||
rsi = mkt_row.get("mkt_rsi", 50)
|
||||
adx = mkt_row.get("mkt_adx", 20)
|
||||
above = mkt_row.get("mkt_above_ma20", 0)
|
||||
dd60 = mkt_row.get("mkt_dd60", 0)
|
||||
|
||||
if above and rsi > 55 and adx > 20:
|
||||
return "bull_trend"
|
||||
if above and rsi > 50:
|
||||
return "bull_osc"
|
||||
if not above and rsi < 45:
|
||||
return "bear"
|
||||
return "neutral"
|
||||
|
||||
# ── 主流程 ──
|
||||
def main():
|
||||
# 1. 回填历史（500 日 ≈ 2 年）
|
||||
backfill_market_indicators(500)
|
||||
|
||||
# 2. 当前状态
|
||||
c = sqlite3.connect(DB_PATH)
|
||||
row = c.execute(
|
||||
"SELECT date, mkt_rsi, mkt_dd60, mkt_adx, mkt_above_ma20, mkt_slope, mkt_roc FROM market_indicators ORDER BY date DESC LIMIT 1"
|
||||
).fetchone()
|
||||
c.close()
|
||||
if not row:
|
||||
print("market_indicators 无数据")
|
||||
return
|
||||
|
||||
current = {
|
||||
"date": row[0], "mkt_rsi": row[1], "mkt_dd60": row[2],
|
||||
"mkt_adx": row[3], "mkt_above_ma20": row[4], "mkt_slope": row[5], "mkt_roc": row[6],
|
||||
}
|
||||
state = classify_state(current)
|
||||
current["state"] = state
|
||||
current["state_desc"] = {
|
||||
"bull_trend": "牛市趋势（龙头/趋势策略重仓）",
|
||||
"bull_osc": "强势震荡（均衡配置）",
|
||||
"bear": "熊市/下跌（弱市均值回复策略 v_weak/v_oversold）",
|
||||
"neutral": "中性震荡（观望/轻仓）",
|
||||
}[state]
|
||||
|
||||
# 3. 历史分态统计（近 250 交易日 ≈ 1 年）
|
||||
c = sqlite3.connect(DB_PATH)
|
||||
rows = c.execute(
|
||||
"SELECT date, mkt_rsi, mkt_adx, mkt_above_ma20, mkt_dd60 FROM market_indicators ORDER BY date DESC LIMIT 250"
|
||||
).fetchall()
|
||||
c.close()
|
||||
from collections import Counter
|
||||
hist = Counter(classify_state({"mkt_rsi": r[1], "mkt_adx": r[2], "mkt_above_ma20": r[3], "mkt_dd60": r[4]}) for r in rows)
|
||||
current["hist_1y"] = dict(hist)
|
||||
current["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
# 4. 写入 market_state.json
|
||||
Path(OUT_PATH).write_text(json.dumps(current, ensure_ascii=False, indent=1), encoding="utf-8")
|
||||
print(f"market_state.json 写入: {state} ({current['state_desc']})")
|
||||
print(f" 近1年分态: {dict(hist)}")
|
||||
print(f" 当前指标: rsi={current['mkt_rsi']:.1f} adx={current['mkt_adx']:.1f} above={current['mkt_above_ma20']} dd60={current['mkt_dd60']:.1f}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,87 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""strategy_alert.py — MoFin 策略失效预警（三振出局，2026-08-13 落地）
|
||||
|
||||
核心理念（机器学习策略攻防体系文）：
|
||||
- 三振出局：黄牌（减半）→ 橙牌（1/4）→ 红牌（清仓淘汰）
|
||||
- 五类失效预警：胜率持续下降 / 盈亏比恶化 / 波动新高 / 信号质量突变 / 风格漂移
|
||||
- 原则：不是 IC 掉到负就删，是降权到观察模式（权重设 0 保留，恢复可启用）
|
||||
|
||||
数据源：strategy_health（已有）+ strategy_research（回测）+ 实盘持仓表现
|
||||
"""
|
||||
import sqlite3
|
||||
import json
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
DB = "/home/hmo/MoFin/data/mofin.db"
|
||||
OUT = Path("/home/hmo/MoFin/data/strategy_alerts.json")
|
||||
|
||||
# 三振阈值（机器学习策略文）：
|
||||
# 黄牌：滚动20日风险调整收益连续3日 < -0.5，或单日波动 > 5%
|
||||
# 橙牌：黄牌后5日未回升零以上 → 权重降至1/4
|
||||
# 红牌：橙牌后5日持续不佳，或累计亏损 > 10% → 清仓移除
|
||||
# MoFin 简化版（基于回测/健康度，实盘数据不足时用回测滚动胜率）：
|
||||
YELLOW_WIN_RATE = 0.40 # 滚动胜率 < 40% → 黄牌（v_weak 近1年 39.3% 已触发）
|
||||
ORANGE_WIN_RATE = 0.35 # 滚动胜率 < 35% → 橙牌
|
||||
RED_WIN_RATE = 0.30 # 滚动胜率 < 30% → 红牌（淘汰）
|
||||
ORANGE_PNL_RATIO = 0.5 # 盈亏比 < 0.5（赚的越来越少亏的越来越多）→ 橙牌
|
||||
|
||||
def rolling_stats(version, days=60):
|
||||
"""从 strategy_research 提取该策略近期交易的滚动胜率/盈亏比"""
|
||||
c = sqlite3.connect(DB)
|
||||
rows = c.execute(
|
||||
"SELECT results_json FROM strategy_research WHERE version=? ORDER BY period_tag DESC LIMIT 1",
|
||||
(version,)
|
||||
).fetchall()
|
||||
c.close()
|
||||
if not rows:
|
||||
return None
|
||||
try:
|
||||
res = json.loads(rows[0][0])
|
||||
trades = res.get("trades", [])
|
||||
cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
|
||||
recent = [t for t in trades if t.get("entry_date", "") >= cutoff]
|
||||
if len(recent) < 5:
|
||||
return None
|
||||
wins = [t for t in recent if t.get("profit_pct", 0) > 0]
|
||||
losses = [t for t in recent if t.get("profit_pct", 0) <= 0]
|
||||
win_rate = len(wins) / len(recent) if recent else 0
|
||||
avg_win = sum(t.get("profit_pct", 0) for t in wins) / len(wins) if wins else 0
|
||||
avg_loss = abs(sum(t.get("profit_pct", 0) for t in losses) / len(losses)) if losses else 1
|
||||
pnl_ratio = avg_win / avg_loss if avg_loss > 0 else 0
|
||||
return {
|
||||
"n": len(recent), "win_rate": win_rate,
|
||||
"avg_win": avg_win, "avg_loss": avg_loss, "pnl_ratio": pnl_ratio,
|
||||
}
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def assess(version, stats):
|
||||
"""三振评估"""
|
||||
if not stats:
|
||||
return {"level": "ok", "action": "数据不足，无法评估", "weight": 1.0}
|
||||
wr, pr = stats["win_rate"], stats["pnl_ratio"]
|
||||
if wr < RED_WIN_RATE:
|
||||
return {"level": "red", "action": f"红牌：滚动胜率{wr:.1%}<30%，淘汰（权重0，观察模式）", "weight": 0.0}
|
||||
if wr < ORANGE_WIN_RATE or pr < ORANGE_PNL_RATIO:
|
||||
return {"level": "orange", "action": f"橙牌：胜率{wr:.1%}或盈亏比{pr:.2f}恶化，权重降1/4", "weight": 0.25}
|
||||
if wr < YELLOW_WIN_RATE:
|
||||
return {"level": "yellow", "action": f"黄牌：滚动胜率{wr:.1%}<40%，权重减半", "weight": 0.5}
|
||||
return {"level": "ok", "action": f"正常：胜率{wr:.1%}，盈亏比{pr:.2f}", "weight": 1.0}
|
||||
|
||||
def main():
|
||||
alerts = {}
|
||||
for version in ["v_weak", "v_oversold"]:
|
||||
stats = rolling_stats(version, days=90)
|
||||
a = assess(version, stats)
|
||||
alerts[version] = {**a, "stats": stats}
|
||||
print(f"{version}: {a['level']} | {a['action']} | weight={a['weight']}")
|
||||
if stats:
|
||||
print(f" 近90日: {stats['n']}笔, 胜率{stats['win_rate']:.1%}, 盈亏比{stats['pnl_ratio']:.2f}")
|
||||
alerts["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
OUT.write_text(json.dumps(alerts, ensure_ascii=False, indent=1), encoding="utf-8")
|
||||
print(f"\nstrategy_alerts.json 写入")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,74 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""strategy_router.py — MoFin 策略动态路由（2026-08-13 一次性落地）
|
||||
|
||||
核心：读 market_state.json 的当前市场状态，决定各策略权重/开关
|
||||
- bear: v_weak/v_oversold 降权 50%（信号质量差，28%胜率）
|
||||
- bull_osc: v_weak/v_oversold 正常（60%胜率，当前状态）
|
||||
- bull_trend: v_weak/v_oversold 降权 50%，启用龙头策略（待研究）
|
||||
- neutral: 轻仓观望
|
||||
|
||||
输出：strategy_weights.json 供扫描器/重评脚本读取
|
||||
"""
|
||||
import json
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
MARKET_STATE = Path("/home/hmo/MoFin/data/market_state.json")
|
||||
OUT = Path("/home/hmo/MoFin/data/strategy_weights.json")
|
||||
|
||||
def load_market_state():
|
||||
if MARKET_STATE.exists():
|
||||
return json.loads(MARKET_STATE.read_text(encoding="utf-8"))
|
||||
return {"state": "neutral", "date": ""}
|
||||
|
||||
def route(state):
|
||||
"""市场状态 → 策略权重。返回 dict"""
|
||||
# 基准权重（等权，程飞：不确定时等权最稳健）
|
||||
base = {"v_weak": 1.0, "v_oversold": 1.0, "leader": 0.0} # leader 待研究，先 0
|
||||
if state == "bear":
|
||||
# 熊市：v_weak/v_oversold 信号质量差（近1年 bear 胜率 28%），降权
|
||||
return {
|
||||
"v_weak": {"weight": 0.5, "action": "降权50%", "reason": "bear 胜率28%，信号质量差"},
|
||||
"v_oversold": {"weight": 0.5, "action": "降权50%", "reason": "bear 胜率28%"},
|
||||
"leader": {"weight": 0.0, "action": "停用", "reason": "bear 不适用龙头"},
|
||||
"state": state,
|
||||
}
|
||||
if state == "bull_osc":
|
||||
# 强势震荡：v_weak/v_oversold 最有效（60%胜率），正常
|
||||
return {
|
||||
"v_weak": {"weight": 1.0, "action": "正常", "reason": "bull_osc 胜率60%，最有效"},
|
||||
"v_oversold": {"weight": 1.0, "action": "正常", "reason": "bull_osc 适用"},
|
||||
"leader": {"weight": 0.0, "action": "停用", "reason": "bull_osc 非趋势市"},
|
||||
"state": state,
|
||||
}
|
||||
if state == "bull_trend":
|
||||
# 牛市趋势：v_weak/v_oversold 失效（30%胜率），启用龙头策略
|
||||
return {
|
||||
"v_weak": {"weight": 0.5, "action": "降权50%", "reason": "bull_trend 胜率30%"},
|
||||
"v_oversold": {"weight": 0.5, "action": "降权50%", "reason": "bull_trend 失效"},
|
||||
"leader": {"weight": 1.0, "action": "启用", "reason": "bull_trend 适用龙头"},
|
||||
"state": state,
|
||||
}
|
||||
# neutral：轻仓观望
|
||||
return {
|
||||
"v_weak": {"weight": 0.5, "action": "轻仓", "reason": "neutral 观望"},
|
||||
"v_oversold": {"weight": 0.5, "action": "轻仓", "reason": "neutral 观望"},
|
||||
"leader": {"weight": 0.0, "action": "停用", "reason": "neutral 观望"},
|
||||
"state": state,
|
||||
}
|
||||
|
||||
def main():
|
||||
ms = load_market_state()
|
||||
state = ms.get("state", "neutral")
|
||||
weights = route(state)
|
||||
weights["market_state"] = ms
|
||||
weights["updated_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
OUT.write_text(json.dumps(weights, ensure_ascii=False, indent=1), encoding="utf-8")
|
||||
print(f"strategy_weights.json 写入: {state}")
|
||||
for k, v in weights.items():
|
||||
if isinstance(v, dict) and "weight" in v:
|
||||
print(f" {k}: weight={v['weight']} ({v['action']}) - {v['reason']}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,151 @@
|
||||
# 事项五解决方案：策略组合 + 市场测温 + 动态切换
|
||||
|
||||
> 2026-08-13 一次性落地。老莫指示：不分短期中期，全部立即完成。
|
||||
|
||||
---
|
||||
|
||||
## 一、根因诊断（数据铁证）
|
||||
|
||||
### 1.1 分市场状态检验结果
|
||||
|
||||
用 market_thermometer 回填 481 日 market_indicators（上证指数），把 v_weak/v_oversold 历史交易按入场日市场状态分组：
|
||||
|
||||
**v_weak 近1年（2025-08-13后，250交易日）**：
|
||||
| 市场状态 | 天数占比 | 笔数 | 胜率 | 均盈 |
|
||||
|---|---|---|---|---|
|
||||
| **bull_osc（强势震荡）** | 43/250 | 304 | **60.2%** | **+5.05%** |
|
||||
| bull_trend（牛市趋势） | 47/250 | 95 | 30.5% | -0.61% |
|
||||
| bear（熊市/下跌） | 111/250 | 223 | 28.3% | -1.88% |
|
||||
| neutral（中性） | 49/250 | 155 | 28.4% | -1.64% |
|
||||
|
||||
**v_weak 全历史（2021+）**：
|
||||
| 市场状态 | 笔数 | 胜率 | 均盈 |
|
||||
|---|---|---|---|
|
||||
| bull_osc | 441 | **54.4%** | +4.18% |
|
||||
| bull_trend | 234 | 44.4% | +2.47% |
|
||||
| neutral | 356 | 44.1% | +1.78% |
|
||||
| bear | 370 | 37.8% | +0.48% |
|
||||
|
||||
### 1.2 核心结论
|
||||
|
||||
1. **v_weak 本质是"强势震荡市反弹策略"**（bull_osc 胜率 60%/54%），不是"熊市抄底策略"（bear 胜率仅 28%/38%）
|
||||
2. **近1年 bear 占一半时间（111/250 天）**——v_weak 一半时间信号质量差，整体胜率被拉低到 39.3%（vs 10y 53.6%）
|
||||
3. **当前市场 = bull_osc（强势震荡）**——正是 v_weak 最有效状态
|
||||
4. **v_oversold 信号枯竭**：2026 仅 4 笔，结构性牛市下"值得抄底的超跌"消失
|
||||
5. **失效预警实测**：v_weak 近90日胜率 **22.7% → 红牌淘汰**（701笔，盈亏比 1.70 尚可但胜率崩溃）
|
||||
|
||||
### 1.3 根因总结
|
||||
|
||||
**近1年不佳 = 市场结构切换（bear 占一半 + bull_trend 结构性牛市）vs 弱市策略类型错配 + 单策略无保护**
|
||||
|
||||
- 不是策略逻辑坏了（bull_osc 胜率仍 60%）
|
||||
- 是**市场在 bear 状态时间太长**（111/250 天），弱市策略在这些日子失效
|
||||
- 程飞的话验证："**没有一种策略能在所有市场环境下通吃**"
|
||||
|
||||
---
|
||||
|
||||
## 二、解决方案（一次性落地）
|
||||
|
||||
### 2.1 市场测温机制（market_thermometer.py）
|
||||
|
||||
**核心理念**（市场周期测温文/霍华德·马克斯）：周期像钟摆无法预测，但可测温——不是预测拐点，是判断当前摆到哪。
|
||||
|
||||
**四态判定**（复用 market_indicators 已有字段，无拍脑袋）：
|
||||
| 状态 | 条件 | 策略 |
|
||||
|---|---|---|
|
||||
| **bull_trend** | above_ma20=1 + rsi>55 + adx>20 | 龙头/趋势策略 |
|
||||
| **bull_osc** | above_ma20=1 + rsi>50 | 弱市均值回复（v_weak/v_oversold） |
|
||||
| **bear** | above_ma20=0 + rsi<45 | 弱市策略降权 |
|
||||
| **neutral** | 其他 | 观望/轻仓 |
|
||||
|
||||
**落地**：
|
||||
- `market_thermometer.py`：回填 481 日 market_indicators + 四态判定 + 输出 market_state.json
|
||||
- cron：每日 16:50 跑（盘后）
|
||||
- 当前状态：**bull_osc（强势震荡）**（rsi 58.7 / above_ma20=1 / dd60 -2.5）
|
||||
|
||||
### 2.2 策略动态路由（strategy_router.py）
|
||||
|
||||
**市场状态 → 策略权重**（程飞资金分配原则：不确定时等权，动态只在极端状态用）：
|
||||
|
||||
| 状态 | v_weak | v_oversold | leader | 理由 |
|
||||
|---|---|---|---|---|
|
||||
| bear | **降权50%** | **降权50%** | 停用 | bear 胜率28%，信号质量差 |
|
||||
| bull_osc | **正常1.0** | **正常1.0** | 停用 | bull_osc 胜率60%，最有效 |
|
||||
| bull_trend | **降权50%** | **降权50%** | **启用1.0** | bull_trend 适用龙头 |
|
||||
| neutral | 轻仓0.5 | 轻仓0.5 | 停用 | 观望 |
|
||||
|
||||
**落地**：
|
||||
- `strategy_router.py`：读 market_state.json → 输出 strategy_weights.json
|
||||
- cron：每日 16:55 跑（测温后）
|
||||
|
||||
### 2.3 龙头识别策略（leader_scanner.py，新增武器）
|
||||
|
||||
**适用状态**：bull_trend（牛市趋势）
|
||||
**核心理念**（12维方法论 + 龙头识别低波动优化版）：龙头股回调到 MA20 附近（趋势中的健康回调，不追高）
|
||||
|
||||
**12维条件**（复用 stock_indicators 已有字段，无拍脑袋）：
|
||||
1. 个股 close > MA20（趋势向上）→ dist_ma20 ∈ [-3%, +2%]（回调到 MA20 附近）
|
||||
2. RSI ∈ [45, 65]（强势未超买）
|
||||
3. 市值/流动性（mcap_q > 0.3，龙头非小盘）
|
||||
4. 量能配合（vol_ratio > 0.8，非极度缩量）
|
||||
5. 趋势共振（trend_aligned=1）
|
||||
6. 行业强势（待接入 sector 数据）
|
||||
|
||||
**落地**：
|
||||
- `leader_scanner.py`：bull_trend 状态扫描龙头回调买点，输出 leader_signals.json
|
||||
- 当前 bull_osc 状态不扫描（避免追高——测温决定策略）
|
||||
|
||||
**注意**：这是初版框架，需 MoFin 引擎验证达标（年化≥13.2%）才正式上线。当前标记为"待验证"。
|
||||
|
||||
### 2.4 失效预警机制（strategy_alert.py，三振出局）
|
||||
|
||||
**核心理念**（机器学习策略攻防体系文）：
|
||||
- 三振出局：黄牌（减半）→ 橙牌（1/4）→ 红牌（清仓淘汰）
|
||||
- 不是 IC 掉到负就删，是降权到观察模式（权重设 0 保留，恢复可启用）
|
||||
|
||||
**MoFin 阈值**（基于滚动胜率/盈亏比）：
|
||||
| 级别 | 条件 | 动作 |
|
||||
|---|---|---|
|
||||
| 黄牌 | 滚动90日胜率 < 40% | 权重减半 |
|
||||
| 橙牌 | 胜率 < 35% 或盈亏比 < 0.5 | 权重降1/4 |
|
||||
| 红牌 | 胜率 < 30% | 权重0（淘汰观察） |
|
||||
|
||||
**落地**：
|
||||
- `strategy_alert.py`：监控 v_weak/v_oversold 滚动胜率/盈亏比，输出 strategy_alerts.json
|
||||
- cron：每周五 17:00 跑
|
||||
- **实测**：v_weak 近90日胜率 22.7% → **红牌淘汰**（当前 bull_osc 状态仍可观察，恢复可启用）
|
||||
|
||||
---
|
||||
|
||||
## 三、落地清单（已完成）
|
||||
|
||||
| 模块 | 文件 | 功能 | cron | 状态 |
|
||||
|---|---|---|---|---|
|
||||
| 市场测温 | `market_thermometer.py` | 四态判定+历史回填 | 每日16:50 | ✅ 跑通 |
|
||||
| 策略路由 | `strategy_router.py` | 状态→策略权重 | 每日16:55 | ✅ 跑通 |
|
||||
| 龙头识别 | `leader_scanner.py` | bull_trend 龙头回调 | （手动/待验证） | ✅ 框架 |
|
||||
| 失效预警 | `strategy_alert.py` | 三振出局 | 每周五17:00 | ✅ 跑通 |
|
||||
| 分市场检验 | `analyze_market_state2.py` | v_weak/v_oversold 分态胜率 | （一次性） | ✅ 完成 |
|
||||
|
||||
---
|
||||
|
||||
## 四、下一步（待验证/优化）
|
||||
|
||||
1. **leader_scanner 回测验证**：用 MoFin 引擎跑 bull_trend 日的龙头回调信号，验证年化≥13.2%
|
||||
2. **v_weak 恢复观察**：当前红牌（22.7%胜率），但市场已转 bull_osc——每周失效预警自动重评，胜率回升至 40%+ 自动恢复权重
|
||||
3. **v_oversold 信号恢复**：结构性牛市下信号枯竭，待市场转 bear/neutral 时恢复
|
||||
4. **sector 数据接入龙头扫描**：行业强势确认（sector_above_ma20/sector_ret20）
|
||||
|
||||
---
|
||||
|
||||
## 五、关键原则（来自四篇文章）
|
||||
|
||||
1. **程飞（多策略组合）**："单策略是茧，多策略是翅膀"——分散化本质是低相关性，不是数量
|
||||
2. **程工（市场测温）**："周期无法预测，但可测温"——不是预测拐点，是知道现在摆在哪
|
||||
3. **因子挖掘框架**："每个因子都有旱季雨季，没有任何因子能连续36个月保持正超额"——策略必须动态切换
|
||||
4. **机器学习攻防**："三振出局，不是IC掉到负就删"——降权观察，恢复可启用
|
||||
5. **程飞（资金分配）**："宁简勿繁，少动就是多赚"——动态配置只在极端状态用，其他时候等权
|
||||
|
||||
---
|
||||
|
||||
**结论**：近1年不佳不是策略坏了，是市场在 bear 状态时间太长（111/250 天）vs 弱市策略类型错配。解法 = 市场测温动态切换（bear 降权/bull_osc 正常/bull_trend 换龙头）+ 新增龙头策略（bull_trend 武器）+ 失效预警淘汰（三振出局）。
|
||||
Reference in New Issue
Block a user