138 lines
4.7 KiB
Python
138 lines
4.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Algo Forge - Tahap 5: evaluasi CSV prob baseline (parity/konsistensi runtime-training).
|
|
|
|
Input : MQL5\\Files\\AlgoForge_bt_prob_XAUUSD_M15.csv
|
|
(dihasilkan AlgoForge_Backtest_Baseline.mq5 mode 0; kolom:
|
|
time;close;atr;pLong;pShort -- delimiter MQL5 FILE_CSV = TAB)
|
|
Output: AUC long/short (label 24 bar x 0.75 ATR, identik train_model.py),
|
|
presisi @0.60, kalibrasi (mean prob vs frekuensi), jumlah sampel.
|
|
|
|
Catatan jujur:
|
|
- Rentang = window backtest (feed XAUUSD 2-digit, bukan XAUUSDc training).
|
|
- Bukan klaim out-of-sample murni: verifikasi konsistensi runtime-training
|
|
+ reproduksi metrik pada feed berbeda.
|
|
- Bandingkan dgn baseline freeze: LONG 0.6270 / SHORT 0.6207.
|
|
"""
|
|
import csv
|
|
import math
|
|
import sys
|
|
|
|
PATH = sys.argv[1] if len(sys.argv) > 1 else r"D:\TradingTerminal\HFM Metatrader 5\MQL5\Files\AlgoForge_bt_prob_XAUUSD_M15.csv"
|
|
HORIZON = 24 # label forward 24 bar
|
|
THR = 0.75 # ambang 0.75 x ATR
|
|
T60 = 0.60 # ambang sinyal utk presisi
|
|
|
|
|
|
def auc_rank(labels, scores):
|
|
"""AUC sederhana (Mann-Whitney U) tanpa sklearn."""
|
|
pos = [s for s, y in zip(scores, labels) if y == 1]
|
|
neg = [s for s, y in zip(scores, labels) if y == 0]
|
|
n_pos, n_neg = len(pos), len(neg)
|
|
if n_pos == 0 or n_neg == 0:
|
|
return float("nan")
|
|
merged = sorted([(s, 1) for s in pos] + [(s, 0) for s in neg], key=lambda x: x[0])
|
|
rank_sum = 0.0
|
|
for i, (_, y) in enumerate(merged, start=1):
|
|
if y == 1:
|
|
rank_sum += i
|
|
# tie correction sederhana (rata-rata rank per grup nilai sama)
|
|
i = 0
|
|
while i < len(merged):
|
|
j = i
|
|
while j < len(merged) and merged[j][0] == merged[i][0]:
|
|
j += 1
|
|
if j - i > 1:
|
|
avg = (i + 1 + j) / 2.0
|
|
cnt_pos = sum(1 for k in range(i, j) if merged[k][1] == 1)
|
|
rank_sum += cnt_pos * (avg - (i + 1))
|
|
i = j
|
|
u = rank_sum - n_pos * (n_pos + 1) / 2.0
|
|
return u / (n_pos * n_neg)
|
|
|
|
|
|
def precision_at(labels, scores, thr):
|
|
"""Presisi pada ambang: P(y=1 | score >= thr)."""
|
|
sel = [(y, s) for y, s in zip(labels, scores) if s >= thr]
|
|
if not sel:
|
|
return float("nan"), 0
|
|
return sum(y for y, _ in sel) / len(sel), len(sel)
|
|
|
|
|
|
def main():
|
|
rows = []
|
|
with open(PATH, "r", encoding="utf-8-sig") as f:
|
|
rdr = csv.reader(f, delimiter="\t")
|
|
header = next(rdr, None)
|
|
if header is None:
|
|
print("EMPTY FILE", PATH)
|
|
return 1
|
|
for r in rdr:
|
|
if len(r) < 5:
|
|
continue
|
|
try:
|
|
t = r[0].strip()
|
|
close = float(r[1])
|
|
atr = float(r[2])
|
|
pl = float(r[3])
|
|
ps = float(r[4])
|
|
except ValueError:
|
|
continue
|
|
rows.append((t, close, atr, pl, ps))
|
|
|
|
n = len(rows)
|
|
if n < HORIZON + 50:
|
|
print(f"TOO FEW ROWS: {n}")
|
|
return 1
|
|
|
|
close = [r[1] for r in rows]
|
|
atr = [r[2] for r in rows]
|
|
pl = [r[3] for r in rows]
|
|
ps = [r[4] for r in rows]
|
|
|
|
y_long, y_short, s_long, s_short = [], [], [], []
|
|
for i in range(n - HORIZON):
|
|
thr = THR * atr[i]
|
|
fwd = close[i + HORIZON] - close[i]
|
|
if fwd >= thr:
|
|
y_long.append(1)
|
|
elif fwd <= -thr:
|
|
y_long.append(0)
|
|
else:
|
|
continue # label 0 dibuang (identik train_model.py)
|
|
s_long.append(pl[i])
|
|
|
|
if fwd <= -thr:
|
|
y_short.append(1)
|
|
elif fwd >= thr:
|
|
y_short.append(0)
|
|
else:
|
|
continue
|
|
s_short.append(ps[i])
|
|
|
|
a_long = auc_rank(y_long, s_long)
|
|
a_short = auc_rank(y_short, s_short)
|
|
p_long, n_long = precision_at(y_long, s_long, T60)
|
|
p_short, n_short = precision_at(y_short, s_short, T60)
|
|
|
|
# kalibrasi sederhana: mean prob vs frekuensi (bin 0.50-0.70)
|
|
bins = [(0.50, 0.55), (0.55, 0.60), (0.60, 0.65), (0.65, 1.01)]
|
|
print("=== BACKTEST BASELINE CSV - EVAL (runtime-training parity) ===")
|
|
print(f"rows={n} berlabel(long)={len(y_long)} berlabel(short)={len(y_short)}")
|
|
print(f"AUC LONG = {a_long:.4f} (baseline freeze 0.6270)")
|
|
print(f"AUC SHORT = {a_short:.4f} (baseline freeze 0.6207)")
|
|
print(f"Presisi LONG @{T60:.2f} = {p_long:.4f} (n={n_long})")
|
|
print(f"Presisi SHORT @{T60:.2f} = {p_short:.4f} (n={n_short})")
|
|
print("--- kalibrasi LONG (mean prob vs frekuensi) ---")
|
|
for lo, hi in bins:
|
|
sel = [(y, s) for y, s in zip(y_long, s_long) if lo <= s < hi]
|
|
if sel:
|
|
mp = sum(s for _, s in sel) / len(sel)
|
|
fr = sum(y for y, _ in sel) / len(sel)
|
|
print(f" prob[{lo:.2f},{hi:.2f}): mean={mp:.4f} freq={fr:.4f} n={len(sel)}")
|
|
print("=== END ===")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|