SniperGold_ML/ml/backtest_eval.py

# -*- 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())