""" flow_study.py -- [2026-09-28] 22-day test: price move vs net buying (from the 9/28 phase check: IMMX/TGB rose on net SELLING and kept going; SG fell on net BUYING and kept falling; NWL's buying burst into its ceiling marked the top). Every 5 minutes 9:45-15:15, every stock on each day's 9:28 list (bars/ + flow/), looking BACK only: ret_W price change over the last W minutes (W = 15, 30), % imb_W net buying (buy - sell) / (buy + sell) over the same W minutes resid_W price change left after what net buying "should" have moved it: residual of a per-day cross-sectional line ret ~ imb (fitted only on the SAME minute's observations across stocks -- no look-ahead) room distance to the nearest 14-day resistance band above (%), day-high distance Outcomes: next 30 / 60 minutes and to the close, vs every stock at the same minute. """ import gzip import json import statistics as st import sys 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") sys.path.insert(0, "/var/www/screener/trade1/sim/analysis") import reference as REF import volatility from separation import auc ET = ZoneInfo("America/New_York") def load(): obs = [] for p in sorted(HERE.glob("bars/bars_2026-*.json.gz")): d = json.load(gzip.open(p, "rt")) day = d["date"] for s in d["top30_current"]: fp = HERE / "flow" / day / f"{s}.json" ref = REF.load(day, s) if not fp.exists(): continue fl = json.loads(fp.read_text()) bars = [b for b in d["symbols"][s]["bars"] if datetime.fromtimestamp(b[0], ET).strftime("%H:%M") < "16:00"] if len(bars) < 200: continue m_of = {} for b in bars: t = datetime.fromtimestamp(b[0], ET) m_of[(t.hour - 9) * 60 + t.minute - 30] = b # forward-filled closes by minute close, high = [None] * 390, [None] * 390 last = bars[0][1] hi = 0 for m in range(390): if m in m_of: last = m_of[m][4] hi = max(hi, m_of[m][2]) close[m] = last high[m] = hi final = close[389] day_open = bars[0][1] oc = (final / day_open - 1) * 100 for m in range(15, 346, 5): o = {"day": day, "sym": s, "m": m, "oc": oc, "big": oc >= 5, "fwd30": (close[min(m + 30, 389)] / close[m] - 1) * 100, "fwd60": (close[min(m + 60, 389)] / close[m] - 1) * 100, "to_close": (final / close[m] - 1) * 100, "from_open": (close[m] / day_open - 1) * 100, "off_high": (close[m] / high[m] - 1) * 100} for W in (15, 30): if m < W: continue o[f"ret{W}"] = (close[m] / close[m - W] - 1) * 100 bv = sum(fl["buy"][m - W + 1:m + 1]) sv = sum(fl["sell"][m - W + 1:m + 1]) o[f"imb{W}"] = (bv - sv) / (bv + sv) if bv + sv else None o[f"netbuy{W}"] = bv - sv bv5, sv5 = sum(fl["buy"][m - 4:m + 1]), sum(fl["sell"][m - 4:m + 1]) o["imb5"] = (bv5 - sv5) / (bv5 + sv5) if bv5 + sv5 else None if ref and ref.get("sr14"): sp = volatility.sr_position(ref["sr14"], close[m]) o["room"] = sp.get("dist_to_resistance_pct") else: o["room"] = None obs.append(o) return obs def add_residuals(obs): """Per (day, minute), across stocks: fit ret_W = a + b * imb_W, keep the residual.""" groups = {} for o in obs: groups.setdefault((o["day"], o["m"]), []).append(o) for W in (15, 30): for g in groups.values(): pts = [(o[f"imb{W}"], o[f"ret{W}"]) for o in g if o.get(f"imb{W}") is not None and o.get(f"ret{W}") is not None] if len(pts) < 8: continue xm = sum(x for x, _ in pts) / len(pts) ym = sum(y for _, y in pts) / len(pts) sxx = sum((x - xm) ** 2 for x, _ in pts) b = sum((x - xm) * (y - ym) for x, y in pts) / sxx if sxx else 0 for o in g: if o.get(f"imb{W}") is not None and o.get(f"ret{W}") is not None: o[f"resid{W}"] = o[f"ret{W}"] - (ym + b * (o[f"imb{W}"] - xm)) def report(obs): days = sorted({o["day"] for o in obs}) h1 = set(days[:len(days) // 2]) base = {} for o in obs: base.setdefault(o["m"], {"fwd30": [], "fwd60": [], "to_close": []}) for k in ("fwd30", "fwd60", "to_close"): base[o["m"]][k].append(o[k]) bm = {m: {k: st.mean(v) for k, v in d.items()} for m, d in base.items()} print(f"{len(obs)} observations ({len({(o['day'], o['sym']) for o in obs})} stock-days, {len(days)} days)") # how strongly price follows net buying overall for W in (15, 30): pts = [(o[f"imb{W}"], o[f"ret{W}"]) for o in obs if o.get(f"imb{W}") is not None] xs, ys = [p[0] for p in pts], [p[1] for p in pts] mx, my = st.mean(xs), st.mean(ys) cor = sum((x - mx) * (y - my) for x, y in pts) / (sum((x - mx) ** 2 for x in xs) * sum((y - my) ** 2 for y in ys)) ** .5 print(f" correlation, price change vs net buying over the same {W} min: {cor:.2f}") print("\n--- Does each measure predict what comes next? (AUC: next-30 / next-60 / to-close rising; 1st/2nd half for next-60)") for f in ("ret15", "imb15", "resid15", "ret30", "imb30", "resid30", "imb5", "room"): row = [] for k in ("fwd30", "fwd60", "to_close"): up = [o[f] for o in obs if o.get(f) is not None and o[k] > 0] dn = [o[f] for o in obs if o.get(f) is not None and o[k] <= 0] row.append(auc(up, dn)) a1 = auc([o[f] for o in obs if o["day"] in h1 and o.get(f) is not None and o["fwd60"] > 0], [o[f] for o in obs if o["day"] in h1 and o.get(f) is not None and o["fwd60"] <= 0]) a2 = auc([o[f] for o in obs if o["day"] not in h1 and o.get(f) is not None and o["fwd60"] > 0], [o[f] for o in obs if o["day"] not in h1 and o.get(f) is not None and o["fwd60"] <= 0]) flag = " <- consistent" if (a1 - .5) * (a2 - .5) > 0 and min(abs(a1 - .5), abs(a2 - .5)) >= .03 else "" print(f" {f:8} {row[0]:.3f} / {row[1]:.3f} / {row[2]:.3f} halves {a1:.3f} / {a2:.3f}{flag}") pats = { "QUIET DEMAND: up >= 1% in 30 min on net selling (imb30 <= 0)": lambda o: o.get("ret30", 0) >= 1 and o.get("imb30") is not None and o["imb30"] <= 0, "HEAVY BUYING + UP: up >= 1% on net buying >= +0.15": lambda o: o.get("ret30", 0) >= 1 and (o.get("imb30") or 0) >= 0.15, "ABSORBED BUYING: down >= 1% on net buying >= 0": lambda o: o.get("ret30", 0) <= -1 and o.get("imb30") is not None and o["imb30"] >= 0, "NORMAL SELLING: down >= 1% on net selling <= -0.15": lambda o: o.get("ret30", 0) <= -1 and (o.get("imb30") or 0) <= -0.15, "UP MORE THAN BUYING EXPLAINS (resid30 >= +1%)": lambda o: (o.get("resid30") or 0) >= 1, "DOWN MORE THAN SELLING EXPLAINS (resid30 <= -1%)": lambda o: (o.get("resid30") or 0) <= -1, "BUYING BURST (imb5 >= +0.5) within 1% of 14-day resistance": lambda o: (o.get("imb5") or 0) >= 0.5 and o.get("room") is not None and o["room"] < 1, "BUYING BURST (imb5 >= +0.5) at the day's high (within 0.5%)": lambda o: (o.get("imb5") or 0) >= 0.5 and o["off_high"] >= -0.5, } print(f"\n--- Patterns (every 5-min observation) vs all stocks at the same minute") print(f" {'pattern':62} {'n':>5} {'next 30':>8} {'next 60':>8} {'to close':>9} {'60m 1st/2nd half':>17} {'rose 60m':>8}") for name, fn in pats.items(): ev = [o for o in obs if fn(o)] if len(ev) < 10: print(f" {name:62} {len(ev):5}") continue ex = {k: st.mean(o[k] - bm[o["m"]][k] for o in ev) for k in ("fwd30", "fwd60", "to_close")} e1 = [o["fwd60"] - bm[o["m"]]["fwd60"] for o in ev if o["day"] in h1] e2 = [o["fwd60"] - bm[o["m"]]["fwd60"] for o in ev if o["day"] not in h1] print(f" {name:62} {len(ev):5} {ex['fwd30']:+7.2f}% {ex['fwd60']:+7.2f}% {ex['to_close']:+8.2f}% " f"{(st.mean(e1) if e1 else 0):+7.2f}/{(st.mean(e2) if e2 else 0):+.2f}% {sum(o['fwd60'] > 0 for o in ev) / len(ev) * 100:7.0f}%") if __name__ == "__main__": obs = load() add_residuals(obs) report(obs)