From 29ec09530ae5e6788a935f75d0b6bb56ffd42b48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=A5=E5=BE=AE?= Date: Thu, 25 Jun 2026 20:04:44 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E4=B8=89=E5=B1=82=E7=AD=96=E7=95=A5?= =?UTF-8?q?=E5=A4=8D=E7=9B=98=E6=A1=86=E6=9E=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 信号层/执行层/综合层独立评估，输出每层判断+整体评级。 当前结果(38条): ✅正确 6 | ❌错误 1 | ⚠️部分 20 | ⏳待定 11 信号层: 方向看反1次 执行层: 止损过紧1次 大部分待定是因为策略太新(<7天)，需要时间积累数据才能做出有效评估。 --- scripts/strategy_review.py | 321 ++++++++++++++++++++++--------------- 1 file changed, 191 insertions(+), 130 deletions(-) diff --git a/scripts/strategy_review.py b/scripts/strategy_review.py index b9d56ba..a959512 100644 --- a/scripts/strategy_review.py +++ b/scripts/strategy_review.py @@ -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}跌破止损" - 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}" + # ─── 综合层：赚钱了吗？ ─── + if cost > 0 and s_type == "持仓策略": + profit_pct = (price - cost) / cost * 100 + if profit_pct > 5: + overall = "盈利" + elif profit_pct > -5: + 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 = f"亏损{profit_pct:.0f}%" + elif tp > 0 and price >= tp: + overall = "触止盈" + elif sl > 0 and price <= sl: + overall = "触止损" else: - return "pending", None, f"运行{days}天，待观察" + 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 ["持有", "观望", "等待", "持股"]) -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 = "待定" + 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: + 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: + signal_verdict = "待定" else: - return {"mode": "unknown", "severity": "low", "detail": "需人工分析"} + # 无明确信号 + 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 - - 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}") + + sv, ev, overall, sf, ef = evaluate_strategy(s, price) + + # 综合评级 + if overall in ("盈利", "触止盈"): + if sv == "正确" or "存疑" not in sv: + stats["correct"] += 1 + else: + 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: - summary_lines.append(f" ⏳ {name}({code}): {detail}") - - # 写入 accuracy_stats + stats["pending"] += 1 + + # 记录失败模式 + 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 - - 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}%") - - 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 summary_lines: + + # 输出 + total_eval = stats["total"] - stats["pending"] + accuracy = stats["correct"] / max(total_eval, 1) * 100 + + 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 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 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)