# -*- 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())