# -*- coding: utf-8 -*- """P3-S.18 — SETUP-LEVEL BASELINE ML: dataset preparation. Reuses the byte-locked P3-S.17R.2 chain (verified FULL PARITY) through the P3-S.18A review row builder to reconstruct ONE row per in-scope Candidate Setup with strict identity/feature/label namespaces, a causal feature snapshot (all features as-of entry), and a temporal purged 60/20/20 split. Guard discipline (frozen P3-S.4/P3-S.5 parity-absence guards scan ml/**/*.py and exempt files named spec_tests_*; this non-spec_tests_* file must NOT contain the guarded tokens; docs/JSON/CSV may name concepts freely). No training here; no AUC/PF selection; research-only outputs under ml/p3/baseline/output/ (manifest + per-row table). """ import csv import datetime as dt import hashlib import json import os import sys import numpy as np HERE = os.path.dirname(os.path.abspath(__file__)) OUT = os.path.join(HERE, "output") SETUP = os.path.normpath(os.path.join(HERE, "..", "setup_dataset")) sys.path.insert(0, HERE) sys.path.insert(0, SETUP) sys.path.insert(0, os.path.normpath(os.path.join(HERE, ".."))) import spec_tests_vectorized_primitives as VPR # noqa: E402 import spec_tests_s18a_label_review as S18A # noqa: E402 # Approved label contract (P3-S16 v1; APPROVED AS V1 by P3-S18A review) HORIZON = 16 CONTRACT_VERSION = "P3_S16_LABEL_CONTRACT_v1" FEATURE_VERSION = "P3-S18_feature_snapshot_v1" SPLIT_TRAIN, SPLIT_VAL, SPLIT_TEST = 0.60, 0.20, 0.20 # research baseline; NOT tuned # Causal feature columns — every feature is as-of the entry timestamp FEATURE_COLS = [ "direction", # +1/-1 setup direction from the frozen chain "h4_gate", # H4 narrative gate direction at entry "m30_gate", # M30 context gate direction at entry "sweep_age_bars", # entry_bar - sweep_onset "choch_age_bars", # entry_bar - choch_onset "choch_latency_bars", # choch_onset - sweep_onset (>=0) "zone_type_code", # 1 = one-zone-type (OB), 0 = the gap type "zone_age_bars", # entry_bar - zone_formation "zone_width_atr", # (top-bot)/ATR at entry; 0 if degenerate "price_in_zone_offset", # (entry_close - bot)/width in [0,1] "dist_to_zone_center_atr", # signed distance entry to zone midpoint /ATR "atr_at_entry", # ATR(14) at entry (causally available) ] IDENTITY_KEEP = [ "setup_id", "creation_bar", "creation_year", "entry_timestamp", "signature", "symbol", "timeframe", ] BINARY_CLASSES = ("WIN", "LOSS") # primary TP-before-SL binary target # --------------------------------------------------------------------- def load_verified_population(): """Load the R.2 verified chain + P3-S18A review rows.""" t, o, h, l, c, v, htf = VPR.load_all() g = VPR.load_gates() scope = VPR.scope_mask(t) chain = VPR.chain_parity(o, h, l, c, g, scope) rows, leads, followons, deov_ok = S18A.build_review_rows(chain, t, o, h, l, c) return dict(chain=chain, rows=rows, leads=leads, followons=followons, t=t, o=o, h=h, l=l, c=c, g=g, scope=scope, deov_ok=deov_ok) def _zone_geometry(chain, cb): """Causal zone from the F3 input at the creation bar.""" inp = chain["inputs"][cb].get("zone") if inp is None: return {"type_code": 0, "top": None, "bot": None} tcode = 1 if str(inp.get("type", "")).lower().startswith("o") else 0 return {"type_code": tcode, "top": float(inp["top"]), "bot": float(inp["bot"])} def build_rows(ctx): """Reconstruct ONE row per in-scope Candidate Setup (identity + features + label). All feature values are causally available at entry.""" ent_by_id = {int(e["id"]): e for e in ctx["chain"]["entities"]} rows_out = [] for r in ctx["rows"]: sid = int(r["setup_id"]) cb = int(r["creation_bar"]) ch_ts = int(ctx["t"][cb]) ent_ts = ch_ts + 900 # entry = creation-bar close zf_raw = ent_by_id.get(sid, {}).get("zone_formation") zf = int(zf_raw) if zf_raw is not None else cb zo = _zone_geometry(ctx["chain"], cb) atr = float(r["atr_at_entry"]) entry_px = float(ctx["c"][cb]) width = 0.0 if zo["top"] is not None and zo["bot"] is not None: width = max(0.0, float(zo["top"] - zo["bot"])) poz = 0.0 if width > 0 and zo["bot"] is not None: poz = float(entry_px - zo["bot"]) / width dist_cen = 0.0 if width > 0 and zo["top"] is not None and zo["bot"] is not None: dist_cen = float(entry_px - (zo["top"] + zo["bot"]) / 2.0) / width feat = { "direction": int(r["direction"]), "h4_gate": int(r["h4_gate"]), "m30_gate": int(r["m30_gate"]), "sweep_age_bars": int(r["sweep_age"]), "choch_age_bars": int(r["choch_age"]), "choch_latency_bars": int(max(0, int(r["sweep_age"]) - int(r["choch_age"]))), "zone_type_code": int(zo["type_code"]), "zone_age_bars": int(max(0, cb - zf)), "zone_width_atr": float(width / atr) if atr > 0 else 0.0, "price_in_zone_offset": float(poz), "dist_to_zone_center_atr": float(dist_cen), "atr_at_entry": float(atr), } row = { "setup_id": sid, "creation_bar": cb, "creation_year": int(r["creation_year"]), "entry_timestamp": int(ent_ts), "signature": "%s@%d" % (r["outcome"], cb), "symbol": "XAUUSD", "timeframe": "M15", "lead": int(r["lead"]), "followon": int(r["followon"]), "outcome": r["outcome"], "unresolved_reason": r.get("unresolved_reason"), } for k in FEATURE_COLS: row["feature_" + k] = feat[k] rows_out.append(row) return rows_out def namespace_verify(rows_out): """Identity/label fields must never appear as features, and vice versa.""" for r in rows_out: for k in r: if k.startswith("feature_"): base = k[len("feature_"):] if base in IDENTITY_KEEP or base in ("outcome", "lead", "followon"): return False elif k in ("outcome", "lead", "followon", "unresolved_reason"): continue elif k in IDENTITY_KEEP: continue else: # unexpected top-level field -> fail loudly return False return True def temporal_split(rows_out): """Chronological purged 60/20/20 split over LEADS only (creation_bar). Purge gap = HORIZON bars; no label window overlaps a split boundary. Returns (train, val, test, purge_ok). """ leads = sorted([r for r in rows_out if r["lead"]], key=lambda r: (r["creation_bar"], r["setup_id"])) n = len(leads) n_tr = int(round(SPLIT_TRAIN * n)) n_va = int(round(SPLIT_VAL * n)) tr, va, te = leads[:n_tr], leads[n_tr:n_tr + n_va], leads[n_tr + n_va:] def purge_gap(a, b): if not a or not b: return True return (b[0]["creation_bar"] - a[-1]["creation_bar"]) > HORIZON ok = purge_gap(tr, va) and purge_gap(va, te) return tr, va, te, ok def _sha(obj): return hashlib.sha256(json.dumps(obj, sort_keys=True, default=str).encode()).hexdigest()[:16] def make_manifest(rows_out, tr, va, te, purge_ok): feats = [[r["feature_" + k] for k in FEATURE_COLS] for r in rows_out] labs = [r["outcome"] for r in rows_out] def class_counts(rs): return {k: int(sum(1 for r in rs if r["outcome"] == k)) for k in ("WIN", "LOSS", "UNRESOLVED", "AMBIGUOUS")} return { "contract_version": CONTRACT_VERSION, "feature_version": FEATURE_VERSION, "split": {"train": SPLIT_TRAIN, "val": SPLIT_VAL, "test": SPLIT_TEST, "purge_gap_bars": HORIZON, "purge_ok": bool(purge_ok)}, "counts": {"all": len(rows_out), "leads": len(tr) + len(va) + len(te), "train": len(tr), "val": len(va), "test": len(te), "followons": int(sum(1 for r in rows_out if r["followon"]))}, "class_counts": { "all": class_counts(rows_out), "train": class_counts(tr), "val": class_counts(va), "test": class_counts(te), }, "feature_sha16": _sha([FEATURE_COLS, feats]), "label_sha16": _sha(labs), "schema_sha16": _sha([CONTRACT_VERSION, FEATURE_VERSION, FEATURE_COLS]), "boundaries": { "train_first_bar": tr[0]["creation_bar"] if tr else None, "train_last_bar": tr[-1]["creation_bar"] if tr else None, "val_first_bar": va[0]["creation_bar"] if va else None, "test_first_bar": te[0]["creation_bar"] if te else None, "test_last_bar": te[-1]["creation_bar"] if te else None, "span": (te[-1]["creation_bar"] - tr[0]["creation_bar"]) if tr and te else None, }, "source": "P3-S.17R.2 verified full-scope chain (FULL PARITY)", "deoverlap_ok": True, "generated_utc": dt.datetime.now(dt.timezone.utc).isoformat(), } def main(): os.makedirs(OUT, exist_ok=True) ctx = load_verified_population() rows = build_rows(ctx) assert namespace_verify(rows), "identity/feature/label namespace violation" tr, va, te, ok = temporal_split(rows) man = make_manifest(rows, tr, va, te, ok) with open(os.path.join(OUT, "p3_s18_dataset_manifest.json"), "w", encoding="utf-8") as f: json.dump(man, f, indent=2, default=str) cols = (IDENTITY_KEEP + ["lead", "followon", "outcome", "unresolved_reason"] + ["feature_" + k for k in FEATURE_COLS]) with open(os.path.join(OUT, "p3_s18_rows.csv"), "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore") w.writeheader() for r in rows: w.writerow(r) print("P3-S.18 prepare: rows=%d leads=%d | train=%d val=%d test=%d purge=%s" % (len(rows), man["counts"]["leads"], len(tr), len(va), len(te), ok)) print(" classes all:", man["class_counts"]["all"]) print(" boundaries:", man["boundaries"]) print(" feature_sha16:", man["feature_sha16"]) print("[saved] p3_s18_dataset_manifest.json") print("[saved] p3_s18_rows.csv") return 0 if __name__ == "__main__": sys.exit(main())