forked from chiki2bum2/SniperGold_ML
258 lines
No EOL
10 KiB
Python
258 lines
No EOL
10 KiB
Python
# -*- 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()) |