521 lines
No EOL
21 KiB
Python
521 lines
No EOL
21 KiB
Python
# -*- 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()) |