Isolate legacy range-expansion factor

app/services/backtest_service.py

@@ -92,6 +92,12 @@ STEP_DAYS = 5          # weekly cadence (≈ 5 trading days)
 MIN_LOOKBACK = 60      # bars needed before D for indicators (EMA cross needs 51)
 HORIZON = 30           # trading days to resolve an outcome (matches the evaluator)
 ATR_MULTIPLIER = 1.5
+RANGE_FACTOR_LOOKBACK = 504
+RANGE_FACTOR_MIN_LOG = 1.0  # approximately a 2.7x high/low span
+RANGE_FACTOR_VARIANTS = {
+    "production_range504",
+    "rewrite_range504_legacy_primary",
+}
 
 # Cross-sectional signal evaluation (factor IC). Each candidate signal is a
 # point-in-time number computed from closes alone (sentiment/fundamentals have no
@@ -146,6 +152,7 @@ SR_RESEARCH_VARIANTS = {
     "legacy_traffic_grid_only",
     "legacy_range_grid_touch",
     "legacy_range_grid_neutral",
+    *RANGE_FACTOR_VARIANTS,
 }
 
 
@@ -158,6 +165,36 @@ def _sr_research_variant() -> str:
     return value
 
 
+def _sr_detector_variant(sr_variant: str) -> str:
+    """Map factor-gated research arms to the detector they hold fixed."""
+    if sr_variant == "production_range504":
+        return "production_control"
+    if sr_variant == "rewrite_range504_legacy_primary":
+        return "rewrite"
+    return sr_variant.removesuffix("_legacy_primary")
+
+
+def _range_504_log(highs: list[float], lows: list[float]) -> float:
+    """Multiplicative high/low range over the last two trading years."""
+    window_highs = highs[-RANGE_FACTOR_LOOKBACK:]
+    window_lows = lows[-RANGE_FACTOR_LOOKBACK:]
+    if not window_highs or not window_lows:
+        return 0.0
+    high = max(window_highs)
+    low = min(window_lows)
+    if high <= 0 or low <= 0 or high < low:
+        return 0.0
+    return math.log(high / low)
+
+
+def _range_factor_allows(sr_variant: str, range_504_log: float) -> bool:
+    """Apply the explicit range factor only in its diagnostic arms."""
+    return (
+        sr_variant not in RANGE_FACTOR_VARIANTS
+        or range_504_log >= RANGE_FACTOR_MIN_LOG
+    )
+
+
 def _apply_zone_strength_variant(zone_levels: list[Any], sr_variant: str) -> list[Any]:
     """Apply post-cluster research controls without changing zone geometry."""
     if sr_variant in {"legacy_geometry_neutral", "legacy_range_grid_neutral"}:
@@ -288,7 +325,8 @@ def _window_setups(
         return []
 
     sr_variant = _sr_research_variant()
-    detector_variant = sr_variant.removesuffix("_legacy_primary")
+    detector_variant = _sr_detector_variant(sr_variant)
+    range_504_log = _range_504_log(highs, lows)
     if sr_variant == "legacy_geometry_neutral":
         detected_levels = detect_sr_levels_legacy(
             highs, lows, closes, volumes, neutral_strength=True
@@ -370,7 +408,7 @@ def _window_setups(
         targets = _prune_floor_pinned_targets(targets)
         primary_min_rr = (
             1.5
-            if sr_variant == "production_control"
+            if sr_variant in {"production_control", "production_range504"}
             or sr_variant.endswith("_legacy_primary")
             or sr_variant.startswith("legacy_")
             else float(activation.get("min_rr", 0.0))
@@ -419,6 +457,10 @@ def _window_setups(
         # week are known. run_backtest ranks momentum and finalizes `qualified`.
         core_config = {**activation, "min_momentum_percentile": 0.0}
         meets_core = setup_qualifies(setup_ns, core_config)
+        meets_core = meets_core and _range_factor_allows(
+            sr_variant,
+            range_504_log,
+        )
         best_prob = best_target_probability(setup_ns)
         out.append({
             "direction": direction,
@@ -445,6 +487,9 @@ def _window_setups(
             ),
             "raw_level_count": len(sr_levels),
             "gate_level_count": len(gate_levels),
+            "range_504_log": range_504_log,
+            "range_504_ratio": math.exp(range_504_log),
+            "range_factor_pass": range_504_log >= RANGE_FACTOR_MIN_LOG,
         })
     return out
 
@@ -608,6 +653,9 @@ def _replay_ticker(
                 "primary_distance_atr": s["primary_distance_atr"],
                 "raw_level_count": s["raw_level_count"],
                 "gate_level_count": s["gate_level_count"],
+                "range_504_log": s["range_504_log"],
+                "range_504_ratio": s["range_504_ratio"],
+                "range_factor_pass": s["range_factor_pass"],
                 "outcome": outcome,
                 "target_hit": target_hit,
                 "realized_r": realized_r,
@@ -677,6 +725,7 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
     rejections: list[int] = []
     raw_counts: list[int] = []
     gate_counts: list[int] = []
+    range_logs: list[float] = []
     for cand in candidates:
         sources = list(cand.get("primary_sources") or [])
         for source in sources:
@@ -688,6 +737,7 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
         rejections.append(int(cand.get("primary_rejection_count", 0) or 0))
         raw_counts.append(int(cand.get("raw_level_count", 0) or 0))
         gate_counts.append(int(cand.get("gate_level_count", 0) or 0))
+        range_logs.append(float(cand.get("range_504_log", 0.0) or 0.0))
 
     def avg(values: list[float] | list[int]) -> float | None:
         return round(sum(values) / len(values), 3) if values else None
@@ -703,6 +753,10 @@ def _sr_variant_diagnostics(candidates: list[dict]) -> dict:
         "avg_primary_rejection_count": avg(rejections),
         "avg_raw_level_count": avg(raw_counts),
         "avg_gate_level_count": avg(gate_counts),
+        "avg_range_504_log": avg(range_logs),
+        "range_factor_pass": sum(
+            1 for value in range_logs if value >= RANGE_FACTOR_MIN_LOG
+        ),
     }
 
 
@@ -738,6 +792,9 @@ def _sr_candidate_audit(candidates: list[dict], min_percentile: float) -> list[d
             "primary_distance_atr": round(float(cand.get("primary_distance_atr", 0.0)), 6),
             "raw_level_count": int(cand.get("raw_level_count", 0) or 0),
             "gate_level_count": int(cand.get("gate_level_count", 0) or 0),
+            "range_504_log": round(float(cand.get("range_504_log", 0.0)), 6),
+            "range_504_ratio": round(float(cand.get("range_504_ratio", 1.0)), 6),
+            "range_factor_pass": bool(cand.get("range_factor_pass")),
             "outcome": cand.get("outcome"),
             "net_r": round(float(cand.get("realized_r", 0.0)) - _cost_r(cand), 6),
             "hold30_r": round(float((cand.get("time_r") or {}).get(30, 0.0)), 6),
@@ -3089,6 +3146,9 @@ async def run_backtest(
             "min_lookback": MIN_LOOKBACK,
             "cost_per_side_pct": round(COST_PER_SIDE * 100, 3),
             "sr_variant": _sr_research_variant(),
+            "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),
             "entry_start": (
                 _backtest_entry_bounds()[0].isoformat()
                 if _backtest_entry_bounds()[0] is not None else None