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

Retire completed GTL tuning harnesses

Browse Source
This commit is contained in:
dennisthiessen
2026-07-13 16:40:29 +02:00
parent 1d84a40c04
commit 8e09f239c8
16 changed files with 59 additions and 2057 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+7 -59
View File
@@ -459,65 +459,14 @@ metrics. Keep the SSH tunnel open only while creating the snapshot; the backtest
run itself is local/offline. `backtest_snapshots/` and generated backtest reports run itself is local/offline. `backtest_snapshots/` and generated backtest reports
are git-ignored. are git-ignored.
### One-command GTL tuning run ### Archived GTL tuning decision
The Gate Target Ladder has a research-only, single-variable matrix for testing The completed replacement, cohort-composition, and strength-sensitivity
whether its useful screening behavior can be made more explicit. It runs 20 matrices found no stable improvement over the frozen Gate Target Ladder. The
complete backtests **sequentially** so each arm gets the full worker pool: temporary matrix runners and tuning hooks have been retired; their three compact
consolidated report pairs remain in `reports/` as the decision audit. Keep the
```bash GTL unchanged and evaluate any future challenger only on new forward data. See
# macOS/Linux the [full research record](docs/research/sr-levels-and-exits.md#gtl-tuning-matrix).
.venv/bin/python scripts/run_gtl_tuning_matrix.py \
backtest_snapshots/prod.sqlite --workers 14
```
The arms cover GTL history length, grid density, pivots, price-traffic scoring,
proposal merging, target-zone width, candidate count, and maximum target
distance. Every arm includes the full-period production book, the fixed
`2024-07-01` train/test split, a candidate-level paired audit, and the same
pre-registered robustness screen. This is one long command, not a Cartesian
parameter search; expect total runtime to be roughly 20 times one full local
backtest.
Progress is checkpointed after every arm to
`reports/backtest-YYYYMMDD-gtl-tuning-matrix.json` and the matching `.md` table.
The large per-arm reports are removed only after successful consolidation. If
the run fails, they remain available for diagnosis; pass `--keep-arm-reports`
to retain them after success too. No arm changes live scanner defaults or
deploys anything.
The replacement matrix found no winning single constant. Its evidence-selected
follow-up keeps control geometry and isolates retained versus added cohorts in
one 13-arm confirmation/union run:
```bash
# macOS/Linux
.venv/bin/python scripts/run_gtl_confirmation_matrix.py \
backtest_snapshots/prod.sqlite --workers 12
```
It first verifies exact control parity with the completed tuning matrix, then
checkpoints consolidated JSON and Markdown reports under
`reports/backtest-YYYYMMDD-gtl-confirmation-matrix.*`.
The confirmation matrix's only near-hit was strength-1000 intersection: it
improved full/train/post-2024 Sharpe but missed the unchanged drawdown guardrail
by 0.3 percentage points. The final narrow sensitivity check is:
```bash
.venv/bin/python scripts/run_gtl_strength_sensitivity.py \
backtest_snapshots/prod.sqlite --workers 12
```
It checks eight coarse scales around 1000, verifies exact control and
strength-1000 replication, and requires two adjacent scales to pass every
original guardrail before calling the result stable.
Final result: control and strength-1000 replication both passed, but there was
no adjacent passing plateau. Scale 1500 passed in isolation while lowering CAGR
and setup expectancy; its neighbors failed. The research decision is therefore
to keep the frozen GTL unchanged and evaluate any future challenger only on new
forward data. See the [full research record](docs/research/sr-levels-and-exits.md#gtl-tuning-matrix).
### Reading a local backtest report ### Reading a local backtest report
@@ -557,7 +506,6 @@ Research-only flags, all off by default (the default report is byte-identical to
| `BACKTEST_HOLDOUT_SPLIT=YYYY-MM-DD` | Adds a `holdout` section: train (entries before) vs test (entries on/after), as disjoint books | | `BACKTEST_HOLDOUT_SPLIT=YYYY-MM-DD` | Adds a `holdout` section: train (entries before) vs test (entries on/after), as disjoint books |
| `BACKTEST_MIN_RR_SWEEP=1` | Sweeps the activation R:R floor against portfolio Sharpe. Combine with `BACKTEST_HOLDOUT_SPLIT` to sweep out-of-sample | | `BACKTEST_MIN_RR_SWEEP=1` | Sweeps the activation R:R floor against portfolio Sharpe. Combine with `BACKTEST_HOLDOUT_SPLIT` to sweep out-of-sample |
| `BACKTEST_SR_VARIANT=<arm>` | Research-only S/R detector/gate arm; see `docs/research/sr-levels-and-exits.md` for the detector and hidden-feature matrices | | `BACKTEST_SR_VARIANT=<arm>` | Research-only S/R detector/gate arm; see `docs/research/sr-levels-and-exits.md` for the detector and hidden-feature matrices |
| `BACKTEST_GTL_CONFIG=<json>` | Parameterizes only the `gtl_tuning` research arm; normally set by `run_gtl_tuning_matrix.py` |
| `BACKTEST_ENTRY_START=YYYY-MM-DD` | Restrict candidate entry dates to a validation window | | `BACKTEST_ENTRY_START=YYYY-MM-DD` | Restrict candidate entry dates to a validation window |
| `BACKTEST_ENTRY_END=YYYY-MM-DD` | Restrict candidate entry dates to a training window | | `BACKTEST_ENTRY_END=YYYY-MM-DD` | Restrict candidate entry dates to a training window |
| `BACKTEST_SR_AUDIT=1` | Add momentum-slice candidate rows for paired S/R cohort comparison | | `BACKTEST_SR_AUDIT=1` | Add momentum-slice candidate rows for paired S/R cohort comparison |
app/services/backtest_service.py
+6 -291
View File
@@ -33,9 +33,7 @@ import statistics
from collections import defaultdict from collections import defaultdict
from collections.abc import Callable from collections.abc import Callable
from concurrent.futures import ProcessPoolExecutor from concurrent.futures import ProcessPoolExecutor
from dataclasses import asdict, dataclass
from datetime import date, datetime, timedelta, timezone from datetime import date, datetime, timedelta, timezone
from functools import lru_cache
from types import SimpleNamespace from types import SimpleNamespace
from typing import Any from typing import Any
@@ -73,7 +71,6 @@ from app.services.recommendation_service import (
_prune_floor_pinned_targets, _prune_floor_pinned_targets,
_risk_level_from_conflicts, _risk_level_from_conflicts,
_select_primary_target, _select_primary_target,
_SR_ZONE_TOLERANCE,
_zone_representative_levels, _zone_representative_levels,
direction_analyzer, direction_analyzer,
get_recommendation_config, get_recommendation_config,
@@ -86,7 +83,6 @@ from app.services.scoring_service import (
compute_technical_from_arrays, compute_technical_from_arrays,
) )
from app.services.sr_service import ( from app.services.sr_service import (
GateTargetLadderConfig,
MAX_LEVELS, MAX_LEVELS,
detect_gate_target_ladder, detect_gate_target_ladder,
detect_sr_levels, detect_sr_levels,
@@ -110,8 +106,6 @@ STRUCTURAL_OVERLAY_SOURCE_VARIANT = (
STRUCTURAL_OVERLAY_WEIGHT = 0.05 STRUCTURAL_OVERLAY_WEIGHT = 0.05
STRUCTURAL_OVERLAY_SCORE_KEY = "structural_overlay_95_5_score" STRUCTURAL_OVERLAY_SCORE_KEY = "structural_overlay_95_5_score"
EXPLICIT_TARGET_LADDER_VARIANT = "explicit_target_ladder" EXPLICIT_TARGET_LADDER_VARIANT = "explicit_target_ladder"
GTL_TUNING_VARIANT = "gtl_tuning"
GTL_CONFIRMATION_VARIANT = "gtl_confirmation"
RANGE_RESIDUAL_VARIANTS = { RANGE_RESIDUAL_VARIANTS = {
"rewrite_range504_structural_legacy_primary", "rewrite_range504_structural_legacy_primary",
"rewrite_range504_structural_primary2", "rewrite_range504_structural_primary2",
@@ -122,69 +116,6 @@ RANGE_FACTOR_VARIANTS = {
*RANGE_RESIDUAL_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,
)
@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 # Cross-sectional signal evaluation (factor IC). Each candidate signal is a
# point-in-time number computed from closes alone (sentiment/fundamentals have no # 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 # history here), sampled one as-of per ISO week, and graded by how its rank
@@ -239,8 +170,6 @@ SR_RESEARCH_VARIANTS = {
"legacy_range_grid_touch", "legacy_range_grid_touch",
"legacy_range_grid_neutral", "legacy_range_grid_neutral",
EXPLICIT_TARGET_LADDER_VARIANT, EXPLICIT_TARGET_LADDER_VARIANT,
GTL_TUNING_VARIANT,
GTL_CONFIRMATION_VARIANT,
STRUCTURAL_OVERLAY_VARIANT, STRUCTURAL_OVERLAY_VARIANT,
*RANGE_FACTOR_VARIANTS, *RANGE_FACTOR_VARIANTS,
} }
@@ -255,72 +184,6 @@ def _sr_research_variant() -> str:
return value 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", ""))
@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: def _sr_detector_variant(sr_variant: str) -> str:
"""Map factor-gated research arms to the detector they hold fixed.""" """Map factor-gated research arms to the detector they hold fixed."""
if sr_variant in {"production_range504", STRUCTURAL_OVERLAY_VARIANT}: if sr_variant in {"production_range504", STRUCTURAL_OVERLAY_VARIANT}:
@@ -361,8 +224,6 @@ def _primary_min_rr_for_variant(sr_variant: str, activation: dict) -> float:
"production_control", "production_control",
"production_range504", "production_range504",
EXPLICIT_TARGET_LADDER_VARIANT, EXPLICIT_TARGET_LADDER_VARIANT,
GTL_TUNING_VARIANT,
GTL_CONFIRMATION_VARIANT,
STRUCTURAL_OVERLAY_VARIANT, STRUCTURAL_OVERLAY_VARIANT,
} }
or sr_variant.endswith("_legacy_primary") or sr_variant.endswith("_legacy_primary")
@@ -477,7 +338,6 @@ def _window_setups(
activation: dict, activation: dict,
*, *,
sr_variant: str | None = None, sr_variant: str | None = None,
gtl_research_config: GTLResearchConfig | None = None,
) -> list[dict]: ) -> list[dict]:
"""Rebuild the setup(s) at the last bar of ``window_records`` (the as-of date), """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.""" using only those bars. Returns one dict per tradeable direction."""
@@ -506,11 +366,6 @@ def _window_setups(
sr_variant = sr_variant or _sr_research_variant() sr_variant = sr_variant or _sr_research_variant()
detector_variant = _sr_detector_variant(sr_variant) detector_variant = _sr_detector_variant(sr_variant)
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) range_504_log = _range_504_log(highs, lows)
if sr_variant == "legacy_geometry_neutral": if sr_variant == "legacy_geometry_neutral":
detected_levels = detect_sr_levels_legacy( detected_levels = detect_sr_levels_legacy(
@@ -530,15 +385,6 @@ def _window_setups(
lows, lows,
closes, 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"}: elif sr_variant in {"legacy_range_grid_touch", "legacy_range_grid_neutral"}:
detected_levels = detect_sr_levels_legacy( detected_levels = detect_sr_levels_legacy(
highs, highs,
@@ -589,20 +435,9 @@ def _window_setups(
gate_levels, gate_levels,
entry, entry,
strength_mode=zone_strength_mode, 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) zone_levels = _apply_zone_strength_variant(zone_levels, sr_variant)
targets = target_generator.generate_targets( targets = target_generator.generate_targets(direction, entry, stop, zone_levels, atr)
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: if not targets:
fallback_k = _atr_target_fallback_k() fallback_k = _atr_target_fallback_k()
if fallback_k is None: if fallback_k is None:
@@ -749,87 +584,6 @@ def _structural_overlay_window_setups(
return tagged 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: 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 """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, bar's open when price gapped through it — so a gap can lose more than −1R,
@@ -934,17 +688,16 @@ def _replay_ticker(
) )
vol_6m = _realized_vol_6m(closes, len(window) - 1) vol_6m = _realized_vol_6m(closes, len(window) - 1)
if sr_variant == STRUCTURAL_OVERLAY_VARIANT: setups = (
setups = _structural_overlay_window_setups(window, config, activation) _structural_overlay_window_setups(window, config, activation)
elif sr_variant == GTL_CONFIRMATION_VARIANT: if sr_variant == STRUCTURAL_OVERLAY_VARIANT
setups = _gtl_confirmation_window_setups(window, config, activation) else _window_setups(
else:
setups = _window_setups(
window, window,
config, config,
activation, activation,
sr_variant=sr_variant, sr_variant=sr_variant,
) )
)
for s in setups: for s in setups:
outcome, outcome_date = evaluate_setup_against_bars( outcome, outcome_date = evaluate_setup_against_bars(
s["direction"], s["stop"], s["target"], forward_bars, HORIZON s["direction"], s["stop"], s["target"], forward_bars, HORIZON
@@ -1010,13 +763,6 @@ def _replay_ticker(
"structural_overlay_gate_level_count": s.get( "structural_overlay_gate_level_count": s.get(
"structural_overlay_gate_level_count" "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, "outcome": outcome,
"target_hit": target_hit, "target_hit": target_hit,
"realized_r": realized_r, "realized_r": realized_r,
@@ -1089,9 +835,6 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
range_logs: list[float] = [] range_logs: list[float] = []
overlay_rows = 0 overlay_rows = 0
overlay_pass = 0 overlay_pass = 0
confirmation_rows = 0
confirmation_pass = 0
confirmation_tuned_additions = 0
for cand in candidates: for cand in candidates:
sources = list(cand.get("primary_sources") or []) sources = list(cand.get("primary_sources") or [])
for source in sources: for source in sources:
@@ -1107,12 +850,6 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
if cand.get("structural_overlay_pass") is not None: if cand.get("structural_overlay_pass") is not None:
overlay_rows += 1 overlay_rows += 1
overlay_pass += int(bool(cand["structural_overlay_pass"])) 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: def avg(values: list[float] | list[int]) -> float | None:
return round(sum(values) / len(values), 3) if values else None return round(sum(values) / len(values), 3) if values else None
@@ -1137,9 +874,6 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
"structural_overlay_weight": ( "structural_overlay_weight": (
STRUCTURAL_OVERLAY_WEIGHT if overlay_rows else None 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,
} }
@@ -1186,15 +920,6 @@ def _sr_candidate_audit(candidates: list[dict], min_percentile: float) -> list[d
"structural_overlay_gate_level_count": int( "structural_overlay_gate_level_count": int(
cand.get("structural_overlay_gate_level_count", 0) or 0 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), "rr": round(float(cand.get("rr", 0.0)), 6),
"primary_prob": round(float(cand.get("primary_prob", 0.0)), 6), "primary_prob": round(float(cand.get("primary_prob", 0.0)), 6),
"primary_sources": list(cand.get("primary_sources") or []), "primary_sources": list(cand.get("primary_sources") or []),
@@ -3601,16 +3326,6 @@ async def run_backtest(
"min_lookback": MIN_LOOKBACK, "min_lookback": MIN_LOOKBACK,
"cost_per_side_pct": round(COST_PER_SIDE * 100, 3), "cost_per_side_pct": round(COST_PER_SIDE * 100, 3),
"sr_variant": _sr_research_variant(), "sr_variant": _sr_research_variant(),
"gtl_config": (
asdict(_gtl_research_config())
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_lookback": RANGE_FACTOR_LOOKBACK,
"range_factor_min_log": RANGE_FACTOR_MIN_LOG, "range_factor_min_log": RANGE_FACTOR_MIN_LOG,
"range_factor_min_ratio": round(math.exp(RANGE_FACTOR_MIN_LOG), 4), "range_factor_min_ratio": round(math.exp(RANGE_FACTOR_MIN_LOG), 4),
app/services/recommendation_service.py
+7 -24
View File
@@ -93,7 +93,6 @@ def _zone_representative_levels(
entry_price: float, entry_price: float,
*, *,
strength_mode: str = "sum", strength_mode: str = "sum",
tolerance: float = _SR_ZONE_TOLERANCE,
) -> list[Any]: ) -> list[Any]:
"""Collapse near-duplicate S/R levels into one representative per zone. """Collapse near-duplicate S/R levels into one representative per zone.
@@ -125,7 +124,7 @@ def _zone_representative_levels(
zones = cluster_sr_zones( zones = cluster_sr_zones(
level_dicts, level_dicts,
entry_price, entry_price,
tolerance=tolerance, tolerance=_SR_ZONE_TOLERANCE,
strength_mode=strength_mode, strength_mode=strength_mode,
) )
@@ -316,16 +315,9 @@ class TargetGenerator:
stop_loss: float, stop_loss: float,
sr_levels: list[SRLevel], sr_levels: list[SRLevel],
atr_value: float, atr_value: float,
*,
max_targets: int | None = 5,
max_atr_multiple_override: float | None = None,
) -> list[dict[str, Any]]: ) -> list[dict[str, Any]]:
if atr_value <= 0: if atr_value <= 0:
return [] 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) risk = abs(entry_price - stop_loss)
if risk <= 0: if risk <= 0:
@@ -334,12 +326,11 @@ class TargetGenerator:
candidates: list[dict[str, Any]] = [] candidates: list[dict[str, Any]] = []
atr_pct = atr_value / entry_price if entry_price > 0 else 0.0 atr_pct = atr_value / entry_price if entry_price > 0 else 0.0
max_atr_multiple: float | None = max_atr_multiple_override max_atr_multiple: float | None = None
if max_atr_multiple is None: if atr_pct > 0.05:
if atr_pct > 0.05: max_atr_multiple = 10.0
max_atr_multiple = 10.0 elif atr_pct < 0.02:
elif atr_pct < 0.02: max_atr_multiple = 3.0
max_atr_multiple = 3.0
for level in sr_levels: for level in sr_levels:
is_candidate = False is_candidate = False
@@ -392,12 +383,6 @@ class TargetGenerator:
if not candidates: if not candidates:
return [] 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 # 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 # top-5 by quality (which biases toward far, high-R:R levels and buries
# every nearby target). Guarantees the nearest level plus a # every nearby target). Guarantees the nearest level plus a
@@ -411,8 +396,6 @@ class TargetGenerator:
selected_ids: set[int] = set() selected_ids: set[int] = set()
def _add(candidate: dict[str, Any] | None) -> None: 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: if candidate is not None and candidate["sr_level_id"] not in selected_ids:
selected.append(candidate) selected.append(candidate)
selected_ids.add(candidate["sr_level_id"]) selected_ids.add(candidate["sr_level_id"])
@@ -425,7 +408,7 @@ class TargetGenerator:
_add(max(bucket, key=lambda c: c["quality"])) _add(max(bucket, key=lambda c: c["quality"]))
# Fill remaining slots with the next-best by quality # Fill remaining slots with the next-best by quality
for candidate in sorted(candidates, key=lambda c: c["quality"], reverse=True): for candidate in sorted(candidates, key=lambda c: c["quality"], reverse=True):
if len(selected) >= max_targets: if len(selected) >= 5:
break break
_add(candidate) _add(candidate)
app/services/sr_service.py
+8 -146
View File
@@ -10,7 +10,6 @@ and persists to DB.
from __future__ import annotations from __future__ import annotations
import math import math
from dataclasses import dataclass
from datetime import datetime from datetime import datetime
from sqlalchemy import delete, select from sqlalchemy import delete, select
@@ -33,40 +32,6 @@ from app.services.price_service import query_ohlcv
DEFAULT_TOLERANCE = 0.005 # fallback when ATR unavailable; also API legacy default 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 VP_LOOKBACK = 252
TOUCH_LOOKBACK = 252 TOUCH_LOOKBACK = 252
PIVOT_LOOKBACK = 504 PIVOT_LOOKBACK = 504
@@ -562,8 +527,6 @@ def detect_gate_target_ladder(
lows: list[float], lows: list[float],
closes: list[float], closes: list[float],
tolerance: float = DEFAULT_TOLERANCE, tolerance: float = DEFAULT_TOLERANCE,
*,
config: GateTargetLadderConfig | None = None,
) -> list[dict]: ) -> list[dict]:
"""Build the scanner's internal, volume-free target proposal ladder. """Build the scanner's internal, volume-free target proposal ladder.
@@ -573,115 +536,14 @@ def detect_gate_target_ladder(
profile calculation. The returned levels are transient and must not be profile calculation. The returned levels are transient and must not be
persisted as chart S/R. persisted as chart S/R.
""" """
if config is None: return detect_sr_levels_legacy(
# Frozen live/default path. Keeping the established helper here makes highs,
# the absence of research configuration an exact compatibility promise. lows,
return detect_sr_levels_legacy( closes,
highs, [0] * len(closes),
lows, tolerance,
closes, explicit_range_grid=True,
[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( def _merge_levels(
docs/research/sr-levels-and-exits.md
+16 -28
View File
@@ -806,10 +806,10 @@ or perform a production deployment.
#### GTL tuning matrix #### GTL tuning matrix
Exact parity establishes a safe, explicit control, but it does not prove that Exact parity established a safe, explicit control but did not prove that the
the inherited GTL constants are optimal. The `gtl_tuning` backtest arm exposes inherited GTL constants were optimal. A temporary offline harness exposed those
only those constants to an offline configuration; the live scanner continues constants without changing the live scanner. That harness has now been retired;
to call the frozen default helper and cannot read this research configuration. the compact consolidated reports remain as the reproducible decision record.
The single-command matrix contains 20 full-period arms. Each non-control arm The single-command matrix contains 20 full-period arms. Each non-control arm
changes exactly one input: changes exactly one input:
@@ -826,11 +826,6 @@ changes exactly one input:
| Pivots | Five-bar swings | None / eleven-bar swings | Do pivots add anything beyond the range ladder? | | Pivots | Five-bar swings | None / eleven-bar swings | Do pivots add anything beyond the range ladder? |
| Traffic strength scale | 500 | 250 / 1000 | Does strength saturation affect probability/selection? | | Traffic strength scale | 500 | 250 / 1000 | Does strength saturation affect probability/selection? |
```bash
.venv/bin/python scripts/run_gtl_tuning_matrix.py \
backtest_snapshots/prod.sqlite --workers 14
```
Arms execute sequentially so multiprocessing pools never compete. Each arm Arms execute sequentially so multiprocessing pools never compete. Each arm
produces full-period production metrics, train/test books split at 2024-07-01, produces full-period production metrics, train/test books split at 2024-07-01,
robust expectancy after removing the top 5% of setups, and retained/added/ robust expectancy after removing the top 5% of setups, and retained/added/
@@ -862,22 +857,16 @@ The paired cohorts do expose a narrower mechanism worth testing:
| Grid without pivots | 428 at +0.232R (+0.100) | 392 at +0.176R (+0.046) | 658 at +0.208R (+0.041) | | Grid without pivots | 428 at +0.232R (+0.100) | 392 at +0.176R (+0.046) | 658 at +0.208R (+0.041) |
Replacement mixes the retained and added cohorts and also discards the removed Replacement mixes the retained and added cohorts and also discards the removed
cohort, so its portfolio result cannot say which part helped. The follow-up cohort, so its portfolio result could not say which part helped. The archived
`gtl_confirmation` matrix therefore preserves frozen control geometry and confirmation matrix therefore preserved frozen control geometry and decomposed
decomposes each selected variant into: each selected variant into:
- **intersection** — only control setups also core-qualified by the variant; - **intersection** — only control setups also core-qualified by the variant;
- **union** — all core-qualified control setups plus genuinely added variant - **union** — all core-qualified control setups plus genuinely added variant
setups, using tuned geometry only for those additions. setups, using tuned geometry only for those additions.
It also tests pre-registered intersections among the three high-breadth It also tests pre-registered intersections among the three high-breadth
confirmers. Its control path must exactly reproduce the completed tuning confirmers. Its control path exactly reproduced the completed tuning matrix.
matrix before any research arm is accepted.
```bash
.venv/bin/python scripts/run_gtl_confirmation_matrix.py \
backtest_snapshots/prod.sqlite --workers 12
```
Result: **13/13 arms completed with exact control parity; no arm passed all six Result: **13/13 arms completed with exact control parity; no arm passed all six
guardrails.** The decomposition does identify one near-hit: guardrails.** The decomposition does identify one near-hit:
@@ -894,17 +883,12 @@ ex-top-5%. This is consistent with a weak tail-dependence filter. It is not yet
a winner: the original drawdown guardrail remains fixed, and 21.7% is worse a winner: the original drawdown guardrail remains fixed, and 21.7% is worse
than 21.4% even though the difference is small. than 21.4% even though the difference is small.
The final parameter test is therefore deliberately one-dimensional. It sweeps The final parameter test was deliberately one-dimensional. It swept
coarse strength scales around 1000 (625, 750, 875, 1000, 1125, 1250, 1500, coarse strength scales around 1000 (625, 750, 875, 1000, 1125, 1250, 1500,
2000), using intersection only. It must reproduce both the frozen control and 2000), using intersection only. It reproduced both the frozen control and the
the completed strength-1000 result exactly. Promotion requires at least two completed strength-1000 result exactly. Promotion required at least two
adjacent non-control scales to pass all six original checks; an isolated winner adjacent non-control scales to pass all six original checks; an isolated winner
is rejected as sensitivity. was rejected as sensitivity.
```bash
.venv/bin/python scripts/run_gtl_strength_sensitivity.py \
backtest_snapshots/prod.sqlite --workers 12
```
Final result: **9/9 arms completed, control parity passed, and the Final result: **9/9 arms completed, control parity passed, and the
strength-1000 replication passed.** Scale 1500 was the only arm to clear all strength-1000 replication passed.** Scale 1500 was the only arm to clear all
@@ -928,6 +912,10 @@ model. This snapshot is now exhausted for GTL fitting; any future challenger
must be pre-registered and evaluated on genuinely new forward data rather than must be pre-registered and evaluated on genuinely new forward data rather than
another iteration over the same history. another iteration over the same history.
The temporary GTL matrix scripts, configurable detector branches, and
confirmation hooks were removed after this decision. The normal snapshot
backtester and frozen `explicit_target_ladder` parity arm remain.
The post-2024 window has been opened and is now analysis data, not a valid final The post-2024 window has been opened and is now analysis data, not a valid final
promotion holdout. These arms can isolate mechanism, but neither may ship without promotion holdout. These arms can isolate mechanism, but neither may ship without
new future data or a separately pre-registered walk-forward protocol. new future data or a separately pre-registered walk-forward protocol.
reports/README.md
+15 -14
View File
@@ -61,32 +61,30 @@ after scanner integration passed: the regenerated report changed only its
`generated_at` timestamp, confirming that the shared helper preserves exact `generated_at` timestamp, confirming that the shared helper preserves exact
parity. Nothing in this research branch deploys the change to production. parity. Nothing in this research branch deploys the change to production.
The next decision point is generated by the one-command GTL parameter run: The archived single-parameter decision point is:
- `backtest-YYYYMMDD-gtl-tuning-matrix.json` - `backtest-20260713-gtl-tuning-matrix.json`
- `backtest-YYYYMMDD-gtl-tuning-matrix.md` - `backtest-20260713-gtl-tuning-matrix.md`
These two consolidated files are the reports worth retaining. The runner uses These two compact consolidated files are retained; the 20 full per-arm reports
a hidden temporary directory for its 20 full per-arm reports and removes it were removed after consolidation.
after successful consolidation unless `--keep-arm-reports` is supplied.
The completed 2026-07-13 matrix found no single-parameter replacement that The completed 2026-07-13 matrix found no single-parameter replacement that
passed all robustness checks. Retain its JSON and Markdown as that decision passed all robustness checks. Retain its JSON and Markdown as that decision
point. The evidence-selected retained-versus-added decomposition writes: point. The retained-versus-added decomposition is retained as:
- `backtest-YYYYMMDD-gtl-confirmation-matrix.json` - `backtest-20260713-gtl-confirmation-matrix.json`
- `backtest-YYYYMMDD-gtl-confirmation-matrix.md` - `backtest-20260713-gtl-confirmation-matrix.md`
Those become the next decision point; detailed per-arm reports remain temporary Detailed per-arm reports were not retained.
unless explicitly retained.
The completed confirmation matrix found one near-hit but no formal winner: The completed confirmation matrix found one near-hit but no formal winner:
strength-1000 intersection improved full/train/post-2024 Sharpe and CAGR, while strength-1000 intersection improved full/train/post-2024 Sharpe and CAGR, while
max drawdown worsened from 21.4% to 21.7%. Retain that matrix as the cohort- max drawdown worsened from 21.4% to 21.7%. Retain that matrix as the cohort-
decomposition decision point. The final stability check writes: decomposition decision point. The final stability check is retained as:
- `backtest-YYYYMMDD-gtl-strength-sensitivity.json` - `backtest-20260713-gtl-strength-sensitivity.json`
- `backtest-YYYYMMDD-gtl-strength-sensitivity.md` - `backtest-20260713-gtl-strength-sensitivity.md`
Its pre-registered decision requires at least two adjacent scales to pass all Its pre-registered decision requires at least two adjacent scales to pass all
six unchanged checks. six unchanged checks.
@@ -96,3 +94,6 @@ Scale 1500 was the sole 6/6 arm, but no adjacent scale passed, so the stable-
plateau rule rejected it. Retain the consolidated JSON/Markdown as the closing plateau rule rejected it. Retain the consolidated JSON/Markdown as the closing
GTL decision point. No confirmation or tuned strength value should be promoted GTL decision point. No confirmation or tuned strength value should be promoted
from this snapshot. from this snapshot.
The temporary matrix runners and their configurable backtest hooks were removed
after consolidation. The normal local snapshot backtester remains.
scripts/run_backtest_snapshot.py
-25
View File
@@ -61,25 +61,10 @@ def _parse_args() -> argparse.Namespace:
"rewrite_range504_structural_primary2", "rewrite_range504_structural_primary2",
"production_structural_overlay", "production_structural_overlay",
"explicit_target_ladder", "explicit_target_ladder",
"gtl_tuning",
"gtl_confirmation",
), ),
default=None, default=None,
help="Research-only S/R detector/gate arm.", help="Research-only S/R detector/gate arm.",
) )
parser.add_argument(
"--gtl-config",
default=None,
help="Research-only GTL configuration as a JSON object (requires --sr-variant gtl_tuning).",
)
parser.add_argument(
"--gtl-confirm-config",
default=None,
help=(
"Research-only GTL intersection/union configuration as JSON "
"(requires --sr-variant gtl_confirmation)."
),
)
parser.add_argument( parser.add_argument(
"--entry-start", "--entry-start",
default=None, default=None,
@@ -215,16 +200,6 @@ async def _main() -> None:
os.environ["BACKTEST_ALLOW_SPAWN"] = "1" os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
if args.sr_variant: if args.sr_variant:
os.environ["BACKTEST_SR_VARIANT"] = args.sr_variant os.environ["BACKTEST_SR_VARIANT"] = args.sr_variant
if args.gtl_config:
if args.sr_variant != "gtl_tuning":
raise SystemExit("--gtl-config requires --sr-variant gtl_tuning")
os.environ["BACKTEST_GTL_CONFIG"] = args.gtl_config
if args.gtl_confirm_config:
if args.sr_variant != "gtl_confirmation":
raise SystemExit(
"--gtl-confirm-config requires --sr-variant gtl_confirmation"
)
os.environ["BACKTEST_GTL_CONFIRM_CONFIG"] = args.gtl_confirm_config
if args.entry_start: if args.entry_start:
os.environ["BACKTEST_ENTRY_START"] = args.entry_start os.environ["BACKTEST_ENTRY_START"] = args.entry_start
if args.entry_end: if args.entry_end:
scripts/run_gtl_confirmation_matrix.py
-360
View File
@@ -1,360 +0,0 @@
"""Run the evidence-selected GTL confirmation/union matrix with one command.
The first GTL tuning matrix tested replacements. This follow-up decomposes the
four informative variants into retained-only intersections and control-plus-
addition unions while preserving frozen control geometry wherever possible.
"""
from __future__ import annotations
import argparse
import json
import math
import subprocess
import sys
from datetime import date, datetime, timezone
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from scripts import run_gtl_tuning_matrix as common # noqa: E402
RUNNER = ROOT / "scripts" / "run_backtest_snapshot.py"
REFERENCE_MATRIX = ROOT / "reports" / "backtest-20260713-gtl-tuning-matrix.json"
TOUCH = {"name": "touch_0_25pct", "touch_tolerance": 0.0025}
STRENGTH = {"name": "strength_1000", "strength_scale": 1000.0}
MERGE = {"name": "merge_0_25pct", "merge_tolerance": 0.0025}
GRID_ONLY = {"name": "pivots_none", "include_pivots": False}
def _arm(
name: str,
description: str,
mode: str,
*confirmations: dict[str, Any],
) -> dict[str, Any]:
return {
"name": name,
"description": description,
"mode": mode,
"confirmations": list(confirmations),
}
# Pre-registered from the replacement matrix's paired cohorts. Replacement
# arms that plainly removed strong control setups or added weak cohorts are not
# repeated here.
GTL_CONFIRMATION_ARMS: tuple[dict[str, Any], ...] = (
_arm("control", "Frozen explicit GTL; composition-path parity control.", "intersection"),
_arm(
"touch_intersection",
"Retain control setups also qualified with 0.25% touch padding.",
"intersection",
TOUCH,
),
_arm(
"touch_union",
"Keep control and admit additions from 0.25% touch padding.",
"union",
TOUCH,
),
_arm(
"strength_intersection",
"Retain control setups also qualified at strength scale 1000.",
"intersection",
STRENGTH,
),
_arm(
"strength_union",
"Keep control and admit additions from strength scale 1000.",
"union",
STRENGTH,
),
_arm(
"merge_intersection",
"Retain control setups also qualified with 0.25% proposal merging.",
"intersection",
MERGE,
),
_arm(
"merge_union",
"Keep control and admit additions from 0.25% proposal merging.",
"union",
MERGE,
),
_arm(
"grid_intersection",
"Retain control setups also qualified by the range grid without pivots.",
"intersection",
GRID_ONLY,
),
_arm(
"grid_union",
"Keep control and admit additions from the range grid without pivots.",
"union",
GRID_ONLY,
),
_arm(
"touch_strength_intersection",
"Require both tighter-touch and faster-strength confirmation.",
"intersection",
TOUCH,
STRENGTH,
),
_arm(
"touch_merge_intersection",
"Require both tighter-touch and tighter-merge confirmation.",
"intersection",
TOUCH,
MERGE,
),
_arm(
"strength_merge_intersection",
"Require both faster-strength and tighter-merge confirmation.",
"intersection",
STRENGTH,
MERGE,
),
_arm(
"touch_strength_merge_intersection",
"Require all three high-breadth confirmation variants.",
"intersection",
TOUCH,
STRENGTH,
MERGE,
),
)
def _args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"snapshot",
nargs="?",
default="backtest_snapshots/prod.sqlite",
help="Local SQLite snapshot path.",
)
parser.add_argument("--workers", type=int, default=7)
parser.add_argument("--holdout-split", default="2024-07-01")
parser.add_argument(
"--out",
default=None,
help=(
"Consolidated JSON path. Defaults to "
"reports/backtest-YYYYMMDD-gtl-confirmation-matrix.json."
),
)
parser.add_argument("--keep-arm-reports", action="store_true")
return parser.parse_args()
def _config(arm: dict[str, Any]) -> dict[str, Any]:
return {
"name": arm["name"],
"mode": arm["mode"],
"confirmations": arm["confirmations"],
}
def _signature(arm: dict) -> dict:
return {
"candidates": arm.get("candidates"),
"qualified": arm.get("qualified"),
"qualified_net_avg_r": arm.get("qualified_net_avg_r"),
"qualified_net_avg_r_ex_top5": arm.get("qualified_net_avg_r_ex_top5"),
"full_book": arm.get("full_book"),
"holdout": arm.get("holdout"),
}
def _reference_control() -> dict | None:
if not REFERENCE_MATRIX.exists():
return None
with REFERENCE_MATRIX.open(encoding="utf-8") as handle:
payload = json.load(handle)
return next(
(arm for arm in payload.get("arms") or [] if arm.get("name") == "control"),
None,
)
def _write_markdown(path: Path, payload: dict) -> None:
rows = [
"# GTL confirmation/union matrix",
"",
f"Status: **{payload['status']}** ",
f"Holdout split: `{payload['holdout_split']}` ",
f"Completed arms: {len(payload['arms'])}/{payload['arm_count']}",
"",
"| Arm | Mode | Qualified | Full Sharpe | CAGR | Max DD | Trades | Train Sharpe | Test Sharpe | Ex-top-5% R | Screen |",
"|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|",
]
for arm in payload["arms"]:
full = arm.get("full_book") or {}
holdout = arm.get("holdout") or {}
screen = arm.get("screen") or {}
rows.append(
"| "
+ " | ".join((
arm["name"],
arm["config"]["mode"],
str(arm.get("qualified") or "-"),
common._fmt(full.get("sharpe")),
common._fmt(full.get("cagr_pct"), 1),
common._fmt(full.get("max_drawdown_pct"), 1),
str(full.get("trades") or "-"),
common._fmt((holdout.get("train") or {}).get("sharpe")),
common._fmt((holdout.get("test") or {}).get("sharpe")),
common._fmt(arm.get("qualified_net_avg_r_ex_top5"), 3),
f"{screen.get('passed', '-')}/{screen.get('total', '-')}",
))
+ " |"
)
rows.extend((
"",
"## Interpretation guardrail",
"",
"Intersections test the retained control cohort; unions test control plus genuinely added setups. The post-2024 interval is a robustness check, not a pristine holdout. Passing does not authorize deployment.",
"",
))
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("\n".join(rows), encoding="utf-8")
def _run_arm(
arm: dict[str, Any],
snapshot: Path,
workers: int,
holdout_split: str,
output: Path,
) -> None:
command = [
sys.executable,
str(RUNNER),
str(snapshot),
"--workers", str(workers),
"--allow-spawn",
"--sr-variant", "gtl_confirmation",
"--gtl-confirm-config", json.dumps(_config(arm), separators=(",", ":")),
"--holdout-split", holdout_split,
"--sr-audit",
"--out", str(output),
]
subprocess.run(command, cwd=ROOT, check=True)
def main() -> None:
args = _args()
snapshot = Path(args.snapshot).resolve()
if not snapshot.exists():
raise SystemExit(f"Snapshot not found: {snapshot}")
if args.workers < 1:
raise SystemExit("--workers must be at least 1")
try:
date.fromisoformat(args.holdout_split)
except ValueError as exc:
raise SystemExit("--holdout-split must use YYYY-MM-DD") from exc
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
default_out = ROOT / "reports" / f"backtest-{stamp[:8]}-gtl-confirmation-matrix.json"
out_path = Path(args.out) if args.out else default_out
if not out_path.is_absolute():
out_path = ROOT / out_path
markdown_path = out_path.with_suffix(".md")
work_dir = ROOT / "reports" / f".gtl-confirmation-work-{stamp}"
work_dir.mkdir(parents=True, exist_ok=False)
reference = _reference_control()
payload: dict[str, Any] = {
"status": "running",
"generated_at": datetime.now(timezone.utc).isoformat(),
"snapshot": str(snapshot),
"workers": args.workers,
"holdout_split": args.holdout_split,
"arm_count": len(GTL_CONFIRMATION_ARMS),
"reference_matrix": str(REFERENCE_MATRIX) if reference else None,
"arms": [],
}
common._write_json(out_path, payload)
_write_markdown(markdown_path, payload)
control_report: dict | None = None
arm_outputs: list[Path] = []
try:
for index, arm in enumerate(GTL_CONFIRMATION_ARMS, start=1):
name = arm["name"]
output = work_dir / f"{index:02d}-{name}.json"
arm_outputs.append(output)
print(f"\n[{index}/{len(GTL_CONFIRMATION_ARMS)}] {name}", flush=True)
print(f" {arm['description']}", flush=True)
_run_arm(arm, snapshot, args.workers, args.holdout_split, output)
report = common._load_report(output)
compact = common._compact_arm(report, _config(arm), control_report)
compact["description"] = arm["description"]
if control_report is None:
control_report = report
compact["screen"] = {
"checks": {}, "passed": 0, "total": 0, "advances": False,
}
if reference is not None and _signature(compact) != _signature(reference):
raise RuntimeError(
"Confirmation-path control does not reproduce the frozen GTL matrix control"
)
payload["control_parity"] = "pass" if reference is not None else "not_checked"
else:
compact["screen"] = common._screen_arm(compact, payload["arms"][0])
payload["arms"].append(compact)
payload["completed_at"] = datetime.now(timezone.utc).isoformat()
common._write_json(out_path, payload)
_write_markdown(markdown_path, payload)
except BaseException as exc:
payload["status"] = "failed"
payload["failed_at"] = datetime.now(timezone.utc).isoformat()
payload["error"] = f"{type(exc).__name__}: {exc}"
payload["work_dir"] = str(work_dir)
common._write_json(out_path, payload)
_write_markdown(markdown_path, payload)
raise
payload["status"] = "complete"
payload["completed_at"] = datetime.now(timezone.utc).isoformat()
payload["advancing_arms"] = [
arm["name"]
for arm in payload["arms"]
if (arm.get("screen") or {}).get("advances")
]
payload["ranking_by_full_sharpe"] = [
arm["name"]
for arm in sorted(
payload["arms"],
key=lambda row: float(
(row.get("full_book") or {}).get("sharpe") or -math.inf
),
reverse=True,
)
]
if args.keep_arm_reports:
payload["arm_report_directory"] = str(work_dir)
common._write_json(out_path, payload)
_write_markdown(markdown_path, payload)
if not args.keep_arm_reports:
for path in arm_outputs:
path.unlink(missing_ok=True)
work_dir.rmdir()
print("\nGTL confirmation matrix complete.")
print(f" JSON: {out_path}")
print(f" Markdown: {markdown_path}")
if payload["advancing_arms"]:
print(" Passing arms: " + ", ".join(payload["advancing_arms"]))
else:
print(" No arm passed every pre-registered screen.")
if __name__ == "__main__":
main()
scripts/run_gtl_strength_sensitivity.py
-287
View File
@@ -1,287 +0,0 @@
"""Run the pre-registered GTL strength-confirmation sensitivity band.
The strength-1000 intersection was the only composition arm to improve
full/train/test Sharpe, but it missed the unchanged drawdown guardrail. This
matrix checks whether that result is a stable one-dimensional plateau rather
than tuning the guardrail or launching another broad parameter search.
"""
from __future__ import annotations
import argparse
import json
import math
import sys
from datetime import date, datetime, timezone
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from scripts import run_gtl_confirmation_matrix as composition # noqa: E402
from scripts import run_gtl_tuning_matrix as common # noqa: E402
REFERENCE_TUNING = ROOT / "reports" / "backtest-20260713-gtl-tuning-matrix.json"
REFERENCE_COMPOSITION = (
ROOT / "reports" / "backtest-20260713-gtl-confirmation-matrix.json"
)
STRENGTH_SCALES = (625.0, 750.0, 875.0, 1000.0, 1125.0, 1250.0, 1500.0, 2000.0)
GTL_STRENGTH_ARMS: tuple[dict[str, Any], ...] = (
composition._arm(
"control",
"Frozen GTL composition-path parity control.",
"intersection",
),
*(
composition._arm(
f"strength_{int(scale)}_intersection",
f"Require control confirmation at traffic-strength scale {scale:g}.",
"intersection",
{"name": f"strength_{int(scale)}", "strength_scale": scale},
)
for scale in STRENGTH_SCALES
),
)
def _args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"snapshot",
nargs="?",
default="backtest_snapshots/prod.sqlite",
help="Local SQLite snapshot path.",
)
parser.add_argument("--workers", type=int, default=7)
parser.add_argument("--holdout-split", default="2024-07-01")
parser.add_argument(
"--out",
default=None,
help=(
"Consolidated JSON path. Defaults to "
"reports/backtest-YYYYMMDD-gtl-strength-sensitivity.json."
),
)
parser.add_argument("--keep-arm-reports", action="store_true")
return parser.parse_args()
def _reference_arm(path: Path, name: str) -> dict | None:
if not path.exists():
return None
with path.open(encoding="utf-8") as handle:
payload = json.load(handle)
return next(
(arm for arm in payload.get("arms") or [] if arm.get("name") == name),
None,
)
def _write_markdown(path: Path, payload: dict) -> None:
rows = [
"# GTL strength-confirmation sensitivity",
"",
f"Status: **{payload['status']}** ",
f"Holdout split: `{payload['holdout_split']}` ",
f"Completed arms: {len(payload['arms'])}/{payload['arm_count']}",
"",
"| Arm | Qualified | Full Sharpe | CAGR | Max DD | Trades | Train Sharpe | Test Sharpe | Ex-top-5% R | Screen |",
"|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|",
]
for arm in payload["arms"]:
full = arm.get("full_book") or {}
holdout = arm.get("holdout") or {}
screen = arm.get("screen") or {}
rows.append(
"| "
+ " | ".join((
arm["name"],
str(arm.get("qualified") or "-"),
common._fmt(full.get("sharpe")),
common._fmt(full.get("cagr_pct"), 1),
common._fmt(full.get("max_drawdown_pct"), 1),
str(full.get("trades") or "-"),
common._fmt((holdout.get("train") or {}).get("sharpe")),
common._fmt((holdout.get("test") or {}).get("sharpe")),
common._fmt(arm.get("qualified_net_avg_r_ex_top5"), 3),
f"{screen.get('passed', '-')}/{screen.get('total', '-')}",
))
+ " |"
)
rows.extend((
"",
"## Pre-registered interpretation",
"",
"The six original guardrails remain unchanged. A stable candidate requires at least two adjacent non-control scales to pass all six; an isolated passing scale is rejected as sensitivity, not promoted.",
"",
))
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("\n".join(rows), encoding="utf-8")
def _stable_plateau_pairs(arms: list[dict]) -> list[list[str]]:
sensitivity = [arm for arm in arms if arm.get("name") != "control"]
return [
[left["name"], right["name"]]
for left, right in zip(sensitivity, sensitivity[1:], strict=False)
if (left.get("screen") or {}).get("advances")
and (right.get("screen") or {}).get("advances")
]
def main() -> None:
args = _args()
snapshot = Path(args.snapshot).resolve()
if not snapshot.exists():
raise SystemExit(f"Snapshot not found: {snapshot}")
if args.workers < 1:
raise SystemExit("--workers must be at least 1")
try:
date.fromisoformat(args.holdout_split)
except ValueError as exc:
raise SystemExit("--holdout-split must use YYYY-MM-DD") from exc
reference_control = _reference_arm(REFERENCE_TUNING, "control")
reference_1000 = _reference_arm(
REFERENCE_COMPOSITION,
"strength_intersection",
)
if reference_control is None or reference_1000 is None:
raise SystemExit(
"The completed GTL tuning and confirmation matrices are required "
"for control/replication checks"
)
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
default_out = (
ROOT / "reports" / f"backtest-{stamp[:8]}-gtl-strength-sensitivity.json"
)
out_path = Path(args.out) if args.out else default_out
if not out_path.is_absolute():
out_path = ROOT / out_path
markdown_path = out_path.with_suffix(".md")
work_dir = ROOT / "reports" / f".gtl-strength-work-{stamp}"
work_dir.mkdir(parents=True, exist_ok=False)
payload: dict[str, Any] = {
"status": "running",
"generated_at": datetime.now(timezone.utc).isoformat(),
"snapshot": str(snapshot),
"workers": args.workers,
"holdout_split": args.holdout_split,
"arm_count": len(GTL_STRENGTH_ARMS),
"strength_scales": list(STRENGTH_SCALES),
"arms": [],
"promotion_rule": (
"At least two adjacent non-control scales must pass all six original "
"guardrails."
),
}
common._write_json(out_path, payload)
_write_markdown(markdown_path, payload)
control_report: dict | None = None
arm_outputs: list[Path] = []
try:
for index, arm in enumerate(GTL_STRENGTH_ARMS, start=1):
name = arm["name"]
output = work_dir / f"{index:02d}-{name}.json"
arm_outputs.append(output)
print(f"\n[{index}/{len(GTL_STRENGTH_ARMS)}] {name}", flush=True)
print(f" {arm['description']}", flush=True)
composition._run_arm(
arm,
snapshot,
args.workers,
args.holdout_split,
output,
)
report = common._load_report(output)
compact = common._compact_arm(
report,
composition._config(arm),
control_report,
)
compact["description"] = arm["description"]
if control_report is None:
control_report = report
compact["screen"] = {
"checks": {}, "passed": 0, "total": 0, "advances": False,
}
if composition._signature(compact) != composition._signature(
reference_control
):
raise RuntimeError(
"Strength sensitivity control does not reproduce frozen GTL control"
)
payload["control_parity"] = "pass"
else:
compact["screen"] = common._screen_arm(
compact,
payload["arms"][0],
)
if name == "strength_1000_intersection":
if composition._signature(compact) != composition._signature(
reference_1000
):
raise RuntimeError(
"Strength-1000 arm does not reproduce the completed composition result"
)
payload["strength_1000_replication"] = "pass"
payload["arms"].append(compact)
payload["completed_at"] = datetime.now(timezone.utc).isoformat()
common._write_json(out_path, payload)
_write_markdown(markdown_path, payload)
except BaseException as exc:
payload["status"] = "failed"
payload["failed_at"] = datetime.now(timezone.utc).isoformat()
payload["error"] = f"{type(exc).__name__}: {exc}"
payload["work_dir"] = str(work_dir)
common._write_json(out_path, payload)
_write_markdown(markdown_path, payload)
raise
payload["status"] = "complete"
payload["completed_at"] = datetime.now(timezone.utc).isoformat()
payload["advancing_arms"] = [
arm["name"]
for arm in payload["arms"]
if (arm.get("screen") or {}).get("advances")
]
payload["stable_plateau_pairs"] = _stable_plateau_pairs(payload["arms"])
payload["stable_candidate"] = bool(payload["stable_plateau_pairs"])
payload["ranking_by_full_sharpe"] = [
arm["name"]
for arm in sorted(
payload["arms"],
key=lambda row: float(
(row.get("full_book") or {}).get("sharpe") or -math.inf
),
reverse=True,
)
]
if args.keep_arm_reports:
payload["arm_report_directory"] = str(work_dir)
common._write_json(out_path, payload)
_write_markdown(markdown_path, payload)
if not args.keep_arm_reports:
for path in arm_outputs:
path.unlink(missing_ok=True)
work_dir.rmdir()
print("\nGTL strength sensitivity complete.")
print(f" JSON: {out_path}")
print(f" Markdown: {markdown_path}")
if payload["stable_candidate"]:
pairs = [" + ".join(pair) for pair in payload["stable_plateau_pairs"]]
print(" Stable passing plateau: " + "; ".join(pairs))
else:
print(" No stable adjacent passing plateau.")
if __name__ == "__main__":
main()
scripts/run_gtl_tuning_matrix.py
-418
View File
@@ -1,418 +0,0 @@
"""Run the complete Gate Target Ladder tuning matrix with one command.
Every arm is a full production-parity backtest. Arms run sequentially so each
one can use the requested worker pool without competing with another arm. Large
per-arm reports live in a temporary run directory and are removed after a
successful consolidation unless ``--keep-arm-reports`` is supplied.
"""
from __future__ import annotations
import argparse
import json
import math
import subprocess
import sys
from datetime import date, datetime, timezone
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
RUNNER = ROOT / "scripts" / "run_backtest_snapshot.py"
BASE_CONFIG: dict[str, Any] = {
"lookback_bars": None,
"grid_bins": 20,
"include_pivots": True,
"pivot_window": 2,
"touch_tolerance": 0.005,
"merge_tolerance": 0.005,
"strength_scale": 500.0,
"zone_tolerance": 0.02,
"candidate_limit": 5,
"max_target_atr": None,
}
# Single-variable arms only. The control value is represented by BASE_CONFIG;
# there is deliberately no Cartesian product.
GTL_TUNING_ARMS: tuple[dict[str, Any], ...] = (
{"name": "control", "description": "Frozen explicit GTL defaults."},
{"name": "lookback_252", "description": "One-year GTL history.", "lookback_bars": 252},
{"name": "lookback_504", "description": "Two-year GTL history.", "lookback_bars": 504},
{"name": "lookback_756", "description": "Three-year GTL history.", "lookback_bars": 756},
{"name": "candidates_8", "description": "Retain up to eight candidates before probability.", "candidate_limit": 8},
{"name": "candidates_all", "description": "Score every eligible target before primary selection.", "candidate_limit": None},
{"name": "max_atr_5_5", "description": "Universal 5.5 ATR maximum target distance.", "max_target_atr": 5.5},
{"name": "max_atr_8", "description": "Universal 8 ATR maximum target distance.", "max_target_atr": 8.0},
{"name": "touch_0", "description": "Strict candle-range crossings with no touch padding.", "touch_tolerance": 0.0},
{"name": "touch_0_25pct", "description": "Use 0.25% padding when counting price traffic.", "touch_tolerance": 0.0025},
{"name": "merge_0_25pct", "description": "Merge GTL proposals within 0.25%.", "merge_tolerance": 0.0025},
{"name": "merge_1pct", "description": "Merge GTL proposals within 1%.", "merge_tolerance": 0.01},
{"name": "zones_1pct", "description": "Cluster target zones within 1%.", "zone_tolerance": 0.01},
{"name": "zones_3pct", "description": "Cluster target zones within 3%.", "zone_tolerance": 0.03},
{"name": "grid_12", "description": "Use 12 evenly spaced range centers.", "grid_bins": 12},
{"name": "grid_32", "description": "Use 32 evenly spaced range centers.", "grid_bins": 32},
{"name": "pivots_none", "description": "Range grid only; omit swing pivots.", "include_pivots": False},
{"name": "pivots_11bar", "description": "Use an 11-bar swing-pivot window.", "pivot_window": 5},
{"name": "strength_250", "description": "Slower traffic-strength saturation.", "strength_scale": 250.0},
{"name": "strength_1000", "description": "Faster traffic-strength saturation.", "strength_scale": 1000.0},
)
BOOK_FIELDS = (
"sharpe",
"cagr_pct",
"max_drawdown_pct",
"trades",
"win_rate",
"avg_hold_days",
"skipped_book_full",
)
def _args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"snapshot",
nargs="?",
default="backtest_snapshots/prod.sqlite",
help="Local SQLite snapshot path.",
)
parser.add_argument("--workers", type=int, default=7)
parser.add_argument(
"--holdout-split",
default="2024-07-01",
help="Disjoint train/test split included in every arm.",
)
parser.add_argument(
"--out",
default=None,
help="Consolidated JSON path. Defaults to reports/backtest-YYYYMMDD-gtl-tuning-matrix.json.",
)
parser.add_argument(
"--keep-arm-reports",
action="store_true",
help="Keep the large temporary per-arm JSON reports after consolidation.",
)
return parser.parse_args()
def _arm_config(arm: dict[str, Any]) -> dict[str, Any]:
config = {**BASE_CONFIG, **{key: value for key, value in arm.items() if key != "description"}}
return config
def _load_report(path: Path) -> dict:
with path.open(encoding="utf-8") as handle:
report = json.load(handle)
if report.get("sr_candidate_audit") is None:
raise ValueError(f"Report lacks sr_candidate_audit: {path}")
return report
def _compact_book(row: dict | None) -> dict | None:
if row is None:
return None
return {key: row.get(key) for key in BOOK_FIELDS}
def _full_book(report: dict) -> dict | None:
runs = ((report.get("portfolio_monitor") or {}).get("runs") or [])
return _compact_book(next(
(
row
for row in runs
if row.get("is_production") and row.get("lookback") == "all"
),
None,
))
def _holdout_books(report: dict) -> dict[str, dict | None]:
rows = ((report.get("holdout") or {}).get("rows") or [])
return {
window: _compact_book(next((row for row in rows if row.get("window") == window), None))
for window in ("train", "test")
}
def _audit_key(row: dict) -> tuple[str, str, str]:
return row["symbol"], row["date"], row["direction"]
def _cohort_stats(rows: list[dict]) -> dict:
net = [float(row.get("net_r", 0.0)) for row in rows]
trimmed = sorted(net, reverse=True)[math.ceil(len(net) * 0.05):]
return {
"count": len(rows),
"net_avg_r": round(sum(net) / len(net), 4) if net else None,
"net_avg_r_ex_top5": round(sum(trimmed) / len(trimmed), 4) if trimmed else None,
}
def _cohort_comparison(control: dict, variant: dict) -> dict:
control_rows = {
_audit_key(row): row for row in control.get("sr_candidate_audit") or []
}
variant_rows = {
_audit_key(row): row for row in variant.get("sr_candidate_audit") or []
}
control_q = {key for key, row in control_rows.items() if row.get("qualified")}
variant_q = {key for key, row in variant_rows.items() if row.get("qualified")}
retained = control_q & variant_q
added = variant_q - control_q
removed = control_q - variant_q
return {
"retained": _cohort_stats([variant_rows[key] for key in retained]),
"added": _cohort_stats([variant_rows[key] for key in added]),
"removed": _cohort_stats([control_rows[key] for key in removed]),
}
def _compact_arm(report: dict, config: dict, control: dict | None) -> dict:
qualified = report.get("overall_qualified") or {}
result = {
"name": config["name"],
"config": config,
"candidates": report.get("candidates"),
"qualified": report.get("qualified"),
"qualified_net_avg_r": qualified.get("net_avg_r"),
"qualified_net_avg_r_ex_top5": qualified.get("net_avg_r_ex_top5"),
"full_book": _full_book(report),
"holdout": _holdout_books(report),
"gtl_diagnostics": report.get("sr_variant_diagnostics"),
"cohort_vs_control": (
_cohort_comparison(control, report) if control is not None else None
),
}
return result
def _screen_arm(arm: dict, control: dict) -> dict:
full = arm.get("full_book") or {}
base_full = control.get("full_book") or {}
train = (arm.get("holdout") or {}).get("train") or {}
base_train = (control.get("holdout") or {}).get("train") or {}
test = (arm.get("holdout") or {}).get("test") or {}
base_test = (control.get("holdout") or {}).get("test") or {}
def at_least(value: Any, baseline: Any) -> bool:
return value is not None and baseline is not None and float(value) >= float(baseline)
control_trades = float(base_full.get("trades") or 0.0)
arm_trades = float(full.get("trades") or 0.0)
checks = {
"full_sharpe_not_worse": at_least(full.get("sharpe"), base_full.get("sharpe")),
"train_sharpe_not_worse": at_least(train.get("sharpe"), base_train.get("sharpe")),
"test_sharpe_not_worse": at_least(test.get("sharpe"), base_test.get("sharpe")),
"drawdown_not_worse": (
full.get("max_drawdown_pct") is not None
and base_full.get("max_drawdown_pct") is not None
and abs(float(full["max_drawdown_pct"]))
<= abs(float(base_full["max_drawdown_pct"]))
),
"retains_80pct_trades": control_trades > 0 and arm_trades >= control_trades * 0.8,
"robust_expectancy_positive": (
arm.get("qualified_net_avg_r_ex_top5") is not None
and float(arm["qualified_net_avg_r_ex_top5"]) > 0
),
}
return {
"checks": checks,
"passed": sum(checks.values()),
"total": len(checks),
"advances": all(checks.values()),
}
def _write_json(path: Path, payload: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as handle:
json.dump(payload, handle, indent=2)
handle.write("\n")
def _fmt(value: Any, digits: int = 2) -> str:
return "-" if value is None else f"{float(value):.{digits}f}"
def _write_markdown(path: Path, payload: dict) -> None:
rows = [
"# GTL tuning matrix",
"",
f"Status: **{payload['status']}** ",
f"Holdout split: `{payload['holdout_split']}` ",
f"Completed arms: {len(payload['arms'])}/{payload['arm_count']}",
"",
"| Arm | Full Sharpe | CAGR | Max DD | Trades | Train Sharpe | Test Sharpe | Ex-top-5% R | Screen |",
"|---|---:|---:|---:|---:|---:|---:|---:|---:|",
]
for arm in payload["arms"]:
full = arm.get("full_book") or {}
holdout = arm.get("holdout") or {}
train = holdout.get("train") or {}
test = holdout.get("test") or {}
screen = arm.get("screen") or {}
rows.append(
"| "
+ " | ".join((
arm["name"],
_fmt(full.get("sharpe")),
_fmt(full.get("cagr_pct"), 1),
_fmt(full.get("max_drawdown_pct"), 1),
str(full.get("trades") or "-"),
_fmt(train.get("sharpe")),
_fmt(test.get("sharpe")),
_fmt(arm.get("qualified_net_avg_r_ex_top5"), 3),
f"{screen.get('passed', '-')}/{screen.get('total', '-')}",
))
+ " |"
)
rows.extend((
"",
"## Interpretation guardrail",
"",
"The post-2024 interval has already informed prior research. The train/test columns are robustness checks, not a pristine holdout. A passing arm is a candidate for forward paper validation, not automatic production promotion.",
"",
))
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("\n".join(rows), encoding="utf-8")
def _run_arm(
*,
arm: dict[str, Any],
snapshot: str,
workers: int,
holdout_split: str,
output: Path,
) -> None:
config = _arm_config(arm)
command = [
sys.executable,
str(RUNNER),
snapshot,
"--workers", str(workers),
"--allow-spawn",
"--sr-variant", "gtl_tuning",
"--gtl-config", json.dumps(config, separators=(",", ":")),
"--holdout-split", holdout_split,
"--sr-audit",
"--out", str(output),
]
subprocess.run(command, cwd=ROOT, check=True)
def main() -> None:
args = _args()
snapshot = Path(args.snapshot).resolve()
if not snapshot.exists():
raise SystemExit(f"Snapshot not found: {snapshot}")
if args.workers < 1:
raise SystemExit("--workers must be at least 1")
try:
date.fromisoformat(args.holdout_split)
except ValueError as exc:
raise SystemExit("--holdout-split must use YYYY-MM-DD") from exc
stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
default_out = ROOT / "reports" / f"backtest-{stamp[:8]}-gtl-tuning-matrix.json"
out_path = Path(args.out) if args.out else default_out
if not out_path.is_absolute():
out_path = ROOT / out_path
markdown_path = out_path.with_suffix(".md")
work_dir = ROOT / "reports" / f".gtl-tuning-work-{stamp}"
work_dir.mkdir(parents=True, exist_ok=False)
payload: dict[str, Any] = {
"status": "running",
"generated_at": datetime.now(timezone.utc).isoformat(),
"snapshot": str(snapshot.resolve()),
"workers": args.workers,
"holdout_split": args.holdout_split,
"arm_count": len(GTL_TUNING_ARMS),
"arms": [],
"caveat": (
"The split is a robustness check, not a pristine holdout; post-2024 "
"data has already informed earlier research."
),
}
_write_json(out_path, payload)
_write_markdown(markdown_path, payload)
control_report: dict | None = None
arm_outputs: list[Path] = []
try:
for index, arm in enumerate(GTL_TUNING_ARMS, start=1):
name = str(arm["name"])
output = work_dir / f"{index:02d}-{name}.json"
arm_outputs.append(output)
print(f"\n[{index}/{len(GTL_TUNING_ARMS)}] GTL arm: {name}", flush=True)
print(f" {arm['description']}", flush=True)
_run_arm(
arm=arm,
snapshot=str(snapshot),
workers=args.workers,
holdout_split=args.holdout_split,
output=output,
)
report = _load_report(output)
config = _arm_config(arm)
compact = _compact_arm(report, config, control_report)
compact["description"] = arm["description"]
if control_report is None:
control_report = report
compact["screen"] = {
"checks": {},
"passed": 0,
"total": 0,
"advances": False,
}
else:
compact["screen"] = _screen_arm(compact, payload["arms"][0])
payload["arms"].append(compact)
payload["completed_at"] = datetime.now(timezone.utc).isoformat()
_write_json(out_path, payload)
_write_markdown(markdown_path, payload)
except Exception as exc:
payload["status"] = "failed"
payload["failed_at"] = datetime.now(timezone.utc).isoformat()
payload["error"] = f"{type(exc).__name__}: {exc}"
payload["work_dir"] = str(work_dir)
_write_json(out_path, payload)
_write_markdown(markdown_path, payload)
raise
payload["status"] = "complete"
payload["completed_at"] = datetime.now(timezone.utc).isoformat()
payload["advancing_arms"] = [
arm["name"] for arm in payload["arms"] if (arm.get("screen") or {}).get("advances")
]
payload["ranking_by_full_sharpe"] = [
arm["name"]
for arm in sorted(
payload["arms"],
key=lambda row: float((row.get("full_book") or {}).get("sharpe") or -math.inf),
reverse=True,
)
]
if args.keep_arm_reports:
payload["arm_report_directory"] = str(work_dir)
_write_json(out_path, payload)
_write_markdown(markdown_path, payload)
if not args.keep_arm_reports:
for path in arm_outputs:
path.unlink(missing_ok=True)
work_dir.rmdir()
print("\nGTL tuning matrix complete.")
print(f" JSON: {out_path}")
print(f" Markdown: {markdown_path}")
if payload["advancing_arms"]:
print(f" Arms passing every pre-registered screen: {', '.join(payload['advancing_arms'])}")
else:
print(" No arm passed every pre-registered screen.")
if __name__ == "__main__":
main()
tests/unit/test_backtest_service.py
-155
View File
@@ -2,7 +2,6 @@
from __future__ import annotations from __future__ import annotations
import json
import math import math
from datetime import date, timedelta from datetime import date, timedelta
from types import SimpleNamespace from types import SimpleNamespace
@@ -804,8 +803,6 @@ def test_sr_research_variant_is_explicit_and_validated(monkeypatch):
"rewrite_range504_structural_primary2", "rewrite_range504_structural_primary2",
"production_structural_overlay", "production_structural_overlay",
"explicit_target_ladder", "explicit_target_ladder",
"gtl_tuning",
"gtl_confirmation",
): ):
monkeypatch.setenv("BACKTEST_SR_VARIANT", variant) monkeypatch.setenv("BACKTEST_SR_VARIANT", variant)
assert bt._sr_research_variant() == variant assert bt._sr_research_variant() == variant
@@ -864,48 +861,6 @@ def test_residual_arms_change_only_the_primary_rr_floor():
"explicit_target_ladder", "explicit_target_ladder",
activation, activation,
) == 1.5 ) == 1.5
assert bt._primary_min_rr_for_variant("gtl_tuning", activation) == 1.5
assert bt._primary_min_rr_for_variant("gtl_confirmation", activation) == 1.5
def test_gtl_research_config_parses_and_rejects_unknown_fields():
config = bt._parse_gtl_research_config(json.dumps({
"name": "lookback_504",
"lookback_bars": 504,
"candidate_limit": None,
}))
assert config.name == "lookback_504"
assert config.lookback_bars == 504
assert config.candidate_limit is None
assert config.ladder_config().grid_bins == 20
with pytest.raises(ValueError, match="Unknown GTL research config fields"):
bt._parse_gtl_research_config('{"mystery_knob": 1}')
with pytest.raises(ValueError, match="valid JSON"):
bt._parse_gtl_research_config("{")
def test_gtl_confirmation_config_parses_and_validates_composition():
config = bt._parse_gtl_confirmation_config(json.dumps({
"name": "touch_strength_intersection",
"mode": "intersection",
"confirmations": [
{"name": "touch", "touch_tolerance": 0.0025},
{"name": "strength", "strength_scale": 1000.0},
],
}))
assert config.name == "touch_strength_intersection"
assert config.mode == "intersection"
assert len(config.confirmations) == 2
assert config.confirmations[0].touch_tolerance == 0.0025
assert config.confirmations[1].strength_scale == 1000.0
with pytest.raises(ValueError, match="exactly one tuned variant"):
bt._parse_gtl_confirmation_config(json.dumps({
"name": "invalid_union",
"mode": "union",
"confirmations": [],
}))
def test_structural_overlay_tags_production_geometry_without_replacing_it(monkeypatch): def test_structural_overlay_tags_production_geometry_without_replacing_it(monkeypatch):
@@ -1017,116 +972,6 @@ def test_window_setups_routes_full_explicit_target_ladder(monkeypatch):
} }
def test_window_setups_routes_gtl_tuning_config(monkeypatch):
captured = {}
def fake_detector(highs, lows, closes, *, config):
captured.update({
"highs": highs,
"lows": lows,
"closes": closes,
"config": config,
})
return []
monkeypatch.setenv("BACKTEST_SR_VARIANT", bt.GTL_TUNING_VARIANT)
monkeypatch.setenv("BACKTEST_GTL_CONFIG", json.dumps({
"name": "lookback_252",
"lookback_bars": 252,
"grid_bins": 12,
}))
monkeypatch.setattr(bt, "detect_gate_target_ladder", fake_detector)
records = [
SimpleNamespace(
date=date(2024, 1, 1) + timedelta(days=i),
open=100.0,
high=101.0,
low=99.0,
close=100.0,
volume=1_000_000,
)
for i in range(bt.MIN_LOOKBACK)
]
assert bt._window_setups(records, {}, {}) == []
assert captured["highs"] == [101.0] * bt.MIN_LOOKBACK
assert captured["lows"] == [99.0] * bt.MIN_LOOKBACK
assert captured["closes"] == [100.0] * bt.MIN_LOOKBACK
assert captured["config"].lookback_bars == 252
assert captured["config"].grid_bins == 12
def test_gtl_confirmation_intersection_keeps_control_geometry(monkeypatch):
production = [{
"direction": "long",
"target": 111.0,
"rr": 2.2,
"meets_core": True,
"sr_variant": bt.EXPLICIT_TARGET_LADDER_VARIANT,
}]
def fake_window_setups(*args, sr_variant=None, gtl_research_config=None, **kwargs):
if sr_variant == bt.EXPLICIT_TARGET_LADDER_VARIANT:
return production
assert sr_variant == bt.GTL_TUNING_VARIANT
return [{
"direction": "long",
"target": 115.0,
"rr": 3.0,
"meets_core": gtl_research_config.name == "pass",
}]
monkeypatch.setattr(bt, "_window_setups", fake_window_setups)
research = bt.GTLConfirmationConfig(
name="two_filters",
mode="intersection",
confirmations=(
bt.GTLResearchConfig(name="pass"),
bt.GTLResearchConfig(name="fail"),
),
)
rows = bt._gtl_confirmation_window_setups(
[], {}, {}, confirmation_config=research
)
assert len(rows) == 1
assert rows[0]["target"] == 111.0
assert rows[0]["rr"] == 2.2
assert rows[0]["meets_core"] is False
assert rows[0]["gtl_confirmation_passes"] == [True, False]
assert production[0]["meets_core"] is True
def test_gtl_confirmation_union_uses_tuned_geometry_only_for_addition(monkeypatch):
production = [
{"direction": "long", "target": 111.0, "meets_core": False},
{"direction": "short", "target": 90.0, "meets_core": True},
]
tuned = [
{"direction": "long", "target": 115.0, "meets_core": True},
{"direction": "short", "target": 85.0, "meets_core": False},
]
def fake_window_setups(*args, sr_variant=None, **kwargs):
return production if sr_variant == bt.EXPLICIT_TARGET_LADDER_VARIANT else tuned
monkeypatch.setattr(bt, "_window_setups", fake_window_setups)
research = bt.GTLConfirmationConfig(
name="strength_union",
mode="union",
confirmations=(bt.GTLResearchConfig(name="strength"),),
)
rows = bt._gtl_confirmation_window_setups(
[], {}, {}, confirmation_config=research
)
by_direction = {row["direction"]: row for row in rows}
assert by_direction["long"]["target"] == 115.0
assert by_direction["long"]["gtl_confirmation_source"] == "tuned_addition"
assert by_direction["short"]["target"] == 90.0
assert by_direction["short"]["gtl_confirmation_source"] == "control"
@pytest.mark.parametrize( @pytest.mark.parametrize(
"variant", "variant",
["legacy_geometry_neutral", "legacy_range_grid_neutral"], ["legacy_geometry_neutral", "legacy_range_grid_neutral"],
tests/unit/test_detect_sr_levels.py
-67
View File
@@ -2,10 +2,7 @@
from __future__ import annotations from __future__ import annotations
import pytest
from app.services.sr_service import ( from app.services.sr_service import (
GateTargetLadderConfig,
MAX_LEVELS, MAX_LEVELS,
_bar_respect_weight, _bar_respect_weight,
_cap_levels, _cap_levels,
@@ -317,70 +314,6 @@ class TestDetectSrLevels:
assert detect_gate_target_ladder(highs, lows, closes) == expected assert detect_gate_target_ladder(highs, lows, closes) == expected
def test_configured_gate_target_ladder_defaults_preserve_parity(self):
highs, lows, closes, volumes = _make_series(n=500)
expected = detect_sr_levels_legacy(
highs,
lows,
closes,
volumes,
explicit_range_grid=True,
)
configured = detect_gate_target_ladder(
highs,
lows,
closes,
config=GateTargetLadderConfig(),
)
assert configured == expected
def test_configured_gate_target_ladder_bounds_history(self):
old_highs = [1_000.0] * 40
old_lows = [10.0] * 40
old_closes = [100.0] * 40
recent_highs = [110.0] * 40
recent_lows = [90.0] * 40
recent_closes = [100.0] * 40
config = GateTargetLadderConfig(
lookback_bars=40,
include_pivots=False,
)
levels = detect_gate_target_ladder(
old_highs + recent_highs,
old_lows + recent_lows,
old_closes + recent_closes,
config=config,
)
assert levels
assert all(90.0 <= level["price_level"] <= 110.0 for level in levels)
def test_configured_gate_target_ladder_separates_merge_tolerance(self):
highs, lows, closes, _ = _make_series(n=300)
tight = detect_gate_target_ladder(
highs,
lows,
closes,
config=GateTargetLadderConfig(merge_tolerance=0.0025),
)
wide = detect_gate_target_ladder(
highs,
lows,
closes,
config=GateTargetLadderConfig(merge_tolerance=0.01),
)
assert len(wide) <= len(tight)
def test_gate_target_ladder_config_rejects_invalid_values(self):
with pytest.raises(ValueError, match="lookback_bars"):
GateTargetLadderConfig(lookback_bars=19)
with pytest.raises(ValueError, match="touch_tolerance"):
GateTargetLadderConfig(touch_tolerance=-0.1)
def test_explicit_range_grid_is_volume_independent(self): def test_explicit_range_grid_is_volume_independent(self):
highs, lows, closes, volumes = _make_series(n=500) highs, lows, closes, volumes = _make_series(n=500)
shifted_volumes = [volume * (i + 1) for i, volume in enumerate(volumes)] shifted_volumes = [volume * (i + 1) for i, volume in enumerate(volumes)]
tests/unit/test_gtl_confirmation_matrix.py
-39
View File
@@ -1,39 +0,0 @@
"""Tests for the evidence-selected GTL composition matrix."""
from __future__ import annotations
from scripts import run_gtl_confirmation_matrix as matrix
def test_confirmation_matrix_is_pre_registered_and_well_formed():
assert len(matrix.GTL_CONFIRMATION_ARMS) == 13
assert matrix.GTL_CONFIRMATION_ARMS[0]["name"] == "control"
names = [arm["name"] for arm in matrix.GTL_CONFIRMATION_ARMS]
assert len(names) == len(set(names))
for arm in matrix.GTL_CONFIRMATION_ARMS:
config = matrix._config(arm)
assert config["mode"] in {"intersection", "union"}
if config["mode"] == "union":
assert len(config["confirmations"]) == 1
for confirmation in config["confirmations"]:
assert confirmation["name"] in {
"touch_0_25pct",
"strength_1000",
"merge_0_25pct",
"pivots_none",
}
def test_signature_uses_only_control_parity_fields():
arm = {
"candidates": 100,
"qualified": 10,
"qualified_net_avg_r": 0.2,
"qualified_net_avg_r_ex_top5": 0.05,
"full_book": {"sharpe": 2.0},
"holdout": {"train": {"sharpe": 1.0}},
"unrelated": "ignored",
}
assert "unrelated" not in matrix._signature(arm)
assert matrix._signature(arm)["full_book"] == {"sharpe": 2.0}
tests/unit/test_gtl_strength_sensitivity.py
-31
View File
@@ -1,31 +0,0 @@
"""Tests for the pre-registered strength-confirmation sensitivity band."""
from __future__ import annotations
from scripts import run_gtl_strength_sensitivity as matrix
def test_strength_sensitivity_is_ordered_and_contains_replication_arm():
assert len(matrix.GTL_STRENGTH_ARMS) == 9
assert matrix.GTL_STRENGTH_ARMS[0]["name"] == "control"
assert list(matrix.STRENGTH_SCALES) == sorted(matrix.STRENGTH_SCALES)
assert "strength_1000_intersection" in {
arm["name"] for arm in matrix.GTL_STRENGTH_ARMS
}
for arm in matrix.GTL_STRENGTH_ARMS[1:]:
config = matrix.composition._config(arm)
assert config["mode"] == "intersection"
assert len(config["confirmations"]) == 1
def test_stable_plateau_requires_adjacent_passing_scales():
arms = [
{"name": "control", "screen": {"advances": False}},
{"name": "strength_625", "screen": {"advances": True}},
{"name": "strength_750", "screen": {"advances": True}},
{"name": "strength_875", "screen": {"advances": False}},
{"name": "strength_1000", "screen": {"advances": True}},
]
assert matrix._stable_plateau_pairs(arms) == [
["strength_625", "strength_750"]
]
tests/unit/test_gtl_tuning_matrix.py
-50
View File
@@ -1,50 +0,0 @@
"""Tests for the single-command GTL research matrix."""
from __future__ import annotations
from scripts import run_gtl_tuning_matrix as matrix
def test_matrix_is_single_variable_and_has_unique_names():
assert len(matrix.GTL_TUNING_ARMS) == 20
assert matrix.GTL_TUNING_ARMS[0]["name"] == "control"
names = [arm["name"] for arm in matrix.GTL_TUNING_ARMS]
assert len(names) == len(set(names))
for arm in matrix.GTL_TUNING_ARMS[1:]:
config = matrix._arm_config(arm)
changed = [
key
for key, value in matrix.BASE_CONFIG.items()
if config[key] != value
]
assert len(changed) == 1, arm["name"]
def _compact_result(*, sharpe: float, drawdown: float, trades: int, robust_r: float) -> dict:
return {
"full_book": {
"sharpe": sharpe,
"max_drawdown_pct": drawdown,
"trades": trades,
},
"holdout": {
"train": {"sharpe": sharpe},
"test": {"sharpe": sharpe},
},
"qualified_net_avg_r_ex_top5": robust_r,
}
def test_screen_requires_every_pre_registered_guardrail():
control = _compact_result(sharpe=2.0, drawdown=20.0, trades=100, robust_r=0.1)
passing = _compact_result(sharpe=2.1, drawdown=19.0, trades=90, robust_r=0.05)
failing = _compact_result(sharpe=2.1, drawdown=19.0, trades=79, robust_r=0.05)
passed = matrix._screen_arm(passing, control)
failed = matrix._screen_arm(failing, control)
assert passed["advances"] is True
assert passed["passed"] == passed["total"] == 6
assert failed["advances"] is False
assert failed["checks"]["retains_80pct_trades"] is False
tests/unit/test_recommendation_service.py
-63
View File
@@ -252,52 +252,6 @@ def test_generate_targets_spreads_across_distance_bands():
assert any(m > 4.6 for m in multiples), "expected an aggressive (far) target" assert any(m > 4.6 for m in multiples), "expected an aggressive (far) target"
def test_generate_targets_research_candidate_limit_can_expand_or_disable_cap():
levels = [
_SRLevelStub(
id=index,
price_level=100.0 + index * 2.0,
type="resistance",
strength=50 + index,
)
for index in range(1, 9)
]
default = target_generator.generate_targets(
"long", 100.0, 97.0, levels, 2.0 # type: ignore[arg-type]
)
expanded = target_generator.generate_targets(
"long", 100.0, 97.0, levels, 2.0, # type: ignore[arg-type]
max_targets=8,
)
uncapped = target_generator.generate_targets(
"long", 100.0, 97.0, levels, 2.0, # type: ignore[arg-type]
max_targets=None,
)
assert len(default) == 5
assert len(expanded) == 8
assert len(uncapped) == 8
assert [row["price"] for row in uncapped] == sorted(
row["price"] for row in uncapped
)
def test_generate_targets_research_max_atr_override_is_universal():
levels = [
_SRLevelStub(id=1, price_level=110.0, type="resistance", strength=60),
_SRLevelStub(id=2, price_level=112.0, type="resistance", strength=60),
]
targets = target_generator.generate_targets(
"long", 100.0, 97.0, levels, 2.0, # type: ignore[arg-type]
max_targets=None,
max_atr_multiple_override=5.5,
)
assert [row["price"] for row in targets] == [110.0]
def test_probability_decreases_with_distance(): def test_probability_decreases_with_distance():
"""A far target must be far less likely than a near one — no 90% at +39%.""" """A far target must be far less likely than a near one — no 90% at +39%."""
config = { config = {
@@ -377,23 +331,6 @@ def test_zone_representative_levels_singletons_unchanged():
assert {round(r.price_level) for r in reps} == {120, 150} assert {round(r.price_level) for r in reps} == {120, 150}
def test_zone_representative_levels_accepts_research_tolerance():
from types import SimpleNamespace
from app.services.recommendation_service import _zone_representative_levels
levels = [
SimpleNamespace(id=1, price_level=110.0, type="resistance", strength=50),
SimpleNamespace(id=2, price_level=111.5, type="resistance", strength=50),
]
tight = _zone_representative_levels(levels, 100.0, tolerance=0.01)
wide = _zone_representative_levels(levels, 100.0, tolerance=0.02)
assert len(tight) == 2
assert len(wide) == 1
assert wide[0].price_level == 110.0
def test_zone_representative_levels_soft_strength_avoids_resaturation(): def test_zone_representative_levels_soft_strength_avoids_resaturation():
from types import SimpleNamespace from types import SimpleNamespace
from app.services.recommendation_service import _zone_representative_levels from app.services.recommendation_service import _zone_representative_levels