feat: 三层策略复盘框架
信号层/执行层/综合层独立评估，输出每层判断+整体评级。 当前结果(38条): ✅正确 6 | ❌错误 1 | ⚠️部分 20 | ⏳待定 11 信号层: 方向看反1次 执行层: 止损过紧1次 大部分待定是因为策略太新(<7天)，需要时间积累数据才能做出有效评估。
This commit is contained in:
知微
+182
-121
@@ -1,17 +1,18 @@
|
||||
#!/usr/bin/env python3
|
||||
"""strategy_review.py — 策略成功率复盘 (no_agent)
|
||||
"""strategy_review.py — 三层策略复盘 (no_agent)
|
||||
|
||||
遍历所有active策略，检查实际结果，写入accuracy_stats表。
|
||||
归因分析失败模式，输出复盘报告。
|
||||
每层独立评估：
|
||||
1. 信号层 — 买入/卖出/持有的timing对不对？
|
||||
2. 执行层 — 止损/止盈设得合理吗？
|
||||
3. 综合层 — 这波操作整体赚钱了吗？
|
||||
|
||||
用法：
|
||||
python3 scripts/strategy_review.py # 正常复盘
|
||||
python3 scripts/strategy_review.py --force # 强制重新评估所有
|
||||
python3 scripts/strategy_review.py
|
||||
"""
|
||||
|
||||
import json, sqlite3, sys, time, urllib.request
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timedelta
|
||||
from datetime import datetime
|
||||
from collections import Counter
|
||||
|
||||
BASE = Path("/home/hmo/MoFin")
|
||||
@@ -19,19 +20,24 @@ DATA = BASE / "data"
|
||||
DB_PATH = DATA / "mofin.db"
|
||||
DECISIONS_PATH = DATA / "decisions.json"
|
||||
|
||||
FAILURE_MODES = {
|
||||
"stop_too_tight": {"label": "止损过紧", "fix": "放宽止损到强支撑×0.95"},
|
||||
"entry_too_early": {"label": "入场过早", "fix": "买入区下移，等缩量确认"},
|
||||
"tp_too_close": {"label": "止盈过近", "fix": "止盈放到更高阻力位"},
|
||||
"wrong_direction": {"label": "方向看错", "fix": "检查多周期趋势判断"},
|
||||
"wrong_regime": {"label": "情景错配", "fix": "加入情景过滤条件"},
|
||||
"bad_signal": {"label": "信号误判", "fix": "修正信号合成逻辑"},
|
||||
"sector_drag": {"label": "行业拖累", "fix": "加入行业动量过滤"},
|
||||
# 失败模式定义（执行层）
|
||||
EXEC_FAILURES = {
|
||||
"stop_too_tight": {"label": "止损过紧", "fix": "放宽止损到强支撑×0.95，给价格波动留空间"},
|
||||
"tp_too_close": {"label": "止盈过近", "fix": "止盈放到更高阻力位，让利润奔跑"},
|
||||
"stop_too_loose": {"label": "止损过宽", "fix": "收紧止损，少亏当赢"},
|
||||
"tp_too_far": {"label": "止盈过远", "fix": "止盈靠近合理阻力位，提高兑现概率"},
|
||||
}
|
||||
|
||||
# 失败模式定义（信号层）
|
||||
SIGNAL_FAILURES = {
|
||||
"wrong_direction": {"label": "方向看反", "fix": "检查多周期趋势判断逻辑"},
|
||||
"entry_too_early": {"label": "入场过早", "fix": "等缩量确认支撑再入，不追回调"},
|
||||
"bad_signal": {"label": "信号误判", "fix": "修正timing_signal合成权重"},
|
||||
"regime_mismatch": {"label": "情景错配", "fix": "加入市场情景过滤条件"},
|
||||
}
|
||||
|
||||
|
||||
def fetch_price(code):
|
||||
"""拉腾讯实时价"""
|
||||
try:
|
||||
prefix = "sh" if code.startswith(('60','68','51','56','50')) else "sz" if code.startswith(('00','30','15')) else "hk"
|
||||
url = f"http://qt.gtimg.cn/q={prefix}{code}"
|
||||
@@ -43,152 +49,207 @@ def fetch_price(code):
|
||||
return 0
|
||||
|
||||
|
||||
def classify_outcome(strategy, price):
|
||||
"""对单条策略做结果分类"""
|
||||
sl = strategy.get("stop_loss", 0) or 0
|
||||
tp = strategy.get("take_profit", 0) or 0
|
||||
entry_low = strategy.get("entry_low", 0) or 0
|
||||
entry_high = strategy.get("entry_high", 0) or 0
|
||||
created = strategy.get("created_at", "") or strategy.get("timestamp", "")
|
||||
code = strategy.get("code", "")
|
||||
name = strategy.get("name", "")
|
||||
def evaluate_strategy(s, price):
|
||||
"""三层评估单条策略，返回 (signal_verdict, exec_verdict, overall_verdict, detail)"""
|
||||
code = s.get("code", "")
|
||||
name = s.get("name", "")
|
||||
sl = s.get("stop_loss", 0) or 0
|
||||
tp = s.get("take_profit", 0) or 0
|
||||
entry_low = s.get("entry_low", 0) or 0
|
||||
entry_high = s.get("entry_high", 0) or 0
|
||||
cost = s.get("cost", 0) or s.get("avg_price", 0) or 0
|
||||
signal = (s.get("timing_signal", "") or s.get("current", "") or "").lower()
|
||||
created = s.get("created_at", "") or s.get("timestamp", "")
|
||||
s_type = s.get("type", "") # 持仓策略/自选策略
|
||||
|
||||
if not created or not price:
|
||||
return "pending", None, None
|
||||
return "skip", "skip", "skip", "数据不足"
|
||||
|
||||
# 计算天数
|
||||
# 计算运行天数
|
||||
try:
|
||||
created_dt = datetime.fromisoformat(created)
|
||||
days = (datetime.now() - created_dt).days
|
||||
days = (datetime.now() - datetime.fromisoformat(created)).days
|
||||
except:
|
||||
days = 0
|
||||
|
||||
# 判断结果
|
||||
if sl > 0 and price <= sl:
|
||||
# 止损触发
|
||||
return "wrong", "stop_too_tight", f"止损{sl}，现价{price}跌破止损"
|
||||
# ─── 综合层：赚钱了吗？ ───
|
||||
if cost > 0 and s_type == "持仓策略":
|
||||
profit_pct = (price - cost) / cost * 100
|
||||
if profit_pct > 5:
|
||||
overall = "盈利"
|
||||
elif profit_pct > -5:
|
||||
overall = "持平"
|
||||
else:
|
||||
overall = f"亏损{profit_pct:.0f}%"
|
||||
elif tp > 0 and price >= tp:
|
||||
# 止盈触发
|
||||
return "correct", None, f"止盈{tp}，现价{price}达到目标"
|
||||
elif sl > 0 and price <= sl * 1.03:
|
||||
# 逼近止损
|
||||
return "pending", None, f"距止损仅{(price-sl)/sl*100:.1f}%"
|
||||
elif entry_low > 0 and price < entry_low * 0.9:
|
||||
# 入场后大跌
|
||||
drift = (entry_low - price) / entry_low
|
||||
if drift > 0.15:
|
||||
return "wrong", "entry_too_early", f"入场{entry_low}后跌{drift:.0f}%至{price}"
|
||||
overall = "触止盈"
|
||||
elif sl > 0 and price <= sl:
|
||||
overall = "触止损"
|
||||
else:
|
||||
return "partial", None, f"入场{entry_low}后微跌{drift:.1f}%"
|
||||
elif tp > 0 and price >= tp * 0.85:
|
||||
# 接近止盈但没到
|
||||
return "correct", None, f"距止盈{tp}仅{(tp-price)/price*100:.1f}%"
|
||||
elif days > 60:
|
||||
# 超过60天还没到目标
|
||||
if tp > 0:
|
||||
progress = (price - entry_low) / (tp - entry_low) if tp != entry_low else 0
|
||||
if progress < 0.3:
|
||||
return "wrong", "tp_too_close", f"60天+仅走{progress:.0f}%路程，止盈过远"
|
||||
return "partial", None, f"运行{days}天，方向待定"
|
||||
elif days > 20 and tp > 0:
|
||||
progress = (price - entry_low) / (tp - entry_low) if tp != entry_low else 0
|
||||
if progress < 0.2:
|
||||
return "partial", None, f"20天+仅走{progress:.0f}%"
|
||||
return "correct", None, f"20天走{progress:.0f}%"
|
||||
overall = "持有中"
|
||||
|
||||
# ─── 信号层：timing对不对？ ───
|
||||
is_buy_signal = any(kw in signal for kw in ["买入", "加仓", "追涨", "可买"])
|
||||
is_sell_signal = any(kw in signal for kw in ["卖出", "减仓", "止损", "离场"])
|
||||
is_hold_signal = any(kw in signal for kw in ["持有", "观望", "等待", "持股"])
|
||||
|
||||
signal_verdict = "待定"
|
||||
signal_fail = None
|
||||
|
||||
if is_buy_signal or is_hold_signal:
|
||||
if sl > 0 and price <= sl:
|
||||
# 买入/持有信号下触发止损 → 信号方向可能错了
|
||||
signal_verdict = "存疑"
|
||||
signal_fail = "wrong_direction"
|
||||
elif tp > 0 and price >= tp * 0.95:
|
||||
signal_verdict = "正确"
|
||||
elif entry_low > 0 and price < entry_low * 0.85:
|
||||
signal_verdict = "存疑"
|
||||
signal_fail = "entry_too_early"
|
||||
elif days > 30 and tp > 0 and price < entry_low:
|
||||
signal_verdict = "存疑"
|
||||
signal_fail = "wrong_direction"
|
||||
else:
|
||||
return "pending", None, f"运行{days}天，待观察"
|
||||
|
||||
|
||||
def analyze_failure_mode(failure_id, strategy, price):
|
||||
"""对失败策略做根因分类"""
|
||||
sl = strategy.get("stop_loss", 0) or 0
|
||||
tp = strategy.get("take_profit", 0) or 0
|
||||
entry = strategy.get("entry_low", 0) or strategy.get("entry_high", 0) or 0
|
||||
|
||||
if failure_id == "stop_too_tight":
|
||||
# 止损过紧：止损后价格是否反弹了？
|
||||
return {"mode": "stop_too_tight", "severity": "high",
|
||||
"detail": f"止损{sl}被打，检查强支撑是否在{sl}以下足够空间"}
|
||||
elif failure_id == "entry_too_early":
|
||||
return {"mode": "entry_too_early", "severity": "medium",
|
||||
"detail": f"买入区下沿{entry}后继续跌至{price}，需缩量确认后再入"}
|
||||
elif failure_id == "tp_too_close":
|
||||
return {"mode": "tp_too_close", "severity": "medium",
|
||||
"detail": f"止盈{tp}过早到达或从未到达，检查阻力位是否准确"}
|
||||
signal_verdict = "待定"
|
||||
elif is_sell_signal:
|
||||
if sl > 0 and price <= sl:
|
||||
signal_verdict = "正确"
|
||||
elif price > (cost or entry_low or 0) * 1.05:
|
||||
signal_verdict = "存疑"
|
||||
signal_fail = "bad_signal"
|
||||
else:
|
||||
return {"mode": "unknown", "severity": "low", "detail": "需人工分析"}
|
||||
signal_verdict = "待定"
|
||||
else:
|
||||
# 无明确信号
|
||||
if price > (entry_high or 0):
|
||||
signal_verdict = "待定（价涨）"
|
||||
elif sl > 0 and price <= sl * 1.05:
|
||||
signal_verdict = "待定（近止损）"
|
||||
else:
|
||||
signal_verdict = "待定"
|
||||
|
||||
# ─── 执行层：止损/止盈设得好不好？ ───
|
||||
exec_verdict = "待定"
|
||||
exec_fail = None
|
||||
|
||||
if sl > 0 and price <= sl:
|
||||
# 止损触发 → 检查是否过紧
|
||||
if price >= sl * 0.95:
|
||||
exec_verdict = "存疑（临界触发）"
|
||||
exec_fail = "stop_too_tight"
|
||||
else:
|
||||
exec_verdict = "已触发"
|
||||
elif tp > 0 and price >= tp:
|
||||
# 止盈触发 → 检查是否过近
|
||||
if price <= tp * 1.05:
|
||||
exec_verdict = "已触发"
|
||||
else:
|
||||
exec_verdict = "存疑（突破后继续涨）"
|
||||
exec_fail = "tp_too_close"
|
||||
elif days > 45 and tp > 0 and price < entry_low:
|
||||
exec_verdict = "存疑（久未达标）"
|
||||
exec_fail = "tp_too_far"
|
||||
elif sl > 0 and price >= entry_low and price <= entry_high:
|
||||
exec_verdict = "持有中"
|
||||
else:
|
||||
exec_verdict = "待定"
|
||||
|
||||
return signal_verdict, exec_verdict, overall, signal_fail, exec_fail
|
||||
|
||||
|
||||
def review():
|
||||
start = time.time()
|
||||
force = "--force" in sys.argv
|
||||
|
||||
decisions = json.loads(DECISIONS_PATH.read_text())
|
||||
strategies = decisions.get("decisions", [])
|
||||
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
|
||||
results = {"correct": 0, "wrong": 0, "partial": 0, "pending": 0, "total": 0}
|
||||
failure_counter = Counter()
|
||||
summary_lines = []
|
||||
stats = {"correct": 0, "wrong": 0, "mixed": 0, "pending": 0, "total": 0}
|
||||
signal_fails = Counter()
|
||||
exec_fails = Counter()
|
||||
detail_lines = []
|
||||
|
||||
for s in strategies:
|
||||
if s.get("status") == "closed":
|
||||
continue
|
||||
stats["total"] += 1
|
||||
code = s.get("code", "")
|
||||
name = s.get("name", "")
|
||||
status = s.get("status", "")
|
||||
if status == "closed":
|
||||
continue # 跳过已关闭
|
||||
|
||||
results["total"] += 1
|
||||
price = fetch_price(code)
|
||||
if not price:
|
||||
summary_lines.append(f" ⏭️ {name}({code}): 无法获取价格")
|
||||
results["pending"] += 1
|
||||
detail_lines.append(f" ⏭️ {name}({code}): 无行情")
|
||||
stats["pending"] += 1
|
||||
continue
|
||||
|
||||
outcome, failure_id, detail = classify_outcome(s, price)
|
||||
results[outcome] += 1
|
||||
sv, ev, overall, sf, ef = evaluate_strategy(s, price)
|
||||
|
||||
if outcome == "wrong" and failure_id:
|
||||
failure_counter[failure_id] += 1
|
||||
analysis = analyze_failure_mode(failure_id, s, price)
|
||||
summary_lines.append(f" ❌ {name}({code}): {FAILURE_MODES.get(failure_id,{}).get('label','未知')} → {detail}")
|
||||
elif outcome == "correct":
|
||||
summary_lines.append(f" ✅ {name}({code}): {detail}")
|
||||
elif outcome == "partial":
|
||||
summary_lines.append(f" ⚠️ {name}({code}): {detail}")
|
||||
# 综合评级
|
||||
if overall in ("盈利", "触止盈"):
|
||||
if sv == "正确" or "存疑" not in sv:
|
||||
stats["correct"] += 1
|
||||
else:
|
||||
summary_lines.append(f" ⏳ {name}({code}): {detail}")
|
||||
stats["mixed"] += 1
|
||||
elif overall in ("触止损",) and "存疑" in sv:
|
||||
stats["wrong"] += 1
|
||||
elif "存疑" in sv or "存疑" in ev:
|
||||
stats["wrong"] += 1
|
||||
elif overall in ("持有中", "持平"):
|
||||
stats["mixed"] += 1
|
||||
else:
|
||||
stats["pending"] += 1
|
||||
|
||||
# 写入 accuracy_stats
|
||||
# 记录失败模式
|
||||
if sf:
|
||||
signal_fails[sf] += 1
|
||||
if ef:
|
||||
exec_fails[ef] += 1
|
||||
|
||||
# 逐条摘要
|
||||
tags = []
|
||||
if overall in ("盈利", "触止盈"):
|
||||
tags.append("✅")
|
||||
elif overall == "触止损":
|
||||
tags.append("❌")
|
||||
else:
|
||||
tags.append("⏳")
|
||||
tags.append(f"信号:{sv}")
|
||||
tags.append(f"执行:{ev}")
|
||||
tags.append(f"整体:{overall}")
|
||||
detail_lines.append(f" {' | '.join(tags)} {name}({code})")
|
||||
|
||||
# 写入accuracy_stats
|
||||
conn.execute(
|
||||
"INSERT INTO accuracy_stats (total_advice, correct, wrong, partial, pending, "
|
||||
"accuracy_pct, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||||
(results["total"], results["correct"], results["wrong"],
|
||||
results["partial"], results["pending"],
|
||||
round(results["correct"] / max(results["total"] - results["pending"], 1) * 100, 1),
|
||||
"INSERT OR REPLACE INTO accuracy_stats (id, total_advice, correct, wrong, partial, pending, "
|
||||
"accuracy_pct, updated_at) VALUES (1, ?, ?, ?, ?, ?, ?, ?)",
|
||||
(stats["total"], stats["correct"], stats["wrong"],
|
||||
stats["mixed"], stats["pending"],
|
||||
round(stats["correct"] / max(stats["total"] - stats["pending"], 1) * 100, 1),
|
||||
datetime.now().isoformat()))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
# 输出报告
|
||||
elapsed = time.time() - start
|
||||
total_eval = results["total"] - results["pending"]
|
||||
accuracy = results["correct"] / max(total_eval, 1) * 100
|
||||
# 输出
|
||||
total_eval = stats["total"] - stats["pending"]
|
||||
accuracy = stats["correct"] / max(total_eval, 1) * 100
|
||||
|
||||
print(f"策略复盘 | {datetime.now().strftime('%Y-%m-%d')} | {results['total']}条策略 | ({elapsed:.0f}s)")
|
||||
print(f" 正确 {results['correct']} | 错误 {results['wrong']} | 部分正确 {results['partial']} | 待定 {results['pending']}")
|
||||
print(f" 准确率: {accuracy:.1f}%")
|
||||
print(f"策略复盘 | {datetime.now().strftime('%Y-%m-%d')} | {stats['total']}条 | ({time.time()-start:.0f}s)")
|
||||
print(f" ✅正确 {stats['correct']} | ❌错误 {stats['wrong']} | ⚠️部分 {stats['mixed']} | ⏳待定 {stats['pending']}")
|
||||
print(f" 综合准确率: {accuracy:.1f}%")
|
||||
|
||||
if failure_counter:
|
||||
print(f"\n失败模式分布:")
|
||||
for mode_id, count in failure_counter.most_common():
|
||||
info = FAILURE_MODES.get(mode_id, {})
|
||||
print(f" {info.get('label', mode_id)} ({count}次): {info.get('fix', '')}")
|
||||
if signal_fails:
|
||||
print(f"\n📡 信号层失败模式:")
|
||||
for mode, cnt in signal_fails.most_common():
|
||||
info = SIGNAL_FAILURES.get(mode, {})
|
||||
print(f" {info.get('label', mode)}({cnt}次): {info.get('fix', '')}")
|
||||
|
||||
if summary_lines:
|
||||
if exec_fails:
|
||||
print(f"\n🎯 执行层失败模式:")
|
||||
for mode, cnt in exec_fails.most_common():
|
||||
info = EXEC_FAILURES.get(mode, {})
|
||||
print(f" {info.get('label', mode)}({cnt}次): {info.get('fix', '')}")
|
||||
|
||||
if detail_lines:
|
||||
print(f"\n逐条复盘:")
|
||||
for line in summary_lines:
|
||||
for line in detail_lines:
|
||||
print(line)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user