forked from chiki2bum2/SniperGold_ML
260 lines
11 KiB
Python
260 lines
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,
| |||
}
|