""" playbook.py -- [2026-09-29] Situation playbook (the user's plan: group each stock by its behavior pattern, find where winners and losers of each group differ, and tell the bot what to do in each situation). Groups and rules from reports/open_window/group_study.py (23 days, 676 stock-days, IN-SAMPLE -- see that script's output before trusting them). Every stock gets ONE group from the shape of its first 45 minutes (decided on the 10:15 bar, i.e. from ~10:16 ET), then a verdict for the day, fixed at the minute the study tested it: 1 OPENING DRIVE (>= +2% over the open, above VWAP, within 1% of the high at 10:15) ALLOW if at 10:20 it's still within near_high_pct of the day high, else AVOID (the drive is over) 2 UP, PULLED BACK (>= +1% over the open, above VWAP, > 1% off the high) ALLOW if at 10:15 steady (R2 > g2_min_r2) AND normal volume (RVOL < g2_max_rvol), else AVOID 3 V RECOVERY (fell >= 2% under the open, back above VWAP) ALLOW if at 10:15 RVOL < g3_max_rvol AND last-15-min volume < g3_max_vol15x its normal, else AVOID (heavy-volume V = fade) 4 SPIKE & FADE (first-30-min high >= +2%, now below VWAP) AVOID all day 5 SELLING SLIDE (<= -2% under the open, below VWAP) AVOID all day 6 WEAK DRIFT (below VWAP, the rest) AVOID all day 7 FLAT / CHOP (above VWAP, the rest) AVOID all day Before the verdict minute: WAIT (before_group: "wait"; "allow" lets v2 buy as before). ALLOW only means v2 may buy -- v2's own rules still decide when. classify() is pure (bars + 14-day reference in, verdict out); monitor.py and simulate.py call the same function and cache the day's verdict per symbol. """ from zoneinfo import ZoneInfo from entry_v2 import _r2_eff ET = ZoneInfo("America/New_York") DEFAULTS = { "enabled": False, "mode": "control", # "control": AVOID/WAIT block v2 buys; "record": log only "before_group": "wait", # before the verdict minute: "wait" (block) or "allow" "group_minute": 45, # 10:15 bar "g1_minute": 50, # 10:20 bar "near_high_pct": 0.66, "g2_min_r2": 0.28, "g2_max_rvol": 1.48, "g3_max_rvol": 2.96, "g3_max_vol15x": 2.51, } GROUPS = {1: "OPENING DRIVE", 2: "UP, PULLED BACK", 3: "V RECOVERY", 4: "SPIKE & FADE", 5: "SELLING SLIDE", 6: "WEAK DRIFT", 7: "FLAT / CHOP"} def _idx(ts) -> int: t = ts.astimezone(ET) return (t.hour - 9) * 60 + t.minute - 30 def _state_at(bars, ref, m): """Measures on the bar of minute m (forward-filled), bars = finished 1-min bars.""" upto = [b for b in bars if 0 <= _idx(b["t"]) <= m] if not upto: return None o = upto[0]["o"] closes, last, j = [], upto[0]["o"], 0 by = {_idx(b["t"]): b for b in upto} for k in range(m + 1): if k in by: last = by[k]["c"] closes.append(last) c = closes[m] hi = max(b["h"] for b in upto) lo = min(b["l"] for b in upto) vol = sum(b["v"] for b in upto) pv = sum((b["h"] + b["l"] + b["c"]) / 3 * b["v"] for b in upto) vwap = pv / vol if vol else c first30 = [b["h"] for b in upto if _idx(b["t"]) < 30] s = {"minute": m, "price": c, "open": o, "ret_open": (c / o - 1) * 100, "vwap_pct": (c / vwap - 1) * 100, "off_high": (c / hi - 1) * 100, "low_pct": (lo / o - 1) * 100, "first30_hi_pct": ((max(first30) / o - 1) * 100) if first30 else 0.0} r2, eff = _r2_eff(closes[m - 14:m + 1]) s.update(r2=r2, eff=eff) if ref: s["rvol"] = vol / max(ref["cum_vol"][m], 1) norm15 = sum(ref["vol_per_min"][m - 14:m + 1]) v15 = sum(b["v"] for b in upto if m - 14 <= _idx(b["t"]) <= m) s["vol15x"] = v15 / norm15 if norm15 else None return {k: (round(v, 4) if isinstance(v, float) else v) for k, v in s.items()} def _group(s): if s["ret_open"] >= 2 and s["vwap_pct"] >= 0 and s["off_high"] >= -1: return 1 if s["ret_open"] >= 1 and s["vwap_pct"] >= 0: return 2 if s["low_pct"] <= -2 and s["vwap_pct"] >= 0: return 3 if s["first30_hi_pct"] >= 2 and s["vwap_pct"] < 0: return 4 if s["ret_open"] <= -2 and s["vwap_pct"] < 0: return 5 if s["vwap_pct"] < 0: return 6 return 7 def classify(bars, ref, cfg=None): """bars: today's 1-min bars ({"t","o","h","l","c","v"}), the forming one may be last. Returns {"action": ALLOW|AVOID|WAIT, "final": bool, "group", "group_name", "reason", "at_1015", "at_1020"}; final=True means the verdict holds for the rest of the day.""" c = {**DEFAULTS, **(cfg or {})} done = bars[:-1] if bars else [] last = _idx(done[-1]["t"]) if done else -1 G, G1 = c["group_minute"], c["g1_minute"] if last < G: return {"action": "WAIT", "final": False, "group": None, "group_name": None, "reason": f"playbook: group decided on the 10:15 bar ({max(last + 1, 0)} of {G + 1} bars)"} if not ref: return {"action": "AVOID", "final": True, "group": None, "group_name": None, "reason": "playbook: no 14-day reference file"} s = _state_at(done, ref, G) g = _group(s) out = {"group": g, "group_name": GROUPS[g], "at_1015": s, "final": True} if g == 1: if last < G1: out.update(action="WAIT", final=False, reason="playbook: OPENING DRIVE -- checking the high on the 10:20 bar") return out s2 = _state_at(done, ref, G1) out["at_1020"] = s2 if s2["off_high"] >= -c["near_high_pct"]: out.update(action="ALLOW", reason=f"playbook: OPENING DRIVE still at its high at 10:20 ({s2['off_high']:+.2f}%)") else: out.update(action="AVOID", reason=f"playbook: OPENING DRIVE over -- {s2['off_high']:+.2f}% off the high at 10:20") elif g == 2: rv = s.get("rvol") if s["r2"] > c["g2_min_r2"] and rv is not None and rv < c["g2_max_rvol"]: out.update(action="ALLOW", reason=f"playbook: UP, PULLED BACK, steady (R2 {s['r2']:.2f}) on normal volume (RVOL {rv:.2f})") else: why = [] if s["r2"] <= c["g2_min_r2"]: why.append(f"not steady (R2 {s['r2']:.2f})") if rv is None or rv >= c["g2_max_rvol"]: why.append(f"heavy volume (RVOL {rv:.2f})" if rv is not None else "no RVOL") out.update(action="AVOID", reason="playbook: UP, PULLED BACK -- " + ", ".join(why)) elif g == 3: rv, vx = s.get("rvol"), s.get("vol15x") if rv is not None and rv < c["g3_max_rvol"] and vx is not None and vx < c["g3_max_vol15x"]: out.update(action="ALLOW", reason=f"playbook: V RECOVERY on normal volume (RVOL {rv:.2f}, 15-min {vx:.2f}x)") else: out.update(action="AVOID", reason=f"playbook: V RECOVERY on heavy volume (RVOL {rv}, 15-min {vx}x)") else: out.update(action="AVOID", reason=f"playbook: {GROUPS[g]} -- avoid all day") return out class Playbook: """Per-day cache of final verdicts (a verdict never changes once final).""" def __init__(self, cfg=None): self.cfg = {**DEFAULTS, **(cfg or {})} self._final = {} def check(self, symbol, bars, ref): v = self._final.get(symbol) if v is None: v = classify(bars, ref, self.cfg) if v.get("final"): self._final[symbol] = v return v def blocks(self, verdict) -> bool: """True when this verdict must stop a v2 buy (control mode only).""" if not self.cfg.get("enabled") or self.cfg.get("mode") != "control": return False if verdict["action"] == "WAIT" and verdict.get("group") is None and self.cfg.get("before_group") == "allow": return False return verdict["action"] != "ALLOW"