""" rule_events.py -- [2026-09-30] Playbook library, layer 2: EVERY time each candidate rule's situation appeared (win or lose), not just the hindsight winners. Candidates come from the 9/29-9/30 trade reviews (SG, AGEN, SSL and the 8 trades of 9/30): ENTRY (buy at the next bar's open; triple barrier +2% / -1% / 60 min, and to 15:55) E1 base after the low : >= 30 min since the day's low with no new low, first close back above VWAP (was below within the last 5 bars) E2 opening dip held : 9:31-10:00, the day's low so far >= open - 0.5%, first close back above both VWAP and the open after trading below one of them E3 VWAP pullback held : was >= 1% over VWAP within the last 30 min, a bar's low comes within 0.2% of VWAP (or under) and it closes back above B1 buy after a climax : a close >= 1% over VWAP within 30 min after a >= 10x-volume bar that made the day's high (what the rule would BLOCK) BASE : every 5th minute, any stock with price over VWAP (the reference) EXIT (for someone holding: what happens next if they DON'T sell) X1 volume climax : a bar with volume >= 10x its 14-day normal that made the day's high X2 stretched : close >= 3% over VWAP and the last 5 closes not rising X3 failed new high : day already >= +2% over the open, a new high in the last 15 min, now >= 0.5% under it XBASE : every 5th minute, any stock >= +2% over the open (the reference) Exit outcome: next 30 min change, and "-1% before +1%" (selling would have been right). Days are split: A = up to 2025-09-30, B = 2025-10-01..2026-06-30, HOLDOUT = 2026-07-01 on (reported separately, not used to pick anything). 0.1% round-trip cost on entries. """ import gzip import json import statistics as st import sys from collections import defaultdict from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo HERE = Path(__file__).resolve().parent sys.path.insert(0, "/var/www/screener/trade") import reference as REF ET = ZoneInfo("America/New_York") COST = 0.1 def period(day): return "A" if day <= "2025-09-30" else "B" if day <= "2026-06-30" else "HOLD" def load(): for p in sorted(HERE.glob("bars/bars_20*.json.gz")): d = json.load(gzip.open(p, "rt")) day = d["date"] for s in d["top30_current"]: ref = REF.load(day, s) raw = d["symbols"].get(s, {}).get("bars", []) if len(raw) < 200 or not ref: continue by = {} for b in raw: t = datetime.fromtimestamp(b[0], ET) m = (t.hour - 9) * 60 + t.minute - 30 if 0 <= m < 390: by[m] = b yield day, s, by, ref REF._CACHE.clear() def series(by, ref): o = by[min(by)][1] C, H, L, V, VW, VX, HI, LO = [], [], [], [], [], [], [], [] last, pv, cv, hi, lo = o, 0.0, 0.0, o, o for m in range(390): b = by.get(m) if b: last = b[4] pv += (b[2] + b[3] + b[4]) / 3 * b[5] cv += b[5] hi, lo = max(hi, b[2]), min(lo, b[3]) C.append(b[4]); H.append(b[2]); L.append(b[3]); V.append(b[5]) else: C.append(last); H.append(last); L.append(last); V.append(0.0) VW.append(pv / cv if cv else last) VX.append(V[-1] / max(ref["vol_per_min"][m], 1)) HI.append(hi); LO.append(lo) return o, C, H, L, V, VW, VX, HI, LO def entry_outcome(by, C, H, L, m): """Buy at the next bar's open. Triple barrier +2/-1 within 60 min; to the close.""" nb = by.get(m + 1) e = nb[1] if nb else C[m] first = "time" for k in range(m + 1, min(m + 61, 386)): if L[k] <= e * 0.99: first = "stop" break if H[k] >= e * 1.02: first = "target" break return {"tb": first, "close": (C[385] / e - 1) * 100 - COST, "fwd60": (C[min(m + 60, 385)] / e - 1) * 100 - COST} def exit_outcome(C, H, L, m): p = C[m] first = "none" for k in range(m + 1, min(m + 31, 386)): if L[k] <= p * 0.99: first = "down" break if H[k] >= p * 1.01: first = "up" break return {"tb": first, "fwd30": (C[min(m + 30, 385)] / p - 1) * 100} def scan(day, s, by, ref, ev): o, C, H, L, V, VW, VX, HI, LO = series(by, ref) pr = period(day) low_m = 0 e1 = e2 = False last_climax = -999 b1_done = False for m in range(1, 346): if L[m] <= LO[m - 1] - 1e-9 or m == 0: low_m = m above = C[m] > VW[m] was_below = any(C[k] <= VW[k] for k in range(max(0, m - 5), m)) # climax at the high if VX[m] >= 10 and H[m] >= HI[m - 1] - 1e-9 and by.get(m): last_climax = m if m >= 15: ev["X1"].append((pr, exit_outcome(C, H, L, m))) if m < 15: continue # E1 base after the low if not e1 and m - low_m >= 30 and above and was_below: e1 = True ev["E1"].append((pr, entry_outcome(by, C, H, L, m))) # E2 opening dip held if not e2 and m <= 30 and LO[m] >= o * 0.995 and C[m] > VW[m] and C[m] > o \ and any(C[k] <= VW[k] or C[k] <= o for k in range(max(0, m - 5), m)): e2 = True ev["E2"].append((pr, entry_outcome(by, C, H, L, m))) # E3 VWAP pullback held if max(C[k] / VW[k] for k in range(max(0, m - 30), m)) >= 1.01 and L[m] <= VW[m] * 1.002 and above \ and not any(L[k] <= VW[k] * 1.002 for k in range(max(0, m - 10), m)): ev["E3"].append((pr, entry_outcome(by, C, H, L, m))) # B1 buying within 30 min after a climax at the high if not b1_done and 0 < m - last_climax <= 30 and C[m] >= VW[m] * 1.01: b1_done = True ev["B1"].append((pr, entry_outcome(by, C, H, L, m))) # X2 stretched if C[m] >= VW[m] * 1.03 and C[m] <= C[m - 5]: ev["X2"].append((pr, exit_outcome(C, H, L, m))) # X3 failed new high if C[m] >= o * 1.02 and any(H[k] >= HI[k] - 1e-9 and H[k] > HI[max(0, k - 1)] for k in range(m - 15, m)) \ and C[m] <= HI[m] * 0.995: ev["X3"].append((pr, exit_outcome(C, H, L, m))) if m % 5 == 0: if above: ev["BASE"].append((pr, entry_outcome(by, C, H, L, m))) if C[m] >= o * 1.02: ev["XBASE"].append((pr, exit_outcome(C, H, L, m))) def report(ev): def ent(rows): if not rows: return "n=0" t = sum(r["tb"] == "target" for r in rows) sp = sum(r["tb"] == "stop" for r in rows) return (f"n={len(rows):5} +2% first {t / len(rows) * 100:4.0f}% -1% first {sp / len(rows) * 100:4.0f}% " f"60m {st.mean(r['fwd60'] for r in rows):+5.2f}% close {st.mean(r['close'] for r in rows):+5.2f}%") def ext(rows): if not rows: return "n=0" d = sum(r["tb"] == "down" for r in rows) u = sum(r["tb"] == "up" for r in rows) return (f"n={len(rows):5} -1% first {d / len(rows) * 100:4.0f}% +1% first {u / len(rows) * 100:4.0f}% " f"next 30m {st.mean(r['fwd30'] for r in rows):+5.2f}%") names = {"BASE": "reference: any minute over VWAP", "E1": "E1 base after the low, VWAP reclaim", "E2": "E2 opening dip held the open", "E3": "E3 VWAP pullback held", "B1": "B1 buy within 30 min after a climax", "XBASE": "reference: any minute >= +2% over open", "X1": "X1 volume climax at the high", "X2": "X2 stretched >= 3% over VWAP, stalling", "X3": "X3 failed new high"} for group, fn in ((("BASE", "E1", "E2", "E3", "B1"), ent), (("XBASE", "X1", "X2", "X3"), ext)): print("\nENTRIES (after 0.1% cost)" if fn is ent else "\nEXITS (what happens if you keep holding)") for k in group: print(f" {names[k]}") for pr in ("A", "B", "HOLD"): rows = [r for p, r in ev[k] if p == pr] print(f" {pr:4} {fn(rows)}") if __name__ == "__main__": ev = defaultdict(list) days = set() for n, (day, s, by, ref) in enumerate(load()): days.add(day) scan(day, s, by, ref, ev) per = defaultdict(int) for d in days: per[period(d)] += 1 print(f"{len(days)} days: A {per['A']}, B {per['B']}, HOLDOUT {per['HOLD']}") report(ev)