Add single-command GTL tuning matrix

app/services/backtest_service.py

@@ -33,7 +33,9 @@ import statistics
 from collections import defaultdict
 from collections.abc import Callable
 from concurrent.futures import ProcessPoolExecutor
+from dataclasses import asdict, dataclass
 from datetime import date, datetime, timedelta, timezone
+from functools import lru_cache
 from types import SimpleNamespace
 from typing import Any
 
@@ -71,6 +73,7 @@ from app.services.recommendation_service import (
     _prune_floor_pinned_targets,
     _risk_level_from_conflicts,
     _select_primary_target,
+    _SR_ZONE_TOLERANCE,
     _zone_representative_levels,
     direction_analyzer,
     get_recommendation_config,
@@ -83,6 +86,7 @@ from app.services.scoring_service import (
     compute_technical_from_arrays,
 )
 from app.services.sr_service import (
+    GateTargetLadderConfig,
     MAX_LEVELS,
     detect_gate_target_ladder,
     detect_sr_levels,
@@ -106,6 +110,7 @@ STRUCTURAL_OVERLAY_SOURCE_VARIANT = (
 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"
 RANGE_RESIDUAL_VARIANTS = {
     "rewrite_range504_structural_legacy_primary",
     "rewrite_range504_structural_primary2",
@@ -116,6 +121,52 @@ RANGE_FACTOR_VARIANTS = {
     *RANGE_RESIDUAL_VARIANTS,
 }
 
+
+@dataclass(frozen=True)
+class GTLResearchConfig:
+    """Research-only controls for one GTL matrix arm.
+
+    The default values reproduce the explicit Gate Target Ladder. Production
+    never reads ``BACKTEST_GTL_CONFIG``.
+    """
+
+    name: str = "control"
+    lookback_bars: int | None = None
+    grid_bins: int = 20
+    include_pivots: bool = True
+    pivot_window: int = 2
+    touch_tolerance: float = 0.005
+    merge_tolerance: float = 0.005
+    strength_scale: float = 500.0
+    zone_tolerance: float = 0.02
+    candidate_limit: int | None = 5
+    max_target_atr: float | None = None
+
+    def __post_init__(self) -> None:
+        if not self.name.strip():
+            raise ValueError("GTL research config name must not be empty")
+        if not isinstance(self.include_pivots, bool):
+            raise ValueError("GTL include_pivots must be a boolean")
+        if not 0.0 <= self.zone_tolerance < 0.10:
+            raise ValueError("GTL zone_tolerance must be in [0, 0.10)")
+        if self.candidate_limit is not None and self.candidate_limit < 1:
+            raise ValueError("GTL candidate_limit must be positive or null")
+        if self.max_target_atr is not None and self.max_target_atr <= 0:
+            raise ValueError("GTL max_target_atr must be positive or null")
+        # Reuse the detector's validation for its subset of fields.
+        self.ladder_config()
+
+    def ladder_config(self) -> GateTargetLadderConfig:
+        return GateTargetLadderConfig(
+            lookback_bars=self.lookback_bars,
+            grid_bins=self.grid_bins,
+            include_pivots=self.include_pivots,
+            pivot_window=self.pivot_window,
+            touch_tolerance=self.touch_tolerance,
+            merge_tolerance=self.merge_tolerance,
+            strength_scale=self.strength_scale,
+        )
+
 # 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
@@ -170,6 +221,7 @@ SR_RESEARCH_VARIANTS = {
     "legacy_range_grid_touch",
     "legacy_range_grid_neutral",
     EXPLICIT_TARGET_LADDER_VARIANT,
+    GTL_TUNING_VARIANT,
     STRUCTURAL_OVERLAY_VARIANT,
     *RANGE_FACTOR_VARIANTS,
 }
@@ -184,6 +236,31 @@ def _sr_research_variant() -> str:
     return value
 
 
+@lru_cache(maxsize=64)
+def _parse_gtl_research_config(raw: str) -> GTLResearchConfig:
+    """Parse one auditable JSON configuration passed by the offline runner."""
+    if not raw.strip():
+        return GTLResearchConfig()
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise ValueError("BACKTEST_GTL_CONFIG must be valid JSON") from exc
+    if not isinstance(payload, dict):
+        raise ValueError("BACKTEST_GTL_CONFIG must be a JSON object")
+    allowed = set(GTLResearchConfig.__dataclass_fields__)
+    unknown = sorted(set(payload) - allowed)
+    if unknown:
+        raise ValueError(f"Unknown GTL research config fields: {', '.join(unknown)}")
+    try:
+        return GTLResearchConfig(**payload)
+    except TypeError as exc:
+        raise ValueError(f"Invalid GTL research config: {exc}") from exc
+
+
+def _gtl_research_config() -> GTLResearchConfig:
+    return _parse_gtl_research_config(os.getenv("BACKTEST_GTL_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}:
@@ -224,6 +301,7 @@ def _primary_min_rr_for_variant(sr_variant: str, activation: dict) -> float:
             "production_control",
             "production_range504",
             EXPLICIT_TARGET_LADDER_VARIANT,
+            GTL_TUNING_VARIANT,
             STRUCTURAL_OVERLAY_VARIANT,
         }
         or sr_variant.endswith("_legacy_primary")
@@ -366,6 +444,7 @@ 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
     range_504_log = _range_504_log(highs, lows)
     if sr_variant == "legacy_geometry_neutral":
         detected_levels = detect_sr_levels_legacy(
@@ -385,6 +464,15 @@ def _window_setups(
             lows,
             closes,
         )
+    elif sr_variant == GTL_TUNING_VARIANT:
+        if gtl_config is None:  # pragma: no cover - guarded by the variant above
+            raise RuntimeError("GTL tuning variant requires a research config")
+        detected_levels = detect_gate_target_ladder(
+            highs,
+            lows,
+            closes,
+            config=gtl_config.ladder_config(),
+        )
     elif sr_variant in {"legacy_range_grid_touch", "legacy_range_grid_neutral"}:
         detected_levels = detect_sr_levels_legacy(
             highs,
@@ -435,9 +523,20 @@ def _window_setups(
             gate_levels,
             entry,
             strength_mode=zone_strength_mode,
+            tolerance=(
+                gtl_config.zone_tolerance if gtl_config else _SR_ZONE_TOLERANCE
+            ),
         )
         zone_levels = _apply_zone_strength_variant(zone_levels, sr_variant)
-        targets = target_generator.generate_targets(direction, entry, stop, zone_levels, atr)
+        targets = target_generator.generate_targets(
+            direction,
+            entry,
+            stop,
+            zone_levels,
+            atr,
+            max_targets=(gtl_config.candidate_limit if gtl_config else 5),
+            max_atr_multiple_override=(gtl_config.max_target_atr if gtl_config else None),
+        )
         if not targets:
             fallback_k = _atr_target_fallback_k()
             if fallback_k is None:
@@ -3326,6 +3425,11 @@ async def run_backtest(
             "min_lookback": MIN_LOOKBACK,
             "cost_per_side_pct": round(COST_PER_SIDE * 100, 3),
             "sr_variant": _sr_research_variant(),
+            "gtl_config": (
+                asdict(_gtl_research_config())
+                if _sr_research_variant() == GTL_TUNING_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),

app/services/recommendation_service.py

@@ -93,6 +93,7 @@ def _zone_representative_levels(
     entry_price: float,
     *,
     strength_mode: str = "sum",
+    tolerance: float = _SR_ZONE_TOLERANCE,
 ) -> list[Any]:
     """Collapse near-duplicate S/R levels into one representative per zone.
 
@@ -124,7 +125,7 @@ def _zone_representative_levels(
     zones = cluster_sr_zones(
         level_dicts,
         entry_price,
-        tolerance=_SR_ZONE_TOLERANCE,
+        tolerance=tolerance,
         strength_mode=strength_mode,
     )
 
@@ -315,9 +316,16 @@ class TargetGenerator:
         stop_loss: float,
         sr_levels: list[SRLevel],
         atr_value: float,
+        *,
+        max_targets: int | None = 5,
+        max_atr_multiple_override: float | None = None,
     ) -> list[dict[str, Any]]:
         if atr_value <= 0:
             return []
+        if max_targets is not None and max_targets < 1:
+            raise ValueError("max_targets must be positive or None")
+        if max_atr_multiple_override is not None and max_atr_multiple_override <= 0:
+            raise ValueError("max_atr_multiple_override must be positive or None")
 
         risk = abs(entry_price - stop_loss)
         if risk <= 0:
@@ -326,11 +334,12 @@ class TargetGenerator:
         candidates: list[dict[str, Any]] = []
         atr_pct = atr_value / entry_price if entry_price > 0 else 0.0
 
-        max_atr_multiple: float | None = None
-        if atr_pct > 0.05:
-            max_atr_multiple = 10.0
-        elif atr_pct < 0.02:
-            max_atr_multiple = 3.0
+        max_atr_multiple: float | None = max_atr_multiple_override
+        if max_atr_multiple is None:
+            if atr_pct > 0.05:
+                max_atr_multiple = 10.0
+            elif atr_pct < 0.02:
+                max_atr_multiple = 3.0
 
         for level in sr_levels:
             is_candidate = False
@@ -383,6 +392,12 @@ class TargetGenerator:
         if not candidates:
             return []
 
+        if max_targets is None:
+            candidates.sort(key=lambda row: row["distance_from_entry"])
+            for target in candidates:
+                target.pop("quality", None)
+            return candidates
+
         # Select up to 5 targets that SPAN the distance range, instead of the
         # top-5 by quality (which biases toward far, high-R:R levels and buries
         # every nearby target). Guarantees the nearest level plus a
@@ -396,6 +411,8 @@ class TargetGenerator:
         selected_ids: set[int] = set()
 
         def _add(candidate: dict[str, Any] | None) -> None:
+            if len(selected) >= max_targets:
+                return
             if candidate is not None and candidate["sr_level_id"] not in selected_ids:
                 selected.append(candidate)
                 selected_ids.add(candidate["sr_level_id"])
@@ -408,7 +425,7 @@ class TargetGenerator:
                 _add(max(bucket, key=lambda c: c["quality"]))
         # Fill remaining slots with the next-best by quality
         for candidate in sorted(candidates, key=lambda c: c["quality"], reverse=True):
-            if len(selected) >= 5:
+            if len(selected) >= max_targets:
                 break
             _add(candidate)
 

app/services/sr_service.py

@@ -10,6 +10,7 @@ and persists to DB.
 from __future__ import annotations
 
 import math
+from dataclasses import dataclass
 from datetime import datetime
 
 from sqlalchemy import delete, select
@@ -32,6 +33,40 @@ from app.services.price_service import query_ohlcv
 
 DEFAULT_TOLERANCE = 0.005  # fallback when ATR unavailable; also API legacy default
 
+
+@dataclass(frozen=True)
+class GateTargetLadderConfig:
+    """Research controls for the transient Gate Target Ladder.
+
+    Live callers omit this object and retain the frozen legacy-parity path.
+    The backtest may pass an explicit configuration to isolate one ladder
+    mechanism at a time without changing Structural S/R.
+    """
+
+    lookback_bars: int | None = None
+    grid_bins: int = 20
+    include_pivots: bool = True
+    pivot_window: int = 2
+    touch_tolerance: float = DEFAULT_TOLERANCE
+    merge_tolerance: float = DEFAULT_TOLERANCE
+    strength_scale: float = 500.0
+
+    def __post_init__(self) -> None:
+        if self.lookback_bars is not None and self.lookback_bars < 20:
+            raise ValueError("GTL lookback_bars must be at least 20 or None")
+        if self.grid_bins < 2:
+            raise ValueError("GTL grid_bins must be at least 2")
+        if self.pivot_window < 1:
+            raise ValueError("GTL pivot_window must be at least 1")
+        for name, value in (
+            ("touch_tolerance", self.touch_tolerance),
+            ("merge_tolerance", self.merge_tolerance),
+        ):
+            if not 0.0 <= value < 0.10:
+                raise ValueError(f"GTL {name} must be in [0, 0.10)")
+        if self.strength_scale <= 0:
+            raise ValueError("GTL strength_scale must be positive")
+
 VP_LOOKBACK = 252
 TOUCH_LOOKBACK = 252
 PIVOT_LOOKBACK = 504
@@ -527,6 +562,8 @@ def detect_gate_target_ladder(
     lows: list[float],
     closes: list[float],
     tolerance: float = DEFAULT_TOLERANCE,
+    *,
+    config: GateTargetLadderConfig | None = None,
 ) -> list[dict]:
     """Build the scanner's internal, volume-free target proposal ladder.
 
@@ -536,14 +573,115 @@ def detect_gate_target_ladder(
     profile calculation. The returned levels are transient and must not be
     persisted as chart S/R.
     """
-    return detect_sr_levels_legacy(
-        highs,
-        lows,
-        closes,
-        [0] * len(closes),
-        tolerance,
-        explicit_range_grid=True,
-    )
+    if config is None:
+        # Frozen live/default path. Keeping the established helper here makes
+        # the absence of research configuration an exact compatibility promise.
+        return detect_sr_levels_legacy(
+            highs,
+            lows,
+            closes,
+            [0] * len(closes),
+            tolerance,
+            explicit_range_grid=True,
+        )
+
+    if tolerance != DEFAULT_TOLERANCE:
+        raise ValueError("Pass either GTL config tolerances or tolerance, not both")
+    if not closes:
+        return []
+    if not (len(highs) == len(lows) == len(closes)):
+        raise ValueError("GTL highs, lows, and closes must have equal lengths")
+
+    if config.lookback_bars is not None:
+        highs = highs[-config.lookback_bars:]
+        lows = lows[-config.lookback_bars:]
+        closes = closes[-config.lookback_bars:]
+
+    candidates: list[tuple[float, str]] = []
+    try:
+        candidates.extend(
+            (float(price), "range_grid")
+            for price in _legacy_range_grid_nodes(
+                highs,
+                lows,
+                closes,
+                num_bins=config.grid_bins,
+            )
+        )
+    except ValidationError:
+        pass
+
+    if config.include_pivots:
+        try:
+            pivots = compute_pivot_points(
+                highs,
+                lows,
+                closes,
+                window=config.pivot_window,
+            )
+            candidates.extend(
+                (float(price), "pivot_point")
+                for price in pivots.get("swing_highs", []) + pivots.get("swing_lows", [])
+            )
+        except ValidationError:
+            pass
+    if not candidates:
+        return []
+
+    total_bars = len(closes)
+    raw: list[dict] = []
+    for price, method in candidates:
+        touch_band = abs(price) * config.touch_tolerance if price else config.touch_tolerance
+        touches = sum(
+            1
+            for low, high in zip(lows, highs, strict=False)
+            if low - touch_band <= price <= high + touch_band
+        )
+        strength = max(
+            0,
+            min(100, int(round((touches / total_bars) * config.strength_scale))),
+        )
+        raw.append({
+            "price_level": price,
+            "strength": strength,
+            "detection_method": method,
+            "type": "",
+            "sources": [method],
+            "rejection_count": touches,
+            "last_rejection_age": None,
+            "weighted_respects": float(touches),
+        })
+
+    merged: list[dict] = []
+    for level in sorted(raw, key=lambda row: row["price_level"]):
+        if not merged:
+            merged.append(dict(level))
+            continue
+        last = merged[-1]
+        ref = last["price_level"]
+        merge_band = abs(ref) * config.merge_tolerance if ref else config.merge_tolerance
+        if abs(level["price_level"] - ref) > merge_band:
+            merged.append(dict(level))
+            continue
+        last["price_level"] = round(
+            (last["price_level"] + level["price_level"]) / 2.0,
+            4,
+        )
+        last["strength"] = min(100, last["strength"] + level["strength"])
+        sources = set(last.get("sources") or [last["detection_method"]])
+        sources |= set(level.get("sources") or [level["detection_method"]])
+        last["sources"] = sorted(sources)
+        last["detection_method"] = (
+            next(iter(sources)) if len(sources) == 1 else "merged"
+        )
+        last["rejection_count"] = max(
+            int(last.get("rejection_count", 0)),
+            int(level.get("rejection_count", 0)),
+        )
+
+    _tag_levels(merged, closes[-1])
+    merged.sort(key=lambda row: row["strength"], reverse=True)
+    return merged
 
 
 def _merge_levels(