animatedread/Warrior_EA
Watch
4
6
Fork
You've already forked Warrior_EA
 7
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
Warrior_EA/research/test_retail.py
AnimateDread c9b489e5d8 research: the retail fade DOES clear cost on EURUSD - correcting the earlier verdict 
The previous commit pooled four instruments with very different spread-to-stop
ratios and concluded the edge never beats the cost. That was too broad. Per
cell (48 cells, M5 path, k=1), 8 clear - and they are not scattered:

   EURUSD  H1  pin     spread/stop 0.042   edge +0.160  cost 0.063  -> +0.097 R
   EURUSD  H1  pin                 0.045        +0.127        0.067     +0.060
   EURUSD  H1  inside              0.047        +0.123        0.060     +0.063
   EURUSD  H1  inside              0.050        +0.102        0.062     +0.040
   EURUSD M15  pin                 0.074        +0.141        0.100     +0.042
   EURUSD M15  pin                 0.072        +0.140        0.099     +0.041

Every clearing cell is on the tightest-spread instrument. XAUUSD carries the
same gross edge (+0.09 to +0.135) and never clears, because its cost is 3x.
That is the mechanism predicting where the effect should survive and being
right - the opposite of the stop-run case, which inverted.

WALK-FORWARD, 4 chronological folds: 6 of 8 hold at >=3/4. EURUSD H1 pin short
side is +0.116 / +0.061 / +0.143 / +0.067 across 23 years, 4/4.

WIDENING THE STOP still says what it said: EURUSD H1 pin goes +0.078 (m=1) ->
+0.028 -> +0.009 -> +0.017 -> -0.003 (m=5). The gross edge collapses ~15x while
the stop widens 5x, so this is NOT drift - it is reversion inside roughly one
setup-risk of a stop order filled at a local extreme. It is only tradeable at
the tight stop, which is exactly where cost bites hardest.

WHAT IS NOT MODELLED, and it decides this: commission and stop slippage. Gross
edge is ~0.139 R = ~2.4 pips on a 17.3-pip stop, against 0.75 pips of spread.
That leaves ~1.6 pips of headroom for commission plus slippage before it is
gone. A demo forward test measuring both is the next step, not more history.

Also fixes a LOOKAHEAD found in the sweep-entry test: the protective stop was
anchored to the low of the very bar that filled the limit order, which is not
known until that bar closes. It was worth ~+0.15 R - larger than any real effect
here - and it inflated the placebo equally, which is how it was caught. With it
removed, buying at retail stop levels is no better than buying at an arbitrary
level the same distance away: the 'stops are a farmable magnet' claim fails its
own control.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-01 22:33:11 -04:00

708 lines
34 KiB
Python
Raw Permalink Blame History

"""Can retail setups be harvested? The three book setups, with their documented stops.
The idea under test is the user's: if price is unpredictable, stop predicting it and trade
against the people who are predicting it badly. We know where they enter and - crucially -
exactly where their stops go, because the books tell them.
From "How To Day Trade Forex For Profit", chapter 5, the three mechanical setups and their
stop rules, quoted rather than paraphrased so the implementation can be checked against
the source:
1. INSIDE BAR "we are looking to buy if the price breaks above the high of the inside
bar... our stop loss order is placed just below the low of the inside bar"
2. PIN BAR "sell short when the price breaks the low of the pin bar. The stop loss
order can be placed just above the high of the bar immediately preceding
the pin bar"
3. ENGULFING "place a stop buy order 1 pip above the high of the engulfing bar... stop
loss order 1 pip below the low of the bar immediately preceding"
All three are trend-filtered by a moving average, which the book insists on.
THREE SEPARATE QUESTIONS, AND THE ECONOMICS THAT CONNECTS THEM
--------------------------------------------------------------
A. Do these setups lose? If they are merely a coin flip that pays the spread, retail's
money goes to the BROKER and the liquidity provider - not to a counterparty who can
systematically capture it. There is nothing to harvest from a fair coin. Harvesting
requires the setups to be genuinely ANTI-predictive, not just unprofitable.
B. Are their stops a magnet? If stop clusters sit at a computable price and get run,
that level should be reached more often than a distance-matched placebo. Same
permuted-offset null used for the volume-profile levels.
C. Does fading the sweep pay? Price takes out the documented stop, then reverses - the
SMC claim. Tested as a complete trade with cost.
A previous test in this project already disconfirmed the generic version of C using swing
extremes as the stop proxy (it passed a family-wise bar and a split-half, then failed its
own mechanism test and 4-fold walk-forward). This is a sharper instrument, not a repeat:
the levels here are the stops a specific, widely-read book instructs its readers to place,
so if the earlier proxy was simply too blunt, this is where that would show up.
"""
import numpy as np, sys, os
sys.stdout.reconfigure(encoding='utf-8', errors='replace')
BARS = 'c:/Users/admin/Documents/Workspaces/Market Data/bars/'
SYMS = ('EURUSD', 'USDJPY', 'XAUUSD', 'SP500')
#--- "1 pip" in the book's sense, per instrument. The buffer is part of the rule, not a
#--- detail: it decides whether a wick that exactly touches the level triggers or not.
PIP = {'EURUSD': 0.0001, 'USDJPY': 0.01, 'XAUUSD': 0.1, 'SP500': 0.25}
def load_bars(sym, tf):
z = np.load(f"{BARS}{sym}_{tf}_ticks.npz", allow_pickle=True)
return z['bars'], {str(c): k for k, c in enumerate(z['columns'])}
def sma(x, n):
out = np.full(len(x), np.nan)
if len(x) >= n:
cs = np.concatenate(([0.0], np.cumsum(x)))
out[n - 1:] = (cs[n:] - cs[:-n]) / n
return out
def setups(sym, tf, ma_n=20, slope_n=5):
"""-> list of (name, fire_index, direction, entry_px, stop_px).
Every level is taken from bars at or before `i`, and the entry can only trigger on a
LATER bar, so nothing here uses information the retail trader would not have had.
"""
a, I = load_bars(sym, tf)
g = lambda k: a[:, I[k]]
o, h, l, c = g('open'), g('high'), g('low'), g('close')
tick = PIP[sym]
m = sma(c, ma_n)
up = np.zeros(len(c), bool); dn = np.zeros(len(c), bool)
up[slope_n:] = (m[slope_n:] > m[:-slope_n]) & (c[slope_n:] > m[slope_n:])
dn[slope_n:] = (m[slope_n:] < m[:-slope_n]) & (c[slope_n:] < m[slope_n:])
rng = np.maximum(h - l, 1e-12)
body = np.abs(c - o)
upw = h - np.maximum(o, c)
dnw = np.minimum(o, c) - l
out = {}
#--- 1. inside bar: contained by its predecessor, traded with the trend
ins = np.zeros(len(c), bool)
ins[1:] = (h[1:] < h[:-1]) & (l[1:] > l[:-1])
out['inside'] = [(np.nonzero(ins & up)[0], +1, 'self'),
(np.nonzero(ins & dn)[0], -1, 'self')]
#--- 2. pin bar: long tail against the trend, small body, close back in the trend's favour
bull_pin = (dnw >= 2 * body) & (dnw >= 0.5 * rng) & (c > (l + 0.5 * rng))
bear_pin = (upw >= 2 * body) & (upw >= 0.5 * rng) & (c < (h - 0.5 * rng))
out['pin'] = [(np.nonzero(bull_pin & up)[0], +1, 'prev'),
(np.nonzero(bear_pin & dn)[0], -1, 'prev')]
#--- 3. engulfing: this body swallows the previous one, after a counter-trend pullback
be = np.zeros(len(c), bool); se = np.zeros(len(c), bool)
be[1:] = (c[1:] > o[1:]) & (c[:-1] < o[:-1]) & (o[1:] <= c[:-1]) & (c[1:] >= o[:-1])
se[1:] = (c[1:] < o[1:]) & (c[:-1] > o[:-1]) & (o[1:] >= c[:-1]) & (c[1:] <= o[:-1])
out['engulf'] = [(np.nonzero(be & up)[0], +1, 'prev'),
(np.nonzero(se & dn)[0], -1, 'prev')]
ev = []
for name, groups in out.items():
for idx, d, stop_from in groups:
idx = idx[(idx > ma_n + slope_n + 2) & (idx < len(c) - 300)]
if not len(idx):
continue
ent = np.where(d > 0, h[idx] + tick, l[idx] - tick)
src = idx if stop_from == 'self' else idx - 1
stp = np.where(d > 0, l[src] - tick, h[src] + tick)
ev.append((name, idx, np.full(len(idx), d), ent, stp))
return ev, o, h, l, c, g('spread_mean'), tick
def triggered(h, l, idx, d, ent, within=1):
"""Did the stop-entry order fill? The book enters on the NEXT bar."""
ok = np.zeros(len(idx), bool)
fill = np.full(len(idx), -1, np.int64)
for k in range(1, within + 1):
j = idx + k
m = (~ok) & (j < len(h))
hit = np.where(d[m] > 0, h[j[m]] >= ent[m], l[j[m]] <= ent[m])
w = np.nonzero(m)[0][hit]
ok[w] = True; fill[w] = j[w]
return ok, fill
def race(h, l, e, d, stop_px, targ_px, H):
"""+1 target first, -1 stop first, 0 unresolved. A bar spanning both books the loss."""
n = len(h)
out = np.zeros(len(e), np.int8)
live = np.ones(len(e), bool)
for k in range(0, H + 1):
j = e + k
ok = live & (j < n)
if not ok.any():
break
jj = j[ok]
lose = np.where(d[ok] > 0, l[jj] <= stop_px[ok], h[jj] >= stop_px[ok])
winb = np.where(d[ok] > 0, h[jj] >= targ_px[ok], l[jj] <= targ_px[ok])
res = np.where(lose, -1, np.where(winb, 1, 0)).astype(np.int8)
w = np.nonzero(ok)[0]
got = res != 0
out[w[got]] = res[got]
live[w[got]] = False
return out
def question_A(sym, tf, H=200, ks=(1.0, 2.0, 3.0), cost=True, agg=None):
"""Do the retail setups actually lose - or are they a coin flip that pays the spread?
THE DECISIVE COMPARISON is this function run twice, with cost on and off.
expR < 0 with cost, ~0 without -> the setups are a fair coin and the loss IS the
spread. That money is already gone to the broker
and the liquidity provider before any counterparty
sees it. There is nothing to harvest.
expR < 0 with cost AND without -> the setups are genuinely anti-predictive, and
taking the other side is a real edge.
Only the second case supports trading against retail. The first is the case that feels
the same and pays nothing.
"""
ev, o, h, l, c, spm, tick = setups(sym, tf)
rows = []
for name, idx, d, ent, stp in ev:
ok, fill = triggered(h, l, idx, d, ent)
if ok.sum() < 100:
continue
i2, d2, e2, s2 = fill[ok], d[ok], ent[ok], stp[ok]
sp = spm[i2] if cost else np.zeros(ok.sum())
risk = np.abs(e2 - s2) + sp
good = risk > 0
i2, d2, e2, s2, sp, risk = (v[good] for v in (i2, d2, e2, s2, sp, risk))
for k in ks:
targ = e2 + d2 * k * risk
r = race(h, l, i2, d2, s2 - d2 * sp, targ + d2 * sp, H)
R = np.where(r > 0, k, np.where(r < 0, -1.0, 0.0))
un = r == 0
if un.any():
j = np.minimum(i2[un] + H, len(c) - 1)
R[un] = (c[j] - e2[un]) * d2[un] / risk[un] - sp[un] / risk[un]
t = R.mean() / max(R.std(ddof=1) / np.sqrt(len(R)), 1e-12)
rows.append((name, k, len(R), 100 * (r > 0).mean(), R.mean(), t))
return rows
def question_A_paired(syms, TFS, ks=(1.0, 2.0)):
"""Every setup priced with and without the spread, side by side. This table is the
whole answer to 'can we farm retail'."""
print(f" {'symbol':>7} {'tf':>4} {'setup':<8} {'k':>4} {'n':>7}"
f"{'expR w/cost':>13}{'t':>7}{'expR NO cost':>14}{'t':>7}{'= spread':>10}")
tot = {}
for tf in TFS:
for s in syms:
wc = question_A(s, tf, ks=ks, cost=True)
nc = question_A(s, tf, ks=ks, cost=False)
for r, q in zip(wc, nc): # same order: long then short per setup
print(f" {s:>7} {tf:>4} {r[0]:<8} {r[1]:>4.1f} {r[2]:>7}"
f"{r[4]:>+13.3f}{r[5]:>+7.2f}{q[4]:>+14.3f}{q[5]:>+7.2f}"
f"{r[4]-q[4]:>+10.3f}")
tot.setdefault((r[0], r[1]), []).append((r[4], q[4]))
print()
for key, v in sorted(tot.items()):
v = np.array(v)
print(f" POOLED {key[0]:<8} k={key[1]:<4} n={len(v):>2} series "
f"expR with cost {v[:, 0].mean():+.3f} without cost {v[:, 1].mean():+.3f}")
def mirror(sym, tf, H=200, ks=(1.0, 2.0), path_tf=None, pess=True, folds=0):
"""Retail's trade and its exact mirror, priced side by side under identical rules.
This is the control that decides whether the losing setups are harvestable, and it is
needed because the two obvious explanations for a negative expR are not edges at all:
SAME-BAR AMBIGUITY. A bar that spans both barriers is booked as a loss everywhere in
this project. Inside-bar stops are by construction tiny, so a large share of bars
span both, and the convention alone manufactures a negative number. It penalises
WHOEVER holds the trade - so if it is the cause, the mirror loses too.
SPREAD. Charged to both sides, likewise.
A real anti-predictive edge is the only thing that makes retail negative AND the mirror
positive. `path_tf` re-walks the same barriers on a finer series (M5 under an H1 setup)
so the ambiguity is measured rather than assumed; `pess=False` flips the convention to
optimistic, bounding the effect from the other side.
"""
ev, o, h, l, c, spm, tick = setups(sym, tf)
ph, pl, pc, pmap = h, l, c, None
if path_tf:
a2, I2 = load_bars(sym, path_tf)
ph, pl, pc = a2[:, I2['high']], a2[:, I2['low']], a2[:, I2['close']]
a1, I1 = load_bars(sym, tf)
pmap = np.searchsorted(a2[:, I2['time']], a1[:, I1['time']])
out = []
for name, idx, d, ent, stp in ev:
ok, fill = triggered(h, l, idx, d, ent)
if ok.sum() < 100:
continue
i2, d2, e2, s2 = fill[ok], d[ok], ent[ok], stp[ok]
sp = spm[i2]
risk = np.abs(e2 - s2) + sp
g = risk > 0
i2, d2, e2, s2, sp, risk = (v[g] for v in (i2, d2, e2, s2, sp, risk))
pi = np.clip(pmap[i2], 0, len(ph) - 1) if pmap is not None else i2
HH = H * (12 if path_tf == 'M5' and tf == 'H1' else
3 if path_tf == 'M5' and tf == 'M15' else 1)
keep = pi + HH < len(ph)
i2, d2, e2, s2, sp, risk, pi = (v[keep] for v in (i2, d2, e2, s2, sp, risk, pi))
if len(pi) < 100:
continue
#--- Accounting, stated once and applied IDENTICALLY to both sides:
#--- barriers sit at the pure price levels (no spread padding), and the round trip
#--- is charged as a single spread deducted in R at the end. Padding the barrier AND
#--- inflating the risk denominator charges it twice, which quietly moves the stop
#--- further away - and since the two sides have different stops, that does not
#--- cancel in the comparison.
risk0 = np.abs(e2 - s2)
#--- A stop closer than two spreads is not a trade anyone takes, and dividing by it
#--- produces R-values in the millions. The book's own pin-bar rule can put the stop
#--- almost on top of the entry when the preceding bar sits right against the pin.
good = risk0 > 2 * sp
i2, d2, e2, s2, sp, risk0, pi = (v[good] for v in
(i2, d2, e2, s2, sp, risk0, pi))
if len(pi) < 100:
continue
fee = sp / risk0
for k in ks:
row = [name, k, len(pi)]
for side in (+1, -1): # +1 = retail, -1 = the mirror
dd = d2 * side
r = race_px(ph, pl, pi, dd, e2 - dd * risk0, e2 + dd * k * risk0, HH, pess)
R = np.where(r > 0, float(k), np.where(r < 0, -1.0, 0.0))
un = r == 0
if un.any():
j = np.minimum(pi[un] + HH, len(pc) - 1)
R[un] = (pc[j] - e2[un]) * dd[un] / risk0[un]
R = R - fee
row += [R.mean(), R.mean() / max(R.std(ddof=1) / np.sqrt(len(R)), 1e-12),
100 * (r == 0).mean()]
if side < 0 and folds:
#--- chronological folds of the FADE side: events are already in time
#--- order, so a plain split is a walk-forward
row.append([float(x.mean()) for x in np.array_split(R, folds)])
row += [sp.mean(), risk0.mean()] # physical units for the same cell
out.append(row)
return out
def race_px(h, l, e, d, stop_px, targ_px, H, pess=True):
"""As race(), but the tie convention is switchable so its contribution can be bounded."""
n = len(h)
out = np.zeros(len(e), np.int8)
live = np.ones(len(e), bool)
for k in range(0, H + 1):
j = e + k
ok = live & (j < n)
if not ok.any():
break
jj = j[ok]
lose = np.where(d[ok] > 0, l[jj] <= stop_px[ok], h[jj] >= stop_px[ok])
winb = np.where(d[ok] > 0, h[jj] >= targ_px[ok], l[jj] <= targ_px[ok])
res = (np.where(lose, -1, np.where(winb, 1, 0)) if pess
else np.where(winb, 1, np.where(lose, -1, 0))).astype(np.int8)
w = np.nonzero(ok)[0]
got = res != 0
out[w[got]] = res[got]
live[w[got]] = False
return out
def widen(sym, tf, H=200, ms=(1.0, 1.5, 2.0, 3.0, 5.0), ks=(1.0, 2.0), path_tf='M5'):
"""The fade, with the stop DECOUPLED from the book's tight one.
The mirror test says the retail setups carry a real anti-predictive edge of ~0.1 R and
a round-trip cost of ~0.14 R, so the fade misses by about a fifth. But those two are not
fixed relative to each other:
cost in R = spread / stop distance
Widening the stop divides the cost while the directional edge - if it is drift over the
horizon rather than an artifact of the barrier geometry - should survive. If the edge is
real, some multiple crosses. If the edge shrinks in lockstep with the cost, it was never
a directional effect and this is the test that says so.
The stop is placed at m x the book's own risk, so the setup still defines the geometry -
it is the same trade, sized differently, not a new one mined on top.
"""
ev, o, h, l, c, spm, tick = setups(sym, tf)
a2, I2 = load_bars(sym, path_tf)
ph, pl, pc = a2[:, I2['high']], a2[:, I2['low']], a2[:, I2['close']]
a1, I1 = load_bars(sym, tf)
pmap = np.searchsorted(a2[:, I2['time']], a1[:, I1['time']])
HH = H * (12 if tf == 'H1' else 3)
out = {}
for name, idx, d, ent, stp in ev:
ok, fill = triggered(h, l, idx, d, ent)
if ok.sum() < 100:
continue
i2, d2, e2, s2 = fill[ok], d[ok], ent[ok], stp[ok]
sp = spm[i2]
risk0 = np.abs(e2 - s2)
g = (risk0 > 2 * sp)
i2, d2, e2, sp, risk0 = (v[g] for v in (i2, d2, e2, sp, risk0))
pi = np.clip(pmap[i2], 0, len(ph) - 1)
keep = pi + HH < len(ph)
i2, d2, e2, sp, risk0, pi = (v[keep] for v in (i2, d2, e2, sp, risk0, pi))
if len(pi) < 150:
continue
dd = -d2 # the fade
for m in ms:
R0 = m * risk0
fee = sp / R0
for k in ks:
r = race_px(ph, pl, pi, dd, e2 - dd * R0, e2 + dd * k * R0, HH)
R = np.where(r > 0, float(k), np.where(r < 0, -1.0, 0.0))
un = r == 0
if un.any():
j = np.minimum(pi[un] + HH, len(pc) - 1)
R[un] = (pc[j] - e2[un]) * dd[un] / R0[un]
R = R - fee
t = R.mean() / max(R.std(ddof=1) / np.sqrt(len(R)), 1e-12)
out.setdefault((name, m, k), []).append((R.mean(), t, len(R), fee.mean()))
return out
def sweep_entry(sym, tf, H=200, js=(0.25, 0.5, 1.0), ks=(1.0, 2.0, 3.0),
path_tf='M5', wait=None, placebo=False, seed=17):
"""BUY THEIR STOPS. A resting limit at the retail stop level, then ride the recovery.
This is a genuinely different trade from the mirror, and the one actually intended:
mirror short from retail's ENTRY, targeting their stop. Profits from the
stop-out itself.
sweep entry a limit order sitting AT the retail stop. Price comes down, fires their
stop, fills us, and we hold for the move retail was originally right
about. Their forced selling is our fill.
The two differ in where you get in, and that changes everything downstream: entering at
the stop level lets the protective stop sit just under the sweep's own low, so R is
small and the same move is worth several times more R than it is from retail's entry.
Fill model. Bars here are MID prices, so a buy limit at L needs the ask down to L, i.e.
mid <= L - spread/2. That half-spread is charged on entry and the other half on exit -
one full round trip, the same as every other test in this project. A resting limit does
not dodge the spread; what it buys is the location.
`placebo=True` re-runs the identical trade against a level the same distance away but
shuffled across events, which is the test of whether the STOP LEVEL is special or just
a price below the market.
"""
ev, o, h, l, c, spm, tick = setups(sym, tf)
a2, I2 = load_bars(sym, path_tf)
ph, pl, pc = a2[:, I2['high']], a2[:, I2['low']], a2[:, I2['close']]
psp = a2[:, I2['spread_mean']]
a1, I1 = load_bars(sym, tf)
pmap = np.searchsorted(a2[:, I2['time']], a1[:, I1['time']])
mult = 12 if tf == 'H1' else 3
wait = wait or H * mult
HH = H * mult
rng = np.random.default_rng(seed)
out = {}
for name, idx, d, ent, stp in ev:
ok, fill = triggered(h, l, idx, d, ent)
if ok.sum() < 100:
continue
i2, d2, e2, s2 = fill[ok], d[ok], ent[ok], stp[ok]
risk0 = np.abs(e2 - s2)
g = risk0 > 2 * spm[i2]
i2, d2, e2, s2, risk0 = (v[g] for v in (i2, d2, e2, s2, risk0))
pi = np.clip(pmap[i2], 0, len(ph) - 1)
keep = pi + wait + HH < len(ph)
i2, d2, e2, s2, risk0, pi = (v[keep] for v in (i2, d2, e2, s2, risk0, pi))
if len(pi) < 150:
continue
lvl = s2
if placebo:
#--- same distance below/above the market, shuffled across events within
#--- direction: destroys "this is where the stops are", keeps the geometry
r = risk0.copy()
for m in (d2 > 0, d2 < 0):
if m.any():
r[m] = rng.permutation(risk0[m])
lvl = e2 - d2 * r
#--- first bar the limit can fill: mid must reach level -/+ half the spread
half = psp[pi] * 0.5
want = lvl - d2 * half
fillbar = np.full(len(pi), -1, np.int64)
live = np.ones(len(pi), bool)
for k in range(0, wait + 1):
j = pi + k
m = live & (j < len(ph))
if not m.any():
break
hit = np.where(d2[m] > 0, pl[j[m]] <= want[m], ph[j[m]] >= want[m])
w = np.nonzero(m)[0][hit]
fillbar[w] = j[w]; live[w] = False
got = (fillbar >= 0) & (fillbar + HH < len(ph))
if got.sum() < 150:
continue
fb, dd, entry = fillbar[got], d2[got], lvl[got]
sp = psp[fb]
for j in js:
#--- Protective stop a fraction of the setup's risk BELOW THE ENTRY PRICE.
#--- It must not be derived from the fill bar's own low: that low is not known
#--- until the bar closes, and anchoring to it guarantees the trade survives the
#--- bar it entered on. That single lookahead was worth ~+0.15 R here - larger
#--- than any effect this project has ever measured - and it inflates the
#--- placebo just as much, which is how it was caught.
R0 = np.maximum(j * risk0[got], 2 * sp)
fee = sp / R0
for k in ks:
r = race_px(ph, pl, fb, dd, entry - dd * R0, entry + dd * k * R0, HH)
R = np.where(r > 0, float(k), np.where(r < 0, -1.0, 0.0))
un = r == 0
if un.any():
q = np.minimum(fb[un] + HH, len(pc) - 1)
R[un] = (pc[q] - entry[un]) * dd[un] / R0[un]
R = R - fee
t = R.mean() / max(R.std(ddof=1) / np.sqrt(len(R)), 1e-12)
out.setdefault((name, j, k), []).append(
(R.mean(), t, len(R), fee.mean(), 100 * got.mean()))
return out
def question_B(sym, tf, H=200, nperm=2000, seed=3):
"""Are the documented stop levels reached more often than a distance-matched placebo?"""
ev, o, h, l, c, spm, tick = setups(sym, tf)
W = np.lib.stride_tricks.sliding_window_view
out = []
for name, idx, d, ent, stp in ev:
ok, fill = triggered(h, l, idx, d, ent)
if ok.sum() < 300:
continue
i2, d2, e2, s2 = fill[ok], d[ok], ent[ok], stp[ok]
keep = i2 + H < len(h)
i2, d2, e2, s2 = i2[keep], d2[keep], e2[keep], s2[keep]
if len(i2) < 300:
continue
#--- the stop sits BELOW entry for a long, so the relevant excursion is downward
cu = np.maximum.accumulate(W(h, H)[i2 + 1], axis=1)[:, -1]
cl = np.minimum.accumulate(W(l, H)[i2 + 1], axis=1)[:, -1]
base = c[i2]
off = s2 - base
upd = off > 0
hit = np.where(upd, off + base <= cu, off + base >= cl)
rate = hit.mean()
#--- placebo: same |offset|, same side, shuffled across events within side
rng = np.random.default_rng(seed)
r = np.abs(off); sgn = np.sign(off)
pr = np.empty(nperm)
for q in range(nperm):
rr = np.empty_like(r)
for m in (upd, ~upd):
if m.any():
rr[m] = rng.permutation(r[m])
p = base + sgn * rr
pr[q] = np.where(upd, p <= cu, p >= cl).mean()
mu, sd = pr.mean(), max(pr.std(ddof=1), 1e-12)
out.append((name, len(i2), rate, mu, (rate - mu) / sd, (pr - mu) / sd))
print(f" {sym:>7} {tf:>4} {name:<8} n={len(i2):>6} stop hit {100*rate:6.2f}% "
f"placebo {100*mu:6.2f}% diff {100*(rate-mu):+6.2f}pp z {(rate-mu)/sd:+7.2f}")
return out
def question_C(sym, tf, H=200, ks=(1.0, 2.0, 3.0), buf=0.05):
"""Fade the sweep: once the documented stop is taken out, trade back the other way."""
ev, o, h, l, c, spm, tick = setups(sym, tf)
rows = []
for name, idx, d, ent, stp in ev:
ok, fill = triggered(h, l, idx, d, ent)
if ok.sum() < 100:
continue
i2, d2, s2 = fill[ok], d[ok], stp[ok]
#--- first bar that takes the retail stop out
sweep = np.full(len(i2), -1, np.int64)
live = np.ones(len(i2), bool)
for k in range(0, H + 1):
j = i2 + k
m = live & (j < len(h))
if not m.any():
break
hit = np.where(d2[m] > 0, l[j[m]] <= s2[m], h[j[m]] >= s2[m])
w = np.nonzero(m)[0][hit]
sweep[w] = j[w]; live[w] = False
m = (sweep >= 0) & (sweep + H + 2 < len(h))
if m.sum() < 100:
continue
sw = sweep[m]; dd = d2[m] # dd = the ORIGINAL retail direction
e3 = sw + 1 # enter at the next open after the sweep
ent3 = o[e3]; sp = spm[e3]
#--- stop beyond the sweep's own extreme - where the liquidity actually was
ext = np.where(dd > 0, l[sw], h[sw])
stop3 = ext - dd * buf * np.abs(ent3 - ext)
risk = np.abs(ent3 - stop3) + sp
good = risk > 0
e3, dd, ent3, stop3, risk, sp = (v[good] for v in (e3, dd, ent3, stop3, risk, sp))
for k in ks:
targ = ent3 + dd * k * risk
r = race(h, l, e3, dd, stop3 - dd * sp, targ + dd * sp, H)
R = np.where(r > 0, k, np.where(r < 0, -1.0, 0.0))
un = r == 0
if un.any():
j = np.minimum(e3[un] + H, len(c) - 1)
R[un] = (c[j] - ent3[un]) * dd[un] / risk[un] - sp[un] / risk[un]
t = R.mean() / max(R.std(ddof=1) / np.sqrt(len(R)), 1e-12)
rows.append((name, k, len(R), 100 * (r > 0).mean(), R.mean(), t))
for r in rows:
print(f" {sym:>7} {tf:>4} {r[0]:<8} k={r[1]:<4} n={r[2]:>6} win {r[3]:5.2f}% "
f"expR {r[4]:+.3f} t {r[5]:+6.2f}")
return rows
if __name__ == '__main__':
which = sys.argv[1] if len(sys.argv) > 1 else 'all'
syms = [s for s in sys.argv[2:] if s in SYMS] or list(SYMS)
TFS = ('M15', 'H1')
if which in ('all', 'A'):
print("=== A. DO THE RETAIL SETUPS LOSE, AND IS THE LOSS HARVESTABLE? ===")
print(" A coin flip that pays the spread has already given its money to the")
print(" broker. Only a loss that SURVIVES removing the spread is fadeable.\n")
question_A_paired(syms, TFS)
if which in ('all', 'M'):
for path_tf, pess, lbl in ((None, True, 'H1/M15 bars, pessimistic ties'),
('M5', True, 'M5 path, pessimistic ties'),
('M5', False, 'M5 path, OPTIMISTIC ties')):
print(f"\n=== MIRROR TEST - {lbl} ===")
print(f" {'symbol':>7} {'tf':>4} {'setup':<8} {'k':>4} {'n':>7}"
f"{'retail expR':>13}{'t':>7}{'MIRROR expR':>13}{'t':>7}{'unres%':>8}")
tot = {}
for tf in TFS:
for s in syms:
for r in mirror(s, tf, path_tf=path_tf, pess=pess):
print(f" {s:>7} {tf:>4} {r[0]:<8} {r[1]:>4.1f} {r[2]:>7}"
f"{r[3]:>+13.3f}{r[4]:>+7.2f}{r[6]:>+13.3f}{r[7]:>+7.2f}"
f"{r[8]:>8.1f}")
tot.setdefault((r[0], r[1]), []).append((r[3], r[6]))
print()
#--- The decomposition that answers the question. Both sides pay the same cost
#--- and suffer the same tie convention, so those cancel in the DIFFERENCE and
#--- double in the SUM:
#--- edge = (mirror - retail)/2 how anti-predictive the setup really is
#--- cost = -(mirror + retail)/2 what it costs to be in the trade at all
#--- Harvesting retail requires edge > cost. Nothing else does.
print(f"\n {'setup':<8}{'k':>5}{'retail':>9}{'mirror':>9}"
f"{'EDGE':>9}{'COST':>9} verdict")
for key, v in sorted(tot.items()):
v = np.array(v)
rt, mr = v[:, 0].mean(), v[:, 1].mean()
edge, cost = (mr - rt) / 2, -(mr + rt) / 2
print(f" {key[0]:<8}{key[1]:>5.1f}{rt:>+9.3f}{mr:>+9.3f}"
f"{edge:>+9.3f}{cost:>+9.3f} "
f"{'EDGE BEATS COST' if edge > cost else 'cost wins'}")
if which in ('all', 'V'):
#--- The battery this project's own rules demand of any survivor:
#--- 1. walk-forward in >=4 folds (a 2-way split hid a losing half once already)
#--- 2. widen the stop - cost falls as 1/m, a real directional edge does not
#--- 3. check the mechanism where it predicts it should be strongest
print("=== VALIDATING THE CELLS THAT CLEARED ===\n")
print("--- 1. WALK-FORWARD, 4 chronological folds (fade expR per fold) ---")
print(f" {'symbol':>7}{'tf':>5} {'setup':<8}{'n':>7}{'all':>8}"
f"{'Q1':>8}{'Q2':>8}{'Q3':>8}{'Q4':>8}{'+ve':>6}")
for tf in TFS:
for s in syms:
for r in mirror(s, tf, path_tf='M5', ks=(1.0,), folds=4):
if r[6] <= 0:
continue
f = r[9] # folds land right after the mirror's three columns
pos = sum(1 for x in f if x > 0)
print(f" {s:>7}{tf:>5} {r[0]:<8}{r[2]:>7}{r[6]:>+8.3f}"
+ "".join(f"{x:>+8.3f}" for x in f)
+ f"{pos:>4}/4" + (" HOLDS" if pos >= 3 else " fails"))
print("\n--- 2. WIDENING THE STOP on the instrument that cleared ---")
print(" cost falls as 1/m. If the edge is drift it survives; if it is")
print(" entry-point microstructure it decays with the cost.\n")
print(f" {'symbol':>7}{'tf':>5} {'setup':<8}{'m':>5}{'cost':>8}{'fade expR':>11}{'t':>8}")
for tf in TFS:
for s in ('EURUSD',):
agg = widen(s, tf, ms=(1.0, 1.5, 2.0, 3.0, 5.0), ks=(1.0,))
for key in sorted(agg):
v = np.array(agg[key])
w = v[:, 2] / v[:, 2].sum()
tp = float((v[:, 1] * np.sqrt(v[:, 2])).sum() / np.sqrt(v[:, 2].sum()))
print(f" {s:>7}{tf:>5} {key[0]:<8}{key[1]:>5.1f}"
f"{(v[:,3]*w).sum():>8.3f}{(v[:,0]*w).sum():>+11.3f}{tp:>+8.2f}")
if which in ('all', 'X'):
#--- Pooling four instruments with different spread-to-stop ratios can hide a cell
#--- where the edge really does clear the cost. This prints every cell, plus the
#--- physical units, so "the spread is tiny" can be checked rather than asserted.
print("=== PER-CELL: does the edge beat the cost ANYWHERE? (M5 path, k=1) ===")
print(" spread and stop are in PIPS/POINTS so 'the spread is tiny' is checkable.\n")
print(f" {'symbol':>7}{'tf':>5} {'setup':<8}{'trades':>8}"
f"{'spread':>8}{'stop':>8}{'sprd/stop':>10}{'EDGE':>8}{'COST':>8}"
f"{'edge-cost':>11}")
clears = 0
for tf in TFS:
for s in syms:
for r in mirror(s, tf, path_tf='M5', ks=(1.0,)):
edge, cost = (r[6] - r[3]) / 2, -(r[6] + r[3]) / 2
pipsz = PIP[s]
ok = edge > cost
clears += ok
print(f" {s:>7}{tf:>5} {r[0]:<8}{r[2]:>8}"
f"{r[9]/pipsz:>8.2f}{r[10]/pipsz:>8.1f}"
f"{r[9]/r[10]:>10.3f}"
f"{edge:>+8.3f}{cost:>+8.3f}{edge-cost:>+11.3f}"
f"{' <-- CLEARS' if ok else ''}")
print(f"\n cells where the edge beats the cost: {clears} / 48")
if which in ('all', 'S'):
print("=== BUY THEIR STOPS: a resting limit AT the retail stop level, then ride "
"the recovery ===")
print(" j = protective stop beyond the sweep's own low, in units of the setup's")
print(" risk. k = target in R. M5 path, one full round-trip spread charged.\n")
for tag, pl_ in (('REAL stop levels', False), ('PLACEBO same distance', True)):
tot = {}
for tf in TFS:
for s in syms:
for key, v in sweep_entry(s, tf, placebo=pl_).items():
tot.setdefault(key, []).extend(v)
print(f" --- {tag} ---")
print(f" {'setup':<8}{'j':>5}{'k':>5}{'trades':>9}{'fill%':>7}"
f"{'cost(R)':>9}{'expR':>9}{'t':>8}")
for key in sorted(tot):
v = np.array(tot[key])
w = v[:, 2] / v[:, 2].sum()
tp = float((v[:, 1] * np.sqrt(v[:, 2])).sum() / np.sqrt(v[:, 2].sum()))
print(f" {key[0]:<8}{key[1]:>5.2f}{key[2]:>5.1f}{int(v[:,2].sum()):>9}"
f"{(v[:,4]*w).sum():>7.1f}{(v[:,3]*w).sum():>9.3f}"
f"{(v[:,0]*w).sum():>+9.3f}{tp:>+8.2f}")
print()
if which in ('all', 'W'):
print("=== WIDENING THE FADE'S STOP: cost falls as 1/m. Does the edge survive? ===")
print(" stop = m x the book's own stop distance; M5 path; spread charged once.\n")
tot = {}
for tf in TFS:
for s in syms:
for key, v in widen(s, tf).items():
tot.setdefault(key, []).extend(v)
print(f" {'setup':<8}{'m':>5}{'k':>5}{'series':>8}{'trades':>9}"
f"{'cost(R)':>9}{'fade expR':>11}{'t':>8}")
for key in sorted(tot):
v = np.array(tot[key])
n = int(v[:, 2].sum())
#--- pooled t across series, weighted by trade count
w = v[:, 2] / v[:, 2].sum()
tp = float((v[:, 1] * np.sqrt(v[:, 2])).sum() / np.sqrt(v[:, 2].sum()))
print(f" {key[0]:<8}{key[1]:>5.1f}{key[2]:>5.1f}{len(v):>8}{n:>9}"
f"{(v[:, 3]*w).sum():>9.3f}{(v[:, 0]*w).sum():>+11.3f}{tp:>+8.2f}")
if which in ('all', 'B'):
print("\n=== B. ARE THE DOCUMENTED STOPS A MAGNET? "
"(vs a placebo at the same distance) ===")
acc = []
for tf in TFS:
for s in syms:
acc += [x[5] for x in question_B(s, tf)]
if acc:
m = min(len(x) for x in acc)
crit = float(np.quantile(np.maximum.reduce([np.abs(x[:m]) for x in acc]), 0.95))
print(f" family-wise |z| bar over {len(acc)} tests: {crit:.2f}")
if which in ('all', 'C'):
print("\n=== C. FADE THE SWEEP: enter after the retail stop is taken out ===")
for tf in TFS:
for s in syms:
question_C(s, tf)
Reference in a new issue View git blame Copy permalink