fix(backtest): enforce timeframe range limits

backend_api_python/app/routes/backtest_center.py

@@ -18,6 +18,7 @@ from app.services.backtest_execution import (
     default_slippage_if_missing,
     parse_rate,
 )
+from app.services.backtest_limits import BacktestRangeLimitError
 from app.services.script_source import get_script_source_service
 from app.services.strategy_v2 import (
     FactorResearchRepository,
@@ -113,6 +114,8 @@ def run_strategy_backtest():
         payload = request.get_json(silent=True) or {}
         run_id, result = _run(payload, int(g.user_id), persist=bool(payload.get("persist", True)))
         return jsonify({"code": 1, "msg": "success", "data": {**result, "runId": run_id}})
+    except BacktestRangeLimitError as exc:
+        return jsonify({"code": 0, "msg": str(exc), "data": exc.details}), 400
     except ValueError as exc:
         return jsonify({"code": 0, "msg": str(exc), "data": None}), 400
     except Exception as exc:
@@ -170,6 +173,8 @@ def run_factor_research():
             code=code,
         )
         return jsonify({"code": 1, "msg": "success", "data": {**result, "runId": run_id}})
+    except BacktestRangeLimitError as exc:
+        return jsonify({"code": 0, "msg": str(exc), "data": exc.details}), 400
     except ValueError as exc:
         return jsonify({"code": 0, "msg": str(exc), "data": None}), 400
     except Exception as exc:

backend_api_python/app/routes/script_source_routes.py

@@ -3,6 +3,7 @@
 from flask import g, jsonify, request
 
 from app.routes.strategy_blueprint import strategy_blp
+from app.services.backtest_limits import backtest_range_policy_metadata
 from app.services.script_source import get_script_source_service
 from app.services.strategy_v2 import (
     StrategyBacktestRepository,
@@ -158,8 +159,27 @@ def compile_script_source_v2():
             code = str((source or {}).get("code") or "").strip()
         if not code:
             return jsonify({"code": 0, "msg": "strategyV2.codeRequired", "data": None}), 400
-        manifest = compile_strategy_v2(code).manifest.metadata()
-        return jsonify({"code": 1, "msg": "success", "data": {"manifest": manifest}})
+        compiled_manifest = compile_strategy_v2(code).manifest
+        manifest = compiled_manifest.metadata()
+        range_policy = backtest_range_policy_metadata(
+            markets=compiled_manifest.markets,
+            timeframe=compiled_manifest.primary_frequency,
+            warmup_bars=compiled_manifest.warmup_bars,
+        )
+        factor_range_policy = backtest_range_policy_metadata(
+            markets=compiled_manifest.markets,
+            timeframe=compiled_manifest.primary_frequency,
+            warmup_bars=max(40, compiled_manifest.warmup_bars),
+        )
+        range_policy.update({
+            "factorMaxSelectedDays": factor_range_policy["maxSelectedDays"],
+            "factorWarmupBars": factor_range_policy["warmupBars"],
+        })
+        return jsonify({
+            "code": 1,
+            "msg": "success",
+            "data": {"manifest": manifest, "backtestRangePolicy": range_policy},
+        })
     except ValueError as exc:
         return jsonify({"code": 0, "msg": str(exc), "data": None}), 400
     except Exception as exc:

backend_api_python/app/services/backtest_limits.py

@@ -4,7 +4,8 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from datetime import datetime, timedelta
-from typing import Any, Dict, Optional
+import math
+from typing import Any, Dict, Iterable, Optional
 
 from app.data_sources.factory import DataSourceFactory
 
@@ -21,6 +22,16 @@ _TIMEFRAME_SECONDS = {
     "1W": 604800,
 }
 
+_TIMEFRAME_ALIASES = {key.lower(): key for key in _TIMEFRAME_SECONDS}
+
+
+class BacktestRangeLimitError(ValueError):
+    """Structured range rejection that routes can return to API clients."""
+
+    def __init__(self, details: Dict[str, Any]) -> None:
+        super().__init__("strategyV2.backtestRangeLimit")
+        self.details = details
+
 
 @dataclass(frozen=True)
 class BacktestRangePolicy:
@@ -74,9 +85,14 @@ _MARKET_LIMITS: Dict[str, Dict[str, BacktestRangePolicy]] = {
 }
 
 
+def normalize_backtest_timeframe(timeframe: str) -> str:
+    raw = str(timeframe or "1D").strip()
+    return _TIMEFRAME_ALIASES.get(raw.lower(), raw)
+
+
 def backtest_range_policy(market: str, timeframe: str) -> BacktestRangePolicy:
     normalized_market = DataSourceFactory.normalize_market(market or "")
-    tf = str(timeframe or "1D").strip()
+    tf = normalize_backtest_timeframe(timeframe)
     return (
         _MARKET_LIMITS.get(normalized_market, {}).get(tf)
         or _DEFAULT_LIMITS.get(tf)
@@ -84,6 +100,56 @@ def backtest_range_policy(market: str, timeframe: str) -> BacktestRangePolicy:
     )
 
 
+def backtest_warmup_calendar_days(timeframe: str, warmup_bars: int) -> int:
+    bars = max(0, int(warmup_bars or 0))
+    if bars == 0:
+        return 0
+    normalized = normalize_backtest_timeframe(timeframe).lower()
+    if normalized.endswith("m") and normalized[:-1].isdigit():
+        minutes = max(1, int(normalized[:-1]))
+        return max(1, math.ceil(bars * minutes * 1.5 / 1440.0))
+    if normalized.endswith("h") and normalized[:-1].isdigit():
+        hours = max(1, int(normalized[:-1]))
+        return max(1, math.ceil(bars * hours * 1.5 / 24.0))
+    if normalized.endswith("d"):
+        return max(2, math.ceil(bars * 7.0 / 5.0 * 1.35))
+    if normalized.endswith("w"):
+        return max(8, bars * 8)
+    return max(1, math.ceil(bars * 1.5))
+
+
+def backtest_range_policy_metadata(
+    *,
+    markets: Iterable[str],
+    timeframe: str,
+    warmup_bars: int = 0,
+) -> Dict[str, Any]:
+    """Return the strictest client-facing policy for a compiled strategy."""
+    normalized_markets = list(dict.fromkeys(
+        DataSourceFactory.normalize_market(market or "")
+        for market in markets
+    )) or [""]
+    policies = [
+        (market, backtest_range_policy(market, timeframe))
+        for market in normalized_markets
+    ]
+    market, policy = min(policies, key=lambda item: item[1].max_days)
+    normalized_timeframe = normalize_backtest_timeframe(timeframe)
+    timeframe_seconds = _TIMEFRAME_SECONDS.get(normalized_timeframe, 86400)
+    normalized_warmup_bars = max(0, int(warmup_bars or 0))
+    warmup_days = backtest_warmup_calendar_days(normalized_timeframe, normalized_warmup_bars)
+    return {
+        "timeframe": normalized_timeframe,
+        "market": market,
+        "maxDays": policy.max_days,
+        "maxSelectedDays": max(0, policy.max_days - warmup_days),
+        "warmupBars": normalized_warmup_bars,
+        "warmupDays": warmup_days,
+        "timeframeSeconds": timeframe_seconds,
+        "maxBars": max(1, (policy.max_days * 86400) // timeframe_seconds),
+    }
+
+
 def _date_limit_start(end_date: datetime, max_days: int, warmup_seconds: int) -> datetime:
     """Return a date-only friendly start that keeps the fetch window under max_days."""
     return end_date - timedelta(days=max(0, int(max_days) - 1)) + timedelta(seconds=warmup_seconds)
@@ -102,12 +168,17 @@ def validate_backtest_range(
     start_date: datetime,
     end_date: datetime,
     warmup_bars: int = 0,
+    fetch_start: Optional[datetime] = None,
 ) -> Optional[Dict[str, Any]]:
     """Return a structured range error, or None when the request is allowed."""
     policy = backtest_range_policy(market, timeframe)
-    tf_seconds = _TIMEFRAME_SECONDS.get(str(timeframe or "1D").strip(), 86400)
+    normalized_timeframe = normalize_backtest_timeframe(timeframe)
+    tf_seconds = _TIMEFRAME_SECONDS.get(normalized_timeframe, 86400)
     warmup_seconds = max(0, int(warmup_bars or 0)) * tf_seconds
-    fetch_start = start_date - timedelta(seconds=warmup_seconds)
+    if fetch_start is None:
+        fetch_start = start_date - timedelta(seconds=warmup_seconds)
+    else:
+        warmup_seconds = max(0, int((start_date - fetch_start).total_seconds()))
     selected_days = max(0, (end_date - start_date).days)
     fetch_days = max(0, (end_date - fetch_start).days)
     if fetch_days <= policy.max_days:

backend_api_python/app/services/strategy_v2/service.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import hashlib
-import math
 import os
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timedelta
@@ -11,6 +10,11 @@ from typing import Any, Callable
 
 import pandas as pd
 
+from app.services.backtest_limits import (
+    BacktestRangeLimitError,
+    backtest_warmup_calendar_days,
+    validate_backtest_range,
+)
 from app.services.fundamental_data import get_fundamental_data_service
 from app.services.universe import UniverseService, get_universe_service
 
@@ -76,7 +80,16 @@ class StrategyV2BacktestService:
                 f"strategyV2.factorResearchUniverseTooSmall:{minimum_symbols}"
             )
         frequency = manifest.primary_frequency
-        fetch_start = start_date - timedelta(days=_warmup_calendar_days(frequency, max(40, manifest.warmup_bars)))
+        warmup_bars = max(40, manifest.warmup_bars)
+        fetch_start = start_date - timedelta(days=backtest_warmup_calendar_days(frequency, warmup_bars))
+        _enforce_backtest_range(
+            candidates=candidates,
+            timeframe=frequency,
+            start_date=start_date,
+            end_date=end_date,
+            warmup_bars=warmup_bars,
+            fetch_start=fetch_start,
+        )
         frames, skipped = self.fetch_frames(candidates, frequency, fetch_start, end_date)
         if not frames:
             raise StrategyV2ContractError("strategyV2.noMarketData")
@@ -143,7 +156,15 @@ class StrategyV2BacktestService:
 
         frequency = manifest.primary_frequency
         fetch_start = start_date - timedelta(
-            days=_warmup_calendar_days(frequency, manifest.warmup_bars)
+            days=backtest_warmup_calendar_days(frequency, manifest.warmup_bars)
+        )
+        _enforce_backtest_range(
+            candidates=candidates,
+            timeframe=frequency,
+            start_date=start_date,
+            end_date=end_date,
+            warmup_bars=manifest.warmup_bars,
+            fetch_start=fetch_start,
         )
         frames, skipped = self.fetch_frames(candidates, frequency, fetch_start, end_date)
         if not frames:
@@ -350,6 +371,37 @@ class StrategyV2BacktestService:
             raise StrategyV2ContractError(f"strategyV2.fundamentalDataMissing:{','.join(missing)}")
 
 
+def _enforce_backtest_range(
+    *,
+    candidates: list[dict[str, Any]],
+    timeframe: str,
+    start_date: datetime,
+    end_date: datetime,
+    warmup_bars: int,
+    fetch_start: datetime,
+) -> None:
+    errors: list[dict[str, Any]] = []
+    checked_markets: set[str] = set()
+    for candidate in candidates:
+        market = str(candidate.get("market") or "")
+        if market in checked_markets:
+            continue
+        checked_markets.add(market)
+        error = validate_backtest_range(
+            market=market,
+            symbol=str(candidate.get("symbol") or ""),
+            timeframe=timeframe,
+            start_date=start_date,
+            end_date=end_date,
+            warmup_bars=warmup_bars,
+            fetch_start=fetch_start,
+        )
+        if error:
+            errors.append(error)
+    if errors:
+        raise BacktestRangeLimitError(min(errors, key=lambda item: int(item["max_days"])))
+
+
 def _instrument_member(item: InstrumentSpec) -> dict[str, Any]:
     return {
         "key": item.key,
@@ -361,24 +413,6 @@ def _instrument_member(item: InstrumentSpec) -> dict[str, Any]:
     }
 
 
-def _warmup_calendar_days(frequency: str, warmup_bars: int) -> int:
-    bars = max(0, int(warmup_bars or 0))
-    if bars == 0:
-        return 0
-    normalized = str(frequency or "1d").strip().lower()
-    if normalized.endswith("m") and normalized[:-1].isdigit():
-        minutes = max(1, int(normalized[:-1]))
-        return max(1, math.ceil(bars * minutes * 1.5 / 1440.0))
-    if normalized.endswith("h") and normalized[:-1].isdigit():
-        hours = max(1, int(normalized[:-1]))
-        return max(1, math.ceil(bars * hours * 1.5 / 24.0))
-    if normalized.endswith("d"):
-        return max(2, math.ceil(bars * 7.0 / 5.0 * 1.35))
-    if normalized.endswith("w"):
-        return max(8, bars * 8)
-    return max(1, math.ceil(bars * 1.5))
-
-
 def _benchmark_for_manifest(manifest: StrategyManifest) -> InstrumentSpec | None:
     if manifest.benchmark is not None:
         return manifest.benchmark