""" review_marks.py -- [2026-10-02] The user: review the whole playbook and confirm the buy/sell marks are in the correct locations; make sure no buy marks sit on a downward slope. For every confirmed mark (day_labels_confirmed.jsonl: buy + leg2): BUY checks held no lower low between the swing low and the buy (by construction) target reached max(2R, +2%) before the stop (by construction) DOWN on a downward slope at the buy = price below VWAP AND VWAP lower than 30 min earlier AND the 15 closes before the buy slope down (linear fit) down15 only the 15-close slope is down SELL checks peak the sell is the highest close between the buy and the sell (by construction) later within 30 min after the sell, a close > 1% above the sell while the low stayed above the stop (= the sell was early) Prints the counts and draws examples of DOWN-flagged buys. """ import gzip import json import random import sys from datetime import datetime, timedelta from pathlib import Path import numpy as np import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import matplotlib.dates as md HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)); sys.path.insert(0, "/var/www/screener/trade") import label_days as LD import reference as REF LABELS = Path("/var/www/screener/trade/data/playbook/day_labels_confirmed.jsonl") def slope(y): x = np.arange(len(y)); return float(np.polyfit(x, y, 1)[0]) if len(y) >= 3 else 0.0 def main(): marks = {(l["date"], l["symbol"]): l for l in map(json.loads, open(LABELS))} stats = {"buys": 0, "held_bad": 0, "target_bad": 0, "DOWN": 0, "down15": 0, "below_vwap": 0, "sells": 0, "peak_bad": 0, "later": 0} flagged, kept = [], [] for p in sorted(LD.BARS.glob("bars_20*.json.gz")): d = json.load(gzip.open(p, "rt")); day = d["date"] for s in d["top30_current"]: mk = marks.get((day, s)) if not mk or not mk.get("buy"): continue ref = REF.load(day, s) if not ref: continue B = LD.build(d["symbols"][s]["bars"], ref) by = {x["m"]: i for i, x in enumerate(B)} for key in ("buy", "leg2"): t = mk.get(key) if not t or t["entry_m"] not in by or t["low_m"] not in by or t["exit_m"] not in by: continue i_low, c, j = by[t["low_m"]], by[t["entry_m"]], by[t["exit_m"]] stats["buys"] += 1; stats["sells"] += 1 L = B[i_low]["l"] if any(B[k]["l"] < L for k in range(i_low + 1, c + 1)): stats["held_bad"] += 1 e, stop = B[c]["c"], t["stop"] need = e + max(2 * (e - stop), e * 0.02) if not any(B[k]["h"] >= need for k in range(c + 1, j + 1)): # target may be hit after the best close; check up to the stop hit = False for k in range(c + 1, len(B)): if B[k]["l"] <= stop: break if B[k]["h"] >= need: hit = True; break if not hit: stats["target_bad"] += 1 closes = [B[k]["c"] for k in range(max(0, c - 14), c + 1)] s15 = slope(closes) < 0 below = B[c]["c"] < B[c]["vwap"] vfall = c >= 30 and B[c]["vwap"] < B[c - 30]["vwap"] down = below and vfall and s15 stats["down15"] += s15; stats["below_vwap"] += below; stats["DOWN"] += down (flagged if down else kept).append((day, s, key, t)) if max(B[k]["c"] for k in range(c, j + 1)) > B[j]["c"] + 1e-9: stats["peak_bad"] += 1 later = False for k in range(j + 1, min(j + 31, len(B))): if B[k]["l"] <= stop: break if B[k]["c"] > B[j]["c"] * 1.01: later = True; break stats["later"] += later REF._CACHE.clear() n = stats["buys"] print(f"marks reviewed: {n} buys / {stats['sells']} sells on {len([1 for v in marks.values() if v.get('buy')])} stock-days") print(f"BUY low not held before the buy: {stats['held_bad']} target not reached: {stats['target_bad']}") print(f"BUY on a DOWNWARD slope (below VWAP + VWAP falling 30 min + 15-close slope down): {stats['DOWN']} ({stats['DOWN'] / n:.1%})") print(f" (only the 15-close slope down: {stats['down15']} ({stats['down15'] / n:.1%}); below VWAP at the buy: {stats['below_vwap']} ({stats['below_vwap'] / n:.1%}))") print(f"SELL not the highest close of the trade: {stats['peak_bad']} price went >1% higher within 30 min (stop not hit): {stats['later']} ({stats['later'] / n:.1%})") json.dump({"stats": stats, "flagged": [(a, b, c) for a, b, c, _ in flagged]}, open(HERE / "review_marks.json", "w")) random.seed(7) for n_, (day, s, key, t) in enumerate(random.sample(flagged, min(6, len(flagged))), 1): d = json.load(gzip.open(LD.BARS / f"bars_{day}.json.gz", "rt")); B = LD.build(d["symbols"][s]["bars"], REF.load(day, s)) T = lambda m: datetime.fromisoformat(day).replace(hour=9, minute=30) + timedelta(minutes=m) ts = [x["t"].replace(tzinfo=None) for x in B] fig, ax = plt.subplots(figsize=(10.5, 4.2)) ax.plot(ts, [x["c"] for x in B], color="#1f3a5f", lw=1); ax.plot(ts, [x["vwap"] for x in B], color="#c0392b", lw=1) ax.scatter([T(t["entry_m"])], [t["entry"]], marker="*", s=200, color="#e67e22", edgecolor="black", zorder=6) ax.scatter([T(t["exit_m"])], [t["exit"]], marker="*", s=200, color="#8e44ad", edgecolor="black", zorder=6) ax.hlines(t["stop"], T(t["low_m"]), T(t["entry_m"] + 10), color="#e74c3c", ls="--", lw=1) ax.annotate(f"FLAGGED buy {t['entry']:.2f}", (T(t["entry_m"]), t["entry"]), xytext=(5, -15), textcoords="offset points", fontsize=8, color="#e67e22", fontweight="bold") ax.annotate(f"sell {t['exit']:.2f} +{t['gain_pct']:.1f}%", (T(t["exit_m"]), t["exit"]), xytext=(5, 7), textcoords="offset points", fontsize=8, color="#8e44ad") ax.set_title(f"{s} {day} -- buy mark on a downward slope (below a falling VWAP, 15-min slope down)", fontsize=9) ax.xaxis.set_major_formatter(md.DateFormatter("%H:%M")); ax.grid(alpha=.3); plt.tight_layout() plt.savefig(HERE / f"flagged_down_{n_}_{s}_{day}.png", dpi=105); plt.close() if __name__ == "__main__": main()