149 lines
4.7 KiB
Python
149 lines
4.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P2.1 VERIFIKASI ROOT CAUSE: EA BTTFBias = window 200 bar TERTUA dari cache 250 bar.
|
|
|
|
Hipotesis (terbukti dari FULLDUMP + isolate):
|
|
EA: GetBar(slot, count-1-i) utk i=0..need-1 (count=250, need=200)
|
|
-> window = bars[249..50] = 200 bar TERTUA cache = tf_bias_asof(E_ea - 50)
|
|
di mana E_ea = bar HTF tertutup terakhir pd waktu komputasi tc = t+900.
|
|
|
|
Uji: EA_hat = tf_bias_asof(E_ea - 50) vs f0/f1/f2 CSV (mode 2) utk SEMUA row.
|
|
"""
|
|
import os
|
|
import sys
|
|
import csv
|
|
import datetime as dt
|
|
|
|
import numpy as np
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
DATA = os.path.normpath(os.path.join(HERE, "..", "..", "..", "..",
|
|
"Files", "AlgoForge", "Data"))
|
|
EA_CSV = os.path.join(HERE, "AlgoForge_bt_features_XAUUSD_M15.csv")
|
|
PERIOD = {"D1": 86400, "H4": 14400, "H1": 3600}
|
|
CACHE = 250
|
|
NEED = 200
|
|
LAG = CACHE - NEED # 50
|
|
|
|
|
|
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 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 bias_series(hh, hl, hc, lag):
|
|
"""bias[k] = bttf window 200 bar ending at k-lag (>= series start)."""
|
|
n = len(hc)
|
|
out = np.zeros(n, dtype=int)
|
|
for k in range(n):
|
|
end = k - lag
|
|
if end < 0:
|
|
continue
|
|
start = max(0, end - NEED + 1)
|
|
out[k] = bttf_window(hh[start:end + 1], hl[start:end + 1],
|
|
hc[start:end + 1], end - start + 1)
|
|
return out
|
|
|
|
|
|
def parse_ea_time(s):
|
|
return int(dt.datetime.strptime(s, "%Y.%m.%d %H:%M")
|
|
.replace(tzinfo=dt.timezone.utc).timestamp())
|
|
|
|
|
|
def main():
|
|
log("Load data...")
|
|
htf = {}
|
|
for key in ("D1", "H4", "H1"):
|
|
ht, hh, hl, hc = load_npz("XAUUSD_" + key)
|
|
m = ht >= int(dt.datetime(2024, 1, 1, tzinfo=dt.timezone.utc).timestamp())
|
|
htf[key] = (ht[m], hh[m], hl[m], hc[m])
|
|
log(f" {key}: n={int(m.sum())}")
|
|
|
|
log("Load EA CSV (mode 2)...")
|
|
rows = []
|
|
with open(EA_CSV, 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:
|
|
t = parse_ea_time(r[0].strip())
|
|
feats = [float(x) for x in r[3:22]]
|
|
except ValueError:
|
|
continue
|
|
rows.append((t, feats))
|
|
log(f" rows={len(rows)}")
|
|
|
|
log("Precompute bias (lag=50) per HTF...")
|
|
B = {}
|
|
for key in ("D1", "H4", "H1"):
|
|
ht, hh, hl, hc = htf[key]
|
|
B[key] = bias_series(hh, hl, hc, LAG)
|
|
log(f" {key} done")
|
|
|
|
log("Evaluate EA_hat = tf_bias_asof(E_ea - 50) vs EA...")
|
|
mis = {k: 0 for k in ("D1", "H4", "H1")}
|
|
tot = {k: 0 for k in ("D1", "H4", "H1")}
|
|
first_mis = {k: 0 for k in ("D1", "H4", "H1")}
|
|
for ri, (t, feats) in enumerate(rows):
|
|
tc = t + 900
|
|
for fi, key in enumerate(("D1", "H4", "H1")):
|
|
ht, hh, hl, hc = htf[key]
|
|
ea_val = int(feats[fi])
|
|
per = PERIOD[key]
|
|
e_ea = int(np.searchsorted(ht, tc - per, side="right")) - 1
|
|
if 0 <= e_ea < len(hc):
|
|
v = int(B[key][e_ea])
|
|
else:
|
|
v = 0
|
|
tot[key] += 1
|
|
if v != ea_val:
|
|
mis[key] += 1
|
|
if ri == 0:
|
|
first_mis[key] += 1
|
|
print(f" MISMATCH row={ri} t={r[0] if False else dt.datetime.fromtimestamp(t, dt.timezone.utc)} "
|
|
f"key={key} EA={ea_val} hat={v} e_ea={dt.datetime.fromtimestamp(int(ht[e_ea]), dt.timezone.utc)}")
|
|
log("")
|
|
log("=== MISMATCH EA_hat(lag=50) vs EA CSV ===")
|
|
for key in ("D1", "H4", "H1"):
|
|
log(f" {key}: mismatch={mis[key]}/{tot[key]} rate={mis[key]/tot[key]:.4f} (first-row-mis={first_mis[key]})")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|