260 lines
No EOL
11 KiB
Python
260 lines
No EOL
11 KiB
Python
# -*- 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,
|
|
} |