110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P2.5: reproduksi HTF bias EA FIXED — E1 (last closed at tc) vs E2 (cache-aware).
|
|
|
|
E2: cache Engine1 hanya di-rebuild saat Bars(HTF) berubah (bar baru muncul).
|
|
newest closed dlm cache pd tc:
|
|
E = bar terbaru dgn time <= tc
|
|
jika E.time + period <= tc (E tertutup):
|
|
jika bar setelah E sudah muncul (ht[E+1] <= tc) -> newest = E
|
|
else (gap setelah E) -> newest = E-1
|
|
jika E forming (E.time+period > tc) -> newest = E-1
|
|
"""
|
|
import os
|
|
import sys
|
|
import csv
|
|
import datetime as dt
|
|
import numpy as np
|
|
|
|
sys.path.insert(0, r"D:\TradingTerminal\HFM Metatrader 5\MQL5\Shared Projects\SniperGold_ML")
|
|
import train_model as TM
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
DATA = os.path.normpath(os.path.join(HERE, "..", "..", "..", "..",
|
|
"Files", "AlgoForge", "Data"))
|
|
EA = os.path.join(HERE, "AlgoForge_bt_features_fixed_XAUUSD_M15.csv")
|
|
PERIOD = {"D1": 86400, "H4": 14400, "H1": 3600}
|
|
|
|
|
|
def log(msg):
|
|
print(msg, flush=True)
|
|
|
|
|
|
def load_npz(name):
|
|
z = np.load(os.path.join(DATA, name + ".npz"))
|
|
return (z["time"].astype(np.int64), z["high"].astype(np.float64),
|
|
z["low"].astype(np.float64), z["close"].astype(np.float64))
|
|
|
|
|
|
def parse_ea_time(s):
|
|
return int(dt.datetime.strptime(s, "%Y.%m.%d %H:%M")
|
|
.replace(tzinfo=dt.timezone.utc).timestamp())
|
|
|
|
|
|
def bias_series(hh, hl, hc):
|
|
n = len(hc)
|
|
out = np.zeros(n, dtype=int)
|
|
for k in range(n):
|
|
out[k] = TM.tf_bias_asof(hh, hl, hc, k)
|
|
return out
|
|
|
|
|
|
def main():
|
|
htf = {}
|
|
for key in ("D1", "H4", "H1"):
|
|
ht_, hh_, hl_, hc_ = load_npz("XAUUSD_" + key)
|
|
htf[key] = (ht_, hh_, hl_, hc_)
|
|
B = {k: bias_series(htf[k][1], htf[k][2], htf[k][3]) for k in ("D1", "H4", "H1")}
|
|
|
|
rows = []
|
|
with open(EA, encoding="utf-8-sig") as f:
|
|
rdr = csv.reader(f, delimiter="\t")
|
|
next(rdr, None)
|
|
for r in rdr:
|
|
if len(r) < 22:
|
|
continue
|
|
try:
|
|
tt = parse_ea_time(r[0].strip())
|
|
fea = [float(x) for x in r[3:22]]
|
|
except ValueError:
|
|
continue
|
|
rows.append((tt, fea))
|
|
log(f"EA rows={len(rows)}")
|
|
|
|
mis = {"E1": {k: 0 for k in ("D1", "H4", "H1")},
|
|
"E2": {k: 0 for k in ("D1", "H4", "H1")}}
|
|
tot = {k: 0 for k in ("D1", "H4", "H1")}
|
|
for (tt, fea) in rows:
|
|
tc = tt + 900
|
|
for fi, key in enumerate(("D1", "H4", "H1")):
|
|
ht_, hh_, hl_, hc_ = htf[key]
|
|
per = PERIOD[key]
|
|
ea_val = int(fea[fi])
|
|
tot[key] += 1
|
|
e1 = int(np.searchsorted(ht_, tc - per, side="right")) - 1
|
|
e2 = e1
|
|
E = int(np.searchsorted(ht_, tc, side="right")) - 1
|
|
if 0 <= E < len(ht_):
|
|
if ht_[E] + per <= tc: # E tertutup
|
|
if E + 1 < len(ht_) and ht_[E + 1] <= tc:
|
|
e2 = E
|
|
else:
|
|
e2 = E - 1
|
|
else: # E forming
|
|
e2 = E - 1
|
|
v1 = int(B[key][e1]) if 0 <= e1 < len(hc_) else 0
|
|
v2 = int(B[key][e2]) if 0 <= e2 < len(hc_) else 0
|
|
if v1 != ea_val:
|
|
mis["E1"][key] += 1
|
|
if v2 != ea_val:
|
|
mis["E2"][key] += 1
|
|
log("")
|
|
log("=== MISMATCH EA fixed vs tf_bias_asof(E) ===")
|
|
for hyp in ("E1", "E2"):
|
|
log(f" {hyp}: D1={mis[hyp]['D1']}/{tot['D1']} ({mis[hyp]['D1']/tot['D1']:.4f}) "
|
|
f"H4={mis[hyp]['H4']}/{tot['H4']} ({mis[hyp]['H4']/tot['H4']:.4f}) "
|
|
f"H1={mis[hyp]['H1']}/{tot['H1']} ({mis[hyp]['H1']/tot['H1']:.4f})")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|