知微
127a0ff021
- 新增 cron: 策略复盘-每日 (20:00, no_agent, deliver=origin) - 执行层增加60日K线最高价检查 - 止盈触发后价格继续涨→标记"卖飞(X%)" + tp_too_close - 德明利这类情况: 综合层=赚钱, 执行层=卖飞(止盈过近) - 优化K线拉取性能(只对有TP的票拉, curl --max-time 3)
276 lines
10 KiB
Python
276 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""strategy_review.py — 三层策略复盘 (no_agent)
|
|
|
|
每层独立评估：
|
|
1. 信号层 — 买入/卖出/持有的timing对不对？
|
|
2. 执行层 — 止损/止盈设得合理吗？
|
|
3. 综合层 — 这波操作整体赚钱了吗？
|
|
|
|
用法：
|
|
python3 scripts/strategy_review.py
|
|
"""
|
|
|
|
import json, sqlite3, sys, time, urllib.request
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from collections import Counter
|
|
|
|
BASE = Path("/home/hmo/MoFin")
|
|
DATA = BASE / "data"
|
|
DB_PATH = DATA / "mofin.db"
|
|
DECISIONS_PATH = DATA / "decisions.json"
|
|
|
|
# 失败模式定义（执行层）
|
|
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}"
|
|
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
|
|
resp = urllib.request.urlopen(req, timeout=5).read().decode('gbk')
|
|
fld = resp.split('=')[1].strip().strip('"').strip(';').split('~')
|
|
return float(fld[3]) if len(fld) > 3 else 0
|
|
except:
|
|
return 0
|
|
|
|
|
|
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 "skip", "skip", "skip", "数据不足"
|
|
|
|
# 计算运行天数
|
|
try:
|
|
days = (datetime.now() - datetime.fromisoformat(created)).days
|
|
except:
|
|
days = 0
|
|
|
|
# ─── 综合层：赚钱了吗？ ───
|
|
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:
|
|
overall = "触止盈"
|
|
elif sl > 0 and price <= sl:
|
|
overall = "触止损"
|
|
else:
|
|
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:
|
|
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:
|
|
# 无明确信号
|
|
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
|
|
|
|
# 取近期最高价（判断是否卖飞，只对有TP的策略拉数据）
|
|
recent_high = 0
|
|
if tp > 0:
|
|
try:
|
|
prefix = "sh" if code.startswith(('60','68','51','56','50')) else "sz" if code.startswith(('00','30','15')) else "hk"
|
|
url = f"http://ifzq.gtimg.cn/appstock/app/fqkline/get?param={prefix}{code},day,,,60,qfq"
|
|
import subprocess as sp
|
|
r = sp.run(["curl", "-s", "--max-time", "3", url], capture_output=True, text=True, timeout=5)
|
|
if r.returncode == 0 and r.stdout:
|
|
data = json.loads(r.stdout)
|
|
day_key = 'qfqday' if prefix != 'hk' else 'day'
|
|
bars = data.get('data', {}).get(f'{prefix}{code}', {}).get(day_key, [])
|
|
if bars:
|
|
recent_high = max(float(b[2]) for b in bars if len(b) > 2)
|
|
except:
|
|
pass
|
|
|
|
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 or recent_high >= tp):
|
|
# 止盈触发或曾触发过
|
|
max_price = max(price, recent_high)
|
|
if max_price <= tp * 1.05:
|
|
exec_verdict = "已触发"
|
|
else:
|
|
overshoot = (max_price - tp) / tp * 100
|
|
exec_verdict = f"卖飞({overshoot:.0f}%)"
|
|
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()
|
|
decisions = json.loads(DECISIONS_PATH.read_text())
|
|
strategies = decisions.get("decisions", [])
|
|
|
|
conn = sqlite3.connect(str(DB_PATH))
|
|
|
|
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", "")
|
|
price = fetch_price(code)
|
|
if not price:
|
|
detail_lines.append(f" ⏭️ {name}({code}): 无行情")
|
|
stats["pending"] += 1
|
|
continue
|
|
|
|
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:
|
|
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 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()
|
|
|
|
# 输出
|
|
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 detail_lines:
|
|
print(line)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
review()
|