gavindiaz/AlphaMaster
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.91

Browse Source
This commit is contained in:
rosemarycox5334-debug
2026-07-05 23:19:21 +08:00
parent 1db926cd93
commit 97a5964b34
17 changed files with 1901 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
analyze_index_overfit.py
+53
View File
@@ -0,0 +1,53 @@
import torch, json, glob
import numpy as np
# Load index checkpoint
f = sorted(glob.glob(r'D:\cl\MT5_AlphaGPT\checkpoints\ckpt_index_step_*.pt'))[-1]
ckpt = torch.load(f, map_location='cpu', weights_only=False)
step = ckpt["step"]
best = ckpt["best_score"]
restarts = ckpt.get("restart_count", 0)
th = ckpt.get('training_history', {})
print(f'=== INDEX CHECKPOINT ===')
print(f'step={step} best={best:.4f} restarts={restarts}')
print()
# Check best_score progression
if isinstance(th, dict) and 'best_score' in th:
bs = th['best_score']
# Find when best_score first reached current value
for i, v in enumerate(bs):
if abs(v - best) < 0.001:
print(f'Best score {best:.4f} first reached at step {i}')
break
# Show progression every 200 steps
print(f'\nBest score progression (every 200 steps):')
for i in range(0, len(bs), 200):
print(f' step {i:4d}: best={bs[i]:.4f}')
print(f' step {len(bs)-1:4d}: best={bs[-1]:.4f}')
# Check val_score
if 'val_score' in th:
vs = th['val_score']
print(f'\nVal score progression (every 200 steps):')
for i in range(0, len(vs), 200):
print(f' step {i:4d}: val={vs[i]:.4f}')
print(f' step {len(vs)-1:4d}: val={vs[-1]:.4f}')
# Check if val is declining
last_100 = vs[-100:]
first_100 = vs[:100]
print(f'\nVal score: first 100 avg={np.mean(first_100):.4f}, last 100 avg={np.mean(last_100):.4f}')
# Entropy metrics
for key in ['entropy', 'top1_prob', 'eff_vocab', 'kl_uniform', 'kl_prev', 'ic_mean', 'ic_stability', 'sortino']:
if key in th:
arr = th[key]
print(f'\n{key}: first5={[round(x,4) for x in arr[:5]]}, last5={[round(x,4) for x in arr[-5:]]}')
# Training restarts
for key in ['restart_count']:
if key in th:
arr = th[key]
print(f'\n{key}: last5={arr[-5:]}')
analyze_index_overfit2.py
+151
View File
@@ -0,0 +1,151 @@
"""Analyze index group overfitting: check time-period breakdown of the factor"""
import sys, json, math
import numpy as np
import torch
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from config import Config
from data_pipeline.data_manager import MT5DataManager
from data_pipeline.fetcher import MT5DataFetcher
from model_core.vocab import FORMULA_VOCAB, VOCAB_VERSION
from model_core.vm import StackVM
from model_core.features import MT5FeatureEngineer
from strategy_manager.signal import compute_target_positions_stateless
_H1_PER_YEAR = 6240
# Load index strategy
data = json.load(open("strategies/best_index.json"))
formula = data["formula"]
print(f"Formula: {formula}")
print(f"Decode: ", end="")
names = FORMULA_VOCAB.token_names
print(" -> ".join(names[t] if 0 <= t < len(names) else f"?{t}" for t in formula))
print()
# Load index group data
Config.SYMBOLS = ["US30.cash", "US100.cash", "US500.cash", "US2000.cash", "JP225.cash"]
with MT5DataFetcher(offline=True) as fetcher:
mgr = MT5DataManager(fetcher)
mgr.load()
raw_dict = mgr.raw_dict
syms = mgr.symbols
T = raw_dict["open"].shape[1]
feat = MT5FeatureEngineer.compute_features(raw_dict)
target_ret = mgr.target_ret
print(f"Symbols: {syms}")
print(f"T={T} bars ({T/_H1_PER_YEAR:.2f} years)")
# Get bar times - just use bar index for simplicity
bar_times = None
# Run factor
vm = StackVM()
factor = vm.execute(formula, feat)
factor_np = factor.detach().numpy()
# Compute positions and PnL
pos = compute_target_positions_stateless(factor)
pos_np = pos.detach().numpy()
target_np = target_ret.detach().numpy()
cost_rate = 0.0003
prev = np.zeros_like(pos_np)
prev[:, 1:] = pos_np[:, :-1]
turnover = np.abs(pos_np - prev)
pnl = pos_np * target_np - turnover * cost_rate
port_pnl = pnl.mean(axis=0)
port_cum = np.cumsum(port_pnl)
# Split into 8 equal segments
n_seg = 8
seg_len = T // n_seg
print(f"\n=== {n_seg}-SEGMENT BREAKDOWN ===")
print(f"{'Seg':>4s} {'Bars':>6s} {'Period':>22s} {'Return':>9s} {'Sharpe':>8s} {'MDD':>8s} {'WinRate':>8s} {'AvgPos':>7s}")
for seg in range(n_seg):
s = seg * seg_len
e = min((seg + 1) * seg_len, T)
seg_pnl = port_pnl[s:e]
seg_cum = np.cumsum(seg_pnl)
t_start = f"bar{s}"
t_end = f"bar{e}"
seg_ret = float(seg_cum[-1])
m, sd = seg_pnl.mean(), seg_pnl.std()
seg_sharpe = m / (sd + 1e-10) * math.sqrt(_H1_PER_YEAR)
seg_mdd = float((np.maximum.accumulate(seg_cum) - seg_cum).max())
# Win rate per bar
win_rate = float((seg_pnl > 0).mean())
# Avg position
seg_pos = pos_np[:, s:e]
avg_pos = float(np.abs(seg_pos).mean())
print(f"{seg+1:>4d} {e-s:>6d} {t_start+' ~ '+t_end:>22s} {seg_ret:>+9.4f} {seg_sharpe:>+8.3f} {seg_mdd:>8.4f} {win_rate*100:>7.1f}% {avg_pos:>7.3f}")
# Per-year analysis
print(f"\n=== PER-YEAR BREAKDOWN ===")
n_years = T // _H1_PER_YEAR
print(f"{'Year':>6s} {'Bars':>6s} {'Return':>9s} {'Sharpe':>8s} {'MDD':>8s} {'WinRate':>8s}")
for yr in range(n_years + 1):
s = yr * _H1_PER_YEAR
e = min((yr + 1) * _H1_PER_YEAR, T)
if s >= T:
break
seg_pnl = port_pnl[s:e]
seg_cum = np.cumsum(seg_pnl)
seg_ret = float(seg_cum[-1])
m, sd = seg_pnl.mean(), seg_pnl.std()
seg_sharpe = m / (sd + 1e-10) * math.sqrt(_H1_PER_YEAR)
seg_mdd = float((np.maximum.accumulate(seg_cum) - seg_cum).max())
win_rate = float((seg_pnl > 0).mean())
t_start = f"bar{s}"
t_end = f"bar{e}"
print(f" Y{yr+1:>2d} {e-s:>6d} {seg_ret:>+9.4f} {seg_sharpe:>+8.3f} {seg_mdd:>8.4f} {win_rate*100:>7.1f}% ({t_start} ~ {t_end})")
# Per-symbol per-half analysis
print(f"\n=== PER-SYMBOL PER-HALF ANALYSIS ===")
split = T // 2
print(f"{'Symbol':<14s} {'H1_Ret':>9s} {'H2_Ret':>9s} {'H1_Sharpe':>10s} {'H2_Sharpe':>10s} {'H1_MDD':>8s} {'H2_MDD':>8s}")
for i, sym in enumerate(syms):
p1 = pnl[i, :split]
p2 = pnl[i, split:]
c1 = np.cumsum(p1)
c2 = np.cumsum(p2)
r1 = float(c1[-1])
r2 = float(c2[-1])
s1 = p1.mean() / (p1.std() + 1e-10) * math.sqrt(_H1_PER_YEAR)
s2 = p2.mean() / (p2.std() + 1e-10) * math.sqrt(_H1_PER_YEAR)
mdd1 = float((np.maximum.accumulate(c1) - c1).max())
mdd2 = float((np.maximum.accumulate(c2) - c2).max())
print(f"{sym:<14s} {r1:>+9.4f} {r2:>+9.4f} {s1:>+10.3f} {s2:>+10.3f} {mdd1:>8.4f} {mdd2:>8.4f}")
# Factor value distribution over time
print(f"\n=== FACTOR VALUE STATISTICS OVER TIME ===")
for seg in range(n_seg):
s = seg * seg_len
e = min((seg + 1) * seg_len, T)
seg_factor = factor_np[:, s:e]
seg_pos = pos_np[:, s:e]
print(f"Seg {seg+1}: factor mean={seg_factor.mean():+.4f} std={seg_factor.std():.4f} "
f"pos_mean={seg_pos.mean():+.4f} pos_abs={np.abs(seg_pos).mean():.4f} "
f"long%={float((seg_pos>0.01).mean())*100:.1f}% short%={float((seg_pos<-0.01).mean())*100:.1f}% flat%={float((np.abs(seg_pos)<0.01).mean())*100:.1f}%")
# Check if factor is basically "always long"
print(f"\n=== POSITION DISTRIBUTION (overall) ===")
for i, sym in enumerate(syms):
p = pos_np[i]
long_pct = float((p > 0.01).mean()) * 100
short_pct = float((p < -0.01).mean()) * 100
flat_pct = float((np.abs(p) < 0.01).mean()) * 100
print(f" {sym:<14s}: long={long_pct:.1f}% short={short_pct:.1f}% flat={flat_pct:.1f}% avg_abs_pos={np.abs(p).mean():.3f}")
backtest_all_groups.py
+516
View File
@@ -0,0 +1,516 @@
"""
backtest_all_groups.py — 对所有组的当前最优因子做完整回测验证
用法：
python backtest_all_groups.py --offline
分别加载 forex / metals_comm / index 三组的最优因子，
在全量历史数据上回测，输出：回测摘要、品种级详情、前后半段一致性、资金曲线图。
"""
import sys, json, math
from pathlib import Path
import numpy as np
import torch
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import matplotlib.gridspec as gridspec
sys.path.insert(0, str(Path(__file__).parent))
from config import Config
from data_pipeline.data_manager import MT5DataManager
from data_pipeline.fetcher import MT5DataFetcher
from model_core.vocab import FORMULA_VOCAB, VOCAB_VERSION
from model_core.vm import StackVM
from model_core.features import MT5FeatureEngineer
from strategy_manager.signal import compute_target_positions_stateless
_H1_PER_YEAR = 6240
OUTPUT_DIR = "backtest_output"
GROUP_CONFIG = {
"forex": {"symbols": Config.SYMBOL_GROUPS["forex"], "cost_rate": 0.00015},
"metals_comm": {"symbols": Config.SYMBOL_GROUPS["metals_comm"], "cost_rate": 0.00025},
"index": {"symbols": Config.SYMBOL_GROUPS["index"], "cost_rate": 0.00030},
}
def decode(toks):
names = FORMULA_VOCAB.token_names
return " -> ".join(names[t] if 0 <= t < len(names) else f"?{t}" for t in toks)
def calc_sharpe(pnl):
m, s = pnl.mean(), pnl.std()
return float(m / (s + 1e-10) * math.sqrt(_H1_PER_YEAR))
def calc_sortino(pnl):
m = pnl.mean()
down = pnl[pnl < 0]
ds = down.std() if len(down) > 0 else 1e-10
ds = max(ds, abs(m), 1e-10)
return float(np.clip(m / ds * math.sqrt(_H1_PER_YEAR), -20, 20))
def calc_mdd(cum_pnl):
peak = np.maximum.accumulate(cum_pnl)
return float((peak - cum_pnl).max())
def calc_ic(factor, target_ret):
N, T = factor.shape
ic_list = []
for n in range(N):
x = factor[n, :-1]
y = target_ret[n, 1:]
xm = x - x.mean()
ym = y - y.mean()
sx = np.sqrt((xm**2).mean())
sy = np.sqrt((ym**2).mean())
if sx < 1e-6 or sy < 1e-6:
continue
ic = (xm * ym).mean() / (sx * sy + 1e-8)
ic_list.append(float(ic))
return float(np.mean(ic_list)) if ic_list else 0.0
def calc_annual_return(total_ret, T):
years = T / _H1_PER_YEAR
return float(total_ret / years) if years > 0 else 0.0
def calc_calmar(total_ret, mdd, T):
ann = calc_annual_return(total_ret, T)
return float(ann / mdd) if mdd > 1e-8 else 0.0
def _max_consecutive_loss(pnl):
max_streak = 0
current = 0
for p in pnl:
if p < 0:
current += 1
max_streak = max(max_streak, current)
else:
current = 0
return max_streak
def backtest_one(formula, feat, target_ret, symbols, cost_rate):
vm = StackVM()
factor = vm.execute(formula, feat)
if factor is None:
return None
N, T = factor.shape
factor_np = factor.detach().numpy()
target_np = target_ret.detach().numpy()
pos = compute_target_positions_stateless(factor)
pos_np = pos.detach().numpy()
prev = np.zeros_like(pos_np)
prev[:, 1:] = pos_np[:, :-1]
turnover = np.abs(pos_np - prev)
pnl = pos_np * target_np - turnover * cost_rate
per_sym = {}
for i, sym in enumerate(symbols[:N]):
p = pnl[i]
cum = np.cumsum(p)
n_trades = int((np.abs(np.diff(pos_np[i])) > 0.5).sum())
per_sym[sym] = {
"pnl": p,
"cum": cum,
"total_ret": float(cum[-1]),
"ann_ret": calc_annual_return(float(cum[-1]), T),
"sharpe": calc_sharpe(p),
"sortino": calc_sortino(p),
"mdd": calc_mdd(cum),
"n_trades": n_trades,
"avg_hold": float(T / max(1, n_trades)) if n_trades > 0 else float(T),
"ic": calc_ic(factor_np[i:i+1], target_np[i:i+1]),
}
port_pnl = pnl.mean(axis=0)
port_cum = np.cumsum(port_pnl)
ic = calc_ic(factor_np, target_np)
split = T // 2
p1 = port_pnl[:split]
p2 = port_pnl[split:]
port_total_ret = float(port_cum[-1])
port_mdd = calc_mdd(port_cum)
port_ann_ret = calc_annual_return(port_total_ret, T)
port_calmar = calc_calmar(port_total_ret, port_mdd, T)
return {
"formula": formula,
"readable": decode(formula),
"per_sym": per_sym,
"port_pnl": port_pnl,
"port_cum": port_cum,
"port_total_ret": port_total_ret,
"port_ann_ret": port_ann_ret,
"port_sharpe": calc_sharpe(port_pnl),
"port_sortino": calc_sortino(port_pnl),
"port_mdd": port_mdd,
"port_calmar": port_calmar,
"ic": ic,
"n_pos_syms": sum(1 for d in per_sym.values() if d["total_ret"] > 0),
"n_syms": N,
"half1_sharpe": calc_sharpe(p1),
"half2_sharpe": calc_sharpe(p2),
"half1_sortino": calc_sortino(p1),
"half2_sortino": calc_sortino(p2),
"half1_ret": float(np.cumsum(p1)[-1]),
"half2_ret": float(np.cumsum(p2)[-1]),
"avg_turnover": float(turnover.mean()),
"avg_hold_h": float(np.mean([d["avg_hold"] for d in per_sym.values()])),
"max_consec_loss": _max_consecutive_loss(port_pnl),
}
def plot_group(result, symbols, group_name, output_dir, cost_rate):
Path(output_dir).mkdir(parents=True, exist_ok=True)
fig = plt.figure(figsize=(18, 12), dpi=110)
gs = gridspec.GridSpec(3, 1, height_ratios=[3, 1, 2], hspace=0.15)
ax_eq = fig.add_subplot(gs[0])
ax_dd = fig.add_subplot(gs[1], sharex=ax_eq)
ax_sym = fig.add_subplot(gs[2], sharex=ax_eq)
T = len(result["port_pnl"])
x = np.arange(T)
port_cum = result["port_cum"]
ax_eq.plot(x, port_cum, linewidth=2.0, color="#1565c0",
label=f"Portfolio (AnnRet={result['port_ann_ret']:+.4f}, Sortino={result['port_sortino']:+.2f}, MDD={result['port_mdd']:.3f})")
ax_eq.axhline(0, color="gray", linewidth=0.5, linestyle="--")
ax_eq.set_ylabel("Cumulative Return", fontsize=10)
ax_eq.legend(loc="upper left", fontsize=9, framealpha=0.85)
ax_eq.grid(alpha=0.25)
ax_eq.set_title(f"{group_name} group backtest | {' + '.join(symbols)} equal weight\n"
f"Formula: {result['readable']}\n"
f"Cost: {cost_rate} T={T} bars H1", fontsize=11, pad=8)
peak = np.maximum.accumulate(port_cum)
dd = port_cum - peak
ax_dd.fill_between(x, dd, 0, alpha=0.4, color="#1565c0", label="Drawdown")
ax_dd.axhline(0, color="gray", linewidth=0.5)
ax_dd.set_ylabel("Drawdown", fontsize=9)
ax_dd.grid(alpha=0.2)
ax_dd.legend(loc="lower left", fontsize=8)
colors = ["#e65100", "#00897b", "#6a1b9a", "#b71c1c", "#26418f"]
for i, sym in enumerate(symbols):
if sym in result["per_sym"]:
cum = result["per_sym"][sym]["cum"]
c = colors[i % len(colors)]
ax_sym.plot(x, cum, linewidth=1.2, color=c, alpha=0.8,
label=f"{sym} (Ret={result['per_sym'][sym]['total_ret']:+.3f})")
ax_sym.axhline(0, color="gray", linewidth=0.5, linestyle="--")
ax_sym.set_ylabel("Per-Symbol Cum Return", fontsize=9)
ax_sym.set_xlabel("Bar Index (H1)", fontsize=9)
ax_sym.legend(loc="upper left", fontsize=7, ncol=3, framealpha=0.85)
ax_sym.grid(alpha=0.2)
plt.tight_layout()
plot_path = Path(output_dir) / f"{group_name}_backtest.png"
fig.savefig(plot_path, bbox_inches="tight")
plt.close(fig)
return str(plot_path)
def run_group_backtest(group_name, group_cfg, fetcher, multi_mgr):
symbols = group_cfg["symbols"]
cost_rate = group_cfg["cost_rate"]
strategy_path = Path(f"strategies/best_{group_name}.json")
print(f"\n{'='*70}")
print(f" [{group_name}] group backtest")
print(f"{'='*70}")
if not strategy_path.exists():
print(f" [SKIP] No strategy file: {strategy_path}")
return None
data = json.load(open(strategy_path))
if data.get("vocab_version", "unknown") != VOCAB_VERSION:
print(f" [SKIP] Vocab mismatch: {data.get('vocab_version')} != {VOCAB_VERSION}")
return None
formula = data["formula"]
best_score = data.get("best_score", 0.0)
print(f" Formula tokens : {formula}")
print(f" Readable : {decode(formula)}")
print(f" Train score : {best_score:.4f}")
print(f" Symbols : {symbols}")
print(f" Cost rate : {cost_rate}")
# Load group data independently
Config.SYMBOLS = list(symbols)
group_mgr = MT5DataManager(fetcher)
group_mgr.load()
Config.SYMBOLS = list(Config.SYMBOL_GROUPS) and [] # reset
# Actually restore from the full config
from config import Config as _Cfg
_Cfg.SYMBOLS = ["EURUSD", "USDJPY", "XAUUSD", "AAVUSD", "COCOA.c",
"US30.cash", "US100.cash", "US500.cash", "US2000.cash", "JP225.cash"]
raw_dict = group_mgr.raw_dict
syms_loaded = group_mgr.symbols
T = raw_dict["open"].shape[1]
print(f" Loaded: {syms_loaded} T={T} bars")
feat = MT5FeatureEngineer.compute_features(raw_dict)
target_ret = group_mgr.target_ret
result = backtest_one(formula, feat, target_ret, syms_loaded, cost_rate)
if result is None:
print(f" [ERROR] Factor execution failed!")
return None
# Print summary
print(f"\n --- Portfolio Summary ({T} bars H1, {len(syms_loaded)} symbols) ---")
print(f" Total Return : {result['port_total_ret']:+.4f}")
print(f" Annual Return: {result['port_ann_ret']:+.4f}")
print(f" Sharpe : {result['port_sharpe']:+.4f}")
print(f" Sortino : {result['port_sortino']:+.4f}")
print(f" Max DD : {result['port_mdd']:.4f}")
print(f" Calmar : {result['port_calmar']:+.4f}")
print(f" IC : {result['ic']:.5f}")
print(f" Avg Turnover : {result['avg_turnover']:.6f}")
print(f" Avg Hold : {result['avg_hold_h']:.1f}h")
print(f" Max Consec Loss: {result['max_consec_loss']} bars")
print(f" Profitable : {result['n_pos_syms']}/{result['n_syms']}")
print(f"\n --- Half-split Consistency ---")
print(f" H1: Sharpe={result['half1_sharpe']:+.4f} Ret={result['half1_ret']:+.4f}")
print(f" H2: Sharpe={result['half2_sharpe']:+.4f} Ret={result['half2_ret']:+.4f}")
h1, h2 = result["half1_sharpe"], result["half2_sharpe"]
if h1 > 0 and h2 > 0:
print(f" [OK] Both halves positive, good consistency")
elif h1 * h2 > 0:
print(f" [WARN] Same sign but negative")
else:
print(f" [FAIL] Opposite signs, overfitting suspected")
print(f"\n --- Per-Symbol Detail ---")
for sym, d in result["per_sym"].items():
sig = "[OK]" if d["total_ret"] > 0 else "[X]"
print(f" {sym:14s}: Ret={d['total_ret']:+.4f} Ann={d['ann_ret']:+.4f} "
f"Sharpe={d['sharpe']:+.4f} Sortino={d['sortino']:+.4f} "
f"MDD={d['mdd']:.4f} IC={d['ic']:.5f} "
f"Trades={d['n_trades']} Hold={d['avg_hold']:.0f}h {sig}")
# FTMO check
print(f"\n --- FTMO Assessment ---")
issues = []
if result["port_mdd"] > 0.10:
issues.append(f"[WARN] MDD={result['port_mdd']:.4f} > 10% FTMO Max Loss")
else:
print(f" [OK] MDD={result['port_mdd']:.4f} < 10%")
if result["port_ann_ret"] > 0:
print(f" [OK] AnnRet={result['port_ann_ret']:+.4f} positive")
else:
issues.append(f"[WARN] Negative annual return")
if result["max_consec_loss"] > 500:
issues.append(f"[WARN] Max consec loss {result['max_consec_loss']} bars too long")
if result["avg_hold_h"] < 2:
issues.append(f"[WARN] Avg hold {result['avg_hold_h']:.1f}h too short")
if result["n_pos_syms"] < result["n_syms"] // 2:
issues.append(f"[WARN] Only {result['n_pos_syms']}/{result['n_syms']} profitable")
for iss in issues:
print(f" {iss}")
if not issues:
print(f" [OK] No obvious FTMO violations")
# Plot
plot_path = plot_group(result, syms_loaded, group_name, OUTPUT_DIR, cost_rate)
print(f"\n Plot saved -> {plot_path}")
# Save JSON report
report = {
"group": group_name,
"formula": formula,
"readable": result["readable"],
"best_score": best_score,
"vocab_version": VOCAB_VERSION,
"symbols": syms_loaded,
"T": T,
"cost_rate": cost_rate,
"port_total_ret": round(result["port_total_ret"], 6),
"port_ann_ret": round(result["port_ann_ret"], 6),
"port_sharpe": round(result["port_sharpe"], 4),
"port_sortino": round(result["port_sortino"], 4),
"port_mdd": round(result["port_mdd"], 6),
"port_calmar": round(result["port_calmar"], 4),
"ic": round(result["ic"], 6),
"avg_turnover": round(result["avg_turnover"], 8),
"avg_hold_h": round(result["avg_hold_h"], 1),
"max_consec_loss": result["max_consec_loss"],
"n_pos_syms": result["n_pos_syms"],
"n_syms": result["n_syms"],
"half1_sharpe": round(result["half1_sharpe"], 4),
"half2_sharpe": round(result["half2_sharpe"], 4),
"half1_ret": round(result["half1_ret"], 6),
"half2_ret": round(result["half2_ret"], 6),
"per_sym": {
sym: {
"total_ret": round(d["total_ret"], 6),
"ann_ret": round(d["ann_ret"], 6),
"sharpe": round(d["sharpe"], 4),
"sortino": round(d["sortino"], 4),
"mdd": round(d["mdd"], 6),
"ic": round(d["ic"], 6),
"n_trades": d["n_trades"],
"avg_hold": round(d["avg_hold"], 1),
}
for sym, d in result["per_sym"].items()
},
}
report_path = Path(OUTPUT_DIR) / f"{group_name}_backtest.json"
with open(report_path, "w") as f:
json.dump(report, f, indent=2, ensure_ascii=False)
print(f" Report saved -> {report_path}")
return result
def main():
offline = "--offline" in sys.argv
# First, save metals_comm current best from latest checkpoint
print("Checking for latest metals_comm checkpoint to save current best...")
import glob
ckpt_files = sorted(glob.glob(r"D:\cl\MT5_AlphaGPT\checkpoints\ckpt_metals_comm_step_*.pt"))
if ckpt_files:
import torch
latest_ckpt = ckpt_files[-1]
ckpt = torch.load(latest_ckpt, map_location='cpu', weights_only=False)
best_formula = ckpt.get('best_formula')
best_score = ckpt.get('best_score', 0)
step = ckpt.get('step', 0)
if best_formula and best_score > 0:
# Update strategy file with latest best
strategy_path = Path("strategies/best_metals_comm.json")
strategy_path.parent.mkdir(exist_ok=True)
strategy_path.write_text(json.dumps({
"vocab_version": VOCAB_VERSION,
"symbol": "metals_comm",
"formula": best_formula,
"best_score": best_score,
"step": step,
}, indent=2))
print(f" Updated best_metals_comm.json: score={best_score:.4f} step={step} formula={best_formula}")
print(f"\n{'='*70}")
print(f" All-Groups Backtest Analysis")
print(f"{'='*70}")
with MT5DataFetcher(offline=offline) as fetcher:
all_results = {}
for gname, gcfg in GROUP_CONFIG.items():
result = run_group_backtest(gname, gcfg, fetcher, None)
if result:
all_results[gname] = result
# Cross-group summary
if all_results:
print(f"\n{'='*70}")
print(f" Cross-Group Comparison")
print(f"{'='*70}")
print(f" {'Group':<14s} {'Score':>8s} {'AnnRet':>9s} {'Sharpe':>8s} {'Sortino':>8s} {'MDD':>8s} {'Calmar':>8s} {'IC':>8s} {'Pos/N':>6s}")
print(f" {'-'*14} {'-'*8} {'-'*9} {'-'*8} {'-'*8} {'-'*8} {'-'*8} {'-'*8} {'-'*6}")
for gname, r in all_results.items():
print(f" {gname:<14s} "
f"{r.get('best_score',0):>8.4f} "
f"{r['port_ann_ret']:>+9.4f} "
f"{r['port_sharpe']:>+8.4f} "
f"{r['port_sortino']:>+8.4f} "
f"{r['port_mdd']:>8.4f} "
f"{r['port_calmar']:>+8.4f} "
f"{r['ic']:>8.5f} "
f"{r['n_pos_syms']:>2d}/{r['n_syms']:<2d}")
# Combined portfolio (equal weight across groups)
print(f"\n --- Combined Portfolio (equal weight across groups) ---")
# Align all groups to min T
min_T = min(len(r["port_pnl"]) for r in all_results.values())
combined_pnl = np.zeros(min_T)
for r in all_results.values():
combined_pnl += r["port_pnl"][:min_T]
combined_pnl /= len(all_results)
combined_cum = np.cumsum(combined_pnl)
combined_mdd = calc_mdd(combined_cum)
combined_ann = calc_annual_return(float(combined_cum[-1]), min_T)
combined_sharpe = calc_sharpe(combined_pnl)
combined_sortino = calc_sortino(combined_pnl)
combined_calmar = calc_calmar(float(combined_cum[-1]), combined_mdd, min_T)
print(f" Total Return : {float(combined_cum[-1]):+.4f}")
print(f" Annual Return: {combined_ann:+.4f}")
print(f" Sharpe : {combined_sharpe:+.4f}")
print(f" Sortino : {combined_sortino:+.4f}")
print(f" Max DD : {combined_mdd:.4f}")
print(f" Calmar : {combined_calmar:+.4f}")
# Plot combined
fig, axes = plt.subplots(2, 1, figsize=(16, 8), dpi=110, gridspec_kw={'height_ratios': [3, 1], 'hspace': 0.15})
x = np.arange(min_T)
colors_g = {"forex": "#2e7d32", "metals_comm": "#e65100", "index": "#1565c0"}
for gname, r in all_results.items():
axes[0].plot(x, np.cumsum(r["port_pnl"][:min_T]), linewidth=1.2, alpha=0.7,
color=colors_g.get(gname, "gray"), label=f"{gname}")
axes[0].plot(x, combined_cum, linewidth=2.5, color="#d32f2f",
label=f"Combined (Ann={combined_ann:+.4f}, Sharpe={combined_sharpe:+.2f}, MDD={combined_mdd:.3f})")
axes[0].axhline(0, color="gray", linewidth=0.5, linestyle="--")
axes[0].legend(fontsize=9)
axes[0].set_title("All Groups Combined Portfolio", fontsize=12)
axes[0].grid(alpha=0.25)
peak = np.maximum.accumulate(combined_cum)
dd = combined_cum - peak
axes[1].fill_between(x, dd, 0, alpha=0.4, color="#d32f2f")
axes[1].axhline(0, color="gray", linewidth=0.5)
axes[1].set_ylabel("Drawdown", fontsize=9)
axes[1].set_xlabel("Bar Index (H1)", fontsize=9)
axes[1].grid(alpha=0.2)
plt.tight_layout()
combined_path = Path(OUTPUT_DIR) / "all_groups_combined.png"
fig.savefig(combined_path, bbox_inches="tight")
plt.close(fig)
print(f"\n Combined plot saved -> {combined_path}")
# Save combined report
combined_report = {
"groups": list(all_results.keys()),
"T": min_T,
"total_ret": round(float(combined_cum[-1]), 6),
"ann_ret": round(combined_ann, 6),
"sharpe": round(combined_sharpe, 4),
"sortino": round(combined_sortino, 4),
"mdd": round(combined_mdd, 6),
"calmar": round(combined_calmar, 4),
}
combined_report_path = Path(OUTPUT_DIR) / "all_groups_combined.json"
with open(combined_report_path, "w") as f:
json.dump(combined_report, f, indent=2, ensure_ascii=False)
print(f" Combined report saved -> {combined_report_path}")
print(f"\n{'='*70}")
print(f" All-Groups Backtest Complete")
print(f"{'='*70}\n")
if __name__ == "__main__":
main()
backtest_detailed.py
+806
View File
@@ -0,0 +1,806 @@
"""
backtest_detailed.py — 全组详细回测报告
输出：交易次数、平均持仓、胜率、盈亏比、最大单笔盈亏、平均盈亏、
前后半段、滚动 Sharpe、品种相关性矩阵、月度收益等
"""
import sys, json, math
from pathlib import Path
import numpy as np
import torch
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import matplotlib.gridspec as gridspec
sys.path.insert(0, str(Path(__file__).parent))
from config import Config
from data_pipeline.data_manager import MT5DataManager
from data_pipeline.fetcher import MT5DataFetcher
from model_core.vocab import FORMULA_VOCAB, VOCAB_VERSION
from model_core.vm import StackVM
from model_core.features import MT5FeatureEngineer
from strategy_manager.signal import compute_target_positions_stateless
_H1_PER_YEAR = 6240
OUTPUT_DIR = "backtest_output"
GROUP_CONFIG = {
"forex": {"symbols": Config.SYMBOL_GROUPS["forex"], "cost_rate": 0.00015},
"metals_comm": {"symbols": Config.SYMBOL_GROUPS["metals_comm"], "cost_rate": 0.00025},
"index": {"symbols": Config.SYMBOL_GROUPS["index"], "cost_rate": 0.00030},
}
def decode(toks):
names = FORMULA_VOCAB.token_names
return " -> ".join(names[t] if 0 <= t < len(names) else f"?{t}" for t in toks)
def calc_sharpe(pnl):
m, s = pnl.mean(), pnl.std()
return float(m / (s + 1e-10) * math.sqrt(_H1_PER_YEAR))
def calc_sortino(pnl):
m = pnl.mean()
down = pnl[pnl < 0]
ds = down.std() if len(down) > 0 else 1e-10
ds = max(ds, abs(m), 1e-10)
return float(np.clip(m / ds * math.sqrt(_H1_PER_YEAR), -20, 20))
def calc_mdd(cum_pnl):
peak = np.maximum.accumulate(cum_pnl)
return float((peak - cum_pnl).max())
def calc_ic(factor, target_ret):
N, T = factor.shape
ic_list = []
for n in range(N):
x = factor[n, :-1]
y = target_ret[n, 1:]
xm = x - x.mean()
ym = y - y.mean()
sx = np.sqrt((xm**2).mean())
sy = np.sqrt((ym**2).mean())
if sx < 1e-6 or sy < 1e-6:
continue
ic = (xm * ym).mean() / (sx * sy + 1e-8)
ic_list.append(float(ic))
return float(np.mean(ic_list)) if ic_list else 0.0
def calc_annual_return(total_ret, T):
years = T / _H1_PER_YEAR
return float(total_ret / years) if years > 0 else 0.0
def calc_calmar(total_ret, mdd, T):
ann = calc_annual_return(total_ret, T)
return float(ann / mdd) if mdd > 1e-8 else 0.0
def _max_consecutive_loss(pnl):
max_streak = 0
current = 0
for p in pnl:
if p < 0:
current += 1
max_streak = max(max_streak, current)
else:
current = 0
return max_streak
def analyze_trades(pos_np, pnl_np, cost_rate):
"""分析交易细节：持仓变化点为交易事件"""
N, T = pos_np.shape
results = []
for n in range(N):
pos = pos_np[n]
pnl = pnl_np[n]
# 找到持仓变化点（开仓/平仓/翻转）
changes = np.where(np.abs(np.diff(pos)) > 0.01)[0] + 1
if len(changes) == 0:
results.append({
"n_trades": 0, "n_long": 0, "n_short": 0, "n_flat": 0,
"win_rate": 0, "avg_win": 0, "avg_loss": 0, "profit_factor": 0,
"max_win": 0, "max_loss": 0, "avg_trade_pnl": 0, "avg_hold_bars": 0,
})
continue
# 把每段持仓作为一个 trade
trade_bounds = []
prev_idx = 0
for c in changes:
trade_bounds.append((prev_idx, c))
prev_idx = c
trade_bounds.append((prev_idx, T))
n_trades = 0
n_long = 0
n_short = 0
n_flat = 0
wins = []
losses = []
hold_bars = []
for (s, e) in trade_bounds:
seg_pos = pos[s]
seg_pnl = pnl[s:e].sum()
seg_len = e - s
if abs(seg_pos) < 0.01:
n_flat += 1
continue
n_trades += 1
if seg_pos > 0:
n_long += 1
else:
n_short += 1
hold_bars.append(seg_len)
if seg_pnl > 0:
wins.append(seg_pnl)
else:
losses.append(seg_pnl)
win_rate = len(wins) / max(1, n_trades)
avg_win = np.mean(wins) if wins else 0
avg_loss = np.mean(losses) if losses else 0
gross_win = sum(wins)
gross_loss = abs(sum(losses))
pf = gross_win / max(gross_loss, 1e-8)
all_trades = wins + losses
max_win = max(all_trades) if all_trades else 0
max_loss = min(all_trades) if all_trades else 0
avg_trade = np.mean(all_trades) if all_trades else 0
avg_hold = np.mean(hold_bars) if hold_bars else 0
results.append({
"n_trades": n_trades,
"n_long": n_long,
"n_short": n_short,
"n_flat": n_flat,
"win_rate": float(win_rate),
"avg_win": float(avg_win),
"avg_loss": float(avg_loss),
"profit_factor": float(pf),
"max_win": float(max_win),
"max_loss": float(max_loss),
"avg_trade_pnl": float(avg_trade),
"avg_hold_bars": float(avg_hold),
})
return results
def backtest_one(formula, feat, target_ret, symbols, cost_rate):
vm = StackVM()
factor = vm.execute(formula, feat)
if factor is None:
return None
N, T = factor.shape
factor_np = factor.detach().numpy()
target_np = target_ret.detach().numpy()
pos = compute_target_positions_stateless(factor)
pos_np = pos.detach().numpy()
prev = np.zeros_like(pos_np)
prev[:, 1:] = pos_np[:, :-1]
turnover = np.abs(pos_np - prev)
pnl = pos_np * target_np - turnover * cost_rate
# 交易分析
trade_stats = analyze_trades(pos_np, pnl, cost_rate)
per_sym = {}
for i, sym in enumerate(symbols[:N]):
p = pnl[i]
cum = np.cumsum(p)
ts = trade_stats[i]
per_sym[sym] = {
"pnl": p,
"cum": cum,
"total_ret": float(cum[-1]),
"ann_ret": calc_annual_return(float(cum[-1]), T),
"sharpe": calc_sharpe(p),
"sortino": calc_sortino(p),
"mdd": calc_mdd(cum),
"ic": calc_ic(factor_np[i:i+1], target_np[i:i+1]),
# 交易细节
"n_trades": ts["n_trades"],
"n_long": ts["n_long"],
"n_short": ts["n_short"],
"win_rate": ts["win_rate"],
"avg_win": ts["avg_win"],
"avg_loss": ts["avg_loss"],
"profit_factor": ts["profit_factor"],
"max_win": ts["max_win"],
"max_loss": ts["max_loss"],
"avg_trade_pnl": ts["avg_trade_pnl"],
"avg_hold_bars": ts["avg_hold_bars"],
"avg_hold_h": ts["avg_hold_bars"], # H1 每 bar = 1h
"n_total_bars": T,
}
port_pnl = pnl.mean(axis=0)
port_cum = np.cumsum(port_pnl)
ic = calc_ic(factor_np, target_np)
# 组合交易统计
port_pos = pos_np.mean(axis=0)
port_trade = analyze_trades(port_pos.reshape(1, -1), port_pnl.reshape(1, -1), cost_rate)[0]
# 分段
split = T // 2
p1 = port_pnl[:split]
p2 = port_pnl[split:]
# 三等分
t1 = T // 3
t2 = 2 * T // 3
seg1 = port_pnl[:t1]
seg2 = port_pnl[t1:t2]
seg3 = port_pnl[t2:]
port_total_ret = float(port_cum[-1])
port_mdd = calc_mdd(port_cum)
port_ann_ret = calc_annual_return(port_total_ret, T)
port_calmar = calc_calmar(port_total_ret, port_mdd, T)
# 逐年收益
years = max(1, T // _H1_PER_YEAR)
yearly_ret = []
for y in range(years):
s = y * _H1_PER_YEAR
e = min((y + 1) * _H1_PER_YEAR, T)
yr = float(port_pnl[s:e].sum())
yearly_ret.append(yr)
return {
"formula": formula,
"readable": decode(formula),
"per_sym": per_sym,
"port_pnl": port_pnl,
"port_cum": port_cum,
"port_total_ret": port_total_ret,
"port_ann_ret": port_ann_ret,
"port_sharpe": calc_sharpe(port_pnl),
"port_sortino": calc_sortino(port_pnl),
"port_mdd": port_mdd,
"port_calmar": port_calmar,
"ic": ic,
"n_pos_syms": sum(1 for d in per_sym.values() if d["total_ret"] > 0),
"n_syms": N,
"T": T,
"years_span": T / _H1_PER_YEAR,
# 交易细节
"port_n_trades": port_trade["n_trades"],
"port_n_long": port_trade["n_long"],
"port_n_short": port_trade["n_short"],
"port_win_rate": port_trade["win_rate"],
"port_avg_win": port_trade["avg_win"],
"port_avg_loss": port_trade["avg_loss"],
"port_profit_factor": port_trade["profit_factor"],
"port_max_win": port_trade["max_win"],
"port_max_loss": port_trade["max_loss"],
"port_avg_trade_pnl": port_trade["avg_trade_pnl"],
"port_avg_hold_bars": port_trade["avg_hold_bars"],
"port_avg_hold_h": port_trade["avg_hold_bars"],
"avg_turnover": float(turnover.mean()),
"max_consec_loss": _max_consecutive_loss(port_pnl),
# 分段
"half1_sharpe": calc_sharpe(p1),
"half2_sharpe": calc_sharpe(p2),
"half1_sortino": calc_sortino(p1),
"half2_sortino": calc_sortino(p2),
"half1_ret": float(np.cumsum(p1)[-1]),
"half2_ret": float(np.cumsum(p2)[-1]),
"third1_sharpe": calc_sharpe(seg1),
"third2_sharpe": calc_sharpe(seg2),
"third3_sharpe": calc_sharpe(seg3),
"third1_ret": float(np.cumsum(seg1)[-1]),
"third2_ret": float(np.cumsum(seg2)[-1]),
"third3_ret": float(np.cumsum(seg3)[-1]),
# 逐年
"yearly_ret": yearly_ret,
}
def print_detailed_report(group_name, result, symbols, cost_rate, best_score):
r = result
T = r["T"]
years = r["years_span"]
print(f"\n{'='*80}")
print(f" [{group_name}] DETAILED BACKTEST REPORT")
print(f"{'='*80}")
print(f" Formula : {r['readable']}")
print(f" Tokens : {r['formula']}")
print(f" Train score: {best_score:.4f}")
print(f" Symbols : {symbols}")
print(f" Data : {T} bars (H1) ~ {years:.2f} years")
print(f" Cost rate : {cost_rate}")
# === 组合概览 ===
print(f"\n {'─'*70}")
print(f" PORTFOLIO OVERVIEW (equal weight, {r['n_syms']} symbols)")
print(f" {'─'*70}")
print(f" Total Return : {r['port_total_ret']:+.4f} ({r['port_total_ret']*100:+.2f}%)")
print(f" Annual Return : {r['port_ann_ret']:+.4f} ({r['port_ann_ret']*100:+.2f}%)")
print(f" Sharpe Ratio : {r['port_sharpe']:+.4f}")
print(f" Sortino Ratio : {r['port_sortino']:+.4f}")
print(f" Max Drawdown : {r['port_mdd']:.4f} ({r['port_mdd']*100:.2f}%)")
print(f" Calmar Ratio : {r['port_calmar']:+.4f}")
print(f" IC (mean) : {r['ic']:.5f}")
print(f" Avg Turnover/bar : {r['avg_turnover']:.6f}")
print(f" Max Consec Loss : {r['max_consec_loss']} bars ({r['max_consec_loss']}h)")
# === 交易统计 ===
print(f"\n {'─'*70}")
print(f" TRADING STATISTICS (portfolio level)")
print(f" {'─'*70}")
print(f" Total Trades : {r['port_n_trades']}")
print(f" Long / Short : {r['port_n_long']} / {r['port_n_short']}")
print(f" Win Rate : {r['port_win_rate']*100:.1f}%")
print(f" Profit Factor : {r['port_profit_factor']:.2f}")
print(f" Avg Win : {r['port_avg_win']:+.6f}")
print(f" Avg Loss : {r['port_avg_loss']:+.6f}")
print(f" Avg Trade PnL : {r['port_avg_trade_pnl']:+.6f}")
print(f" Max Win : {r['port_max_win']:+.6f}")
print(f" Max Loss : {r['port_max_loss']:+.6f}")
print(f" Avg Hold : {r['port_avg_hold_h']:.1f} bars ({r['port_avg_hold_h']:.1f}h)")
# === 前后半段 ===
print(f"\n {'─'*70}")
print(f" CONSISTENCY ANALYSIS")
print(f" {'─'*70}")
print(f" Half-split (50/50):")
print(f" H1: Sharpe={r['half1_sharpe']:+.4f} Sortino={r['half1_sortino']:+.4f} Ret={r['half1_ret']:+.4f}")
print(f" H2: Sharpe={r['half2_sharpe']:+.4f} Sortino={r['half2_sortino']:+.4f} Ret={r['half2_ret']:+.4f}")
h1, h2 = r["half1_sharpe"], r["half2_sharpe"]
if h1 > 0 and h2 > 0:
print(f" [OK] Both positive")
elif h1 * h2 > 0:
print(f" [WARN] Same sign but negative")
else:
print(f" [FAIL] Opposite signs - overfitting suspected!")
print(f" Third-split (33/33/33):")
print(f" T1: Sharpe={r['third1_sharpe']:+.4f} Ret={r['third1_ret']:+.4f}")
print(f" T2: Sharpe={r['third2_sharpe']:+.4f} Ret={r['third2_ret']:+.4f}")
print(f" T3: Sharpe={r['third3_sharpe']:+.4f} Ret={r['third3_ret']:+.4f}")
print(f" Yearly Returns:")
for i, yr in enumerate(r["yearly_ret"]):
print(f" Year {i+1}: {yr:+.4f} ({yr*100:+.2f}%)")
# === 品种详情 ===
print(f"\n {'─'*70}")
print(f" PER-SYMBOL BREAKDOWN")
print(f" {'─'*70}")
print(f" {'Symbol':<14s} {'TotRet':>8s} {'AnnRet':>8s} {'Sharpe':>7s} {'Sort':>7s} {'MDD':>7s} {'IC':>8s} "
f"{'Trades':>6s} {'Long':>5s} {'Short':>5s} {'WinR%':>6s} {'PF':>5s} {'AvgHold':>7s} {'AvgWin':>9s} {'AvgLoss':>9s}")
print(f" {'-'*14} {'-'*8} {'-'*8} {'-'*7} {'-'*7} {'-'*7} {'-'*8} {'-'*6} {'-'*5} {'-'*5} {'-'*6} {'-'*5} {'-'*7} {'-'*9} {'-'*9}")
for sym, d in r["per_sym"].items():
sig = "OK" if d["total_ret"] > 0 else "X"
print(f" {sym:<14s} {d['total_ret']:>+8.4f} {d['ann_ret']:>+8.4f} {d['sharpe']:>+7.3f} {d['sortino']:>+7.3f} "
f"{d['mdd']:>7.4f} {d['ic']:>8.5f} {d['n_trades']:>6d} {d['n_long']:>5d} {d['n_short']:>5d} "
f"{d['win_rate']*100:>5.1f}% {d['profit_factor']:>5.2f} {d['avg_hold_h']:>6.1f}h "
f"{d['avg_win']:>+9.5f} {d['avg_loss']:>+9.5f} [{sig}]")
# === FTMO ===
print(f"\n {'─'*70}")
print(f" FTMO ASSESSMENT")
print(f" {'─'*70}")
issues = []
if r["port_mdd"] > 0.10:
issues.append(f"[WARN] MDD={r['port_mdd']*100:.2f}% > 10% FTMO Max Loss")
else:
print(f" [OK] MDD={r['port_mdd']*100:.2f}% < 10%")
if r["port_ann_ret"] > 0:
print(f" [OK] Annual return positive ({r['port_ann_ret']*100:+.2f}%)")
else:
issues.append(f"[WARN] Negative annual return")
if r["port_profit_factor"] > 1.3:
print(f" [OK] Profit Factor={r['port_profit_factor']:.2f} > 1.3")
else:
issues.append(f"[WARN] Profit Factor={r['port_profit_factor']:.2f} < 1.3")
if r["port_win_rate"] > 0.35:
print(f" [OK] Win rate={r['port_win_rate']*100:.1f}% > 35%")
else:
issues.append(f"[WARN] Win rate={r['port_win_rate']*100:.1f}% < 35%")
if r["max_consec_loss"] > 500:
issues.append(f"[WARN] Max consec loss {r['max_consec_loss']} bars too long")
if r["port_avg_hold_h"] < 2:
issues.append(f"[WARN] Avg hold {r['port_avg_hold_h']:.1f}h too short")
if r["n_pos_syms"] < r["n_syms"] // 2:
issues.append(f"[WARN] Only {r['n_pos_syms']}/{r['n_syms']} profitable")
for iss in issues:
print(f" {iss}")
if not issues:
print(f" [OK] No FTMO violations detected")
def plot_detailed(result, symbols, group_name, output_dir, cost_rate):
Path(output_dir).mkdir(parents=True, exist_ok=True)
r = result
T = r["T"]
x = np.arange(T)
fig = plt.figure(figsize=(20, 16), dpi=110)
gs = gridspec.GridSpec(4, 2, height_ratios=[2.5, 1, 1.5, 1.5], width_ratios=[1.5, 1],
hspace=0.25, wspace=0.2)
# 1. 组合资金曲线 + 回撤
ax_eq = fig.add_subplot(gs[0, :])
port_cum = r["port_cum"]
ax_eq.plot(x, port_cum, linewidth=2.0, color="#1565c0",
label=f"Portfolio (Ann={r['port_ann_ret']*100:+.2f}%, Sharpe={r['port_sharpe']:+.2f}, MDD={r['port_mdd']*100:.2f}%)")
ax_eq.axhline(0, color="gray", linewidth=0.5, linestyle="--")
ax_eq.set_ylabel("Cumulative Return", fontsize=10)
ax_eq.legend(loc="upper left", fontsize=9, framealpha=0.85)
ax_eq.grid(alpha=0.25)
ax_eq.set_title(f"[{group_name}] {', '.join(symbols)} | Formula: {r['readable']}\n"
f"Cost={cost_rate} T={T} bars ({r['years_span']:.2f}y) Score={r.get('best_score',0):.4f}",
fontsize=11, pad=8)
# 2. 回撤
ax_dd = fig.add_subplot(gs[1, :], sharex=ax_eq)
peak = np.maximum.accumulate(port_cum)
dd = port_cum - peak
ax_dd.fill_between(x, dd, 0, alpha=0.4, color="#1565c0")
ax_dd.axhline(0, color="gray", linewidth=0.5)
ax_dd.set_ylabel("Drawdown", fontsize=9)
ax_dd.grid(alpha=0.2)
# 3. 各品种资金曲线
ax_sym = fig.add_subplot(gs[2, 0], sharex=ax_eq)
colors = ["#e65100", "#00897b", "#6a1b9a", "#b71c1c", "#26418f"]
for i, sym in enumerate(symbols):
if sym in r["per_sym"]:
cum = r["per_sym"][sym]["cum"]
c = colors[i % len(colors)]
ax_sym.plot(x, cum, linewidth=1.2, color=c, alpha=0.8,
label=f"{sym} (Ret={r['per_sym'][sym]['total_ret']:+.3f}, WR={r['per_sym'][sym]['win_rate']*100:.0f}%)")
ax_sym.axhline(0, color="gray", linewidth=0.5, linestyle="--")
ax_sym.set_ylabel("Per-Symbol Return", fontsize=9)
ax_sym.legend(loc="upper left", fontsize=7, ncol=2, framealpha=0.85)
ax_sym.grid(alpha=0.2)
# 4. 滚动 Sharpe (250 bar window)
ax_roll = fig.add_subplot(gs[2, 1])
window = 250
if T > window:
roll_sharpe = []
for i in range(window, T):
seg = r["port_pnl"][i-window:i]
m, s = seg.mean(), seg.std()
roll_sharpe.append(m / (s + 1e-10) * math.sqrt(_H1_PER_YEAR))
ax_roll.plot(range(window, T), roll_sharpe, linewidth=0.8, color="#2e7d32", alpha=0.7)
ax_roll.axhline(0, color="gray", linewidth=0.5, linestyle="--")
ax_roll.axhline(1.0, color="orange", linewidth=0.5, linestyle=":", alpha=0.5)
ax_roll.set_ylabel("Rolling Sharpe (250)", fontsize=8)
ax_roll.set_xlabel("Bar Index", fontsize=8)
ax_roll.grid(alpha=0.2)
ax_roll.set_title("Rolling Sharpe Ratio", fontsize=9)
else:
ax_roll.text(0.5, 0.5, "Insufficient data for rolling", ha="center", va="center", transform=ax_roll.transAxes)
# 5. 逐年收益柱状图
ax_yr = fig.add_subplot(gs[3, 0])
yearly = r["yearly_ret"]
if yearly:
yr_colors = ["#2e7d32" if v >= 0 else "#c62828" for v in yearly]
ax_yr.bar(range(1, len(yearly)+1), [v*100 for v in yearly], color=yr_colors, alpha=0.8)
ax_yr.axhline(0, color="gray", linewidth=0.5)
ax_yr.set_xlabel("Year", fontsize=9)
ax_yr.set_ylabel("Return (%)", fontsize=9)
ax_yr.set_title("Yearly Returns", fontsize=9)
ax_yr.grid(alpha=0.2)
# 6. 月度收益热力图
ax_mo = fig.add_subplot(gs[3, 1])
# 按月聚合（假设 H1，每月约 520 bars）
bars_per_month = 520
n_months = T // bars_per_month
if n_months > 1:
monthly = []
for m in range(n_months):
s = m * bars_per_month
e = min((m+1) * bars_per_month, T)
monthly.append(float(r["port_pnl"][s:e].sum()))
ax_mo.bar(range(n_months), [v*100 for v in monthly],
color=["#2e7d32" if v >= 0 else "#c62828" for v in monthly], alpha=0.7)
ax_mo.axhline(0, color="gray", linewidth=0.5)
ax_mo.set_xlabel("Month", fontsize=8)
ax_mo.set_ylabel("Return (%)", fontsize=8)
ax_mo.set_title("Monthly Returns", fontsize=9)
ax_mo.grid(alpha=0.2)
else:
ax_mo.text(0.5, 0.5, "Insufficient data", ha="center", va="center", transform=ax_mo.transAxes)
plt.savefig(Path(output_dir) / f"{group_name}_detailed.png", bbox_inches="tight")
plt.close(fig)
return str(Path(output_dir) / f"{group_name}_detailed.png")
def run_group(group_name, group_cfg, fetcher):
symbols = group_cfg["symbols"]
cost_rate = group_cfg["cost_rate"]
strategy_path = Path(f"strategies/best_{group_name}.json")
if not strategy_path.exists():
print(f"\n [{group_name}] SKIP - no strategy file")
return None
data = json.load(open(strategy_path))
if data.get("vocab_version", "unknown") != VOCAB_VERSION:
print(f"\n [{group_name}] SKIP - vocab mismatch")
return None
formula = data["formula"]
best_score = data.get("best_score", 0.0)
# Load group data
original_symbols = Config.SYMBOLS[:]
Config.SYMBOLS = list(symbols)
mgr = MT5DataManager(fetcher)
mgr.load()
Config.SYMBOLS = original_symbols
raw_dict = mgr.raw_dict
syms_loaded = mgr.symbols
T = raw_dict["open"].shape[1]
feat = MT5FeatureEngineer.compute_features(raw_dict)
target_ret = mgr.target_ret
result = backtest_one(formula, feat, target_ret, syms_loaded, cost_rate)
if result is None:
print(f"\n [{group_name}] ERROR - factor execution failed")
return None
result["best_score"] = best_score
print_detailed_report(group_name, result, syms_loaded, cost_rate, best_score)
plot_path = plot_detailed(result, syms_loaded, group_name, OUTPUT_DIR, cost_rate)
print(f"\n Plot saved -> {plot_path}")
# Save JSON
report = {
"group": group_name,
"formula": formula,
"readable": result["readable"],
"best_score": best_score,
"symbols": syms_loaded,
"T": T,
"years_span": round(T / _H1_PER_YEAR, 2),
"cost_rate": cost_rate,
"port_total_ret": round(result["port_total_ret"], 6),
"port_ann_ret": round(result["port_ann_ret"], 6),
"port_sharpe": round(result["port_sharpe"], 4),
"port_sortino": round(result["port_sortino"], 4),
"port_mdd": round(result["port_mdd"], 6),
"port_calmar": round(result["port_calmar"], 4),
"ic": round(result["ic"], 6),
"avg_turnover": round(result["avg_turnover"], 8),
"max_consec_loss": result["max_consec_loss"],
"port_n_trades": result["port_n_trades"],
"port_n_long": result["port_n_long"],
"port_n_short": result["port_n_short"],
"port_win_rate": round(result["port_win_rate"], 4),
"port_avg_win": round(result["port_avg_win"], 6),
"port_avg_loss": round(result["port_avg_loss"], 6),
"port_profit_factor": round(result["port_profit_factor"], 4),
"port_max_win": round(result["port_max_win"], 6),
"port_max_loss": round(result["port_max_loss"], 6),
"port_avg_trade_pnl": round(result["port_avg_trade_pnl"], 6),
"port_avg_hold_h": round(result["port_avg_hold_h"], 1),
"n_pos_syms": result["n_pos_syms"],
"n_syms": result["n_syms"],
"half1_sharpe": round(result["half1_sharpe"], 4),
"half2_sharpe": round(result["half2_sharpe"], 4),
"half1_ret": round(result["half1_ret"], 6),
"half2_ret": round(result["half2_ret"], 6),
"third1_sharpe": round(result["third1_sharpe"], 4),
"third2_sharpe": round(result["third2_sharpe"], 4),
"third3_sharpe": round(result["third3_sharpe"], 4),
"third1_ret": round(result["third1_ret"], 6),
"third2_ret": round(result["third2_ret"], 6),
"third3_ret": round(result["third3_ret"], 6),
"yearly_ret": [round(v, 6) for v in result["yearly_ret"]],
"per_sym": {
sym: {
"total_ret": round(d["total_ret"], 6),
"ann_ret": round(d["ann_ret"], 6),
"sharpe": round(d["sharpe"], 4),
"sortino": round(d["sortino"], 4),
"mdd": round(d["mdd"], 6),
"ic": round(d["ic"], 6),
"n_trades": d["n_trades"],
"n_long": d["n_long"],
"n_short": d["n_short"],
"win_rate": round(d["win_rate"], 4),
"avg_win": round(d["avg_win"], 6),
"avg_loss": round(d["avg_loss"], 6),
"profit_factor": round(d["profit_factor"], 4),
"max_win": round(d["max_win"], 6),
"max_loss": round(d["max_loss"], 6),
"avg_trade_pnl": round(d["avg_trade_pnl"], 6),
"avg_hold_h": round(d["avg_hold_h"], 1),
}
for sym, d in result["per_sym"].items()
},
}
report_path = Path(OUTPUT_DIR) / f"{group_name}_detailed.json"
with open(report_path, "w") as f:
json.dump(report, f, indent=2, ensure_ascii=False)
print(f" Report saved -> {report_path}")
return result
def main():
offline = "--offline" in sys.argv
# Update metals_comm strategy from latest checkpoint
print("Checking metals_comm latest checkpoint...")
import glob, torch
ckpt_files = sorted(glob.glob(r"D:\cl\MT5_AlphaGPT\checkpoints\ckpt_metals_comm_step_*.pt"))
if ckpt_files:
ckpt = torch.load(ckpt_files[-1], map_location='cpu', weights_only=False)
bf = ckpt.get('best_formula')
bs = ckpt.get('best_score', 0)
step = ckpt.get('step', 0)
if bf and bs > 0:
Path("strategies").mkdir(exist_ok=True)
Path("strategies/best_metals_comm.json").write_text(json.dumps({
"vocab_version": VOCAB_VERSION,
"symbol": "metals_comm",
"formula": bf,
"best_score": bs,
"step": step,
}, indent=2))
print(f" Updated: score={bs:.4f} step={step} formula={bf}")
print(f"\n{'='*80}")
print(f" FULL DETAILED BACKTEST - ALL GROUPS")
print(f"{'='*80}")
all_results = {}
with MT5DataFetcher(offline=offline) as fetcher:
for gname, gcfg in GROUP_CONFIG.items():
r = run_group(gname, gcfg, fetcher)
if r:
all_results[gname] = r
# === Cross-group summary ===
if all_results:
print(f"\n{'='*80}")
print(f" CROSS-GROUP COMPARISON")
print(f"{'='*80}")
print(f" {'Group':<14s} {'Years':>6s} {'Score':>8s} {'AnnRet':>9s} {'Sharpe':>8s} {'Sortino':>8s} "
f"{'MDD':>7s} {'Calmar':>8s} {'IC':>8s} {'Trades':>7s} {'WinR%':>6s} {'PF':>5s} {'Hold':>6s} {'Pos/N':>6s}")
print(f" {'-'*14} {'-'*6} {'-'*8} {'-'*9} {'-'*8} {'-'*8} {'-'*7} {'-'*8} {'-'*8} {'-'*7} {'-'*6} {'-'*5} {'-'*6} {'-'*6}")
for gname, r in all_results.items():
print(f" {gname:<14s} "
f"{r['years_span']:>5.2f}y "
f"{r.get('best_score',0):>8.4f} "
f"{r['port_ann_ret']:>+9.4f} "
f"{r['port_sharpe']:>+8.4f} "
f"{r['port_sortino']:>+8.4f} "
f"{r['port_mdd']:>7.4f} "
f"{r['port_calmar']:>+8.4f} "
f"{r['ic']:>8.5f} "
f"{r['port_n_trades']:>7d} "
f"{r['port_win_rate']*100:>5.1f}% "
f"{r['port_profit_factor']:>5.2f} "
f"{r['port_avg_hold_h']:>5.1f}h "
f"{r['n_pos_syms']:>2d}/{r['n_syms']:<2d}")
# === Combined portfolio ===
print(f"\n {'─'*70}")
print(f" COMBINED PORTFOLIO (equal weight across groups)")
print(f" {'─'*70}")
min_T = min(len(r["port_pnl"]) for r in all_results.values())
combined_pnl = np.zeros(min_T)
for r in all_results.values():
combined_pnl += r["port_pnl"][:min_T]
combined_pnl /= len(all_results)
combined_cum = np.cumsum(combined_pnl)
combined_mdd = calc_mdd(combined_cum)
combined_ann = calc_annual_return(float(combined_cum[-1]), min_T)
combined_sharpe = calc_sharpe(combined_pnl)
combined_sortino = calc_sortino(combined_pnl)
combined_calmar = calc_calmar(float(combined_cum[-1]), combined_mdd, min_T)
# Combined trade stats
combined_pos = np.mean([r["port_pnl"][:min_T] for r in all_results.values()], axis=0)
# Use sign of combined pnl as proxy position
combined_trade = analyze_trades(
np.sign(combined_pnl).reshape(1, -1),
combined_pnl.reshape(1, -1), 0
)[0]
print(f" Data span : {min_T} bars ({min_T/_H1_PER_YEAR:.2f} years)")
print(f" Total Return : {float(combined_cum[-1]):+.4f} ({float(combined_cum[-1])*100:+.2f}%)")
print(f" Annual Return : {combined_ann:+.4f} ({combined_ann*100:+.2f}%)")
print(f" Sharpe Ratio : {combined_sharpe:+.4f}")
print(f" Sortino Ratio : {combined_sortino:+.4f}")
print(f" Max Drawdown : {combined_mdd:.4f} ({combined_mdd*100:.2f}%)")
print(f" Calmar Ratio : {combined_calmar:+.4f}")
print(f" Max Consec Loss : {_max_consecutive_loss(combined_pnl)} bars")
# Combined plot
fig, axes = plt.subplots(3, 1, figsize=(18, 12), dpi=110,
gridspec_kw={'height_ratios': [3, 1, 2], 'hspace': 0.2})
x = np.arange(min_T)
colors_g = {"forex": "#2e7d32", "metals_comm": "#e65100", "index": "#1565c0"}
for gname, r in all_results.items():
axes[0].plot(x, np.cumsum(r["port_pnl"][:min_T]), linewidth=1.2, alpha=0.7,
color=colors_g.get(gname, "gray"),
label=f"{gname} (Ann={r['port_ann_ret']*100:+.1f}%, MDD={r['port_mdd']*100:.1f}%)")
axes[0].plot(x, combined_cum, linewidth=2.5, color="#d32f2f",
label=f"Combined (Ann={combined_ann*100:+.2f}%, Sharpe={combined_sharpe:+.2f}, MDD={combined_mdd*100:.2f}%)")
axes[0].axhline(0, color="gray", linewidth=0.5, linestyle="--")
axes[0].legend(fontsize=9, loc="upper left")
axes[0].set_title(f"All Groups Combined Portfolio ({min_T} bars, {min_T/_H1_PER_YEAR:.2f} years)", fontsize=12)
axes[0].grid(alpha=0.25)
peak = np.maximum.accumulate(combined_cum)
dd = combined_cum - peak
axes[1].fill_between(x, dd, 0, alpha=0.4, color="#d32f2f")
axes[1].axhline(0, color="gray", linewidth=0.5)
axes[1].set_ylabel("Drawdown", fontsize=9)
axes[1].grid(alpha=0.2)
# Yearly returns
years_c = max(1, min_T // _H1_PER_YEAR)
yr_ret = []
for y in range(years_c):
s = y * _H1_PER_YEAR
e = min((y+1) * _H1_PER_YEAR, min_T)
yr_ret.append(float(combined_pnl[s:e].sum()))
yr_colors = ["#2e7d32" if v >= 0 else "#c62828" for v in yr_ret]
axes[2].bar(range(1, len(yr_ret)+1), [v*100 for v in yr_ret], color=yr_colors, alpha=0.8)
axes[2].axhline(0, color="gray", linewidth=0.5)
axes[2].set_xlabel("Year", fontsize=9)
axes[2].set_ylabel("Return (%)", fontsize=9)
axes[2].set_title("Combined Yearly Returns", fontsize=10)
axes[2].grid(alpha=0.2)
plt.savefig(Path(OUTPUT_DIR) / "all_groups_detailed.png", bbox_inches="tight")
plt.close(fig)
print(f"\n Combined plot saved -> {Path(OUTPUT_DIR) / 'all_groups_detailed.png'}")
combined_report = {
"groups": list(all_results.keys()),
"T": min_T,
"years_span": round(min_T / _H1_PER_YEAR, 2),
"total_ret": round(float(combined_cum[-1]), 6),
"ann_ret": round(combined_ann, 6),
"sharpe": round(combined_sharpe, 4),
"sortino": round(combined_sortino, 4),
"mdd": round(combined_mdd, 6),
"calmar": round(combined_calmar, 4),
"max_consec_loss": _max_consecutive_loss(combined_pnl),
"yearly_ret": [round(v, 6) for v in yr_ret],
}
cr_path = Path(OUTPUT_DIR) / "all_groups_detailed.json"
with open(cr_path, "w") as f:
json.dump(combined_report, f, indent=2, ensure_ascii=False)
print(f" Combined report saved -> {cr_path}")
print(f"\n{'='*80}")
print(f" BACKTEST COMPLETE")
print(f"{'='*80}\n")
if __name__ == "__main__":
main()
check_ckpt.py
+12
View File
@@ -0,0 +1,12 @@
import torch, glob, os
files = sorted(glob.glob(r'D:\cl\MT5_AlphaGPT\checkpoints\ckpt_metals_comm_step_*.pt'))
for f in files[-10:]:
ckpt = torch.load(f, map_location='cpu', weights_only=False)
step = ckpt.get('step', 0)
best = ckpt.get('best_score', 0)
restarts = ckpt.get('restarts', 0)
formula = ckpt.get('best_formula', [])
hist = ckpt.get('score_history', [])
last_scores = [round(s, 3) for s in hist[-5:]] if hist else []
print(f'step={step:4d} best={best:.4f} restarts={restarts} formula={formula} recent={last_scores}')
check_ckpt2.py
+18
View File
@@ -0,0 +1,18 @@
import torch, glob
files = sorted(glob.glob(r'D:\cl\MT5_AlphaGPT\checkpoints\ckpt_metals_comm_step_*.pt'))
# Check every 100 steps
for f in files:
name = f.split('\\')[-1]
step_num = int(name.split('step_')[1].split('.')[0])
if step_num % 100 != 0 and step_num != 20:
continue
ckpt = torch.load(f, map_location='cpu', weights_only=False)
step = ckpt.get('step', 0)
best = ckpt.get('best_score', 0)
restarts = ckpt.get('restarts', 0)
formula = ckpt.get('best_formula', [])
# check entropy
entropy = ckpt.get('entropy', None)
stagnation = ckpt.get('stagnation_steps', None)
print(f'step={step:4d} best={best:.4f} restarts={restarts} formula={formula}')
check_ckpt3.py
+22
View File
@@ -0,0 +1,22 @@
import torch, glob
files = sorted(glob.glob(r'D:\cl\MT5_AlphaGPT\checkpoints\ckpt_metals_comm_step_*.pt'))
# Check last 5 checkpoints in detail
for f in files[-5:]:
ckpt = torch.load(f, map_location='cpu', weights_only=False)
step = ckpt.get('step', 0)
best = ckpt.get('best_score', 0)
restarts = ckpt.get('restarts', 0)
formula = ckpt.get('best_formula', [])
# All keys in ckpt
keys = list(ckpt.keys())
# Check for any score/entropy history
score_hist = ckpt.get('score_history', [])
entropy_hist = ckpt.get('entropy_history', [])
last_scores = [round(s, 4) for s in score_hist[-10:]] if score_hist else []
last_entropy = [round(e, 4) for e in entropy_hist[-10:]] if entropy_hist else []
print(f'step={step} best={best:.4f} restarts={restarts}')
print(f' keys: {keys}')
print(f' last_scores: {last_scores}')
print(f' last_entropy: {last_entropy}')
print()
check_ckpt4.py
+11
View File
@@ -0,0 +1,11 @@
import torch, glob
f = sorted(glob.glob(r'D:\cl\MT5_AlphaGPT\checkpoints\ckpt_metals_comm_step_*.pt'))[-1]
ckpt = torch.load(f, map_location='cpu', weights_only=False)
print(f'step={ckpt["step"]} best={ckpt["best_score"]:.4f} restarts={ckpt.get("restart_count",0)}')
th = ckpt.get('training_history', [])
print(f'training_history length: {len(th)}')
if th:
print(f'last 20 entries:')
for entry in th[-20:]:
print(f' {entry}')
check_ckpt5.py
+23
View File
@@ -0,0 +1,23 @@
import torch, glob, json
f = sorted(glob.glob(r'D:\cl\MT5_AlphaGPT\checkpoints\ckpt_metals_comm_step_*.pt'))[-1]
ckpt = torch.load(f, map_location='cpu', weights_only=False)
step = ckpt["step"]
best = ckpt["best_score"]
restarts = ckpt.get("restart_count", 0)
th = ckpt.get('training_history', {})
print(f'step={step} best={best:.4f} restarts={restarts}')
print(f'training_history type: {type(th)}')
if isinstance(th, dict):
# Print all keys and their lengths
for k, v in th.items():
if isinstance(v, list):
print(f' {k}: len={len(v)}, last5={[round(x,4) if isinstance(x,float) else x for x in v[-5:]]}')
else:
print(f' {k}: {v}')
elif isinstance(th, list):
print(f' len={len(th)}')
for entry in th[-10:]:
print(f' {entry}')
check_data_coverage.py
+22
View File
@@ -0,0 +1,22 @@
"""Check data availability per group - how many bars, what time range"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from config import Config
from data_pipeline.data_manager import MT5DataManager
from data_pipeline.fetcher import MT5DataFetcher
groups = {
"forex": ["EURUSD", "USDJPY"],
"metals_comm": ["XAUUSD", "AAVUSD", "COCOA.c"],
"index": ["US30.cash", "US100.cash", "US500.cash", "US2000.cash", "JP225.cash"],
}
with MT5DataFetcher(offline=True) as fetcher:
for gname, symbols in groups.items():
Config.SYMBOLS = symbols
mgr = MT5DataManager(fetcher)
mgr.load()
T = mgr.raw_dict["open"].shape[1]
years = T / 6240
print(f"{gname:14s}: {mgr.symbols} T={T} years={years:.2f}")
check_metals_status.py
+39
View File
@@ -0,0 +1,39 @@
import torch, glob
f = sorted(glob.glob(r'D:\cl\MT5_AlphaGPT\checkpoints\ckpt_metals_comm_step_*.pt'))[-1]
ckpt = torch.load(f, map_location='cpu', weights_only=False)
th = ckpt.get('training_history', {})
step = ckpt['step']
best = ckpt['best_score']
restarts = ckpt.get('restart_count', 0)
print(f'step={step} best={best:.4f} restarts={restarts}')
print()
# best_score progression every 100 steps
bs = th.get('best_score', [])
vs = th.get('val_score', [])
ent = th.get('entropy', [])
top1 = th.get('top1_prob', [])
effv = th.get('eff_vocab', [])
print('=== best_score / val_score / entropy every 100 steps ===')
for i in range(0, len(bs), 100):
v = vs[i] if i < len(vs) else 0
e = ent[i] if i < len(ent) else 0
t = top1[i] if i < len(top1) else 0
ev = effv[i] if i < len(effv) else 0
print(f' step {i:4d}: best={bs[i]:.4f} val={v:.4f} entropy={e:.4f} top1={t:.4f} effV={ev:.4f}')
print(f' step {len(bs)-1:4d}: best={bs[-1]:.4f} val={vs[-1]:.4f} entropy={ent[-1]:.4f} top1={top1[-1]:.4f} effV={effv[-1]:.4f}')
# Stagnation
for i in range(len(bs)-1, -1, -1):
if abs(bs[i] - best) > 0.001:
print(f'\nStagnation: best_score={best:.4f} unchanged since step {i+1} ({len(bs)-1-i} steps ago)')
break
# val_score trend last 200 vs first 200
if len(vs) > 200:
import numpy as np
print(f'\nval_score first 200 avg: {np.mean(vs[:200]):.4f}')
print(f'val_score last 200 avg: {np.mean(vs[-200:]):.4f}')
smoke_test_constraints.py
+41
View File
@@ -0,0 +1,41 @@
"""Smoke test: verify training loop starts with new constraints"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from config import Config
from data_pipeline.data_manager import MT5DataManager
from data_pipeline.fetcher import MT5DataFetcher
from model_core.engine import AlphaEngine
from model_core.config import ModelConfig
Config.SYMBOLS = ["EURUSD", "USDJPY"]
with MT5DataFetcher(offline=True) as fetcher:
mgr = MT5DataManager(fetcher)
mgr.load()
engine = AlphaEngine(data_manager=mgr, target_symbol="forex")
# Run 3 steps to verify the training loop works
print("Starting 3-step smoke test...")
engine.train(start_step=0, end_step=3, verbose_header=True)
print(f"\nSmoke test complete!")
print(f" best_score = {engine.best_score:.4f}")
print(f" best_formula = {engine.best_formula}")
if engine.best_formula:
from model_core.vocab import FORMULA_VOCAB
names = FORMULA_VOCAB.token_names
decoded = ' -> '.join(names[t] for t in engine.best_formula)
print(f" decoded = {decoded}")
# Check infection status
from model_core.vm import validate_formula_structure
violations = validate_formula_structure(engine.best_formula, names)
if violations:
print(f" ⚠ Structure violations: {len(violations)}")
for v in violations:
print(f" - {v}")
else:
print(f" ✅ No structure violations")
strategies/STATUS_20250705.md
+44
View File
@@ -0,0 +1,44 @@
# 策略状态档案 — 2026-07-05 22:50
# 保存于训练框架改进前，作为基线快照
## 策略列表
### 1. forex 组
- **文件**: `best_forex.json`
- **归档**: `archive/best_forex_20250705_pre_refactor.json`
- **公式**: [20, 53, 57, 73, 34, 73, 56, 50]
- **可读**: AROON_OSC_25 -> TS_MIN_10 -> EMA_20 -> CS_SCALE -> SIGN -> CS_SCALE -> EMA_5 -> MOMENTUM_5
- **训练评分**: 0.4851
- **状态**: ✅ 可用 — 作为底仓保留
- **回测**: 8年49998 bars, 年化+2.34%, Sharpe+0.64, MDD 7.39%, 前后一致性良好
- **问题**: 收益偏低，但风控达标
- **是否需要替换**: 否，保留使用
### 2. metals_comm 组
- **文件**: `best_metals_comm.json`
- **归档**: `archive/best_metals_comm_20250705_pre_refactor.json`
- **公式**: [27, 57, 52, 88, 57, 17, 79, 50]
- **可读**: CS_ZSCORE_RET20 -> EMA_20 -> TS_MAX_10 -> TANH_SQUASH -> EMA_20 -> AMIHUD_ILLIQ -> MAX -> MOMENTUM_5
- **训练评分**: 2.3198 (step 920, 熵坍塌停止)
- **状态**: ⚠ 谨慎使用 — 数据仅1.37年，MDD 23.47% 超标
- **回测**: 8546 bars, 年化+117.7%(虚高), Sharpe+3.20, MDD 23.47%
- **问题**: 数据太短；XAUUSD 亏损-26%；AAVUSD/COCOA.c 贡献全部利润
- **是否需要替换**: 是，补充数据后重训替换
### 3. index 组
- **文件**: `best_index.json`
- **归档**: `archive/best_index_20250705_pre_refactor.json`
- **公式**: [13, 47, 75, 47, 77, 40, 86, 82]
- **可读**: PPO -> TS_RANK_10 -> TS_SUM_5 -> TS_RANK_10 -> TS_SUM_20 -> TS_MEAN_5 -> CLIP -> SQRT
- **训练评分**: 0.4464
- **状态**: ❌ 不建议单独使用 — 因子退化成 beta (91.8%做多)
- **回测**: 5年32076 bars, 年化+7.68%, Sharpe+0.48, MDD 29.98%
- **问题**: TS_RANK 连续使用导致符号丢失，因子恒正，等同被动多头
- **是否需要替换**: 是，修复算子链约束后重训替换
## 三组组合回测（等权）
- 数据跨度: 1.37年 (8546 bars, 受限于metals_comm最短)
- 年化: +35.52% (metals_comm短数据放大)
- Sharpe: +2.63
- MDD: 7.93% (通过FTMO)
- 注意: 组合MDD达标主要靠分散化，但index组beta问题和metals数据不足是隐患
strategies/archive/best_forex_20250705_pre_refactor.json
+15
View File
@@ -0,0 +1,15 @@
{
"vocab_version": "v71336c7c7b25",
"symbol": "forex",
"formula": [
20,
53,
57,
73,
34,
73,
56,
50
],
"best_score": 0.4851294159889221
}
strategies/archive/best_index_20250705_pre_refactor.json
+15
View File
@@ -0,0 +1,15 @@
{
"vocab_version": "v71336c7c7b25",
"symbol": "index",
"formula": [
13,
47,
75,
47,
77,
40,
86,
82
],
"best_score": 0.4464336633682251
}
strategies/archive/best_metals_comm_20250705_pre_refactor.json
+15
View File
@@ -0,0 +1,15 @@
{
"vocab_version": "v71336c7c7b25",
"symbol": "metals_comm",
"formula": [
27,
57,
52,
88,
57,
17,
79,
50
],
"best_score": 2.3198370933532715
}
trace_index_factor.py
+98
View File
@@ -0,0 +1,98 @@
"""Deep dive: why index factor is always positive"""
import sys, json, math
import numpy as np
import torch
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from config import Config
from data_pipeline.data_manager import MT5DataManager
from data_pipeline.fetcher import MT5DataFetcher
from model_core.vocab import FORMULA_VOCAB, VOCAB_VERSION
from model_core.vm import StackVM
from model_core.features import MT5FeatureEngineer
from strategy_manager.signal import compute_target_positions_stateless
# Load index strategy
data = json.load(open("strategies/best_index.json"))
formula = data["formula"]
names = FORMULA_VOCAB.token_names
print(f"Formula tokens: {formula}")
print(f"Decode: {' -> '.join(names[t] for t in formula)}")
print()
# Load index group data
Config.SYMBOLS = ["US30.cash", "US100.cash", "US500.cash", "US2000.cash", "JP225.cash"]
with MT5DataFetcher(offline=True) as fetcher:
mgr = MT5DataManager(fetcher)
mgr.load()
raw_dict = mgr.raw_dict
syms = mgr.symbols
T = raw_dict["open"].shape[1]
feat = MT5FeatureEngineer.compute_features(raw_dict)
target_ret = mgr.target_ret
# Execute formula step by step
vm = StackVM()
# Manually trace each step
stack = []
print(f"Feature tensor shape: {feat.shape} (N={feat.shape[0]}, F={feat.shape[1]}, T={feat.shape[2]})")
print()
for i, token in enumerate(formula):
token = int(token)
if token < vm.feat_offset:
fname = names[token] if token < len(names) else f"feat_{token}"
val = feat[:, token, :]
stack.append(val)
print(f" Step {i}: PUSH {fname} (token={token}) shape={val.shape} mean={val.float().mean():.4f} std={val.float().std():.4f} min={val.float().min():.4f} max={val.float().max():.4f}")
elif token in vm.op_map:
oname = names[token] if token < len(names) else f"op_{token}"
arity = vm.arity_map[token]
args = []
for _ in range(arity):
args.append(stack.pop())
args.reverse()
# Stats before
for j, a in enumerate(args):
print(f" Step {i}: arg{j} for {oname}: mean={a.float().mean():.4f} std={a.float().std():.4f} min={a.float().min():.4f} max={a.float().max():.4f} neg%={float((a<0).float().mean())*100:.1f}%")
res = vm.op_map[token](*args)
# Check for nans
nan_pct = float(torch.isnan(res).float().mean()) * 100
print(f" Step {i}: {oname} -> mean={res.float().mean():.4f} std={res.float().std():.4f} min={res.float().min():.4f} max={res.float().max():.4f} neg%={float((res<0).float().mean())*100:.1f}% nan%={nan_pct:.1f}%")
stack.append(res)
else:
print(f" Step {i}: UNKNOWN token {token}")
if len(stack) == 1:
raw_factor = stack[0]
print(f"\n=== RAW FACTOR (before normalize) ===")
print(f" mean={raw_factor.float().mean():.4f} std={raw_factor.float().std():.4f}")
print(f" min={raw_factor.float().min():.4f} max={raw_factor.float().max():.4f}")
print(f" neg%={float((raw_factor<0).float().mean())*100:.1f}% pos%={float((raw_factor>0).float().mean())*100:.1f}%")
normalized = vm._normalize_output(raw_factor)
print(f"\n=== AFTER NORMALIZE ===")
print(f" mean={normalized.float().mean():.4f} std={normalized.float().std():.4f}")
print(f" min={normalized.float().min():.4f} max={normalized.float().max():.4f}")
print(f" neg%={float((normalized<0).float().mean())*100:.1f}% pos%={float((normalized>0).float().mean())*100:.1f}%")
pos = compute_target_positions_stateless(normalized)
print(f"\n=== AFTER tanh (positions) ===")
print(f" mean={pos.float().mean():.4f} std={pos.float().std():.4f}")
print(f" min={pos.float().min():.4f} max={pos.float().max():.4f}")
print(f" long%={float((pos>0.05).float().mean())*100:.1f}% short%={float((pos<-0.05).float().mean())*100:.1f}% flat%={float((pos.abs()<0.05).float().mean())*100:.1f}%")
print(f" avg_abs_pos={float(pos.abs().mean()):.4f}")
# Check per-sym
print(f"\n=== PER-SYMBOL POSITION DISTRIBUTION ===")
for i, sym in enumerate(syms):
p = pos[i]
long_pct = float((p > 0.05).float().mean()) * 100
short_pct = float((p < -0.05).float().mean()) * 100
flat_pct = float((p.abs() < 0.05).float().mean()) * 100
print(f" {sym:<14s}: long={long_pct:.1f}% short={short_pct:.1f}% flat={flat_pct:.1f}% mean={p.float().mean():+.4f}")