224 行
11 KiB
Python
224 行
11 KiB
Python
 | """
| ||
Pattern discovery from raw bars, both directions, no fixed windows (2026-10-03).
| |||
| |||
PRE-REGISTERED (written before any result was seen):
| |||
* Universe: EURUSD GBPUSD USDJPY AUDUSD USDCAD XAUUSD XAGUSD (broker M1), SP500 UK100 (Dukascopy M1,
| |||
mid prices). H1 bars. Everything dated >= 2025-01-01 is SEALED: never loaded into any model or table.
| |||
* Events: causal CUSUM on log close, threshold = 2 x ATR100/close (event-driven sampling, AFML ch.2).
| |||
* Features: ONLY information up to the close of bar t, scale-free (in ATR units), computed at SIX
| |||
log-spaced look-backs (8..256) so the model chooses its own scale; soft/continuous - sweeps, reclaim,
| |||
range position/compression, efficiency, effort-vs-result, wicks. No true/false rules.
| |||
* Labels: entry at the OPEN of t+1. Long and short are labelled SEPARATELY from the same event:
| |||
barrier +TP*ATR / -SL*ATR, vertical T bars, both-touched-in-one-bar = stop first. Net of cost.
| |||
Trial A = (TP 2, SL 1.5, T 48). Trial B = (4, 3, 96). Two trials, both reported.
| |||
* Model: gradient boosting regressor on net outcome, POOLED across instruments with no instrument id
| |||
(a pattern must generalise), walk-forward by year 2010..2024, expanding train, 96-bar purge.
| |||
Trades = top decile of OOS predictions per (instrument, year), thinned to non-overlapping.
| |||
* Control: the unconditional thinned mean of the same direction on the same events (random entry).
| |||
* Cost (full spread, bp of price, deliberately pessimistic): see COST_BP. Real-tick test comes later.
| |||
"""
| |||
from __future__ import annotations
| |||
import os, sys, time
| |||
import numpy as np
| |||
import pandas as pd
| |||
| |||
sys.path.insert(0, os.path.dirname(__file__))
| |||
SQXDEC = os.environ.get("SQXDEC", r"C:\Users\admin\AppData\Local\Temp\2\claude\c--Users-admin-Documents-Workspaces-Warrior-EA"
| |||
r"\5debf25a-7b4e-42ef-8752-8db935abfa95\scratchpad\sqxdec")
| |||
SEAL = pd.Timestamp("2025-01-01")
| |||
LOOKBACKS = (8, 16, 32, 64, 128, 256)
| |||
COST_BP = {"EURUSD": 1.0, "GBPUSD": 1.5, "USDJPY": 1.0, "AUDUSD": 1.5, "USDCAD": 1.5, "XAUUSD": 3.0,
| |||
"XAGUSD": 8.0, "SP500": 1.5, "UK100": 2.0}
| |||
BROKER = ["EURUSD", "GBPUSD", "USDJPY", "AUDUSD", "USDCAD", "XAUUSD", "XAGUSD"]
| |||
DUKA = {"SP500": "USA500IDXUSD_dukascopy_the5ers", "UK100": "GBRIDXGBP_dukascopy_the5ers"}
| |||
START = {"EURUSD": 2000, "GBPUSD": 2000, "USDJPY": 2000, "AUDUSD": 2000, "USDCAD": 2000, "XAUUSD": 2005,
| |||
"XAGUSD": 2010, "SP500": 2014, "UK100": 2014}
| |||
TRIALS = {"A": (2.0, 1.5, 48), "B": (4.0, 3.0, 96)}
| |||
CACHE = os.path.join(os.path.dirname(SQXDEC), "disc_cache")
| |||
| |||
| |||
def h1_from_m1(df: pd.DataFrame) -> pd.DataFrame:
| |||
r = df.resample("60min")
| |||
out = pd.DataFrame({"o": r.o.first(), "h": r.h.max(), "l": r.l.min(), "c": r.c.last(), "tv": r.tv.sum(),
| |||
"n": r.o.count()})
| |||
return out[out.n >= 20].drop(columns="n")
| |||
| |||
| |||
def load_h1(sym: str) -> pd.DataFrame:
| |||
os.makedirs(CACHE, exist_ok=True)
| |||
p = os.path.join(CACHE, sym + "_h1.pkl")
| |||
if os.path.exists(p):
| |||
return pd.read_pickle(p)
| |||
if sym in BROKER:
| |||
import moves_catalogue as mc
| |||
m1 = mc.load(sym)
| |||
h1 = h1_from_m1(m1)
| |||
else:
| |||
sys.path.insert(0, SQXDEC)
| |||
import sqxbars
| |||
bars, _ = sqxbars.load(DUKA[sym], "M1", verbose=False)
| |||
m1 = pd.DataFrame({"o": bars[:, 1], "h": bars[:, 2], "l": bars[:, 3], "c": bars[:, 4], "tv": bars[:, 5]},
| |||
index=pd.to_datetime(bars[:, 0], unit="ms"))
| |||
m1 = m1[~m1.index.duplicated()]
| |||
h1 = h1_from_m1(m1)
| |||
h1 = h1[h1.index.year >= START[sym]]
| |||
h1.to_pickle(p)
| |||
return h1
| |||
| |||
| |||
def cusum_events(c: np.ndarray, thr: np.ndarray) -> np.ndarray:
| |||
lc = np.log(c)
| |||
d = np.diff(lc, prepend=lc[0])
| |||
sp = sn = 0.0
| |||
ev = []
| |||
for i in range(1, len(c)):
| |||
if not np.isfinite(thr[i]):
| |||
continue
| |||
sp = max(0.0, sp + d[i]); sn = min(0.0, sn + d[i])
| |||
if sp > thr[i] or sn < -thr[i]:
| |||
ev.append(i); sp = sn = 0.0
| |||
return np.array(ev, dtype=int)
| |||
| |||
| |||
def features(b: pd.DataFrame) -> tuple[pd.DataFrame, np.ndarray]:
| |||
o, h, l, c, tv = b.o, b.h, b.l, b.c, b.tv.astype(float).replace(0, np.nan)
| |||
pc = c.shift(1)
| |||
tr = np.maximum(h - l, np.maximum((h - pc).abs(), (l - pc).abs()))
| |||
atr = tr.rolling(100, min_periods=50).mean()
| |||
atr20 = tr.rolling(20, min_periods=10).mean()
| |||
F = {}
| |||
F["volratio"] = atr20 / atr
| |||
F["volpct"] = atr.rolling(500, min_periods=100).rank(pct=True)
| |||
rng = (h - l).replace(0, np.nan)
| |||
F["upwick"] = (h - np.maximum(o, c)) / rng
| |||
F["lowick"] = (np.minimum(o, c) - l) / rng
| |||
F["body"] = (c - o) / rng
| |||
F["closeloc"] = (c - l) / rng
| |||
hr = b.index.hour + b.index.minute / 60
| |||
F["hr_s"] = pd.Series(np.sin(2 * np.pi * hr / 24), index=b.index)
| |||
F["hr_c"] = pd.Series(np.cos(2 * np.pi * hr / 24), index=b.index)
| |||
F["dow"] = pd.Series(b.index.dayofweek.astype(float), index=b.index)
| |||
k = 8
| |||
for L in LOOKBACKS:
| |||
hh = h.rolling(L).max(); ll = l.rolling(L).min()
| |||
F[f"ret{L}"] = (c - c.shift(L)) / atr
| |||
w = (hh - ll)
| |||
F[f"pos{L}"] = (c - ll) / w.replace(0, np.nan)
| |||
F[f"width{L}"] = w / atr
| |||
path = c.diff().abs().rolling(L).sum()
| |||
F[f"eff{L}"] = (c - c.shift(L)).abs() / path.replace(0, np.nan)
| |||
if L > k:
| |||
ph = h.shift(k).rolling(L - k).max(); pl = l.shift(k).rolling(L - k).min() # prior range, excl. last k bars
| |||
F[f"sweephi{L}"] = ((h.rolling(k).max() - ph) / atr).clip(lower=0)
| |||
F[f"sweeplo{L}"] = ((pl - l.rolling(k).min()) / atr).clip(lower=0)
| |||
F[f"reclaimhi{L}"] = (c - ph) / atr
| |||
F[f"reclaimlo{L}"] = (c - pl) / atr
| |||
if tv.notna().mean() > 0.5:
| |||
vr = tv.rolling(k).mean() / tv.rolling(L).mean()
| |||
rr = (h - l).rolling(k).mean() / (h - l).rolling(L).mean()
| |||
F[f"vol{L}"] = vr
| |||
F[f"effort{L}"] = vr / rr.replace(0, np.nan)
| |||
else:
| |||
F[f"vol{L}"] = pd.Series(np.nan, index=b.index); F[f"effort{L}"] = F[f"vol{L}"]
| |||
X = pd.DataFrame(F)
| |||
return X, atr.to_numpy()
| |||
| |||
| |||
def label(b: pd.DataFrame, ev: np.ndarray, atr: np.ndarray, cost_bp: float, tp: float, sl: float, T: int):
| |||
"""Net outcome in ATR units, long and short, entry at open t+1. Also exit offset for thinning."""
| |||
o, h, l, c = (b[x].to_numpy() for x in ("o", "h", "l", "c"))
| |||
n = len(b)
| |||
ev = ev[ev + T + 2 < n]
| |||
out = np.full((len(ev), 5), np.nan)
| |||
for j, i in enumerate(ev):
| |||
e = o[i + 1]; s = atr[i]
| |||
if not np.isfinite(s) or s <= 0:
| |||
continue
| |||
hh, ll = h[i + 1:i + 1 + T], l[i + 1:i + 1 + T]
| |||
cst = cost_bp * 1e-4 * e / s
| |||
# long
| |||
tpi = np.argmax(hh >= e + tp * s) if (hh >= e + tp * s).any() else T
| |||
sli = np.argmax(ll <= e - sl * s) if (ll <= e - sl * s).any() else T
| |||
if sli <= tpi and sli < T: rl, xl = -sl, sli
| |||
elif tpi < T: rl, xl = tp, tpi
| |||
else: rl, xl = (c[i + T] - e) / s, T - 1
| |||
# short
| |||
tpi = np.argmax(ll <= e - tp * s) if (ll <= e - tp * s).any() else T
| |||
sli = np.argmax(hh >= e + sl * s) if (hh >= e + sl * s).any() else T
| |||
if sli <= tpi and sli < T: rs, xs = -sl, sli
| |||
elif tpi < T: rs, xs = tp, tpi
| |||
else: rs, xs = (e - c[i + T]) / s, T - 1
| |||
out[j] = (rl - cst, rs - cst, xl, xs, cst)
| |||
return ev, out
| |||
| |||
| |||
def build(sym: str) -> pd.DataFrame:
| |||
p = os.path.join(CACHE, f"{sym}_events.pkl")
| |||
if os.path.exists(p):
| |||
return pd.read_pickle(p)
| |||
b = load_h1(sym)
| |||
b = b[b.index < SEAL] # SEALED: nothing later is ever read
| |||
X, atr = features(b)
| |||
ev = cusum_events(b.c.to_numpy(), 2.0 * atr / b.c.to_numpy())
| |||
parts = []
| |||
for name, (tp, sl, T) in TRIALS.items():
| |||
e2, y = label(b, ev, atr, COST_BP[sym], tp, sl, T)
| |||
d = X.iloc[e2].copy()
| |||
d["sym"] = sym; d["t"] = b.index[e2]; d["trial"] = name
| |||
d["yL"], d["yS"], d["xL"], d["xS"], d["cst"] = y.T
| |||
parts.append(d)
| |||
df = pd.concat(parts)
| |||
df["year"] = df.t.dt.year
| |||
df.to_pickle(p)
| |||
return df
| |||
| |||
| |||
def thin(t: np.ndarray, x: np.ndarray, mask: np.ndarray) -> np.ndarray:
| |||
"""Greedy non-overlap: indices (into arrays) of kept trades, in time order."""
| |||
keep, busy = [], -1
| |||
for j in np.flatnonzero(mask):
| |||
if j > busy:
| |||
keep.append(j); busy = j + int(x[j]) + 1
| |||
return np.array(keep, dtype=int)
| |||
| |||
| |||
def main():
| |||
from sklearn.ensemble import HistGradientBoostingRegressor
| |||
t0 = time.time()
| |||
syms = sys.argv[1:] or list(COST_BP)
| |||
all_ev = pd.concat([build(s) for s in syms])
| |||
print(f"events built {time.time() - t0:.0f}s: " + ", ".join(f"{s}:{(all_ev.sym == s).sum() // 2}" for s in syms))
| |||
feats = [c for c in all_ev.columns if c not in ("sym", "t", "trial", "yL", "yS", "xL", "xS", "year", "cst")]
| |||
res = []
| |||
for trial in TRIALS:
| |||
ev = all_ev[all_ev.trial == trial].sort_values("t").reset_index(drop=True)
| |||
T = TRIALS[trial][2]
| |||
for side, y, xo in (("long", "yL", "xL"), ("short", "yS", "xS")):
| |||
for yr in range(2010, 2025):
| |||
tr = ev[(ev.t < pd.Timestamp(f"{yr}-01-01") - pd.Timedelta(hours=T + 2)) & ev[y].notna()]
| |||
te = ev[(ev.year == yr) & ev[y].notna()]
| |||
if len(tr) < 5000 or te.empty:
| |||
continue
| |||
m = HistGradientBoostingRegressor(max_iter=150, learning_rate=0.05, max_depth=4,
| |||
min_samples_leaf=200, l2_regularization=5.0, random_state=0)
| |||
m.fit(tr[feats], tr[y])
| |||
pred = m.predict(te[feats])
| |||
te = te.assign(pred=pred)
| |||
for sym, g in te.groupby("sym"):
| |||
g = g.sort_values("t")
| |||
thr = np.quantile(g.pred, 0.90)
| |||
sel = thin(g.t.values, g[xo].values, (g.pred >= thr).to_numpy())
| |||
allk = thin(g.t.values, g[xo].values, np.ones(len(g), bool))
| |||
res.append((trial, side, sym, yr, len(sel), g[y].values[sel].sum(), len(allk), g[y].values[allk].sum(), g.cst.values[sel].sum(), g.cst.values[allk].sum()))
| |||
print(f"trial {trial} {side} done {time.time() - t0:.0f}s", flush=True)
| |||
r = pd.DataFrame(res, columns=["trial", "side", "sym", "year", "n", "sumR", "n0", "sumR0", "cst", "cst0"])
| |||
r.to_pickle(os.path.join(CACHE, "disc_results.pkl"))
| |||
g = r.groupby(["trial", "side", "sym"]).agg(n=("n", "sum"), R=("sumR", "sum"), n0=("n0", "sum"), R0=("sumR0", "sum"),
| |||
yrs_pos=("sumR", lambda s: int((s > 0).sum())), yrs=("sumR", "size"))
| |||
g["mean"] = g.R / g.n; g["ctrl"] = g.R0 / g.n0; g["lift"] = g["mean"] - g["ctrl"]
| |||
pd.set_option("display.width", 200)
| |||
print(g.round(3).to_string())
| |||
| |||
| |||
if __name__ == "__main__":
| |||
main()
|