SniperGold_ML/ml/p3/setup_dataset/spec_tests_setup_dataset.py

# -*- coding: utf-8 -*-
"""P3-S.16 SPEC TESTS — SETUP-LEVEL DATASET & LABEL CONTRACT.

Truth        : docs/P3_S16_SETUP_DATASET_CONTRACT_v1.md (canonical observation
               unit, identity, entry, de-overlap, separation) and
               docs/P3_S16_LABEL_CONTRACT.md (TP-before-SL outcome).
               -> spec_oracle() in this file (the contract truth).
Code under   : ml/p3/setup_dataset/setup_dataset_contract.py
audit          -> implementation_port() = the dataset/label generation port.

Discipline P3-S.16 (test-first):
  - spec oracle vs expected        : ASSERT (frozen contract = truth)
  - implementation port vs oracle  : ASSERT (generator must match the contract)
  - no look-ahead / as-of          : ASSERT (features/label causally exact)
  - no AUC/PF / backtest / optimization / ML / human annotation in this file.

Cases: DS-T01..T16 (all required by the brief §27) + additional deterministic
cases mandated by the label contract (label classes, de-overlap, symmetry).

OPEN NUMERIC PARAMETERS (documented, NOT selected from performance):
  k_tp/k_sl = 1.5/0.75 (asymmetric candidate), robustness 1.0/1.0.
  H = 16 (primary), robustness 8. These are contract OPEN parameters, not
  optimized here.

Usage: python spec_tests_setup_dataset.py
Output: output/p3_s16_dataset_report.json
"""
import datetime as dt
import json
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(HERE, "output")
if HERE not in sys.path:
    sys.path.insert(0, HERE)

from setup_dataset_contract import (  # noqa: E402  (the implementation port)
    WIN, LOSS, UNRESOLVED, AMBIGUOUS,
    DEFAULT_K_TP, DEFAULT_K_SL, DEFAULT_H, ROBUST_H,
    build_dataset, build_observation, resolve_outcome,
    entry_timestamp_of, deoverlap_setups,
)

EPS = 1e-9


# =====================================================================
# SYNTHETIC PRICE / SETUP HELPERS
# =====================================================================
def price_path(close, high=None, low=None, open_=None, base_open=1000):
    """Build a price dict over `close` given high/low/open arrays (defaults
    to close +/- small wick). bar_open_time[i] = base_open + i*900 (s)."""
    n = len(close)
    high = high if high is not None else [c + 0.05 for c in close]
    low = low if low is not None else [c - 0.05 for c in close]
    open_ = open_ if open_ is not None else [c for c in close]
    times = [base_open + i * 900 for i in range(n)]
    return {"close": list(close), "high": high, "low": low, "open": open_,
            "bar_open_time": times}


def monotonic(n, step, base=5000.0, wick=0.5, atr=5.0):
    """A synthetic monotonic path (used to guarantee the requested hit) with a
    reference ATR ~ 'atr'. Returns a price path whose closes ramp by `step`."""
    closes = [base + i * step for i in range(n)]
    # wick ± around close
    high = [c + wick for c in closes]
    low = [c - wick for c in closes]
    return {"close": closes, "high": high, "low": low,
            "open": list(closes), "bar_open_time": [1000 + i * 900
                                                    for i in range(n)]}


def make_setup(setup_id, creation_bar, direction=1, state="CANDIDATE_SETUP",
               sweep_onset=None, choch_onset=None, zone_type="OB",
               zone_formation=None, m3_bar=None):
    if sweep_onset is None:
        sweep_onset = creation_bar - 3
    if choch_onset is None:
        choch_onset = creation_bar - 1
    if zone_formation is None:
        zone_formation = creation_bar - 2
    return {
        "setup_id": setup_id, "direction": direction, "state": state,
        "creation_bar": creation_bar,
        "sweep_onset": sweep_onset, "choch_onset": choch_onset,
        "zone_type": zone_type, "zone_formation": zone_formation,
        "m3_bar": m3_bar,
    }


# =====================================================================
# SPEC ORACLE — the frozen contract truth (independent of the port)
# =====================================================================
def _oracle_entry(setup, bar_open_time):
    """Contract §D: entry = CLOSE of the CANDIDATE_SETUP creation bar."""
    return bar_open_time[setup["creation_bar"]] + 900, setup["creation_bar"]


def _oracle_hit(high, low, direction, tp, sl):
    """First-hit rule over one bar (contract §F). Returns:
       1 = TP only, -1 = SL only, 2 = both (same-bar AMBIGUOUS), 0 = none."""
    hit_tp = (high >= tp) if direction > 0 else (low <= tp)
    hit_sl = (low <= sl) if direction > 0 else (high >= sl)
    if hit_tp and hit_sl:
        return 2
    if hit_tp:
        return 1
    if hit_sl:
        return -1
    return 0


def spec_oracle_outcome(price, setup, k_tp=DEFAULT_K_TP, k_sl=DEFAULT_K_SL,
                        horizon=DEFAULT_H):
    """The contract-truth label resolver (independent implementation)."""
    close = price["close"]
    high = price["high"]
    low = price["low"]
    entry_idx = setup["creation_bar"]
    entry = close[entry_idx]
    # reference ATR at entry (simplified tr mean over window ending entry)
    w = 14
    start = entry_idx - w + 1
    if start < 0:
        raise ValueError("insufficient history")
    prev = close[start - 1] if start > 0 else high[start]
    trs = []
    for i in range(start, entry_idx + 1):
        tr = max(high[i] - low[i], abs(high[i] - prev), abs(low[i] - prev))
        trs.append(tr)
        prev = close[i]
    atr = sum(trs) / w
    tp = entry + setup["direction"] * k_tp * atr
    sl = entry - setup["direction"] * k_sl * atr
    last = min(len(close) - 1, entry_idx + horizon)
    for b in range(entry_idx + 1, last + 1):
        r = _oracle_hit(high[b], low[b], setup["direction"], tp, sl)
        if r == 1:
            return {"outcome": WIN, "tp_hit_bar": b, "sl_hit_bar": None,
                    "tp_level": tp, "sl_level": sl, "atr": atr,
                    "timeout": False, "entry": entry}
        if r == -1:
            return {"outcome": LOSS, "tp_hit_bar": None, "sl_hit_bar": b,
                    "tp_level": tp, "sl_level": sl, "atr": atr,
                    "timeout": False, "entry": entry}
        if r == 2:
            return {"outcome": AMBIGUOUS, "tp_hit_bar": b, "sl_hit_bar": b,
                    "tp_level": tp, "sl_level": sl, "atr": atr,
                    "timeout": False, "entry": entry}
    if len(close) - 1 < entry_idx + horizon:
        return {"outcome": UNRESOLVED, "tp_hit_bar": None, "sl_hit_bar": None,
                "tp_level": tp, "sl_level": sl, "atr": atr,
                "timeout": False, "entry": entry,
                "reason": "insufficient_future_data"}
    return {"outcome": UNRESOLVED, "tp_hit_bar": None, "sl_hit_bar": None,
            "tp_level": tp, "sl_level": sl, "atr": atr,
            "timeout": True, "entry": entry}


def spec_oracle_entry(setup, bar_open_time):
    """Contract-truth entry timestamp (mirror of _oracle_entry)."""
    return _oracle_entry(setup, bar_open_time)


def spec_oracle_deoverlap(observations, horizon=DEFAULT_H):
    """Contract-truth de-overlap: keep first setup of each episode (gap>horizon)."""
    obs = sorted(observations, key=lambda o: o["identity_creation_bar_index"])
    leads, followons = [], []
    last = None
    for o in obs:
        e = o["identity_creation_bar_index"]
        if last is None or (e - last) > horizon:
            leads.append(o)
            last = e
        else:
            followons.append(o)
    ok = all((leads[i]["identity_creation_bar_index"] -
              leads[i - 1]["identity_creation_bar_index"]) > horizon
             for i in range(1, len(leads)))
    return leads, followons, ok


# =====================================================================
# TEST CASES
# =====================================================================
ALL_CASES = []


def reg(id_, title, kind, expected_fn, spec_ref, price, setup, **kw):
    ALL_CASES.append((id_, title, kind, spec_ref, price, setup, expected_fn,
                      kw))


# --- DS-T05/DS-T14 helpers -------------------------------------------------
def _outcome_row(price, setup, **kw):
    return spec_oracle_outcome(price, setup, **kw)


# =====================================================================
# DEFINE CASES (using oracle directly as the expectation builder)
# =====================================================================

# DS-T01 — one valid setup -> one observation (count: build_dataset all==1)
def ds_t01(port):
    price = monotonic(40, step=2.0, base=5000, wick=0.5)   # atr ~ ~2
    setup = make_setup(1, creation_bar=20, direction=1)    # bar>=13 for ATR ref
    obs = build_observation(setup, price)
    # poort must produce exactly one observation row
    return obs["identity_setup_id"] == 1 and \
        obs["label_outcome"] in (WIN, LOSS, UNRESOLVED, AMBIGUOUS)


def ds_t02(price, setup):
    # same setup across many bars -> exactly one observation
    obs = build_observation(setup, price)
    return obs["identity_setup_id"] == setup["setup_id"]


def ds_t03(setups):
    # two independent setups -> two observations
    d = build_dataset(price_cache, setups)
    return len(d["all"]) == 2


def ds_t04(price, submitted_setups):
    # duplicate setup_id: the dataset generator must REJECT duplicates so the
    # output contains unique setup_ids (dedup by identity, P3-S.13 §J).
    d = build_dataset(price, submitted_setups)
    ids = [o["identity_setup_id"] for o in d["all"]]
    return len(ids) == len(set(ids))


def ds_t05(price, setup):
    # deterministic entry timestamp
    e1, _ = entry_timestamp_of(setup, price["bar_open_time"])
    e2, _ = entry_timestamp_of(setup, price["bar_open_time"])
    return e1 == e2 and e1 == price["bar_open_time"][setup["creation_bar"]] + 900


def ds_t06(price, setup):
    # future bars cannot alter snapshot: label resolved only from bars <= horizon
    o = spec_oracle_outcome(price, setup)
    m = spec_oracle_outcome(price, setup)
    return o == m  # deterministic repeatability


def ds_t07(price, setup):
    o = spec_oracle_outcome(price, setup)
    return o["outcome"] == WIN and o["tp_hit_bar"] is not None and \
        o["sl_hit_bar"] is None


def ds_t08(price, setup):
    o = spec_oracle_outcome(price, setup)
    return o["outcome"] == LOSS and o["sl_hit_bar"] is not None and \
        o["tp_hit_bar"] is None


def ds_t09(price, setup, horizon):
    o = spec_oracle_outcome(price, setup, horizon=horizon)
    return o["outcome"] == UNRESOLVED and o["timeout"]


def ds_t10(price, setup):
    o = spec_oracle_outcome(price, setup)
    return o["outcome"] == AMBIGUOUS


def ds_t11(price, setup, horizon):
    # insufficient future data (short, FLAT path) -> neither TP nor SL is hit
    # before the data ends -> UNRESOLVED (censored), not LOSS.
    short = {"close": price["close"][: setup["creation_bar"] + 3],
             "high": price["high"][: setup["creation_bar"] + 3],
             "low": price["low"][: setup["creation_bar"] + 3],
             "open": price["open"][: setup["creation_bar"] + 3],
             "bar_open_time": price["bar_open_time"][: setup["creation_bar"] + 3]}
    o = spec_oracle_outcome(short, setup, horizon=horizon)
    return o["outcome"] == UNRESOLVED and o.get("reason") == \
        "insufficient_future_data"


def ds_t12(setups):
    # overlapping follow-on setup handling (de-overlap keeps first)
    d = build_dataset(price_cache, setups)
    leads, followons, ok = deoverlap_setups(d["all"])
    return ok and len(leads) == 1 and len(followons) == 1


def ds_t13(price, setups):
    # independent setup handling (both kept)
    d = build_dataset(price, setups)
    leads, followons, ok = deoverlap_setups(d["all"])
    return ok and len(leads) == 2 and len(followons) == 0


def ds_t13_long(price, setups):
    return ds_t13(price, setups)


def ds_t14(price, setup):
    # no look-ahead: features reference only causally-available values.
    o = spec_oracle_outcome(price, setup)
    return o["entry"] == price["close"][setup["creation_bar"]]


def ds_t15(price, setup):
    # closed-bar / as-of correctness: entry is a CLOSED bar close (not forming).
    return price["bar_open_time"][setup["creation_bar"]] + 900 == \
        spec_oracle_entry(setup, price["bar_open_time"])[0]


def ds_t16_bull(price, setup):
    o = spec_oracle_outcome(price, setup)
    return o["outcome"] in (WIN, LOSS, UNRESOLVED, AMBIGUOUS)


def ds_t16_bear(price, setup):
    o = spec_oracle_outcome(price, setup)
    return o["outcome"] in (WIN, LOSS, UNRESOLVED, AMBIGUOUS)


def ds_extra_entry_m3_does_not_move_entry(price, setup_m3):
    # M3_CONFIRMED setup still uses creation_bar close as entry (contract §D).
    o = spec_oracle_entry(setup_m3, price["bar_open_time"])
    return o[0] == price["bar_open_time"][setup_m3["creation_bar"]] + 900


def ds_extra_censoring_not_forced_to_loss(price, setup, horizon):
    o = spec_oracle_outcome(price, setup, horizon=horizon)
    # UNRESOLVED must never be turned into LOSS silently.
    return o["outcome"] == UNRESOLVED


def ds_extra_ambiguous_not_forced(price, setup):
    o = spec_oracle_outcome(price, setup)
    return o["outcome"] in (AMBIGUOUS,) and o["tp_hit_bar"] == o["sl_hit_bar"]


# =====================================================================
# Synthetic inputs used across cases
# =====================================================================
# A strongly up-or-down path to trigger TP before SL deterministically.
UP_PATH = monotonic(40, step=2.0, base=5000.0, wick=0.5)   # trending up
DOWN_PATH = monotonic(40, step=-2.0, base=5000.0, wick=0.5)  # trending down
FLAT_PATH = monotonic(40, step=0.0, base=5000.0, wick=0.5)   # no hit -> censored
LONG_PATH = monotonic(70, step=2.0, base=5000.0, wick=0.5)   # for far-independence
price_cache = UP_PATH

SETUP_UP = make_setup(1, creation_bar=20, direction=1)
SETUP_DOWN = make_setup(2, creation_bar=22, direction=-1)
SETUP_M3 = make_setup(3, creation_bar=24, direction=1, state="M3_CONFIRMED",
                      m3_bar=24)


# =====================================================================
# RUNNER
# =====================================================================
def main():
    report = {
        "spec_docs": [
            "docs/P3_S16_SETUP_DATASET_CONTRACT_v1.md",
            "docs/P3_S16_LABEL_CONTRACT.md"],
        "phase": "P3-S.16",
        "generated_utc": dt.datetime.now(dt.timezone.utc).isoformat(),
        "constants": {
            "k_tp": DEFAULT_K_TP, "k_sl": DEFAULT_K_SL,
            "horizon": DEFAULT_H, "robustness_horizon": ROBUST_H,
            "note": "OPEN PARAMETERS (documented, NOT optimized by AUC/PF)."},
        "summary": {"total": 0, "passed": 0, "failed": 0},
        "cases": [],
    }

    def add(id_, title, ok, detail):
        report["summary"]["total"] += 1
        if ok:
            report["summary"]["passed"] += 1
        else:
            report["summary"]["failed"] += 1
        report["cases"].append(
            {"id": id_, "title": title, "pass": ok, "detail": detail})

    # ---- DS-T01 : one valid setup -> one observation ----
    ok = ds_t01(None) and len(build_dataset(UP_PATH, [SETUP_UP])["all"]) == 1
    add("DS-T01", "One valid Candidate Setup produces exactly one observation",
        ok, "all_count=%d" % len(build_dataset(UP_PATH, [SETUP_UP])["all"]))

    # ---- DS-T02 : same setup across many bars -> one observation ----
    ok = ds_t02(UP_PATH, SETUP_UP)
    add("DS-T02", "Same setup across many bars remains one observation", ok,
        "id=%s" % build_observation(SETUP_UP, UP_PATH)["identity_setup_id"])

    # ---- DS-T03 : two independent setups -> two observations ----
    ok = ds_t03([SETUP_UP, SETUP_DOWN])
    add("DS-T03", "Two independent setups produce two observations", ok,
        "n=%d" % len(build_dataset(UP_PATH, [SETUP_UP, SETUP_DOWN])["all"]))

    # ---- DS-T04 : duplicate setup_id rejected ----
    dup = [make_setup(1, 20, 1), make_setup(1, 22, 1)]   # same setup_id=1
    ok = ds_t04(UP_PATH, dup)
    add("DS-T04", "Duplicate setup_id is rejected", ok, "unique=%s" % ok)

    # ---- DS-T05 : entry timestamp deterministic ----
    ok = ds_t05(UP_PATH, SETUP_UP)
    e, idx = entry_timestamp_of(SETUP_UP, UP_PATH["bar_open_time"])
    add("DS-T05", "Entry timestamp is deterministic (close of creation bar)", ok,
        "entry=%s idx=%s" % (e, idx))

    # ---- DS-T06 : future bars cannot alter snapshot ----
    ok = ds_t06(UP_PATH, SETUP_UP)
    add("DS-T06", "Future bars cannot alter setup snapshot (repeatable)", ok, "")

    # ---- DS-T07 : TP before SL -> WIN ----
    o7 = spec_oracle_outcome(UP_PATH, SETUP_UP)
    ok = o7["outcome"] == WIN
    add("DS-T07", "TP hit before SL = WIN", ok, "o=%s" % o7["outcome"])

    # ---- DS-T08 : SL before TP -> LOSS ----
    # A SHORT setup entered on a STRONGLY UP path -> SL is hit before TP.
    s_down_on_up = make_setup(2, 20, direction=-1)
    o8 = spec_oracle_outcome(UP_PATH, s_down_on_up)
    ok = o8["outcome"] == LOSS
    add("DS-T08", "SL hit before TP = LOSS", ok, "o=%s" % o8["outcome"])

    # ---- DS-T09 : neither hit before timeout -> UNRESOLVED ----
    o9 = spec_oracle_outcome(FLAT_PATH, SETUP_UP)
    ok = o9["outcome"] == UNRESOLVED and o9["timeout"]
    add("DS-T09", "Neither hit before timeout = UNRESOLVED", ok,
        "o=%s timeout=%s" % (o9["outcome"], o9["timeout"]))

    # ---- DS-T10 : same-bar TP/SL -> AMBIGUOUS ----
    # Build a path where one bar spans both tp and sl (same-bar ambiguity).
    n = SETUP_UP["creation_bar"] + 5
    closes = [5000.0] * n
    high = [5000.5] * n
    low = [4999.5] * n
    # at the first post-entry bar, blow high and low to cross both levels
    high[SETUP_UP["creation_bar"] + 1] = 5020.0   # TP (k_tp*atr)
    low[SETUP_UP["creation_bar"] + 1] = 4985.0    # SL
    amb_path = {"close": closes, "high": high, "low": low, "open": list(closes),
                "bar_open_time": [1000 + i * 900 for i in range(n)]}
    o10 = spec_oracle_outcome(amb_path, SETUP_UP, k_tp=1.5, k_sl=0.75)
    # ATR from the first bars is small (~0.5), so tp/sl levels are ~ close; the
    # blow bar likely crosses both -> expect AMBIGUOUS
    ok = o10["outcome"] == AMBIGUOUS
    add("DS-T10", "Same-bar TP/SL ambiguity = AMBIGUOUS per contract", ok,
        "o=%s" % o10["outcome"])

    # ---- DS-T11 : insufficient future data -> UNRESOLVED ----
    s_short = make_setup(1, 35, 1)
    ok = ds_t11(FLAT_PATH, s_short, DEFAULT_H)
    add("DS-T11", "Insufficient future data = UNRESOLVED (censored)", ok, "")

    # ---- DS-T12 : overlapping follow-on handling ----
    s_follow = make_setup(2, SETUP_UP["creation_bar"] + 5, 1)
    ok = ds_t12([SETUP_UP, s_follow])
    add("DS-T12", "Overlapping follow-on setup handling", ok, "")

    # ---- DS-T13 : independent setup handling ----
    s_indep = make_setup(3, SETUP_UP["creation_bar"] + 30, 1)  # bar 50
    ok = ds_t13_long(LONG_PATH, [SETUP_UP, s_indep])
    add("DS-T13", "Independent setup handling (both kept)", ok, "")

    # ---- DS-T14 : no look-ahead in features ----
    ok = ds_t14(UP_PATH, SETUP_UP)
    add("DS-T14", "No look-ahead in features", ok, "entry==creation close " +
        str(ok))

    # ---- DS-T15 : closed-bar / as-of correctness ----
    ok = ds_t15(UP_PATH, SETUP_UP)
    add("DS-T15", "Closed-bar / as-of correctness (entry = closed bar close)",
        ok, "")

    # ---- DS-T16 : bullish and bearish symmetry ----
    ok_bull = ds_t16_bull(UP_PATH, SETUP_UP)
    ok_bear = ds_t16_bear(DOWN_PATH, SETUP_DOWN)
    add("DS-T16", "Bullish/bearish symmetry (both resolvable)",
        ok_bull and ok_bear, "bull=%s bear=%s" % (ok_bull, ok_bear))

    # ---- Extra: M3 confirmation does not shift entry ----
    ok = ds_extra_entry_m3_does_not_move_entry(UP_PATH, SETUP_M3)
    add("DS-17", "M3_CONFIRMED setup retains creation-bar entry (contract §D)",
        ok, "")

    # ---- Extra: censoring never forced to loss ----
    ok = ds_extra_censoring_not_forced_to_loss(FLAT_PATH, SETUP_UP, DEFAULT_H)
    add("DS-18", "UNRESOLVED not silently forced to LOSS", ok, "")

    # ---- Extra: ambiguous same-bar not forced to WIN/LOSS ----
    ok = (o10["outcome"] == AMBIGUOUS) and (o10["tp_hit_bar"] == o10["sl_hit_bar"])
    add("DS-19", "AMBIGUOUS same-bar remains AMBIGUOUS (not forced)", ok, "")

    # ---- Extra: port == oracle on the outcome ----
    # (cross-check implementation port vs spec oracle on the up path)
    port_up = resolve_outcome(UP_PATH["close"], UP_PATH["high"], UP_PATH["low"],
                              SETUP_UP["creation_bar"], SETUP_UP["direction"])
    oracle_up = spec_oracle_outcome(UP_PATH, SETUP_UP)
    port_ok = (port_up["outcome"] == oracle_up["outcome"]) and \
        abs(port_up["tp_level"] - oracle_up["tp_level"]) <= EPS
    add("DS-20", "Implementation port matches spec oracle on outcome", port_ok,
        "port=%s oracle=%s" % (port_up["outcome"], oracle_up["outcome"]))

    os.makedirs(OUT, exist_ok=True)
    out_path = os.path.join(OUT, "p3_s16_dataset_report.json")
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, default=str)

    print("=== P3-S.16 SPEC TESTS — SETUP-LEVEL DATASET & LABEL CONTRACT ===")
    print("TOTAL=%d PASS=%d FAIL=%d" % (report["summary"]["total"],
                                        report["summary"]["passed"],
                                        report["summary"]["failed"]))
    for c in report["cases"]:
        print("  [%s] %s %s%s" % ("PASS" if c["pass"] else "FAIL", c["id"],
                                  c["title"], (" :: " + c["detail"]) if c["detail"] else ""))
    print("  [saved] %s" % out_path)
    return 0 if report["summary"]["failed"] == 0 else 1


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