223 lines
8.9 KiB
Python
223 lines
8.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P2.2 EQH/EQL adjudication: rekonstruksi semantik EA/v4.4 DetectEQ di Python.
|
|
|
|
EA/v4.4 DetectEQ:
|
|
- pasangan pivot BERURUTAN dalam list (g_sp[i-1], g_sp[i]), hanya bila
|
|
keduanya bertipe sama (high-high / low-low)
|
|
- tol = EQ_TOL_ATR * ATR SAAT INI (bar row t)
|
|
- jarak bar >= EQ_BARS
|
|
- sweep: bar b in (p2, total] dengan high[b] > p2.price (EQH)
|
|
atau low[b] < p2.price (EQL)
|
|
|
|
Python training (build_features) saat ini:
|
|
- pasangan same-type BERURUTAN (melewati pivot tipe berlawanan)
|
|
- tol = EQ_TOL_ATR * ATR DI BAR PIVOT
|
|
|
|
Uji: rekonstruksi EA vs CSV mode 2 (f10/f11), atas pivot list full history.
|
|
"""
|
|
import os
|
|
import sys
|
|
import csv
|
|
import datetime as dt
|
|
|
|
import numpy as np
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
SRC_TM = os.path.normpath(os.path.join(HERE, "..", "..", "..", "SniperGold_ML"))
|
|
sys.path.insert(0, SRC_TM)
|
|
import train_model as TM # noqa: E402
|
|
|
|
DATA = os.path.normpath(os.path.join(HERE, "..", "..", "..", "..",
|
|
"Files", "AlgoForge", "Data"))
|
|
EA_CSV = os.path.join(HERE, "AlgoForge_bt_features_XAUUSD_M15.csv")
|
|
|
|
|
|
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["open"].astype(np.float64),
|
|
z["high"].astype(np.float64), z["low"].astype(np.float64),
|
|
z["close"].astype(np.float64), z["tick_volume"].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 eq_swept_ea(sp, h, l, A, n, eq_thr=TM.EQ_TOL_ATR, eq_bars=TM.EQ_BARS):
|
|
"""Rekonstruksi DetectEQ EA/v4.4. sp = list pivot kronologis [(p, price, is_high)]."""
|
|
eqh = np.zeros(n, dtype=int)
|
|
eql = np.zeros(n, dtype=int)
|
|
for i in range(1, len(sp)):
|
|
p1, pr1, h1 = sp[i - 1]
|
|
p2, pr2, h2 = sp[i]
|
|
if abs(p2 - p1) < eq_bars:
|
|
continue
|
|
# ATR basis: ATR SAAT INI = A[bar row]. Di sini A adalah series; utk flag
|
|
# kumulatif dipakai A[bar sweep] - nilai tol dihitung pd bar p2 (waktu pair
|
|
# terbentuk) spt EA (g_atr = ATR saat bar diproses). EA menghitung ulang
|
|
# tiap row dgn g_atr terkini; rekonstruksi kumulatif: gunakan A[p2].
|
|
if h1 and h2 and abs(pr2 - pr1) <= eq_thr * A[p2]:
|
|
q = np.where(h[p2 + 1:] > pr2)[0]
|
|
if len(q):
|
|
eqh[p2 + 1 + q[0]:] = 1
|
|
if (not h1) and (not h2) and abs(pr2 - pr1) <= eq_thr * A[p2]:
|
|
q = np.where(l[p2 + 1:] < pr2)[0]
|
|
if len(q):
|
|
eql[p2 + 1 + q[0]:] = 1
|
|
return eqh, eql
|
|
|
|
|
|
def eq_swept_ea_current(sp, h, l, A, n, eq_thr=TM.EQ_TOL_ATR, eq_bars=TM.EQ_BARS):
|
|
"""Varian tol = ATR pada SETIAP bar row (g_atr terkini per row)."""
|
|
eqh = np.zeros(n, dtype=int)
|
|
eql = np.zeros(n, dtype=int)
|
|
for i in range(1, len(sp)):
|
|
p1, pr1, h1 = sp[i - 1]
|
|
p2, pr2, h2 = sp[i]
|
|
if abs(p2 - p1) < eq_bars:
|
|
continue
|
|
if h1 and h2:
|
|
# cari crossing pertama; flag dari crossing s.d. akhir
|
|
q = np.where(h[p2 + 1:] > pr2)[0]
|
|
if len(q):
|
|
c = p2 + 1 + q[0]
|
|
# pasangan memenuhi tol bila |pr2-pr1| <= eq_thr * A[b] utk SEMUA b >= c?
|
|
# EA: tol dihitung dgn g_atr TERKINI (bar row). Flag di bar row r berlaku
|
|
# bila |pr2-pr1| <= eq_thr * A[r] DAN crossing <= r.
|
|
# -> pasangan dgn selisih besar baru "jadi EQH" saat ATR naik.
|
|
# Rekonstruksi: eqh[r] = 1 utk r >= c bila |pr2-pr1| <= eq_thr*A[r].
|
|
eqh[c:] = 1
|
|
# koreksi: baris yg ATR-nya terlalu kecil utk memenuhi tol -> 0
|
|
eqh[c:] = (np.abs(pr2 - pr1) <= eq_thr * A[c:]).astype(int)
|
|
if (not h1) and (not h2):
|
|
q = np.where(l[p2 + 1:] < pr2)[0]
|
|
if len(q):
|
|
c = p2 + 1 + q[0]
|
|
eql[c:] = (np.abs(pr2 - pr1) <= eq_thr * A[c:]).astype(int)
|
|
return eqh, eql
|
|
|
|
|
|
def main():
|
|
log("Load data...")
|
|
t, o, h, l, c, v = load_npz("XAUUSD_M15")
|
|
keep = t >= int(dt.datetime(2017, 1, 1, tzinfo=dt.timezone.utc).timestamp())
|
|
t, o, h, l, c, v = t[keep], o[keep], h[keep], l[keep], c[keep], v[keep]
|
|
n = len(c)
|
|
A = np.maximum(TM.atr_series(h, l, c), 1e-9)
|
|
|
|
log("Build structure (swing) full history...")
|
|
swing_at = np.zeros(n, dtype=int)
|
|
sw = TM.build_structure(o, h, l, c, TM.SWING_LEN, False, swing_at, begin=100)
|
|
sp = sw["pivots"]
|
|
log(f" pivots={len(sp)}")
|
|
|
|
log("Reconstruct EQH/EQL (varian A: tol ATR@pivot; varian B: tol ATR@row)...")
|
|
eqhA, eqlA = eq_swept_ea(sp, h, l, A, n)
|
|
eqhB, eqlB = eq_swept_ea_current(sp, h, l, A, n)
|
|
|
|
# RECON-C: pair list dibatasi window EA (startBar=max(100, r-600); pivot valid
|
|
# bila startBar+len <= p <= r-len), consecutive-list, tol ATR@row.
|
|
# Optimasi: pair consecutive dihitung sekali dari full list (subset window tetap
|
|
# consecutive), first-crossing dihitung sekali; per row hanya cek window+tol.
|
|
log("RECON-C: windowed pair list (600 bar, spt EA)...")
|
|
eqhC = np.zeros(n, dtype=int)
|
|
eqlC = np.zeros(n, dtype=int)
|
|
pairs = [] # (p1, pr1, p2, pr2, is_high, first_cross)
|
|
for k in range(1, len(sp)):
|
|
p1, pr1, h1 = sp[k - 1]
|
|
p2, pr2, h2 = sp[k]
|
|
if abs(p2 - p1) < TM.EQ_BARS:
|
|
continue
|
|
if h1 and h2:
|
|
q = np.where(h[p2 + 1:] > pr2)[0]
|
|
fc = p2 + 1 + q[0] if len(q) else n + 1
|
|
pairs.append((p1, p2, 1, fc, abs(pr2 - pr1)))
|
|
if (not h1) and (not h2):
|
|
q = np.where(l[p2 + 1:] < pr2)[0]
|
|
fc = p2 + 1 + q[0] if len(q) else n + 1
|
|
pairs.append((p1, p2, 0, fc, abs(pr2 - pr1)))
|
|
pairs = np.array(pairs, dtype=np.float64)
|
|
log(f" pairs={len(pairs)}")
|
|
for r in range(2 * TM.SWING_LEN, n):
|
|
# cache M15 EA = 700 bar (maxBars), arrays full cache oldest-first.
|
|
# begin = max(startBar, 2*len); startBar = max(100, total-600) = 100 utk total=700.
|
|
# pivot valid: p_rel in [startBar-len, total-len-1] -> absolut
|
|
# lo = r - total + 1 + (startBar - len), hi = r - len
|
|
total = 700
|
|
start_bar = max(2 * TM.SWING_LEN, total - 600)
|
|
lo = r - total + 1 + (start_bar - TM.SWING_LEN)
|
|
hi = r - TM.SWING_LEN
|
|
if hi <= lo:
|
|
continue
|
|
m = (pairs[:, 0] >= lo) & (pairs[:, 1] <= hi) & (pairs[:, 3] <= r)
|
|
if m.any():
|
|
tol = TM.EQ_TOL_ATR * A[r]
|
|
for row in pairs[m]:
|
|
if row[4] <= tol:
|
|
if row[2] == 1:
|
|
eqhC[r] = 1
|
|
else:
|
|
eqlC[r] = 1
|
|
log(" RECON-C done")
|
|
|
|
# Python training as-is (same-type pairs, ATR@pivot)
|
|
eqh_py, eql_py = np.zeros(n, dtype=int), np.zeros(n, dtype=int)
|
|
lastH = lastL = None
|
|
for (p, pr, is_high) in sp:
|
|
tol = TM.EQ_TOL_ATR * A[p]
|
|
if is_high:
|
|
if lastH is not None and p - lastH[0] >= TM.EQ_BARS and abs(pr - lastH[1]) <= tol:
|
|
if p + 1 < n:
|
|
q = np.where(h[p + 1:] > pr)[0]
|
|
if len(q):
|
|
eqh_py[p + 1 + q[0]:] = 1
|
|
lastH = (p, pr)
|
|
else:
|
|
if lastL is not None and p - lastL[0] >= TM.EQ_BARS and abs(pr - lastL[1]) <= tol:
|
|
if p + 1 < n:
|
|
q = np.where(l[p + 1:] < pr)[0]
|
|
if len(q):
|
|
eql_py[p + 1 + q[0]:] = 1
|
|
lastL = (p, pr)
|
|
|
|
log("Load EA CSV + join...")
|
|
py_idx = {int(tt): i for i, tt in enumerate(t)}
|
|
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:
|
|
tt = parse_ea_time(r[0].strip())
|
|
fea = [float(x) for x in r[3:22]]
|
|
atr_ea = float(r[2])
|
|
except ValueError:
|
|
continue
|
|
if tt in py_idx:
|
|
rows.append((py_idx[tt], fea, atr_ea))
|
|
log(f" joined={len(rows)}")
|
|
|
|
for name, eqh, eql in (("PY-training", eqh_py, eql_py),
|
|
("RECON-A (ATR@pivot)", eqhA, eqlA),
|
|
("RECON-B (ATR@row)", eqhB, eqlB),
|
|
("RECON-C (window+ATR@row)", eqhC, eqlC)):
|
|
m10 = sum(1 for bi, fea, _ in rows if eqh[bi] != int(fea[10]))
|
|
m11 = sum(1 for bi, fea, _ in rows if eql[bi] != int(fea[11]))
|
|
log(f" {name:22s} f10_eqh mismatch={m10}/{len(rows)} ({m10/len(rows):.4f}) "
|
|
f"f11_eql mismatch={m11}/{len(rows)} ({m11/len(rows):.4f})")
|
|
|
|
# sanity ATR vs EA
|
|
d_atr = [abs(A[bi] - atr) for bi, _, atr in rows]
|
|
log(f" sanity ATR py vs EA: max|d|={max(d_atr):.5f} mean|d|={sum(d_atr)/len(d_atr):.5f}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|