"""Flexible basket rule engine for the 50 iteration study (3 Oct 2026). Every basket line is a daily equal weight point in time index (dead coins count).""" import json, math, statistics as S, collections, datetime as dt from stages import stage_series D = json.load(open('bk_clean.json')) PX = json.load(open('majors_px.json')) BTC = {d: v for d, v in D['btc']['line']} KEYS = [k for k in D if k != 'btc'] CHAIN = {k: ('Solana' if k.startswith('sol') else 'Base' if k.startswith('base') else 'TON' if k.startswith('ton') else 'Ethereum' if k.startswith('eth') else 'BNB' if k.startswith('bnb') else 'Majors') for k in KEYS} MEME = {k: ('meme' in k or k in ('sol_pump', 'sol_bonk')) for k in KEYS} SPLIT = '2025-01-01' def sma_map(px, n): ds = sorted(px); out = {} for i, d in enumerate(ds): w = [px[x] for x in ds[max(0, i - n + 1):i + 1]] if len(w) >= int(n * 0.9): out[d] = sum(w) / len(w) return out BMA = {n: sma_map(BTC, n) for n in (20, 50, 100, 200)} SOLBTC = {d: PX['SOL'][d] / PX['BTC'][d] for d in PX['SOL'] if d in PX['BTC']} def donch(L, nin, nout): s = False; o = [] for i in range(len(L)): if i >= nin and L[i] > max(L[i - nin:i]): s = True elif i >= nout and L[i] < min(L[i - nout:i]): s = False o.append(s) return o SERIES = {k: ([d for d, _ in D[k]['line']], [v for _, v in D[k]['line']]) for k in KEYS} MAJ_ON = {} _ds, _L = SERIES['majors'] for nin, nout in ((30, 15), (20, 10), (50, 20)): MAJ_ON[(nin, nout)] = dict(zip(_ds, donch(_L, nin, nout))) # majors phase map for "majors not in hold half" test is built lazily ON_ALL = {k: dict(zip(SERIES[k][0], donch(SERIES[k][1], 30, 15))) for k in KEYS} STAGES = {k: stage_series(SERIES[k][1]) for k in KEYS} def gate_btc(d, kind): if kind is None: return True n = int(kind) return d in BTC and d in BMA[n] and BTC[d] > BMA[n][d] def run(k, nin=30, nout=15, btc='100', majors=True, delay=0, ext_cap=None, half='hot', exit_rule='low', max_days=None, froth_double=None, solbtc_froth=None, breadth=None, cooldown=0, cost=0.01, counter=False, two_closes=False, hot_z=2.0, sell_low=None, take_all_day=None, half_day=None, trail_after_half=None): ds, L = SERIES[k]; st = STAGES[k]; out = []; i = max(nin, nout); last_exit = -999 lg = [math.log(x) for x in L] while i < len(L) - 1: d = ds[i] sig = L[i] > max(L[i - nin:i]) if sig and two_closes: sig = L[i - 1] > max(L[i - 1 - nin:i - 1]) if i - 1 - nin >= 0 else False if sig and not gate_btc(d, btc): sig = False if sig and majors and k != 'majors' and not MAJ_ON[(30, 15)].get(d): sig = False if sig and froth_double and i >= 30 and L[i] / L[i - 30] >= froth_double: sig = False if sig and solbtc_froth and CHAIN[k] == 'Solana': d0 = (dt.date.fromisoformat(d) - dt.timedelta(28)).isoformat() if d in SOLBTC and d0 in SOLBTC and SOLBTC[d] / SOLBTC[d0] - 1 >= solbtc_froth: sig = False if sig and breadth is not None: n_on = sum(1 for kk in KEYS if kk != k and kk != 'majors' and ON_ALL[kk].get(d)) n_all = sum(1 for kk in KEYS if kk != k and kk != 'majors' and d in ON_ALL[kk]) if n_all and n_on / n_all < breadth: sig = False if sig and counter and not gate_btc(d, '50'): sig = False if sig and i - last_exit <= cooldown: sig = False if not sig: i += 1; continue brk = max(L[i - nin:i]) j0 = i + delay if j0 >= len(L) - 1: break # still valid at the delayed entry? (risk on not lost in between) lost = any(L[j] < min(L[j - nout:j]) for j in range(i + 1, j0 + 1)) if lost: i = j0 + 1; continue if ext_cap is not None and L[j0] / brk - 1 > ext_cap: i += 1; continue ent = L[j0]; w = 1.0; acc = 0.0; ex = None; trimmed = False; lown = sell_low or nout for j in range(j0 + 1, len(L)): age = j - i if take_all_day and age >= take_all_day: ex = j; break if exit_rule == 'hold5' and age >= 5: ex = j; break if exit_rule == 'low' and L[j] < min(L[j - lown:j]): ex = j; break if trimmed and trail_after_half and L[j] < min(L[j - trail_after_half:j]): ex = j; break if max_days and age >= max_days: ex = j; break if not trimmed and half: hot = st[j][0] == 'HOT' if hot_z == 2.0 else (st[j][1].get('z', 0) >= hot_z and st[j][1].get('pos', 0) > 0.8) fire = (half == 'hot' and hot) or (half == 'day' and age >= half_day) or (half == 'first' and (hot or age >= half_day)) if fire: acc += 0.5 * (L[j] / ent - 1); w = 0.5; trimmed = True if ex is None: break r = acc + w * (L[ex] / ent - 1) - cost out.append(dict(k=k, sig=d, entry=ds[j0], exit=ds[ex], ret=r, days=ex - j0)) last_exit = ex; i = ex + 1 return out def stats(ts): xs = [t['ret'] for t in ts] if not xs: return dict(n=0, mean=0, med=0, win=0, pf=0) wn = [x for x in xs if x > 0]; ls = [x for x in xs if x <= 0] return dict(n=len(xs), mean=round(S.mean(xs) * 100, 1), med=round(S.median(xs) * 100, 1), win=round(len(wn) / len(xs) * 100), pf=round(sum(wn) / -sum(ls), 2) if ls and sum(ls) < 0 else 9.99) def weekly(ts): w = collections.defaultdict(list) for t in ts: w[dt.date.fromisoformat(t['entry']).isocalendar()[:2]].append(t['ret']) xs = [S.mean(v) for v in w.values()] if not xs: return dict(weeks=0, pf=0) p = sum(x for x in xs if x > 0); n = -sum(x for x in xs if x <= 0) return dict(weeks=len(xs), pf=round(p / n, 2) if n else 9.99, mean=round(S.mean(xs) * 100, 1)) def evaluate(**kw): ts = [] for k in KEYS: ts += run(k, **kw) a = [t for t in ts if t['entry'] < SPLIT]; b = [t for t in ts if t['entry'] >= SPLIT] sol = [t for t in ts if CHAIN[t['k']] == 'Solana']; oth = [t for t in ts if CHAIN[t['k']] != 'Solana'] return dict(all=stats(ts), A=stats(a), B=stats(b), wk=weekly(ts), wkB=weekly(b), sol=stats(sol), other=stats(oth), trades=ts) def line(name, r): def f(o): return "n%3d avg%+5.1f med%+5.1f w%3d pf%5.2f" % (o['n'], o['mean'], o['med'], o['win'], o['pf']) return "%-46s ALL %s | 2023-24 %s | 2025-26 %s | wkPF %.2f/%.2f" % (name[:46], f(r['all']), f(r['A']), f(r['B']), r['wk']['pf'], r['wkB']['pf'])