diff --git a/README.md b/README.md index 6e70d69..804ee9c 100644 --- a/README.md +++ b/README.md @@ -459,6 +459,33 @@ 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 + +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. + ### Reading a local backtest report The deployed **Signals → Track Record** page is deliberately trimmed to validation @@ -497,6 +524,7 @@ 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 7b2b601..dfecd1f 100644 --- a/app/services/backtest_service.py +++ b/app/services/backtest_service.py @@ -33,7 +33,9 @@ import statistics from collections import defaultdict from collections.abc import Callable from concurrent.futures import ProcessPoolExecutor +from dataclasses import asdict, dataclass from datetime import date, datetime, timedelta, timezone +from functools import lru_cache from types import SimpleNamespace from typing import Any @@ -71,6 +73,7 @@ from app.services.recommendation_service import ( _prune_floor_pinned_targets, _risk_level_from_conflicts, _select_primary_target, + _SR_ZONE_TOLERANCE, _zone_representative_levels, direction_analyzer, get_recommendation_config, @@ -83,6 +86,7 @@ from app.services.scoring_service import ( compute_technical_from_arrays, ) from app.services.sr_service import ( + GateTargetLadderConfig, MAX_LEVELS, detect_gate_target_ladder, detect_sr_levels, @@ -106,6 +110,7 @@ STRUCTURAL_OVERLAY_SOURCE_VARIANT = ( STRUCTURAL_OVERLAY_WEIGHT = 0.05 STRUCTURAL_OVERLAY_SCORE_KEY = "structural_overlay_95_5_score" EXPLICIT_TARGET_LADDER_VARIANT = "explicit_target_ladder" +GTL_TUNING_VARIANT = "gtl_tuning" RANGE_RESIDUAL_VARIANTS = { "rewrite_range504_structural_legacy_primary", "rewrite_range504_structural_primary2", @@ -116,6 +121,52 @@ RANGE_FACTOR_VARIANTS = { *RANGE_RESIDUAL_VARIANTS, } + +@dataclass(frozen=True) +class GTLResearchConfig: + """Research-only controls for one GTL matrix arm. + + The default values reproduce the explicit Gate Target Ladder. Production + never reads ``BACKTEST_GTL_CONFIG``. + """ + + name: str = "control" + lookback_bars: int | None = None + grid_bins: int = 20 + include_pivots: bool = True + pivot_window: int = 2 + touch_tolerance: float = 0.005 + merge_tolerance: float = 0.005 + strength_scale: float = 500.0 + zone_tolerance: float = 0.02 + candidate_limit: int | None = 5 + max_target_atr: float | None = None + + def __post_init__(self) -> None: + if not self.name.strip(): + raise ValueError("GTL research config name must not be empty") + if not isinstance(self.include_pivots, bool): + raise ValueError("GTL include_pivots must be a boolean") + if not 0.0 <= self.zone_tolerance < 0.10: + raise ValueError("GTL zone_tolerance must be in [0, 0.10)") + if self.candidate_limit is not None and self.candidate_limit < 1: + raise ValueError("GTL candidate_limit must be positive or null") + if self.max_target_atr is not None and self.max_target_atr <= 0: + raise ValueError("GTL max_target_atr must be positive or null") + # Reuse the detector's validation for its subset of fields. + self.ladder_config() + + def ladder_config(self) -> GateTargetLadderConfig: + return GateTargetLadderConfig( + lookback_bars=self.lookback_bars, + grid_bins=self.grid_bins, + include_pivots=self.include_pivots, + pivot_window=self.pivot_window, + touch_tolerance=self.touch_tolerance, + merge_tolerance=self.merge_tolerance, + strength_scale=self.strength_scale, + ) + # Cross-sectional signal evaluation (factor IC). Each candidate signal is a # point-in-time number computed from closes alone (sentiment/fundamentals have no # history here), sampled one as-of per ISO week, and graded by how its rank @@ -170,6 +221,7 @@ SR_RESEARCH_VARIANTS = { "legacy_range_grid_touch", "legacy_range_grid_neutral", EXPLICIT_TARGET_LADDER_VARIANT, + GTL_TUNING_VARIANT, STRUCTURAL_OVERLAY_VARIANT, *RANGE_FACTOR_VARIANTS, } @@ -184,6 +236,31 @@ def _sr_research_variant() -> str: return value +@lru_cache(maxsize=64) +def _parse_gtl_research_config(raw: str) -> GTLResearchConfig: + """Parse one auditable JSON configuration passed by the offline runner.""" + if not raw.strip(): + return GTLResearchConfig() + try: + payload = json.loads(raw) + except json.JSONDecodeError as exc: + raise ValueError("BACKTEST_GTL_CONFIG must be valid JSON") from exc + if not isinstance(payload, dict): + raise ValueError("BACKTEST_GTL_CONFIG must be a JSON object") + allowed = set(GTLResearchConfig.__dataclass_fields__) + unknown = sorted(set(payload) - allowed) + if unknown: + raise ValueError(f"Unknown GTL research config fields: {', '.join(unknown)}") + try: + return GTLResearchConfig(**payload) + except TypeError as exc: + raise ValueError(f"Invalid GTL research config: {exc}") from exc + + +def _gtl_research_config() -> GTLResearchConfig: + return _parse_gtl_research_config(os.getenv("BACKTEST_GTL_CONFIG", "")) + + def _sr_detector_variant(sr_variant: str) -> str: """Map factor-gated research arms to the detector they hold fixed.""" if sr_variant in {"production_range504", STRUCTURAL_OVERLAY_VARIANT}: @@ -224,6 +301,7 @@ def _primary_min_rr_for_variant(sr_variant: str, activation: dict) -> float: "production_control", "production_range504", EXPLICIT_TARGET_LADDER_VARIANT, + GTL_TUNING_VARIANT, STRUCTURAL_OVERLAY_VARIANT, } or sr_variant.endswith("_legacy_primary") @@ -366,6 +444,7 @@ def _window_setups( sr_variant = sr_variant or _sr_research_variant() detector_variant = _sr_detector_variant(sr_variant) + gtl_config = _gtl_research_config() if sr_variant == GTL_TUNING_VARIANT else None range_504_log = _range_504_log(highs, lows) if sr_variant == "legacy_geometry_neutral": detected_levels = detect_sr_levels_legacy( @@ -385,6 +464,15 @@ def _window_setups( lows, closes, ) + elif sr_variant == GTL_TUNING_VARIANT: + if gtl_config is None: # pragma: no cover - guarded by the variant above + raise RuntimeError("GTL tuning variant requires a research config") + detected_levels = detect_gate_target_ladder( + highs, + lows, + closes, + config=gtl_config.ladder_config(), + ) elif sr_variant in {"legacy_range_grid_touch", "legacy_range_grid_neutral"}: detected_levels = detect_sr_levels_legacy( highs, @@ -435,9 +523,20 @@ def _window_setups( gate_levels, entry, strength_mode=zone_strength_mode, + tolerance=( + gtl_config.zone_tolerance if gtl_config else _SR_ZONE_TOLERANCE + ), ) zone_levels = _apply_zone_strength_variant(zone_levels, sr_variant) - targets = target_generator.generate_targets(direction, entry, stop, zone_levels, atr) + targets = target_generator.generate_targets( + direction, + entry, + stop, + zone_levels, + atr, + max_targets=(gtl_config.candidate_limit if gtl_config else 5), + max_atr_multiple_override=(gtl_config.max_target_atr if gtl_config else None), + ) if not targets: fallback_k = _atr_target_fallback_k() if fallback_k is None: @@ -3326,6 +3425,11 @@ async def run_backtest( "min_lookback": MIN_LOOKBACK, "cost_per_side_pct": round(COST_PER_SIDE * 100, 3), "sr_variant": _sr_research_variant(), + "gtl_config": ( + asdict(_gtl_research_config()) + if _sr_research_variant() == GTL_TUNING_VARIANT + else None + ), "range_factor_lookback": RANGE_FACTOR_LOOKBACK, "range_factor_min_log": RANGE_FACTOR_MIN_LOG, "range_factor_min_ratio": round(math.exp(RANGE_FACTOR_MIN_LOG), 4), diff --git a/app/services/recommendation_service.py b/app/services/recommendation_service.py index d2d6f87..ae33286 100644 --- a/app/services/recommendation_service.py +++ b/app/services/recommendation_service.py @@ -93,6 +93,7 @@ def _zone_representative_levels( entry_price: float, *, strength_mode: str = "sum", + tolerance: float = _SR_ZONE_TOLERANCE, ) -> list[Any]: """Collapse near-duplicate S/R levels into one representative per zone. @@ -124,7 +125,7 @@ def _zone_representative_levels( zones = cluster_sr_zones( level_dicts, entry_price, - tolerance=_SR_ZONE_TOLERANCE, + tolerance=tolerance, strength_mode=strength_mode, ) @@ -315,9 +316,16 @@ class TargetGenerator: stop_loss: float, sr_levels: list[SRLevel], atr_value: float, + *, + max_targets: int | None = 5, + max_atr_multiple_override: float | None = None, ) -> list[dict[str, Any]]: if atr_value <= 0: return [] + if max_targets is not None and max_targets < 1: + raise ValueError("max_targets must be positive or None") + if max_atr_multiple_override is not None and max_atr_multiple_override <= 0: + raise ValueError("max_atr_multiple_override must be positive or None") risk = abs(entry_price - stop_loss) if risk <= 0: @@ -326,11 +334,12 @@ class TargetGenerator: candidates: list[dict[str, Any]] = [] atr_pct = atr_value / entry_price if entry_price > 0 else 0.0 - max_atr_multiple: float | None = None - if atr_pct > 0.05: - max_atr_multiple = 10.0 - elif atr_pct < 0.02: - max_atr_multiple = 3.0 + max_atr_multiple: float | None = max_atr_multiple_override + if max_atr_multiple is None: + if atr_pct > 0.05: + max_atr_multiple = 10.0 + elif atr_pct < 0.02: + max_atr_multiple = 3.0 for level in sr_levels: is_candidate = False @@ -383,6 +392,12 @@ class TargetGenerator: if not candidates: return [] + if max_targets is None: + candidates.sort(key=lambda row: row["distance_from_entry"]) + for target in candidates: + target.pop("quality", None) + return candidates + # Select up to 5 targets that SPAN the distance range, instead of the # top-5 by quality (which biases toward far, high-R:R levels and buries # every nearby target). Guarantees the nearest level plus a @@ -396,6 +411,8 @@ class TargetGenerator: selected_ids: set[int] = set() def _add(candidate: dict[str, Any] | None) -> None: + if len(selected) >= max_targets: + return if candidate is not None and candidate["sr_level_id"] not in selected_ids: selected.append(candidate) selected_ids.add(candidate["sr_level_id"]) @@ -408,7 +425,7 @@ class TargetGenerator: _add(max(bucket, key=lambda c: c["quality"])) # Fill remaining slots with the next-best by quality for candidate in sorted(candidates, key=lambda c: c["quality"], reverse=True): - if len(selected) >= 5: + if len(selected) >= max_targets: break _add(candidate) diff --git a/app/services/sr_service.py b/app/services/sr_service.py index b6ac7c7..b1090fc 100644 --- a/app/services/sr_service.py +++ b/app/services/sr_service.py @@ -10,6 +10,7 @@ and persists to DB. from __future__ import annotations import math +from dataclasses import dataclass from datetime import datetime from sqlalchemy import delete, select @@ -32,6 +33,40 @@ from app.services.price_service import query_ohlcv DEFAULT_TOLERANCE = 0.005 # fallback when ATR unavailable; also API legacy default + +@dataclass(frozen=True) +class GateTargetLadderConfig: + """Research controls for the transient Gate Target Ladder. + + Live callers omit this object and retain the frozen legacy-parity path. + The backtest may pass an explicit configuration to isolate one ladder + mechanism at a time without changing Structural S/R. + """ + + lookback_bars: int | None = None + grid_bins: int = 20 + include_pivots: bool = True + pivot_window: int = 2 + touch_tolerance: float = DEFAULT_TOLERANCE + merge_tolerance: float = DEFAULT_TOLERANCE + strength_scale: float = 500.0 + + def __post_init__(self) -> None: + if self.lookback_bars is not None and self.lookback_bars < 20: + raise ValueError("GTL lookback_bars must be at least 20 or None") + if self.grid_bins < 2: + raise ValueError("GTL grid_bins must be at least 2") + if self.pivot_window < 1: + raise ValueError("GTL pivot_window must be at least 1") + for name, value in ( + ("touch_tolerance", self.touch_tolerance), + ("merge_tolerance", self.merge_tolerance), + ): + if not 0.0 <= value < 0.10: + raise ValueError(f"GTL {name} must be in [0, 0.10)") + if self.strength_scale <= 0: + raise ValueError("GTL strength_scale must be positive") + VP_LOOKBACK = 252 TOUCH_LOOKBACK = 252 PIVOT_LOOKBACK = 504 @@ -527,6 +562,8 @@ def detect_gate_target_ladder( lows: list[float], closes: list[float], tolerance: float = DEFAULT_TOLERANCE, + *, + config: GateTargetLadderConfig | None = None, ) -> list[dict]: """Build the scanner's internal, volume-free target proposal ladder. @@ -536,14 +573,115 @@ def detect_gate_target_ladder( profile calculation. The returned levels are transient and must not be persisted as chart S/R. """ - return detect_sr_levels_legacy( - highs, - lows, - closes, - [0] * len(closes), - tolerance, - explicit_range_grid=True, - ) + if config is None: + # Frozen live/default path. Keeping the established helper here makes + # the absence of research configuration an exact compatibility promise. + return detect_sr_levels_legacy( + highs, + lows, + closes, + [0] * len(closes), + tolerance, + explicit_range_grid=True, + ) + + if tolerance != DEFAULT_TOLERANCE: + raise ValueError("Pass either GTL config tolerances or tolerance, not both") + if not closes: + return [] + if not (len(highs) == len(lows) == len(closes)): + raise ValueError("GTL highs, lows, and closes must have equal lengths") + + if config.lookback_bars is not None: + highs = highs[-config.lookback_bars:] + lows = lows[-config.lookback_bars:] + closes = closes[-config.lookback_bars:] + + candidates: list[tuple[float, str]] = [] + try: + candidates.extend( + (float(price), "range_grid") + for price in _legacy_range_grid_nodes( + highs, + lows, + closes, + num_bins=config.grid_bins, + ) + ) + except ValidationError: + pass + + if config.include_pivots: + try: + pivots = compute_pivot_points( + highs, + lows, + closes, + window=config.pivot_window, + ) + candidates.extend( + (float(price), "pivot_point") + for price in pivots.get("swing_highs", []) + pivots.get("swing_lows", []) + ) + except ValidationError: + pass + if not candidates: + return [] + + total_bars = len(closes) + raw: list[dict] = [] + for price, method in candidates: + touch_band = abs(price) * config.touch_tolerance if price else config.touch_tolerance + touches = sum( + 1 + for low, high in zip(lows, highs, strict=False) + if low - touch_band <= price <= high + touch_band + ) + strength = max( + 0, + min(100, int(round((touches / total_bars) * config.strength_scale))), + ) + raw.append({ + "price_level": price, + "strength": strength, + "detection_method": method, + "type": "", + "sources": [method], + "rejection_count": touches, + "last_rejection_age": None, + "weighted_respects": float(touches), + }) + + merged: list[dict] = [] + for level in sorted(raw, key=lambda row: row["price_level"]): + if not merged: + merged.append(dict(level)) + continue + last = merged[-1] + ref = last["price_level"] + merge_band = abs(ref) * config.merge_tolerance if ref else config.merge_tolerance + if abs(level["price_level"] - ref) > merge_band: + merged.append(dict(level)) + continue + last["price_level"] = round( + (last["price_level"] + level["price_level"]) / 2.0, + 4, + ) + last["strength"] = min(100, last["strength"] + level["strength"]) + sources = set(last.get("sources") or [last["detection_method"]]) + sources |= set(level.get("sources") or [level["detection_method"]]) + last["sources"] = sorted(sources) + last["detection_method"] = ( + next(iter(sources)) if len(sources) == 1 else "merged" + ) + last["rejection_count"] = max( + int(last.get("rejection_count", 0)), + int(level.get("rejection_count", 0)), + ) + + _tag_levels(merged, closes[-1]) + merged.sort(key=lambda row: row["strength"], reverse=True) + return merged def _merge_levels( diff --git a/docs/research/sr-levels-and-exits.md b/docs/research/sr-levels-and-exits.md index b609b55..32a5e48 100644 --- a/docs/research/sr-levels-and-exits.md +++ b/docs/research/sr-levels-and-exits.md @@ -804,6 +804,47 @@ statistics, and portfolio results are unchanged. This closes the local backtest gate for the dual-purpose implementation. It does not itself authorize 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. + +The single-command matrix contains 20 full-period arms. Each non-control arm +changes exactly one input: + +| Knob | Frozen control | Alternatives | Question isolated | +|---|---:|---|---| +| History | All available bars | 252 / 504 / 756 bars | Is recent or long-cycle range geometry useful? | +| Candidate cap | 5 | 8 / unlimited | Does early pruning discard the useful headline? | +| Max target distance | Existing volatility-dependent rule | 5.5 / 8 ATR universally | Is the medium-volatility unlimited branch the hidden edge? | +| Traffic touch padding | 0.5% | 0 / 0.25% | Does padded price traffic carry information? | +| Proposal merge | 0.5% | 0.25% / 1% | Is proposal density or consolidation important? | +| Target zones | 2% | 1% / 3% | Does the reachable near edge manufacture the gate geometry? | +| Range centers | 20 | 12 / 32 | Is coarse ladder density the useful feature? | +| 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/ +removed cohorts against control. The consolidated JSON and Markdown table are +checkpointed after every arm; successful runs delete temporary per-arm reports +unless `--keep-arm-reports` is set. + +The pre-registered screen requires all of the following versus control: +full/train/test Sharpe not worse, full-period drawdown not worse, at least 80% +of production trades retained, and positive qualified expectancy after removing +the top 5%. Passing identifies a candidate for forward paper validation, not an +automatic deployment. The post-2024 interval has already influenced this +research, so the split is a robustness check rather than a pristine holdout. + 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 655170c..0fecf75 100644 --- a/reports/README.md +++ b/reports/README.md @@ -60,3 +60,12 @@ book (Sharpe 2.03, CAGR 50.0%, max drawdown 21.4%, 321 trades). The final rerun 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: + +- `backtest-YYYYMMDD-gtl-tuning-matrix.json` +- `backtest-YYYYMMDD-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. diff --git a/scripts/run_backtest_snapshot.py b/scripts/run_backtest_snapshot.py index f87bb67..21330e3 100644 --- a/scripts/run_backtest_snapshot.py +++ b/scripts/run_backtest_snapshot.py @@ -11,7 +11,7 @@ import asyncio import json import os import sys -from datetime import datetime +from datetime import date, datetime from pathlib import Path from typing import Any @@ -61,10 +61,16 @@ def _parse_args() -> argparse.Namespace: "rewrite_range504_structural_primary2", "production_structural_overlay", "explicit_target_ladder", + "gtl_tuning", ), 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( "--entry-start", default=None, @@ -80,6 +86,11 @@ def _parse_args() -> argparse.Namespace: action="store_true", help="Include candidate-level S/R audit rows for paired comparison.", ) + parser.add_argument( + "--holdout-split", + default=None, + help="Add a disjoint train/test portfolio report split at YYYY-MM-DD.", + ) return parser.parse_args() @@ -195,12 +206,22 @@ 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.entry_start: os.environ["BACKTEST_ENTRY_START"] = args.entry_start if args.entry_end: os.environ["BACKTEST_ENTRY_END"] = args.entry_end if args.sr_audit: os.environ["BACKTEST_SR_AUDIT"] = "1" + if args.holdout_split: + try: + date.fromisoformat(args.holdout_split) + except ValueError as exc: + raise SystemExit("--holdout-split must use YYYY-MM-DD") from exc + os.environ["BACKTEST_HOLDOUT_SPLIT"] = args.holdout_split from app.config import settings from app.services.backtest_service import run_backtest diff --git a/scripts/run_gtl_tuning_matrix.py b/scripts/run_gtl_tuning_matrix.py new file mode 100644 index 0000000..b838d33 --- /dev/null +++ b/scripts/run_gtl_tuning_matrix.py @@ -0,0 +1,418 @@ +"""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 8047788..e35172b 100644 --- a/tests/unit/test_backtest_service.py +++ b/tests/unit/test_backtest_service.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json import math from datetime import date, timedelta from types import SimpleNamespace @@ -803,6 +804,7 @@ def test_sr_research_variant_is_explicit_and_validated(monkeypatch): "rewrite_range504_structural_primary2", "production_structural_overlay", "explicit_target_ladder", + "gtl_tuning", ): monkeypatch.setenv("BACKTEST_SR_VARIANT", variant) assert bt._sr_research_variant() == variant @@ -861,6 +863,24 @@ 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 + + +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_structural_overlay_tags_production_geometry_without_replacing_it(monkeypatch): @@ -972,6 +992,45 @@ 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 + + @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 3f8676a..e3d14bc 100644 --- a/tests/unit/test_detect_sr_levels.py +++ b/tests/unit/test_detect_sr_levels.py @@ -2,7 +2,10 @@ from __future__ import annotations +import pytest + from app.services.sr_service import ( + GateTargetLadderConfig, MAX_LEVELS, _bar_respect_weight, _cap_levels, @@ -314,6 +317,70 @@ 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_tuning_matrix.py b/tests/unit/test_gtl_tuning_matrix.py new file mode 100644 index 0000000..df20c3c --- /dev/null +++ b/tests/unit/test_gtl_tuning_matrix.py @@ -0,0 +1,50 @@ +"""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 e2c41b3..d9f789f 100644 --- a/tests/unit/test_recommendation_service.py +++ b/tests/unit/test_recommendation_service.py @@ -252,6 +252,52 @@ 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 = { @@ -331,6 +377,23 @@ 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