""" star_follow.py -- [2026-10-01] "STAR FOLLOW", the user's playbook-reference method (agreed 2026-10-01 on ARHS). Follows the marked best buys/sells (the stars) of the closest past days. 1 Library: every stock-day in reports/open_window/bars with its marked best trade(s) (data/playbook/day_labels.jsonl: best holdable buy/sell where a 1% stop is never hit, plus a second move). Only days BEFORE the traded day are used. 2 Match: price vs open and price vs VWAP (both / the stock's daily ATR%) and 0.5*log(1 + volume vs its 14-day normal), minute by minute from 9:30 to NOW, averaged over the whole day so far. 3 Every minute the 150 closest days are found again (references drop out / come in). 4 Not holding, from 9:35 to 15:15: BUY at the next minute's open when >= `trigger` of the current references have a marked best buy within the last `window` minutes. 5 Holding: SELL at the next minute's open when >= `trigger` references that were already in their marked trade (marked buy earlier than the window) have a marked best sell within the last `window` minutes; safety stop `stop_pct` below the buy; 15:55 sell-all. 6 After a sell, back to 4. Cost 0.1% per trade. python3 star_follow.py 2026-10-01 CXM [--trigger 10 --window 5 --k 150 --stop 1.0] """ import argparse import json import sys from datetime import datetime from pathlib import Path from zoneinfo import ZoneInfo import numpy as np HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)); sys.path.insert(0, "/var/www/screener/trade") import reference_trader as RT import label_days as LD ET = ZoneInfo("America/New_York") def hm(m): return f"{9 + (30 + m) // 60}:{(30 + m) % 60:02d}" def load_marks(): marks = {} for l in map(json.loads, open(LD.OUT)): marks[f"{l['date']}|{l['symbol']}"] = [(l[k]["entry_m"], l[k]["exit_m"]) for k in ("best_held", "leg2") if l.get(k)] return marks def day_bars(day, sym): import gzip p = RT.BARS / f"bars_{day}.json.gz" if p.exists(): d = json.load(gzip.open(p, "rt")) if sym in d["symbols"]: return d["symbols"][sym]["bars"] from alpaca_client import get_client d0 = datetime.fromisoformat(day).replace(tzinfo=ET) raw = get_client().get_minute_bars(sym, start=d0.replace(hour=9, minute=30), end=d0.replace(hour=16), limit=1000) return [[b.timestamp, float(b.open), float(b.high), float(b.low), float(b.close), float(b.volume)] for b in raw] def star_follow(lib, marks, day, a, k=150, trigger=10, window=5, stop_pct=1.0, start_m=5, last_buy_m=345, cost=0.1, verbose=True, can_buy=None): lib.start(day) pos, prev, trades, log = None, None, [], [] for m in range(start_m, 385): idx, _ = lib.match(a["F"], m, day, k) cur = [lib.keys[i] for i in idx] S = set(cur) kept = len(S & prev) if prev else k prev = S w0 = m - window + 1 buys = sum(1 for key in cur for (bm, sm) in marks.get(key, []) if w0 <= bm <= m) sells = sum(1 for key in cur for (bm, sm) in marks.get(key, []) if bm < w0 and w0 <= sm <= m) row = {"m": m, "t": hm(m), "price": float(a["C"][m]), "kept": kept, "buys": buys, "sells": sells, "holding": pos is not None, "event": ""} if pos is None: if buys >= trigger and m <= last_buy_m and (can_buy is None or can_buy(m)): e = float(a["O"][m + 1]) pos = {"buy_m": m + 1, "buy": e, "stop": e * (1 - stop_pct / 100), "signal_m": m, "signal_n": buys} row["event"] = f"BUY signal ({buys} reference buys) -> buy {hm(m + 1)} at {e:.3f}" else: if a["L"][m] <= pos["stop"]: trades.append({**pos, "sell_m": m, "sell": pos["stop"], "why": f"{stop_pct}% stop"}) row["event"] = f"STOP at {pos['stop']:.3f}" pos = None elif sells >= trigger: s_ = float(a["O"][m + 1]) trades.append({**pos, "sell_m": m + 1, "sell": s_, "why": f"references exit ({sells})"}) row["event"] = f"EXIT signal ({sells} reference sells) -> sell {hm(m + 1)} at {s_:.3f}" pos = None log.append(row) if pos: trades.append({**pos, "sell_m": 385, "sell": float(a["C"][385]), "why": "15:55"}) for t in trades: t["pl_pct"] = round((t["sell"] / t["buy"] - 1) * 100 - cost, 2) return trades, log def main(): ap = argparse.ArgumentParser() ap.add_argument("day"); ap.add_argument("symbol") ap.add_argument("--k", type=int, default=150); ap.add_argument("--trigger", type=int, default=10) ap.add_argument("--window", type=int, default=5); ap.add_argument("--stop", type=float, default=1.0) args = ap.parse_args() ref = RT.REF.load(args.day, args.symbol) a = RT.day_arrays(day_bars(args.day, args.symbol), ref) lib = RT.Lib() marks = load_marks() trades, log = star_follow(lib, marks, args.day, a, args.k, args.trigger, args.window, args.stop) lab = LD.label(args.day, args.symbol, LD.build([[int(x.timestamp()) if isinstance(x, datetime) else x, *r[1:]] for r in day_bars(args.day, args.symbol) for x in [r[0]]], ref)) print(f"STAR FOLLOW {args.symbol} {args.day} (k={args.k}, trigger={args.trigger} in {args.window} min, stop {args.stop}%)") print(f"{'time':>5} {'price':>7} {'state':>5} {'kept':>4} {'ref buys':>8} {'ref sells':>9} event") for r in log: if r["event"] or r["m"] % 15 == 0: print(f"{r['t']:>5} {r['price']:7.3f} {'HOLD' if r['holding'] else 'flat':>5} {r['kept']:4} {r['buys']:8} {r['sells']:9} {r['event']}") print("\nTRADES (0.1% cost each):") for t in trades: print(f" buy {hm(t['buy_m'])} {t['buy']:.3f} -> sell {hm(t['sell_m'])} {t['sell']:.3f} {t['pl_pct']:+.2f}% [{t['why']}]") print(f" total {sum(t['pl_pct'] for t in trades):+.2f}%") bh = lab.get("best_held") if bh: print(f"marked best trade of the day: buy {hm(bh['entry_m'])} {bh['entry']['price']:.3f} -> sell {hm(bh['exit_m'])} " f"{bh['exit']['price']:.3f} +{bh['gain_pct']:.2f}%") json.dump({"trades": trades, "log": log}, open(HERE / f"starfollow_{args.symbol}_{args.day}.json", "w"), indent=0) if __name__ == "__main__": main()