Warrior_EA/research/test_depth.py
AnimateDread f1b7dcf7f3 fix: correct MI sample alignment and improve BN weight diagnostic report
The MI sample builder used `MathAbs(labelBarOffset)` as a padding, causing rows from offset and non-offset builds to be paired with a double shift. This broke the positive control, failed the 5× gate, and voided all reported mutual‑information figures. Replace with the fixed `MiShiftPad` constant to ensure builds enumerate the same set of bars and row-k alignment is preserved.

Add `BatchOptionsTotal()` to `CNeuronBatchNormOCL` and split the packed BN weight array in the learning report into separate norms for the outgoing dense matrix, gamma, beta, running statistics, and Adam moment buffers. This turns an ambiguous single‑norm reading into precise diagnostics that distinguish weight divergence from scaling issues.
2026-08-02 08:12:47 -04:00

99 lines
4.4 KiB
Python

"""How deep should you wait for the retest? The whole curve, not three points.
Established: at three discrete retest locations the ordering was price edge > value-area edge
> VPOC, monotone in all 8 symbol/timeframe combinations, and the proposed mechanism is
adverse selection - a pullback that reaches deeper into the old range is disproportionately a
breakout that has already failed.
If that mechanism is right it is a CONTINUUM, not three points, and it makes a prediction
that can be checked without choosing anything: expR must fall monotonically as the order is
placed deeper. It also says where to look for the near-zero base the context modifier needs -
on the SHALLOW side, past the price edge, where nobody in either book places an order.
phi < 1 a shallow pullback that never reaches the broken edge - nobody trades here
phi = 1 the price edge itself - the classic Last Point of Support
phi > 1 through the edge into the old range: value-area edge and VPOC territory
ARMS
----
real limit at price_now - phi*(price_now - edge)
placebo limit at the same DISTANCE from the same starting price, distances permuted
across events. Geometry identical, level identity destroyed. Without it, a
shallow-side profit is indistinguishable from "buying small dips works".
The placebo is the arm that matters most. A leak or a drift effect lifts BOTH curves; only a
gap between them is about the level.
"""
import numpy as np, sys, time
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
import fills, book, wyckoff
SYMS = ('EURUSD', 'USDJPY', 'XAUUSD', 'SP500')
DEPTHS = (0.25, 0.5, 0.75, 1.0, 1.25, 1.5, 2.0)
def curve(sym, tf, kR=2.0, mrisk=2.0, wait=40, H=200, depths=DEPTHS, seed=11):
bk = fills.Book(sym)
f = book.frame(sym, tf, bk)
ev = wyckoff.breakouts(f)
if ev is None:
return None
rows = []
for u in depths:
a = wyckoff.retest(sym, tf, u, mrisk=mrisk, kR=kR, wait=wait, H=H,
bk=bk, f=f, ev=ev)
b = wyckoff.retest(sym, tf, u, mrisk=mrisk, kR=kR, wait=wait, H=H,
bk=bk, f=f, ev=ev, placebo=seed)
rows.append((u, a, b))
return rows, ev
def show(sym, tf, rows):
print(f"\n --- {sym} {tf} ---")
print(f" {'phi':>6}{'n':>7}{'ind':>7}{'fill%':>7}{'REAL':>9}{'t':>7}"
f"{'placebo':>9}{'t':>7}{'real-plac':>11}{'unres%':>8}")
us, re, pl = [], [], []
for u, a, b in rows:
if a is None:
continue
A = a['R'][a['indep']]
B = b['R'][b['indep']] if b is not None else np.array([0.0])
us.append(u); re.append(A.mean()); pl.append(B.mean())
print(f" {u:>+6.2f}{a['n']:>7}{int(a['indep'].sum()):>7}"
f"{100*a['n']/max(a['placed'],1):>6.1f}%{A.mean():>+9.4f}{book.tstat(A):>+7.2f}"
f"{B.mean():>+9.4f}{book.tstat(B):>+7.2f}"
f"{A.mean()-B.mean():>+11.4f}{100*a['unresolved']:>7.1f}%")
if len(us) >= 4:
s1, t1 = book.slope_t(np.array(re), np.array(us))
s2, t2 = book.slope_t(np.array(pl), np.array(us))
print(f" slope vs depth: real {s1:+.4f} (t {t1:+.2f}) "
f"placebo {s2:+.4f} (t {t2:+.2f})")
return us, re, pl
if __name__ == '__main__':
syms = [s for s in sys.argv[1:] if s in SYMS] or list(SYMS)
print("=== RETEST DEPTH CURVE ===")
print(" prediction from the adverse-selection mechanism: expR falls as phi RISES.")
print(" phi<1 is the shallow side nobody trades - where a near-zero base could live.")
allrows = []
for sym in syms:
for tf in ('H1', 'H4'):
t0 = time.time()
out = curve(sym, tf)
if out is None:
print(f"\n --- {sym} {tf} --- no events"); continue
rows, ev = out
us, re, pl = show(sym, tf, rows)
print(f" ({len(ev['i']):,} breakouts, {time.time()-t0:.0f}s)")
allrows.append((sym, tf, us, re, pl))
if allrows:
print("\n=== POOLED SHAPE ===")
print(f" {'phi':>6}{'mean real':>11}{'mean placebo':>14}{'cells real>plac':>17}")
for k, u in enumerate(DEPTHS):
r = [re[k] for _, _, us, re, pl in allrows if k < len(re)]
p = [pl[k] for _, _, us, re, pl in allrows if k < len(pl)]
if not r:
continue
w = sum(1 for x, y in zip(r, p) if x > y)
print(f" {u:>+6.2f}{np.mean(r):>+11.4f}{np.mean(p):>+14.4f}{w:>10}/{len(r)}")