diff --git a/README.md b/README.md index 258c157..734f8b6 100644 --- a/README.md +++ b/README.md @@ -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 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 -whether its useful screening behavior can be made more explicit. It runs 20 -complete backtests **sequentially** so each arm gets the full worker pool: - -```bash -# macOS/Linux -.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). +The completed replacement, cohort-composition, and strength-sensitivity +matrices found no stable improvement over the frozen Gate Target Ladder. The +temporary matrix runners and tuning hooks have been retired; their three compact +consolidated report pairs remain in `reports/` as the decision audit. Keep the +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 @@ -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_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=` | 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=` | 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_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 | diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index c718cee..7b2b601 100644 --- a/app/services/backtest_service.py +++ b/app/services/backtest_service.py @@ -33,9 +33,7 @@ import statistics from collections import defaultdict from collections.abc import Callable from concurrent.futures import ProcessPoolExecutor -from dataclasses import asdict, dataclass from datetime import date, datetime, timedelta, timezone -from functools import lru_cache from types import SimpleNamespace from typing import Any @@ -73,7 +71,6 @@ from app.services.recommendation_service import ( _prune_floor_pinned_targets, _risk_level_from_conflicts, _select_primary_target, - _SR_ZONE_TOLERANCE, _zone_representative_levels, direction_analyzer, get_recommendation_config, @@ -86,7 +83,6 @@ from app.services.scoring_service import ( compute_technical_from_arrays, ) from app.services.sr_service import ( - GateTargetLadderConfig, MAX_LEVELS, detect_gate_target_ladder, detect_sr_levels, @@ -110,8 +106,6 @@ STRUCTURAL_OVERLAY_SOURCE_VARIANT = ( STRUCTURAL_OVERLAY_WEIGHT = 0.05 STRUCTURAL_OVERLAY_SCORE_KEY = "structural_overlay_95_5_score" EXPLICIT_TARGET_LADDER_VARIANT = "explicit_target_ladder" -GTL_TUNING_VARIANT = "gtl_tuning" -GTL_CONFIRMATION_VARIANT = "gtl_confirmation" RANGE_RESIDUAL_VARIANTS = { "rewrite_range504_structural_legacy_primary", "rewrite_range504_structural_primary2", @@ -122,69 +116,6 @@ RANGE_FACTOR_VARIANTS = { *RANGE_RESIDUAL_VARIANTS, } - -@dataclass(frozen=True) -class GTLResearchConfig: - """Research-only controls for one GTL matrix arm. - - The default values reproduce the explicit Gate Target Ladder. Production - never reads ``BACKTEST_GTL_CONFIG``. - """ - - name: str = "control" - lookback_bars: int | None = None - grid_bins: int = 20 - include_pivots: bool = True - pivot_window: int = 2 - touch_tolerance: float = 0.005 - merge_tolerance: float = 0.005 - strength_scale: float = 500.0 - zone_tolerance: float = 0.02 - candidate_limit: int | None = 5 - max_target_atr: float | None = None - - def __post_init__(self) -> None: - if not self.name.strip(): - raise ValueError("GTL research config name must not be empty") - if not isinstance(self.include_pivots, bool): - raise ValueError("GTL include_pivots must be a boolean") - if not 0.0 <= self.zone_tolerance < 0.10: - raise ValueError("GTL zone_tolerance must be in [0, 0.10)") - if self.candidate_limit is not None and self.candidate_limit < 1: - raise ValueError("GTL candidate_limit must be positive or null") - if self.max_target_atr is not None and self.max_target_atr <= 0: - raise ValueError("GTL max_target_atr must be positive or null") - # Reuse the detector's validation for its subset of fields. - self.ladder_config() - - def ladder_config(self) -> GateTargetLadderConfig: - return GateTargetLadderConfig( - lookback_bars=self.lookback_bars, - grid_bins=self.grid_bins, - include_pivots=self.include_pivots, - pivot_window=self.pivot_window, - touch_tolerance=self.touch_tolerance, - merge_tolerance=self.merge_tolerance, - strength_scale=self.strength_scale, - ) - - -@dataclass(frozen=True) -class GTLConfirmationConfig: - """Research-only composition of the frozen GTL and tuned variants.""" - - name: str = "control" - mode: str = "intersection" - confirmations: tuple[GTLResearchConfig, ...] = () - - def __post_init__(self) -> None: - if not self.name.strip(): - raise ValueError("GTL confirmation config name must not be empty") - if self.mode not in {"intersection", "union"}: - raise ValueError("GTL confirmation mode must be intersection or union") - if self.mode == "union" and len(self.confirmations) != 1: - raise ValueError("GTL union mode requires exactly one tuned variant") - # Cross-sectional signal evaluation (factor IC). Each candidate signal is a # point-in-time number computed from closes alone (sentiment/fundamentals have no # history here), sampled one as-of per ISO week, and graded by how its rank @@ -239,8 +170,6 @@ SR_RESEARCH_VARIANTS = { "legacy_range_grid_touch", "legacy_range_grid_neutral", EXPLICIT_TARGET_LADDER_VARIANT, - GTL_TUNING_VARIANT, - GTL_CONFIRMATION_VARIANT, STRUCTURAL_OVERLAY_VARIANT, *RANGE_FACTOR_VARIANTS, } @@ -255,72 +184,6 @@ def _sr_research_variant() -> str: return value -@lru_cache(maxsize=64) -def _parse_gtl_research_config(raw: str) -> GTLResearchConfig: - """Parse one auditable JSON configuration passed by the offline runner.""" - if not raw.strip(): - return GTLResearchConfig() - try: - payload = json.loads(raw) - except json.JSONDecodeError as exc: - raise ValueError("BACKTEST_GTL_CONFIG must be valid JSON") from exc - if not isinstance(payload, dict): - raise ValueError("BACKTEST_GTL_CONFIG must be a JSON object") - allowed = set(GTLResearchConfig.__dataclass_fields__) - unknown = sorted(set(payload) - allowed) - if unknown: - raise ValueError(f"Unknown GTL research config fields: {', '.join(unknown)}") - try: - return GTLResearchConfig(**payload) - except TypeError as exc: - raise ValueError(f"Invalid GTL research config: {exc}") from exc - - -def _gtl_research_config() -> GTLResearchConfig: - return _parse_gtl_research_config(os.getenv("BACKTEST_GTL_CONFIG", "")) - - -@lru_cache(maxsize=32) -def _parse_gtl_confirmation_config(raw: str) -> GTLConfirmationConfig: - """Parse one composition arm for the offline confirmation matrix.""" - if not raw.strip(): - return GTLConfirmationConfig() - try: - payload = json.loads(raw) - except json.JSONDecodeError as exc: - raise ValueError("BACKTEST_GTL_CONFIRM_CONFIG must be valid JSON") from exc - if not isinstance(payload, dict): - raise ValueError("BACKTEST_GTL_CONFIRM_CONFIG must be a JSON object") - allowed = {"name", "mode", "confirmations"} - unknown = sorted(set(payload) - allowed) - if unknown: - raise ValueError( - f"Unknown GTL confirmation config fields: {', '.join(unknown)}" - ) - raw_confirmations = payload.get("confirmations", []) - if not isinstance(raw_confirmations, list): - raise ValueError("GTL confirmations must be a JSON array") - confirmations: list[GTLResearchConfig] = [] - for item in raw_confirmations: - if not isinstance(item, dict): - raise ValueError("Each GTL confirmation must be a JSON object") - confirmations.append(_parse_gtl_research_config(json.dumps(item))) - try: - return GTLConfirmationConfig( - name=payload.get("name", "control"), - mode=payload.get("mode", "intersection"), - confirmations=tuple(confirmations), - ) - except (AttributeError, TypeError, ValueError) as exc: - raise ValueError(f"Invalid GTL confirmation config: {exc}") from exc - - -def _gtl_confirmation_config() -> GTLConfirmationConfig: - return _parse_gtl_confirmation_config( - os.getenv("BACKTEST_GTL_CONFIRM_CONFIG", "") - ) - - def _sr_detector_variant(sr_variant: str) -> str: """Map factor-gated research arms to the detector they hold fixed.""" if sr_variant in {"production_range504", STRUCTURAL_OVERLAY_VARIANT}: @@ -361,8 +224,6 @@ def _primary_min_rr_for_variant(sr_variant: str, activation: dict) -> float: "production_control", "production_range504", EXPLICIT_TARGET_LADDER_VARIANT, - GTL_TUNING_VARIANT, - GTL_CONFIRMATION_VARIANT, STRUCTURAL_OVERLAY_VARIANT, } or sr_variant.endswith("_legacy_primary") @@ -477,7 +338,6 @@ def _window_setups( activation: dict, *, sr_variant: str | None = None, - gtl_research_config: GTLResearchConfig | None = None, ) -> list[dict]: """Rebuild the setup(s) at the last bar of ``window_records`` (the as-of date), using only those bars. Returns one dict per tradeable direction.""" @@ -506,11 +366,6 @@ def _window_setups( sr_variant = sr_variant or _sr_research_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) if sr_variant == "legacy_geometry_neutral": detected_levels = detect_sr_levels_legacy( @@ -530,15 +385,6 @@ def _window_setups( lows, closes, ) - elif sr_variant == GTL_TUNING_VARIANT: - if gtl_config is None: # pragma: no cover - guarded by the variant above - raise RuntimeError("GTL tuning variant requires a research config") - detected_levels = detect_gate_target_ladder( - highs, - lows, - closes, - config=gtl_config.ladder_config(), - ) elif sr_variant in {"legacy_range_grid_touch", "legacy_range_grid_neutral"}: detected_levels = detect_sr_levels_legacy( highs, @@ -589,20 +435,9 @@ def _window_setups( gate_levels, entry, strength_mode=zone_strength_mode, - tolerance=( - gtl_config.zone_tolerance if gtl_config else _SR_ZONE_TOLERANCE - ), ) zone_levels = _apply_zone_strength_variant(zone_levels, sr_variant) - targets = target_generator.generate_targets( - direction, - entry, - stop, - zone_levels, - atr, - 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), - ) + targets = target_generator.generate_targets(direction, entry, stop, zone_levels, atr) if not targets: fallback_k = _atr_target_fallback_k() if fallback_k is None: @@ -749,87 +584,6 @@ def _structural_overlay_window_setups( return tagged -def _gtl_confirmation_window_setups( - window_records: list, - config: dict, - activation: dict, - *, - confirmation_config: GTLConfirmationConfig | None = None, -) -> list[dict]: - """Compose tuned GTLs around the frozen ladder without changing it silently. - - ``intersection`` retains the frozen setup geometry and requires every tuned - variant to clear the core gate in the same direction. ``union`` preserves a - frozen setup whenever it already clears the core gate, and otherwise admits - the one tuned variant's setup. This makes retained, removed, and added - cohorts explicit instead of conflating them in a replacement arm. - """ - research = confirmation_config or _gtl_confirmation_config() - production = _window_setups( - window_records, - config, - activation, - sr_variant=EXPLICIT_TARGET_LADDER_VARIANT, - ) - tuned_sets = [ - _window_setups( - window_records, - config, - activation, - sr_variant=GTL_TUNING_VARIANT, - gtl_research_config=tuned_config, - ) - for tuned_config in research.confirmations - ] - tuned_by_direction = [ - {row["direction"]: row for row in rows} - for rows in tuned_sets - ] - - def annotate(row: dict, passes: list[bool], source: str) -> dict: - tagged = dict(row) - tagged["sr_variant"] = GTL_CONFIRMATION_VARIANT - tagged["gtl_confirmation_name"] = research.name - tagged["gtl_confirmation_mode"] = research.mode - tagged["gtl_confirmation_source"] = source - tagged["gtl_confirmation_passes"] = passes - tagged["gtl_confirmation_all_pass"] = all(passes) - return tagged - - if research.mode == "intersection": - tagged: list[dict] = [] - for production_row in production: - direction = production_row["direction"] - passes = [ - bool(rows.get(direction) and rows[direction].get("meets_core")) - for rows in tuned_by_direction - ] - row = annotate(production_row, passes, "control") - row["meets_core"] = bool(production_row.get("meets_core")) and all( - passes - ) - tagged.append(row) - return tagged - - # Union mode is validated to contain exactly one tuned variant. Keep one - # setup per direction: frozen geometry wins whenever it already qualifies; - # tuned geometry is used only for a genuinely added core-qualified setup. - production_by_direction = {row["direction"]: row for row in production} - tuned_by_dir = tuned_by_direction[0] - tagged = [] - for direction in sorted(set(production_by_direction) | set(tuned_by_dir)): - production_row = production_by_direction.get(direction) - tuned_row = tuned_by_dir.get(direction) - tuned_pass = bool(tuned_row and tuned_row.get("meets_core")) - if production_row is not None and production_row.get("meets_core"): - tagged.append(annotate(production_row, [tuned_pass], "control")) - elif tuned_pass and tuned_row is not None: - tagged.append(annotate(tuned_row, [True], "tuned_addition")) - elif production_row is not None: - tagged.append(annotate(production_row, [tuned_pass], "control")) - return tagged - - def _stop_fill_r(direction: str, entry: float, stop: float, bar) -> float: """Realized R when the stop is hit on ``bar``: filled at the stop, or at the bar's open when price gapped through it — so a gap can lose more than −1R, @@ -934,17 +688,16 @@ def _replay_ticker( ) vol_6m = _realized_vol_6m(closes, len(window) - 1) - if sr_variant == STRUCTURAL_OVERLAY_VARIANT: - setups = _structural_overlay_window_setups(window, config, activation) - elif sr_variant == GTL_CONFIRMATION_VARIANT: - setups = _gtl_confirmation_window_setups(window, config, activation) - else: - setups = _window_setups( + setups = ( + _structural_overlay_window_setups(window, config, activation) + if sr_variant == STRUCTURAL_OVERLAY_VARIANT + else _window_setups( window, config, activation, sr_variant=sr_variant, ) + ) for s in setups: outcome, outcome_date = evaluate_setup_against_bars( s["direction"], s["stop"], s["target"], forward_bars, HORIZON @@ -1010,13 +763,6 @@ def _replay_ticker( "structural_overlay_gate_level_count": s.get( "structural_overlay_gate_level_count" ), - "gtl_confirmation_name": s.get("gtl_confirmation_name"), - "gtl_confirmation_mode": s.get("gtl_confirmation_mode"), - "gtl_confirmation_source": s.get("gtl_confirmation_source"), - "gtl_confirmation_passes": s.get("gtl_confirmation_passes"), - "gtl_confirmation_all_pass": s.get( - "gtl_confirmation_all_pass" - ), "outcome": outcome, "target_hit": target_hit, "realized_r": realized_r, @@ -1089,9 +835,6 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict: range_logs: list[float] = [] overlay_rows = 0 overlay_pass = 0 - confirmation_rows = 0 - confirmation_pass = 0 - confirmation_tuned_additions = 0 for cand in candidates: sources = list(cand.get("primary_sources") or []) for source in sources: @@ -1107,12 +850,6 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict: if cand.get("structural_overlay_pass") is not None: overlay_rows += 1 overlay_pass += int(bool(cand["structural_overlay_pass"])) - if cand.get("gtl_confirmation_all_pass") is not None: - confirmation_rows += 1 - confirmation_pass += int(bool(cand["gtl_confirmation_all_pass"])) - confirmation_tuned_additions += int( - cand.get("gtl_confirmation_source") == "tuned_addition" - ) def avg(values: list[float] | list[int]) -> float | None: return round(sum(values) / len(values), 3) if values else None @@ -1137,9 +874,6 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict: "structural_overlay_weight": ( STRUCTURAL_OVERLAY_WEIGHT if overlay_rows else None ), - "gtl_confirmation_rows": confirmation_rows, - "gtl_confirmation_pass": confirmation_pass, - "gtl_confirmation_tuned_additions": confirmation_tuned_additions, } @@ -1186,15 +920,6 @@ def _sr_candidate_audit(candidates: list[dict], min_percentile: float) -> list[d "structural_overlay_gate_level_count": int( cand.get("structural_overlay_gate_level_count", 0) or 0 ), - "gtl_confirmation_name": cand.get("gtl_confirmation_name"), - "gtl_confirmation_mode": cand.get("gtl_confirmation_mode"), - "gtl_confirmation_source": cand.get("gtl_confirmation_source"), - "gtl_confirmation_passes": list( - cand.get("gtl_confirmation_passes") or [] - ), - "gtl_confirmation_all_pass": cand.get( - "gtl_confirmation_all_pass" - ), "rr": round(float(cand.get("rr", 0.0)), 6), "primary_prob": round(float(cand.get("primary_prob", 0.0)), 6), "primary_sources": list(cand.get("primary_sources") or []), @@ -3601,16 +3326,6 @@ async def run_backtest( "min_lookback": MIN_LOOKBACK, "cost_per_side_pct": round(COST_PER_SIDE * 100, 3), "sr_variant": _sr_research_variant(), - "gtl_config": ( - asdict(_gtl_research_config()) - if _sr_research_variant() == GTL_TUNING_VARIANT - else None - ), - "gtl_confirmation_config": ( - asdict(_gtl_confirmation_config()) - if _sr_research_variant() == GTL_CONFIRMATION_VARIANT - else None - ), "range_factor_lookback": RANGE_FACTOR_LOOKBACK, "range_factor_min_log": RANGE_FACTOR_MIN_LOG, "range_factor_min_ratio": round(math.exp(RANGE_FACTOR_MIN_LOG), 4), diff --git a/app/services/recommendation_service.py b/app/services/recommendation_service.py index ae33286..d2d6f87 100644 --- a/app/services/recommendation_service.py +++ b/app/services/recommendation_service.py @@ -93,7 +93,6 @@ def _zone_representative_levels( entry_price: float, *, strength_mode: str = "sum", - tolerance: float = _SR_ZONE_TOLERANCE, ) -> list[Any]: """Collapse near-duplicate S/R levels into one representative per zone. @@ -125,7 +124,7 @@ def _zone_representative_levels( zones = cluster_sr_zones( level_dicts, entry_price, - tolerance=tolerance, + tolerance=_SR_ZONE_TOLERANCE, strength_mode=strength_mode, ) @@ -316,16 +315,9 @@ class TargetGenerator: stop_loss: float, sr_levels: list[SRLevel], atr_value: float, - *, - max_targets: int | None = 5, - max_atr_multiple_override: float | None = None, ) -> list[dict[str, Any]]: if atr_value <= 0: return [] - if max_targets is not None and max_targets < 1: - raise ValueError("max_targets must be positive or None") - if max_atr_multiple_override is not None and max_atr_multiple_override <= 0: - raise ValueError("max_atr_multiple_override must be positive or None") risk = abs(entry_price - stop_loss) if risk <= 0: @@ -334,12 +326,11 @@ class TargetGenerator: candidates: list[dict[str, Any]] = [] atr_pct = atr_value / entry_price if entry_price > 0 else 0.0 - max_atr_multiple: float | None = max_atr_multiple_override - if max_atr_multiple is None: - if atr_pct > 0.05: - max_atr_multiple = 10.0 - elif atr_pct < 0.02: - max_atr_multiple = 3.0 + max_atr_multiple: float | None = None + if atr_pct > 0.05: + max_atr_multiple = 10.0 + elif atr_pct < 0.02: + max_atr_multiple = 3.0 for level in sr_levels: is_candidate = False @@ -392,12 +383,6 @@ class TargetGenerator: if not candidates: return [] - if max_targets is None: - candidates.sort(key=lambda row: row["distance_from_entry"]) - for target in candidates: - target.pop("quality", None) - return candidates - # Select up to 5 targets that SPAN the distance range, instead of the # top-5 by quality (which biases toward far, high-R:R levels and buries # every nearby target). Guarantees the nearest level plus a @@ -411,8 +396,6 @@ class TargetGenerator: selected_ids: set[int] = set() def _add(candidate: dict[str, Any] | None) -> None: - if len(selected) >= max_targets: - return if candidate is not None and candidate["sr_level_id"] not in selected_ids: selected.append(candidate) selected_ids.add(candidate["sr_level_id"]) @@ -425,7 +408,7 @@ class TargetGenerator: _add(max(bucket, key=lambda c: c["quality"])) # Fill remaining slots with the next-best by quality for candidate in sorted(candidates, key=lambda c: c["quality"], reverse=True): - if len(selected) >= max_targets: + if len(selected) >= 5: break _add(candidate) diff --git a/app/services/sr_service.py b/app/services/sr_service.py index b1090fc..b6ac7c7 100644 --- a/app/services/sr_service.py +++ b/app/services/sr_service.py @@ -10,7 +10,6 @@ and persists to DB. from __future__ import annotations import math -from dataclasses import dataclass from datetime import datetime 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 - -@dataclass(frozen=True) -class GateTargetLadderConfig: - """Research controls for the transient Gate Target Ladder. - - Live callers omit this object and retain the frozen legacy-parity path. - The backtest may pass an explicit configuration to isolate one ladder - mechanism at a time without changing Structural S/R. - """ - - lookback_bars: int | None = None - grid_bins: int = 20 - include_pivots: bool = True - pivot_window: int = 2 - touch_tolerance: float = DEFAULT_TOLERANCE - merge_tolerance: float = DEFAULT_TOLERANCE - strength_scale: float = 500.0 - - def __post_init__(self) -> None: - if self.lookback_bars is not None and self.lookback_bars < 20: - raise ValueError("GTL lookback_bars must be at least 20 or None") - if self.grid_bins < 2: - raise ValueError("GTL grid_bins must be at least 2") - if self.pivot_window < 1: - raise ValueError("GTL pivot_window must be at least 1") - for name, value in ( - ("touch_tolerance", self.touch_tolerance), - ("merge_tolerance", self.merge_tolerance), - ): - if not 0.0 <= value < 0.10: - raise ValueError(f"GTL {name} must be in [0, 0.10)") - if self.strength_scale <= 0: - raise ValueError("GTL strength_scale must be positive") - VP_LOOKBACK = 252 TOUCH_LOOKBACK = 252 PIVOT_LOOKBACK = 504 @@ -562,8 +527,6 @@ def detect_gate_target_ladder( lows: list[float], closes: list[float], tolerance: float = DEFAULT_TOLERANCE, - *, - config: GateTargetLadderConfig | None = None, ) -> list[dict]: """Build the scanner's internal, volume-free target proposal ladder. @@ -573,115 +536,14 @@ def detect_gate_target_ladder( profile calculation. The returned levels are transient and must not be persisted as chart S/R. """ - if config is None: - # Frozen live/default path. Keeping the established helper here makes - # the absence of research configuration an exact compatibility promise. - return detect_sr_levels_legacy( - highs, - lows, - closes, - [0] * len(closes), - tolerance, - explicit_range_grid=True, - ) - - if tolerance != DEFAULT_TOLERANCE: - raise ValueError("Pass either GTL config tolerances or tolerance, not both") - if not closes: - return [] - if not (len(highs) == len(lows) == len(closes)): - raise ValueError("GTL highs, lows, and closes must have equal lengths") - - if config.lookback_bars is not None: - highs = highs[-config.lookback_bars:] - lows = lows[-config.lookback_bars:] - closes = closes[-config.lookback_bars:] - - candidates: list[tuple[float, str]] = [] - try: - candidates.extend( - (float(price), "range_grid") - for price in _legacy_range_grid_nodes( - highs, - lows, - closes, - num_bins=config.grid_bins, - ) - ) - except ValidationError: - pass - - if config.include_pivots: - try: - pivots = compute_pivot_points( - highs, - lows, - closes, - window=config.pivot_window, - ) - candidates.extend( - (float(price), "pivot_point") - for price in pivots.get("swing_highs", []) + pivots.get("swing_lows", []) - ) - except ValidationError: - pass - if not candidates: - return [] - - total_bars = len(closes) - raw: list[dict] = [] - for price, method in candidates: - touch_band = abs(price) * config.touch_tolerance if price else config.touch_tolerance - touches = sum( - 1 - for low, high in zip(lows, highs, strict=False) - if low - touch_band <= price <= high + touch_band - ) - strength = max( - 0, - min(100, int(round((touches / total_bars) * config.strength_scale))), - ) - raw.append({ - "price_level": price, - "strength": strength, - "detection_method": method, - "type": "", - "sources": [method], - "rejection_count": touches, - "last_rejection_age": None, - "weighted_respects": float(touches), - }) - - merged: list[dict] = [] - for level in sorted(raw, key=lambda row: row["price_level"]): - if not merged: - merged.append(dict(level)) - continue - last = merged[-1] - ref = last["price_level"] - merge_band = abs(ref) * config.merge_tolerance if ref else config.merge_tolerance - if abs(level["price_level"] - ref) > merge_band: - merged.append(dict(level)) - continue - last["price_level"] = round( - (last["price_level"] + level["price_level"]) / 2.0, - 4, - ) - last["strength"] = min(100, last["strength"] + level["strength"]) - sources = set(last.get("sources") or [last["detection_method"]]) - sources |= set(level.get("sources") or [level["detection_method"]]) - last["sources"] = sorted(sources) - last["detection_method"] = ( - next(iter(sources)) if len(sources) == 1 else "merged" - ) - last["rejection_count"] = max( - int(last.get("rejection_count", 0)), - int(level.get("rejection_count", 0)), - ) - - _tag_levels(merged, closes[-1]) - merged.sort(key=lambda row: row["strength"], reverse=True) - return merged + return detect_sr_levels_legacy( + highs, + lows, + closes, + [0] * len(closes), + tolerance, + explicit_range_grid=True, + ) def _merge_levels( diff --git a/docs/research/sr-levels-and-exits.md b/docs/research/sr-levels-and-exits.md index b110eea..35c650d 100644 --- a/docs/research/sr-levels-and-exits.md +++ b/docs/research/sr-levels-and-exits.md @@ -806,10 +806,10 @@ or perform a production deployment. #### GTL tuning matrix -Exact parity establishes a safe, explicit control, but it does not prove that -the inherited GTL constants are optimal. The `gtl_tuning` backtest arm exposes -only those constants to an offline configuration; the live scanner continues -to call the frozen default helper and cannot read this research configuration. +Exact parity established a safe, explicit control but did not prove that the +inherited GTL constants were optimal. A temporary offline harness exposed those +constants without changing the live scanner. That harness has now been retired; +the compact consolidated reports remain as the reproducible decision record. The single-command matrix contains 20 full-period arms. Each non-control arm 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? | | 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 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/ @@ -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) | 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 -`gtl_confirmation` matrix therefore preserves frozen control geometry and -decomposes each selected variant into: +cohort, so its portfolio result could not say which part helped. The archived +confirmation matrix therefore preserved frozen control geometry and decomposed +each selected variant into: - **intersection** — only control setups also core-qualified by the variant; - **union** — all core-qualified control setups plus genuinely added variant setups, using tuned geometry only for those additions. It also tests pre-registered intersections among the three high-breadth -confirmers. Its control path must exactly reproduce the completed tuning -matrix before any research arm is accepted. - -```bash -.venv/bin/python scripts/run_gtl_confirmation_matrix.py \ - backtest_snapshots/prod.sqlite --workers 12 -``` +confirmers. Its control path exactly reproduced the completed tuning matrix. Result: **13/13 arms completed with exact control parity; no arm passed all six 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 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, -2000), using intersection only. It must reproduce both the frozen control and -the completed strength-1000 result exactly. Promotion requires at least two +2000), using intersection only. It reproduced both the frozen control and the +completed strength-1000 result exactly. Promotion required at least two adjacent non-control scales to pass all six original checks; an isolated winner -is rejected as sensitivity. - -```bash -.venv/bin/python scripts/run_gtl_strength_sensitivity.py \ - backtest_snapshots/prod.sqlite --workers 12 -``` +was rejected as sensitivity. Final result: **9/9 arms completed, control parity passed, and the 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 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 promotion holdout. These arms can isolate mechanism, but neither may ship without new future data or a separately pre-registered walk-forward protocol. diff --git a/reports/README.md b/reports/README.md index 5173d91..42cc996 100644 --- a/reports/README.md +++ b/reports/README.md @@ -61,32 +61,30 @@ after scanner integration passed: the regenerated report changed only its `generated_at` timestamp, confirming that the shared helper preserves exact 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-YYYYMMDD-gtl-tuning-matrix.md` +- `backtest-20260713-gtl-tuning-matrix.json` +- `backtest-20260713-gtl-tuning-matrix.md` -These two consolidated files are the reports worth retaining. The runner uses -a hidden temporary directory for its 20 full per-arm reports and removes it -after successful consolidation unless `--keep-arm-reports` is supplied. +These two compact consolidated files are retained; the 20 full per-arm reports +were removed after consolidation. The completed 2026-07-13 matrix found no single-parameter replacement that 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-YYYYMMDD-gtl-confirmation-matrix.md` +- `backtest-20260713-gtl-confirmation-matrix.json` +- `backtest-20260713-gtl-confirmation-matrix.md` -Those become the next decision point; detailed per-arm reports remain temporary -unless explicitly retained. +Detailed per-arm reports were not retained. The completed confirmation matrix found one near-hit but no formal winner: 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- -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-YYYYMMDD-gtl-strength-sensitivity.md` +- `backtest-20260713-gtl-strength-sensitivity.json` +- `backtest-20260713-gtl-strength-sensitivity.md` Its pre-registered decision requires at least two adjacent scales to pass all 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 GTL decision point. No confirmation or tuned strength value should be promoted from this snapshot. + +The temporary matrix runners and their configurable backtest hooks were removed +after consolidation. The normal local snapshot backtester remains. diff --git a/scripts/run_backtest_snapshot.py b/scripts/run_backtest_snapshot.py index 77f3487..ac3d0f0 100644 --- a/scripts/run_backtest_snapshot.py +++ b/scripts/run_backtest_snapshot.py @@ -61,25 +61,10 @@ def _parse_args() -> argparse.Namespace: "rewrite_range504_structural_primary2", "production_structural_overlay", "explicit_target_ladder", - "gtl_tuning", - "gtl_confirmation", ), default=None, 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( "--entry-start", default=None, @@ -215,16 +200,6 @@ async def _main() -> None: os.environ["BACKTEST_ALLOW_SPAWN"] = "1" if 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: os.environ["BACKTEST_ENTRY_START"] = args.entry_start if args.entry_end: diff --git a/scripts/run_gtl_confirmation_matrix.py b/scripts/run_gtl_confirmation_matrix.py deleted file mode 100644 index cc8ee2f..0000000 --- a/scripts/run_gtl_confirmation_matrix.py +++ /dev/null @@ -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() diff --git a/scripts/run_gtl_strength_sensitivity.py b/scripts/run_gtl_strength_sensitivity.py deleted file mode 100644 index ea6e378..0000000 --- a/scripts/run_gtl_strength_sensitivity.py +++ /dev/null @@ -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() diff --git a/scripts/run_gtl_tuning_matrix.py b/scripts/run_gtl_tuning_matrix.py deleted file mode 100644 index b838d33..0000000 --- a/scripts/run_gtl_tuning_matrix.py +++ /dev/null @@ -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() diff --git a/tests/unit/test_backtest_service.py b/tests/unit/test_backtest_service.py index 18df51f..8047788 100644 --- a/tests/unit/test_backtest_service.py +++ b/tests/unit/test_backtest_service.py @@ -2,7 +2,6 @@ from __future__ import annotations -import json import math from datetime import date, timedelta from types import SimpleNamespace @@ -804,8 +803,6 @@ def test_sr_research_variant_is_explicit_and_validated(monkeypatch): "rewrite_range504_structural_primary2", "production_structural_overlay", "explicit_target_ladder", - "gtl_tuning", - "gtl_confirmation", ): monkeypatch.setenv("BACKTEST_SR_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", activation, ) == 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): @@ -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( "variant", ["legacy_geometry_neutral", "legacy_range_grid_neutral"], diff --git a/tests/unit/test_detect_sr_levels.py b/tests/unit/test_detect_sr_levels.py index e3d14bc..3f8676a 100644 --- a/tests/unit/test_detect_sr_levels.py +++ b/tests/unit/test_detect_sr_levels.py @@ -2,10 +2,7 @@ from __future__ import annotations -import pytest - from app.services.sr_service import ( - GateTargetLadderConfig, MAX_LEVELS, _bar_respect_weight, _cap_levels, @@ -317,70 +314,6 @@ class TestDetectSrLevels: 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): highs, lows, closes, volumes = _make_series(n=500) shifted_volumes = [volume * (i + 1) for i, volume in enumerate(volumes)] diff --git a/tests/unit/test_gtl_confirmation_matrix.py b/tests/unit/test_gtl_confirmation_matrix.py deleted file mode 100644 index d6d763a..0000000 --- a/tests/unit/test_gtl_confirmation_matrix.py +++ /dev/null @@ -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} diff --git a/tests/unit/test_gtl_strength_sensitivity.py b/tests/unit/test_gtl_strength_sensitivity.py deleted file mode 100644 index 9d3ce11..0000000 --- a/tests/unit/test_gtl_strength_sensitivity.py +++ /dev/null @@ -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"] - ] diff --git a/tests/unit/test_gtl_tuning_matrix.py b/tests/unit/test_gtl_tuning_matrix.py deleted file mode 100644 index df20c3c..0000000 --- a/tests/unit/test_gtl_tuning_matrix.py +++ /dev/null @@ -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 diff --git a/tests/unit/test_recommendation_service.py b/tests/unit/test_recommendation_service.py index d9f789f..e2c41b3 100644 --- a/tests/unit/test_recommendation_service.py +++ b/tests/unit/test_recommendation_service.py @@ -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" -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(): """A far target must be far less likely than a near one — no 90% at +39%.""" config = { @@ -377,23 +331,6 @@ def test_zone_representative_levels_singletons_unchanged(): 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(): from types import SimpleNamespace from app.services.recommendation_service import _zone_representative_levels