# -*- coding: utf-8 -*- """ 深度分析：每笔开仓的因子组合 vs 后续盈亏结果 找出"有预测价值"的因子，给出优化权重 """ import re from collections import defaultdict import os as _os; LOG = _os.path.join(_os.path.dirname(_os.path.dirname(_os.path.dirname(_os.path.abspath(__file__)))), "my_trader.log") WEIGHTS_CURRENT = { "TR": 0.10, "OB": 0.06, "TK": 0.12, "OI": 0.05, "FR": 0.02, "MP": 0.01, "VD": 0.09, "BTC": 0.07, "GM": 0.06, "IV": 0.01, "EX": 0.07, "LC": 0.15, "MR": 0.06, "NEWS": 0.13, "SM": 0.00, } FK = ["TR","OB","TK","OI","FR","MP","VD","BTC","GM","IV","EX","LC","MR","SM"] factor_re = re.compile( r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*" r"TR:([+-][0-9.]+) OB:([+-][0-9.]+) FR:([+-][0-9.]+) TK:([+-][0-9.]+) " r"OI:([+-][0-9.]+) MP:([+-][0-9.]+) VD:([+-][0-9.]+) BTC:([+-][0-9.]+) " r"GM:([+-][0-9.]+) IV:([+-][0-9.]+) EX:([+-][0-9.]+) LC:([+-][0-9.]+) " r"MR:([+-][0-9.]+) SM:([+-][0-9.]+).*=> ([+-][0-9.]+) -> (LONG|SHORT|WAIT)" ) open_re = re.compile(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*触发 (LONG|SHORT).*执行开仓") pos_re = re.compile(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*\[持仓\] ETH (long|short) ([0-9.]+)张 @ \$([0-9.]+) P.L:\$([+-][0-9.]+)") eq_re = re.compile(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*Cycle \d+: equity=\$([0-9.]+)") close_re = re.compile(r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*(?:止损触发|止盈触发|平仓成功|仓位已关闭|强制平仓|position closed|减仓)") print("读取日志...") lines = open(LOG, encoding="utf-8", errors="replace").readlines() print(f"共 {len(lines)} 行") # 解析所有记录 all_factors = [] # (ts, fdict, score, direction) all_opens = [] # ts all_pos = [] # (ts, side, qty, entry, pnl) all_equity = [] # (ts, eq) for line in lines: mf = factor_re.search(line) if mf: g = mf.groups() fdict = {k: float(v) for k, v in zip(["TR","OB","FR","TK","OI","MP","VD","BTC","GM","IV","EX","LC","MR","SM"], g[1:15])} all_factors.append((g[0], fdict, float(g[15]), g[16])) mo = open_re.search(line) if mo: all_opens.append(mo.group(1)) mp = pos_re.search(line) if mp: all_pos.append((mp.group(1), mp.group(2), float(mp.group(3)), float(mp.group(4)), float(mp.group(5)))) me = eq_re.search(line) if me: all_equity.append((me.group(1), float(me.group(2)))) print(f"因子记录: {len(all_factors)} 开仓: {len(all_opens)} 持仓快照: {len(all_pos)} equity: {len(all_equity)}") # ── 把开仓事件和对应因子配对 ── # 对每次开仓，找最近的因子行（开仓前30秒内） def ts2sec(ts): from datetime import datetime return datetime.strptime(ts, "%Y-%m-%d %H:%M:%S").timestamp() factor_idx = 0 trades = [] # {ts, direction, factors, score, open_eq, pnl_list} for open_ts in all_opens: ots = ts2sec(open_ts) best = None for (fts, fd, sc, di) in all_factors: ft = ts2sec(fts) if -60 < ft - ots < 5: # 开仓前60s到后5s best = (fts, fd, sc, di) if best is None: continue # 找开仓时的equity open_eq = None for (ets, eq) in all_equity: if abs(ts2sec(ets) - ots) < 120: open_eq = eq break # 收集开仓后的PnL（取最大/最终） pnl_series = [] for (pts, side, qty, entry, pnl) in all_pos: pt = ts2sec(pts) if 0 < pt - ots < 3600: # 开仓后1小时内的持仓快照 pnl_series.append((pts, pnl)) direction = best[3] trades.append({ "ts": open_ts, "direction": direction, "factors": best[1], "score": best[2], "open_eq": open_eq, "pnl_series": pnl_series, }) print(f"\n配对成功的交易: {len(trades)}") # ── 为每笔交易判断盈亏 ── print("\n" + "="*70) print("=== 每笔交易详情 ===") print("="*70) for i, t in enumerate(trades): pnl_list = [p for _, p in t["pnl_series"]] max_pnl = max(pnl_list) if pnl_list else 0 fin_pnl = pnl_list[-1] if pnl_list else 0 result = "盈" if fin_pnl > 0.001 else ("亏" if fin_pnl < -0.001 else "平") f = t["factors"] print(f"\n【{i+1}】{t['ts'][5:16]} {t['direction']} 总分={t['score']:+.3f} 结果={result}(${fin_pnl:+.4f})") # 按贡献排序 contribs = [(k, f.get(k,0), WEIGHTS_CURRENT.get(k,0), f.get(k,0)*WEIGHTS_CURRENT.get(k,0)) for k in FK] contribs.sort(key=lambda x: abs(x[3]), reverse=True) for k,v,w,c in contribs[:6]: if abs(c) < 0.001: continue direction_match = (c > 0) == (t["direction"] == "LONG") mark = "✓支撑" if direction_match else "✗反向" print(f" {k:4} {v:+.3f}×{w:.2f}={c:+.4f} {mark}") # ── 全局因子胜率分析 ── print("\n" + "="*70) print("=== 因子方向一致性分析（预测胜率）===") print("="*70) factor_stats = defaultdict(lambda: {"correct":0,"total":0,"sum_val":0.0}) for t in trades: pnl_list = [p for _, p in t["pnl_series"]] if not pnl_list: continue fin_pnl = pnl_list[-1] actual_up = fin_pnl > 0 if t["direction"]=="LONG" else fin_pnl > 0 for k in FK: v = t["factors"].get(k, 0) if abs(v) < 0.05: continue # 中性不统计 # 因子预测方向 pred_up = (v > 0) # 正值=看多 if k in ("OB",): pred_up = not pred_up # OB反向 correct = (pred_up == (t["direction"]=="LONG" and fin_pnl>0 or t["direction"]=="SHORT" and fin_pnl>0)) factor_stats[k]["correct"] += 1 if correct else 0 factor_stats[k]["total"] += 1 factor_stats[k]["sum_val"] += abs(v) print(f"\n{'因子':5} {'权重':6} {'出现次':6} {'平均|值|':8} {'预测准确率':10} {'建议权重'}") suggested = {} for k in FK: st = factor_stats[k] if st["total"] == 0: suggested[k] = WEIGHTS_CURRENT[k] continue win_rate = st["correct"] / st["total"] avg_abs = st["sum_val"] / st["total"] # 建议：胜率>60%且avg_abs>0.2 → 提权；胜率<40% → 降权 cur_w = WEIGHTS_CURRENT[k] if win_rate >= 0.65 and avg_abs >= 0.15: new_w = round(min(cur_w * 1.3, 0.20), 3) elif win_rate <= 0.40: new_w = round(max(cur_w * 0.6, 0.005), 3) else: new_w = cur_w suggested[k] = new_w flag = "⬆" if new_w > cur_w else ("⬇" if new_w < cur_w else "—") print(f" {k:5} {cur_w:.2f} {st['total']:4} {avg_abs:.3f} {win_rate*100:.0f}% {new_w:.3f} {flag}") # 归一化建议权重 total_w = sum(suggested.values()) norm = {k: round(v/total_w, 4) for k,v in suggested.items()} norm_total = sum(norm.values()) # 余量给 LC norm["LC"] = round(norm["LC"] + (1.0 - norm_total), 4) print(f"\n=== 归一化后建议权重（合计={sum(norm.values()):.4f}）===") for k in FK: cur = WEIGHTS_CURRENT[k] new = norm[k] flag = "⬆" if new > cur+0.005 else ("⬇" if new < cur-0.005 else "—") print(f" {k:5} {cur:.3f} → {new:.3f} {flag}")