SniperGold_ML/ml/p3/baseline/spec_tests_baseline_ml.py

# -*- coding: utf-8 -*-
"""P3-S.18 — SETUP-LEVEL BASELINE ML: deterministic spec tests (ML-T01..T08).

Truth      : docs/P3_S16_SETUP_DATASET_CONTRACT_v1.md (observation unit,
             identity/feature/label separation, temporal purged split) +
             docs/P3_S16_LABEL_CONTRACT.md (primary TP-before-SL).
Observed   : ml/p3/baseline/prepare_dataset.py (dataset reconstruction).
Discipline : deterministic tests, no training, no AUC/PF, no tuning.

Filename is spec_tests_* (guard-exempt) per the frozen P3-S.4/S.5 parity-
absence policy.
"""
import json
import os
import sys

import numpy as np

HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(HERE, "output")
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.normpath(os.path.join(HERE, "..", "setup_dataset")))

import prepare_dataset as PD          # noqa: E402

SEED = 42


def load_ctx():
    return PD.load_verified_population()


def ml_t01():
    """ONE Candidate Setup = ONE row; no per-bar duplication."""
    ctx = load_ctx()
    rows = PD.build_rows(ctx)
    ids = [r["setup_id"] for r in rows]
    ok = len(ids) == len(set(ids))
    return ok, {"rows": len(rows), "unique": len(set(ids))}


def ml_t02():
    """No duplicate setup_id across the whole dataset."""
    ctx = load_ctx()
    rows = PD.build_rows(ctx)
    dup = len(rows) - len({r["setup_id"] for r in rows})
    return dup == 0, {"duplicates": dup}


def ml_t03():
    """No target leakage: label fields are not features; feature snapshot
    holds only causal values (no post-entry / outcome-derived fields)."""
    ctx = load_ctx()
    rows = PD.build_rows(ctx)
    ok = PD.namespace_verify(rows)
    return ok, {"namespace_ok": ok}


def ml_t04():
    """Preprocessing (split + feature table) is deterministic; no random
    shuffle across time: chronological creation_bar ordering in each split."""
    ctx = load_ctx()
    rows = PD.build_rows(ctx)
    tr, va, te, ok = PD.temporal_split(rows)
    chrono = all(tr[i]["creation_bar"] <= tr[i + 1]["creation_bar"]
                 for i in range(len(tr) - 1))
    ok2 = chrono and ok
    return ok2, {"chronological_train": chrono, "purge_ok": ok}


def ml_t05():
    """Temporal ordering preserved across train/val/test (sequential)."""
    ctx = load_ctx()
    rows = PD.build_rows(ctx)
    tr, va, te, _ = PD.temporal_split(rows)
    if not (tr and va and te):
        return False, {"error": "empty split"}
    a = tr[-1]["creation_bar"]
    b = va[0]["creation_bar"]
    c = va[-1]["creation_bar"]
    d = te[0]["creation_bar"]
    ok = a < b and c < d
    return ok, {"boundaries": (a, b, c, d)}


def ml_t06():
    """Purge gap enforced: train|val and val|test gap > HORIZON."""
    ctx = load_ctx()
    rows = PD.build_rows(ctx)
    tr, va, te, ok = PD.temporal_split(rows)
    g1 = va[0]["creation_bar"] - tr[-1]["creation_bar"] if tr and va else None
    g2 = te[0]["creation_bar"] - va[-1]["creation_bar"] if va and te else None
    ok2 = ok and (g1 is None or g1 > PD.HORIZON) and \
        (g2 is None or g2 > PD.HORIZON)
    return bool(ok2), {"gap1": g1, "gap2": g2, "horizon": PD.HORIZON}


def ml_t07():
    """Same seed / same inputs -> identical manifest and hashes."""
    ctx = load_ctx()
    rows = PD.build_rows(ctx)
    tr1, va1, te1, ok1 = PD.temporal_split(rows)
    tr2, va2, te2, ok2 = PD.temporal_split(rows)
    m1 = PD.make_manifest(rows, tr1, va1, te1, ok1)
    m2 = PD.make_manifest(rows, tr2, va2, te2, ok2)
    same = (m1["feature_sha16"] == m2["feature_sha16"] and
            m1["label_sha16"] == m2["label_sha16"] and
            m1["counts"] == m2["counts"])
    return bool(same), {"sha": m1["feature_sha16"]}


def ml_t08():
    """Provenance preserved per row (identity kept, not a feature)."""
    ctx = load_ctx()
    rows = PD.build_rows(ctx)
    keep = all(r.get("setup_id") is not None and r.get("entry_timestamp")
               is not None and r.get("symbol") == "XAUUSD" for r in rows)
    no_leak = all("feature_setup_id" not in r and
                  "feature_entry_timestamp" not in r for r in rows)
    return bool(keep and no_leak), {"provenance_ok": keep,
                                    "feature_leak": not no_leak}


def main():
    tests = [
        ("ML-T01", "one setup = one row", ml_t01),
        ("ML-T02", "no duplicate setup ID", ml_t02),
        ("ML-T03", "no target leakage", ml_t03),
        ("ML-T04", "preprocessing deterministic / train only", ml_t04),
        ("ML-T05", "temporal ordering preserved", ml_t05),
        ("ML-T06", "purge gap enforced", ml_t06),
        ("ML-T07", "same seed = same result", ml_t07),
        ("ML-T08", "provenance preserved", ml_t08),
    ]
    results = []
    for tid, title, fn in tests:
        try:
            ok, detail = fn()
        except Exception as e:                            # noqa: BLE001
            ok, detail = False, {"error": repr(e)}
        results.append({"id": tid, "title": title, "pass": bool(ok),
                        "detail": detail})
        print("  [%s] %s %s" % ("PASS" if ok else "FAIL", tid, title))
    n_pass = sum(1 for r in results if r["pass"])
    print("TOTAL=%d PASS=%d FAIL=%d" % (len(results), n_pass,
                                        len(results) - n_pass))
    with open(os.path.join(OUT, "p3_s18_ml_tests.json"), "w",
              encoding="utf-8") as f:
        json.dump({"tests": results, "total": len(results),
                   "passed": n_pass}, f, indent=2)
    return 0 if n_pass == len(results) else 1


if __name__ == "__main__":
    sys.exit(main())