# -*- coding: utf-8 -*- """P3-S.16 SETUP-LEVEL DATASET CONTRACT IMPLEMENTATION (port). Research-only. This module is the DETERMINISTIC reference implementation of: docs/P3_S16_SETUP_DATASET_CONTRACT_v1.md docs/P3_S16_LABEL_CONTRACT.md It turns ONE frozen Candidate Setup + a causally-applicable price path into ONE training observation with a setup-outcome label (WIN / LOSS / UNRESOLVED / AMBIGUOUS), and provides the setup-level de-overlap logic. It does NOT train a model, does NOT export weights, does NOT touch the runtime, does NOT modify FEATURE_CONTRACT.md, and selects NO numeric by AUC/PF. Numeric constants below are the OPEN parameters recorded by the contract; they are semantic candidates, not optimized. This module is the 'implementation port' whose behaviour is asserted against the 'spec oracle' in spec_tests_setup_dataset.py (§28 convention: spec_oracle = truth ; implementation_port = observed implementation). """ import os # --------------------------------------------------------------------- # CONTRACT CONSTANTS (label contract §D/§E) — OPEN PARAMETERS, NOT optimized # --------------------------------------------------------------------- DEFAULT_K_TP = 1.5 # TP in ATR units (asymmetric candidate 2:1 R:R) DEFAULT_K_SL = 0.75 # SL in ATR units ROBUST_K_TP = 1.0 # symmetric robustness ROBUST_K_SL = 1.0 DEFAULT_H = 16 # primary horizon (M15 bars), semantic lifetime basis ROBUST_H = 8 # robustness horizon # Contract ATR window (FEATURE_CONTRACT §0) ATR_WINDOW = 14 # label classes WIN = "WIN" LOSS = "LOSS" UNRESOLVED = "UNRESOLVED" AMBIGUOUS = "AMBIGUOUS" # --------------------------------------------------------------------- # Entry timestamp (dataset contract §D) # --------------------------------------------------------------------- def entry_timestamp_of(setup, bar_open_time, bar_period_sec=900): """ENtry timestamp = CLOSE of the CANDIDATE_SETUP creation bar (M15). Causality: the setup completes at creation_bar; the first closed bar at/ after completion is the executable point. M3 is optional/after-creation and cannot be the entry (contract §D). Returns the entry clock time = creation_bar open_time + period. """ assert setup.get("state") in ("CANDIDATE_SETUP", "M3_CONFIRMED"), \ "entry defined only for an ACTIVE CANDIDATE_SETUP" ibar = setup["creation_bar"] return bar_open_time[ibar] + bar_period_sec, ibar def entry_price_of(close_series, entry_bar_index): """Entry price = close of the creation bar (reference mid, contract §H).""" return close_series[entry_bar_index] def atr_at_entry(close_series, high_series, low_series, entry_bar_index, window=ATR_WINDOW): """Rolling ATR over the `window` closed bars ENDING at entry. Causally available at entry; used for TP/SL reference (label §D).""" start = entry_bar_index - window + 1 if start < 0: raise ValueError("insufficient history for ATR at entry") tr_sum = 0.0 prev_close = close_series[start - 1] if start > 0 else \ open_series_fallback(close_series, high_series, low_series, start) for i in range(start, entry_bar_index + 1): tr = max(high_series[i] - low_series[i], abs(high_series[i] - prev_close), abs(low_series[i] - prev_close)) tr_sum += tr prev_close = close_series[i] return tr_sum / window def open_series_fallback(close_series, high_series, low_series, i): # low/mid of the first bar as a prev-close proxy when no prior bar exists. return (high_series[i] + low_series[i]) / 2.0 # --------------------------------------------------------------------- # Outcome label — first-hit TP-before-SL (label contract §C/§D/§E/§F/§G) # --------------------------------------------------------------------- def resolve_outcome(close_series, high_series, low_series, entry_bar_index, direction, k_tp=DEFAULT_K_TP, k_sl=DEFAULT_K_SL, horizon=DEFAULT_H, atr=None): """Resolve the setup-outcome label for ONE setup. Arguments are the price-path for bars [entry_bar_index .. entry+horizon] and the entry reference. Returns a dict {outcome, tp_level, sl_level, tp_hit_bar, sl_hit_bar, timeout, invalidated_early}. """ # reference ATR at entry (contract §D) if atr is None: atr = atr_at_entry(close_series, high_series, low_series, entry_bar_index) entry = close_series[entry_bar_index] tp_level = entry + direction * k_tp * atr sl_level = entry - direction * k_sl * atr result = { "outcome": None, "entry": entry, "atr": atr, "tp_level": tp_level, "sl_level": sl_level, "tp_hit_bar": None, "sl_hit_bar": None, "timeout": False, "invalidated_early": False, "entry_bar_index": entry_bar_index, "horizon": horizon, } # scan strictly AFTER entry bar (first-hit rule, label §F) last = min(len(close_series) - 1, entry_bar_index + horizon) for b in range(entry_bar_index + 1, last + 1): # TP touch: long needs high>=tp; short needs low<=tp. hit_tp = (high_series[b] >= tp_level) if direction > 0 \ else (low_series[b] <= tp_level) # SL touch: long needs low<=sl; short needs high>=sl. hit_sl = (low_series[b] <= sl_level) if direction > 0 \ else (high_series[b] >= sl_level) if hit_tp and hit_sl: result["outcome"] = AMBIGUOUS # same-bar TP+SL (§F) result["tp_hit_bar"] = b result["sl_hit_bar"] = b return result if hit_tp: result["outcome"] = WIN result["tp_hit_bar"] = b return result if hit_sl: result["outcome"] = LOSS result["sl_hit_bar"] = b return result # neither hit up to available data within horizon if len(close_series) - 1 < entry_bar_index + horizon: result["outcome"] = UNRESOLVED # insufficient future data (§C/§G) result["reason"] = "insufficient_future_data" else: result["outcome"] = UNRESOLVED # timeout, true censoring (§G) result["timeout"] = True return result # --------------------------------------------------------------------- # Observation building — ONE setup -> ONE observation row (contract §B/§C) # --------------------------------------------------------------------- def build_observation(setup, price, atr=None, k_tp=DEFAULT_K_TP, k_sl=DEFAULT_K_SL, horizon=DEFAULT_H): """Turn ONE verified Candidate Setup + price-path into ONE observation. `price` is a dict with lists: open/high/low/close (bar arrays) and `bar_open_time` (array of open times). `setup` is the immutable identity dict (Store 1) + the causal fields. Returns a row separating identity_ / feature_ / label_ namespaces. """ entry_idx = setup["creation_bar"] entry_ts, _ = entry_timestamp_of(setup, price["bar_open_time"]) entry = entry_price_of(price["close"], entry_idx) # creation-bar close lbl = resolve_outcome(price["close"], price["high"], price["low"], entry_idx, setup["direction"], k_tp=k_tp, k_sl=k_sl, horizon=horizon, atr=atr) # feature snapshot (Store 2) — causally available at entry. # Detailed feature numerics live in the future feature contract; here we # record the causally-available gates / references the contract requires. swee_age = setup["creation_bar"] - setup.get("sweep_onset", setup["creation_bar"]) chore_age = None if setup.get("choch_onset") is not None: chore_age = setup["creation_bar"] - setup["choch_onset"] return { "identity_setup_id": setup["setup_id"], "identity_direction": setup["direction"], "identity_creation_timestamp": entry_ts - 900, # creation open "identity_creation_bar_index": entry_idx, "identity_setup_state": setup["state"], "identity_sweep_onset_index": setup.get("sweep_onset"), "identity_choch_onset_index": setup.get("choch_onset"), "identity_zone_type": setup.get("zone_type"), "identity_zone_formation_timestamp": setup.get("zone_formation"), "identity_m3_confirmation_ts": setup.get("m3_bar"), "identity_entry_timestamp": entry_ts, "feature_sweep_age_bars": swee_age, "feature_choch_age_bars": chore_age, "feature_atr_at_entry": lbl["atr"], "feature_tp_level_atr_mult": k_tp, "feature_sl_level_atr_mult": k_sl, "feature_horizon_bars": horizon, "label_outcome": lbl["outcome"], "label_tp_level": lbl["tp_level"], "label_sl_level": lbl["sl_level"], "label_tp_hit_bar": lbl["tp_hit_bar"], "label_sl_hit_bar": lbl["sl_hit_bar"], "label_timeout": lbl["timeout"], "label_invalidated_early": False, "label_contract_version": "P3_S16_LABEL_CONTRACT_v1", } # --------------------------------------------------------------------- # Setup-level de-overlap (§F / §18) — lead-setup-per-episode # --------------------------------------------------------------------- def deoverlap_setups(observations, horizon=DEFAULT_H): """Given a list of observations (each with creation_bar_index entry), apply setup-level de-overlap: retain the FIRST setup of each episode (lead), report the follow-ons. Returns (leads, followons, verified). """ obs = sorted(observations, key=lambda o: o["identity_creation_bar_index"]) leads = [] followons = [] last_entry = None for o in obs: entry = o["identity_creation_bar_index"] if last_entry is None or (entry - last_entry) > horizon: leads.append(o) last_entry = entry else: followons.append(o) # verify separation > horizon verified = True for i in range(1, len(leads)): if (leads[i]["identity_creation_bar_index"] - leads[i - 1]["identity_creation_bar_index"]) <= horizon: verified = False return leads, followons, verified def build_dataset(price, setups, atr=None, k_tp=DEFAULT_K_TP, k_sl=DEFAULT_K_SL, horizon=DEFAULT_H): """Build the full observation list (one per CANDIDATE_SETUP) then apply setup-level de-overlap. Rejects duplicate setup_id (dedup by identity, P3-S.13 §J). Returns {all, leads, followons, deoverlap_ok}.""" seen = set() deduped = [] for s in setups: if s["setup_id"] in seen: continue # duplicate setup_id rejected seen.add(s["setup_id"]) deduped.append(s) all_obs = [build_observation(s, price, atr=atr, k_tp=k_tp, k_sl=k_sl, horizon=horizon) for s in deduped] leads, followons, ok = deoverlap_setups(all_obs, horizon=horizon) return { "all": all_obs, "leads": leads, "followons": followons, "deoverlap_ok": ok, }