dennisthiessen/signal-platform
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add GTL cohort composition backtest

Browse Source
This commit is contained in:
dennisthiessen
2026-07-13 14:22:56 +02:00
parent 3f86aec0be
commit 623dc08875
8 changed files with 755 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/backtest_service.py
+187 -6
View File
@@ -111,6 +111,7 @@ STRUCTURAL_OVERLAY_WEIGHT = 0.05
STRUCTURAL_OVERLAY_SCORE_KEY = "structural_overlay_95_5_score"
EXPLICIT_TARGET_LADDER_VARIANT = "explicit_target_ladder"
GTL_TUNING_VARIANT = "gtl_tuning"
GTL_CONFIRMATION_VARIANT = "gtl_confirmation"
RANGE_RESIDUAL_VARIANTS = {
"rewrite_range504_structural_legacy_primary",
"rewrite_range504_structural_primary2",
@@ -167,6 +168,23 @@ class GTLResearchConfig:
strength_scale=self.strength_scale,
)
@dataclass(frozen=True)
class GTLConfirmationConfig:
"""Research-only composition of the frozen GTL and tuned variants."""
name: str = "control"
mode: str = "intersection"
confirmations: tuple[GTLResearchConfig, ...] = ()
def __post_init__(self) -> None:
if not self.name.strip():
raise ValueError("GTL confirmation config name must not be empty")
if self.mode not in {"intersection", "union"}:
raise ValueError("GTL confirmation mode must be intersection or union")
if self.mode == "union" and len(self.confirmations) != 1:
raise ValueError("GTL union mode requires exactly one tuned variant")
# Cross-sectional signal evaluation (factor IC). Each candidate signal is a
# point-in-time number computed from closes alone (sentiment/fundamentals have no
# history here), sampled one as-of per ISO week, and graded by how its rank
@@ -222,6 +240,7 @@ SR_RESEARCH_VARIANTS = {
"legacy_range_grid_neutral",
EXPLICIT_TARGET_LADDER_VARIANT,
GTL_TUNING_VARIANT,
GTL_CONFIRMATION_VARIANT,
STRUCTURAL_OVERLAY_VARIANT,
*RANGE_FACTOR_VARIANTS,
}
@@ -261,6 +280,47 @@ def _gtl_research_config() -> GTLResearchConfig:
return _parse_gtl_research_config(os.getenv("BACKTEST_GTL_CONFIG", ""))
@lru_cache(maxsize=32)
def _parse_gtl_confirmation_config(raw: str) -> GTLConfirmationConfig:
"""Parse one composition arm for the offline confirmation matrix."""
if not raw.strip():
return GTLConfirmationConfig()
try:
payload = json.loads(raw)
except json.JSONDecodeError as exc:
raise ValueError("BACKTEST_GTL_CONFIRM_CONFIG must be valid JSON") from exc
if not isinstance(payload, dict):
raise ValueError("BACKTEST_GTL_CONFIRM_CONFIG must be a JSON object")
allowed = {"name", "mode", "confirmations"}
unknown = sorted(set(payload) - allowed)
if unknown:
raise ValueError(
f"Unknown GTL confirmation config fields: {', '.join(unknown)}"
)
raw_confirmations = payload.get("confirmations", [])
if not isinstance(raw_confirmations, list):
raise ValueError("GTL confirmations must be a JSON array")
confirmations: list[GTLResearchConfig] = []
for item in raw_confirmations:
if not isinstance(item, dict):
raise ValueError("Each GTL confirmation must be a JSON object")
confirmations.append(_parse_gtl_research_config(json.dumps(item)))
try:
return GTLConfirmationConfig(
name=payload.get("name", "control"),
mode=payload.get("mode", "intersection"),
confirmations=tuple(confirmations),
)
except (AttributeError, TypeError, ValueError) as exc:
raise ValueError(f"Invalid GTL confirmation config: {exc}") from exc
def _gtl_confirmation_config() -> GTLConfirmationConfig:
return _parse_gtl_confirmation_config(
os.getenv("BACKTEST_GTL_CONFIRM_CONFIG", "")
)
def _sr_detector_variant(sr_variant: str) -> str:
"""Map factor-gated research arms to the detector they hold fixed."""
if sr_variant in {"production_range504", STRUCTURAL_OVERLAY_VARIANT}:
@@ -302,6 +362,7 @@ def _primary_min_rr_for_variant(sr_variant: str, activation: dict) -> float:
"production_range504",
EXPLICIT_TARGET_LADDER_VARIANT,
GTL_TUNING_VARIANT,
GTL_CONFIRMATION_VARIANT,
STRUCTURAL_OVERLAY_VARIANT,
}
or sr_variant.endswith("_legacy_primary")
@@ -416,6 +477,7 @@ def _window_setups(
activation: dict,
*,
sr_variant: str | None = None,
gtl_research_config: GTLResearchConfig | None = None,
) -> list[dict]:
"""Rebuild the setup(s) at the last bar of ``window_records`` (the as-of date),
using only those bars. Returns one dict per tradeable direction."""
@@ -444,7 +506,11 @@ def _window_setups(
sr_variant = sr_variant or _sr_research_variant()
detector_variant = _sr_detector_variant(sr_variant)
gtl_config = _gtl_research_config() if sr_variant == GTL_TUNING_VARIANT else None
gtl_config = (
gtl_research_config or _gtl_research_config()
if sr_variant == GTL_TUNING_VARIANT
else None
)
range_504_log = _range_504_log(highs, lows)
if sr_variant == "legacy_geometry_neutral":
detected_levels = detect_sr_levels_legacy(
@@ -683,6 +749,87 @@ def _structural_overlay_window_setups(
return tagged
def _gtl_confirmation_window_setups(
window_records: list,
config: dict,
activation: dict,
*,
confirmation_config: GTLConfirmationConfig | None = None,
) -> list[dict]:
"""Compose tuned GTLs around the frozen ladder without changing it silently.
``intersection`` retains the frozen setup geometry and requires every tuned
variant to clear the core gate in the same direction. ``union`` preserves a
frozen setup whenever it already clears the core gate, and otherwise admits
the one tuned variant's setup. This makes retained, removed, and added
cohorts explicit instead of conflating them in a replacement arm.
"""
research = confirmation_config or _gtl_confirmation_config()
production = _window_setups(
window_records,
config,
activation,
sr_variant=EXPLICIT_TARGET_LADDER_VARIANT,
)
tuned_sets = [
_window_setups(
window_records,
config,
activation,
sr_variant=GTL_TUNING_VARIANT,
gtl_research_config=tuned_config,
)
for tuned_config in research.confirmations
]
tuned_by_direction = [
{row["direction"]: row for row in rows}
for rows in tuned_sets
]
def annotate(row: dict, passes: list[bool], source: str) -> dict:
tagged = dict(row)
tagged["sr_variant"] = GTL_CONFIRMATION_VARIANT
tagged["gtl_confirmation_name"] = research.name
tagged["gtl_confirmation_mode"] = research.mode
tagged["gtl_confirmation_source"] = source
tagged["gtl_confirmation_passes"] = passes
tagged["gtl_confirmation_all_pass"] = all(passes)
return tagged
if research.mode == "intersection":
tagged: list[dict] = []
for production_row in production:
direction = production_row["direction"]
passes = [
bool(rows.get(direction) and rows[direction].get("meets_core"))
for rows in tuned_by_direction
]
row = annotate(production_row, passes, "control")
row["meets_core"] = bool(production_row.get("meets_core")) and all(
passes
)
tagged.append(row)
return tagged
# Union mode is validated to contain exactly one tuned variant. Keep one
# setup per direction: frozen geometry wins whenever it already qualifies;
# tuned geometry is used only for a genuinely added core-qualified setup.
production_by_direction = {row["direction"]: row for row in production}
tuned_by_dir = tuned_by_direction[0]
tagged = []
for direction in sorted(set(production_by_direction) | set(tuned_by_dir)):
production_row = production_by_direction.get(direction)
tuned_row = tuned_by_dir.get(direction)
tuned_pass = bool(tuned_row and tuned_row.get("meets_core"))
if production_row is not None and production_row.get("meets_core"):
tagged.append(annotate(production_row, [tuned_pass], "control"))
elif tuned_pass and tuned_row is not None:
tagged.append(annotate(tuned_row, [True], "tuned_addition"))
elif production_row is not None:
tagged.append(annotate(production_row, [tuned_pass], "control"))
return tagged
def _stop_fill_r(direction: str, entry: float, stop: float, bar) -> float:
"""Realized R when the stop is hit on ``bar``: filled at the stop, or at the
bar's open when price gapped through it — so a gap can lose more than −1R,
@@ -787,16 +934,17 @@ def _replay_ticker(
)
vol_6m = _realized_vol_6m(closes, len(window) - 1)
setups = (
_structural_overlay_window_setups(window, config, activation)
if sr_variant == STRUCTURAL_OVERLAY_VARIANT
else _window_setups(
if sr_variant == STRUCTURAL_OVERLAY_VARIANT:
setups = _structural_overlay_window_setups(window, config, activation)
elif sr_variant == GTL_CONFIRMATION_VARIANT:
setups = _gtl_confirmation_window_setups(window, config, activation)
else:
setups = _window_setups(
window,
config,
activation,
sr_variant=sr_variant,
)
)
for s in setups:
outcome, outcome_date = evaluate_setup_against_bars(
s["direction"], s["stop"], s["target"], forward_bars, HORIZON
@@ -862,6 +1010,13 @@ def _replay_ticker(
"structural_overlay_gate_level_count": s.get(
"structural_overlay_gate_level_count"
),
"gtl_confirmation_name": s.get("gtl_confirmation_name"),
"gtl_confirmation_mode": s.get("gtl_confirmation_mode"),
"gtl_confirmation_source": s.get("gtl_confirmation_source"),
"gtl_confirmation_passes": s.get("gtl_confirmation_passes"),
"gtl_confirmation_all_pass": s.get(
"gtl_confirmation_all_pass"
),
"outcome": outcome,
"target_hit": target_hit,
"realized_r": realized_r,
@@ -934,6 +1089,9 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
range_logs: list[float] = []
overlay_rows = 0
overlay_pass = 0
confirmation_rows = 0
confirmation_pass = 0
confirmation_tuned_additions = 0
for cand in candidates:
sources = list(cand.get("primary_sources") or [])
for source in sources:
@@ -949,6 +1107,12 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
if cand.get("structural_overlay_pass") is not None:
overlay_rows += 1
overlay_pass += int(bool(cand["structural_overlay_pass"]))
if cand.get("gtl_confirmation_all_pass") is not None:
confirmation_rows += 1
confirmation_pass += int(bool(cand["gtl_confirmation_all_pass"]))
confirmation_tuned_additions += int(
cand.get("gtl_confirmation_source") == "tuned_addition"
)
def avg(values: list[float] | list[int]) -> float | None:
return round(sum(values) / len(values), 3) if values else None
@@ -973,6 +1137,9 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
"structural_overlay_weight": (
STRUCTURAL_OVERLAY_WEIGHT if overlay_rows else None
),
"gtl_confirmation_rows": confirmation_rows,
"gtl_confirmation_pass": confirmation_pass,
"gtl_confirmation_tuned_additions": confirmation_tuned_additions,
}
@@ -1019,6 +1186,15 @@ def _sr_candidate_audit(candidates: list[dict], min_percentile: float) -> list[d
"structural_overlay_gate_level_count": int(
cand.get("structural_overlay_gate_level_count", 0) or 0
),
"gtl_confirmation_name": cand.get("gtl_confirmation_name"),
"gtl_confirmation_mode": cand.get("gtl_confirmation_mode"),
"gtl_confirmation_source": cand.get("gtl_confirmation_source"),
"gtl_confirmation_passes": list(
cand.get("gtl_confirmation_passes") or []
),
"gtl_confirmation_all_pass": cand.get(
"gtl_confirmation_all_pass"
),
"rr": round(float(cand.get("rr", 0.0)), 6),
"primary_prob": round(float(cand.get("primary_prob", 0.0)), 6),
"primary_sources": list(cand.get("primary_sources") or []),
@@ -3430,6 +3606,11 @@ async def run_backtest(
if _sr_research_variant() == GTL_TUNING_VARIANT
else None
),
"gtl_confirmation_config": (
asdict(_gtl_confirmation_config())
if _sr_research_variant() == GTL_CONFIRMATION_VARIANT
else None
),
"range_factor_lookback": RANGE_FACTOR_LOOKBACK,
"range_factor_min_log": RANGE_FACTOR_MIN_LOG,
"range_factor_min_ratio": round(math.exp(RANGE_FACTOR_MIN_LOG), 4),