127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P2.1: verifikasi bttf_window pada bar FULLDUMP EA (input eksak EA).
|
|
|
|
Untuk pick 2026-03-18 01:00: EA D1=+1 tetapi tf_bias_asof(npz, window ending 03-17)=-1.
|
|
FULLDUMP berisi 250 bar D1 persis seperti cache EA (idx 0 = terbaru).
|
|
Uji: bttf_window pada 200 bar terbaru dump == +1 atau -1?
|
|
Jika -1 -> replica saya buggy ATAU window EA bukan 200 bar ending 03-17.
|
|
"""
|
|
import os
|
|
import csv
|
|
import sys
|
|
import datetime as dt
|
|
import numpy as np
|
|
|
|
DEBUG = r"D:\TradingTerminal\HFM Metatrader 5\Tester\Agent-127.0.0.1-3000\MQL5\Files\AlgoForge_htf_debug_XAUUSD_M15.csv"
|
|
DATA = r"D:\TradingTerminal\HFM Metatrader 5\MQL5\Files\AlgoForge\Data"
|
|
|
|
|
|
def parse(s):
|
|
return int(dt.datetime.strptime(s, "%Y.%m.%d %H:%M")
|
|
.replace(tzinfo=dt.timezone.utc).timestamp())
|
|
|
|
|
|
def bttf_window(h, l, c, need):
|
|
if need < 120:
|
|
return 0
|
|
s = 3
|
|
m = need - 1
|
|
up, dn = float("inf"), -float("inf")
|
|
upb = dnb = -1
|
|
trend = 0
|
|
for i in range(s + 1, m):
|
|
p = i - s
|
|
if p >= s:
|
|
isH = isL = True
|
|
for kk in range(1, s + 1):
|
|
if h[p] <= h[p + kk] or h[p] <= h[p - kk]:
|
|
isH = False
|
|
if l[p] >= l[p + kk] or l[p] >= l[p - kk]:
|
|
isL = False
|
|
if isH:
|
|
up, upb = h[p], p
|
|
if isL:
|
|
dn, dnb = l[p], p
|
|
if upb >= 0 and c[i] > up:
|
|
trend = 1
|
|
up, upb = float("inf"), -1
|
|
if dnb >= 0 and c[i] < dn:
|
|
trend = -1
|
|
dn, dnb = -float("inf"), -1
|
|
return trend
|
|
|
|
|
|
def tf_bias_asof(hh, ll, cc, k):
|
|
need = 200
|
|
s = 3
|
|
start = max(0, k - need + 1)
|
|
h = hh[start:k + 1]
|
|
l = ll[start:k + 1]
|
|
c = cc[start:k + 1]
|
|
return bttf_window(h, l, c, len(h))
|
|
|
|
|
|
def load_npz(name):
|
|
z = np.load(os.path.join(DATA, name + ".npz"))
|
|
return (z["time"].astype(np.int64), z["open"].astype(np.float64),
|
|
z["high"].astype(np.float64), z["low"].astype(np.float64),
|
|
z["close"].astype(np.float64))
|
|
|
|
|
|
def main():
|
|
# ambil FULLDUMP D1 @ 2026-03-18 01:00
|
|
pick = parse("2026.03.18 01:00")
|
|
bars = None
|
|
cur = None
|
|
with open(DEBUG, encoding="utf-8-sig") as f:
|
|
rdr = csv.reader(f, delimiter="\t")
|
|
for r in rdr:
|
|
if not r:
|
|
continue
|
|
if r[0] == "=== FULLDUMP":
|
|
cur = (r[1], parse(r[2]))
|
|
if cur == ("D1", pick):
|
|
bars = []
|
|
continue
|
|
if cur == ("D1", pick) and len(r) == 7 and r[0] == "D1":
|
|
bars.append((int(r[1]), parse(r[2]), float(r[3]), float(r[4]), float(r[5]), float(r[6])))
|
|
elif cur == ("D1", pick) and r[0] != "D1":
|
|
break
|
|
if not bars:
|
|
print("FULLDUMP D1 @ 03-18 01:00 tidak ditemukan")
|
|
return 1
|
|
print(f"D1 dump bars={len(bars)} idx0={dt.datetime.fromtimestamp(bars[0][1], dt.timezone.utc)} "
|
|
f"idx249={dt.datetime.fromtimestamp(bars[-1][1], dt.timezone.utc)}")
|
|
|
|
# EA: window = 200 bar TERBARU dari cache (idx 0..199 di dump), h[0]=tertua
|
|
win = bars[:200][::-1] # balik: 0=tertua
|
|
h = np.array([b[3] for b in win])
|
|
l = np.array([b[4] for b in win])
|
|
c = np.array([b[5] for b in win])
|
|
print(f"window 200 bar: {dt.datetime.fromtimestamp(win[0][1], dt.timezone.utc)} .. "
|
|
f"{dt.datetime.fromtimestamp(win[-1][1], dt.timezone.utc)}")
|
|
print("bttf_window(dump 200 terbaru) =", bttf_window(h, l, c, 200))
|
|
|
|
# juga npz: tf_bias_asof ending 03-17 (E_ea) dan containing-1
|
|
ht, ho, hh, hl, hc = load_npz("XAUUSD_D1")
|
|
t = pick
|
|
tc = t + 900
|
|
cidx = int(np.searchsorted(ht, t, side="right")) - 1
|
|
e_ea = int(np.searchsorted(ht, tc - 86400, side="right")) - 1
|
|
print(f"npz: cidx={cidx} ({dt.datetime.fromtimestamp(int(ht[cidx]), dt.timezone.utc).date()}) "
|
|
f"e_ea={e_ea} ({dt.datetime.fromtimestamp(int(ht[e_ea]), dt.timezone.utc).date()})")
|
|
print("tf_bias_asof(cidx-1) =", tf_bias_asof(hh, hl, hc, cidx - 1))
|
|
print("tf_bias_asof(e_ea) =", tf_bias_asof(hh, hl, hc, e_ea))
|
|
|
|
# bandingkan bar dump dgn npz di 200-window (cek identik)
|
|
diff = 0
|
|
for (iea, tb, o, hh_, ll_, cc_) in win:
|
|
j = int(np.searchsorted(ht, tb))
|
|
if j >= len(ht) or ht[j] != tb or abs(hh_ - hh[j]) > 1e-9 or abs(ll_ - hl[j]) > 1e-9 or abs(cc_ - hc[j]) > 1e-9:
|
|
diff += 1
|
|
print(f"dump-vs-npz diff bars di window: {diff}/200")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|