feat: add selectable daily backtest cadence

app/routers/admin.py

@@ -387,6 +387,7 @@ async def trigger_job(
         db,
         job_name,
         target_model=body.target_model if body is not None else None,
+        cadence=body.cadence if body is not None else None,
     )
     return APIEnvelope(status="success", data=result)
 

app/routers/trades.py

@@ -36,6 +36,7 @@ async def list_trade_setups(
         recommended_action=recommended_action,
         live_recommendation=True,
         exclude_open_trade_tickers=True,
+        exclude_reentry_lockdown_tickers=True,
     )
 
     data = []
@@ -98,6 +99,7 @@ async def get_ticker_trade_setups(
         db,
         symbol=symbol,
         live_recommendation=True,
+        include_reentry_lockdown=True,
     )
     data = []
     for row in rows:

app/scheduler.py

@@ -37,8 +37,10 @@ from app.services import fundamental_service, ingestion_service, sentiment_servi
 from app.services.alert_service import dispatch_alerts
 from app.services.backtest_service import (
     BACKTEST_TARGET_MODELS,
+    DEFAULT_BACKTEST_CADENCE,
     PRODUCTION_GTL_TARGET_MODEL,
     run_and_store as run_backtest_and_store,
+    validate_backtest_cadence,
     validate_backtest_target_model,
 )
 from app.services.benchmark_service import refresh_benchmark_prices
@@ -112,6 +114,7 @@ def _idle_runtime() -> dict[str, object]:
 
 _job_runtime: dict[str, dict[str, object]] = {name: _idle_runtime() for name in _JOB_NAMES}
 _next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
+_next_backtest_cadence = DEFAULT_BACKTEST_CADENCE
 
 
 # ---------------------------------------------------------------------------
@@ -119,23 +122,44 @@ _next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
 # ---------------------------------------------------------------------------
 
 
-def queue_backtest_target_model(target_model: str | None) -> str:
+def queue_backtest_options(
-    """Select the model for the next manual backtest run only.
+    target_model: str | None,
+    cadence: str | None,
+) -> tuple[str, str]:
+    """Select model and cadence for the next manual backtest run only.
 
-    Scheduled runs and subsequent manual runs return to the production GTL.
+    Scheduled and subsequent manual runs return to production GTL at the
+    resource-safe weekly cadence.
     """
-    global _next_backtest_target_model
+    global _next_backtest_target_model, _next_backtest_cadence
-    selected = validate_backtest_target_model(
+    selected_model = validate_backtest_target_model(
         target_model or PRODUCTION_GTL_TARGET_MODEL
     )
-    _next_backtest_target_model = selected
+    selected_cadence = validate_backtest_cadence(
+        cadence or DEFAULT_BACKTEST_CADENCE
+    )
+    _next_backtest_target_model = selected_model
+    _next_backtest_cadence = selected_cadence
+    return selected_model, selected_cadence
+
+
+def queue_backtest_target_model(target_model: str | None) -> str:
+    """Compatibility wrapper for callers selecting only the target model."""
+    selected, _ = queue_backtest_options(target_model, DEFAULT_BACKTEST_CADENCE)
+    return selected
+
+
+def _consume_backtest_options() -> tuple[str, str]:
+    global _next_backtest_target_model, _next_backtest_cadence
+    selected = (_next_backtest_target_model, _next_backtest_cadence)
+    _next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
+    _next_backtest_cadence = DEFAULT_BACKTEST_CADENCE
     return selected
 
 
 def _consume_backtest_target_model() -> str:
-    global _next_backtest_target_model
+    """Compatibility wrapper consuming all queued one-run options."""
-    selected = _next_backtest_target_model
+    selected, _ = _consume_backtest_options()
-    _next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
     return selected
 
 
@@ -1028,12 +1052,13 @@ async def compute_regime_monitor() -> None:
 async def run_backtest_job() -> None:
     """Replay the price-derived engine over history and cache the report."""
     job_name = "backtest"
-    target_model = _consume_backtest_target_model()
+    target_model, cadence = _consume_backtest_options()
     _log_event(
         logging.INFO,
         "job_start",
         job=job_name,
         target_model=target_model,
+        cadence=cadence,
     )
     _runtime_start(job_name)
 
@@ -1051,6 +1076,7 @@ async def run_backtest_job() -> None:
                 db,
                 _on_progress,
                 target_model=target_model,
+                cadence=cadence,
             )
 
         _runtime_finish(
@@ -1058,6 +1084,7 @@ async def run_backtest_job() -> None:
             processed=report.get("tickers", 0), total=report.get("tickers", 0),
             message=(
                 f"{BACKTEST_TARGET_MODELS[target_model]}: "
+                f"{cadence} cadence, "
                 f"{report.get('candidates', 0)} setups, "
                 f"{report.get('qualified', 0)} qualified"
             ),

app/schemas/admin.py

@@ -46,6 +46,7 @@ class JobToggle(BaseModel):
 class JobTriggerRequest(BaseModel):
     """Optional parameters for a one-time manual job run."""
     target_model: Literal["production_gtl", "structural_sr"] | None = None
+    cadence: Literal["weekly", "daily"] | None = None
 
 
 class RecommendationConfigUpdate(BaseModel):

app/schemas/trade_setup.py

@@ -59,5 +59,6 @@ class TradeSetupResponse(BaseModel):
     momentum_percentile: float | None = None
     strategy_rank: float | None = None
     volatility_percentile: float | None = None
+    reentry_lockdown_remaining_sessions: int | None = None
     context_as_of: TradeSetupContextAsOfResponse | None = None
     recommendation_summary: RecommendationSummaryResponse | None = None

app/services/admin_service.py

@@ -607,6 +607,7 @@ async def trigger_job(
     job_name: str,
     *,
     target_model: str | None = None,
+    cadence: str | None = None,
 ) -> dict[str, str]:
     """Trigger a manual job run via the scheduler.
 
@@ -616,6 +617,8 @@ async def trigger_job(
         raise ValidationError(f"Unknown job: {job_name}. Valid jobs: {', '.join(sorted(VALID_JOB_NAMES))}")
     if target_model is not None and job_name != "backtest":
         raise ValidationError("target_model is supported only for the backtest job")
+    if cadence is not None and job_name != "backtest":
+        raise ValidationError("cadence is supported only for the backtest job")
 
     from app.scheduler import get_job_runtime_snapshot, scheduler
 
@@ -643,9 +646,9 @@ async def trigger_job(
         return {"job": job_name, "status": "not_found", "message": f"Job '{job_name}' is not registered in the scheduler"}
 
     if job_name == "backtest":
-        from app.scheduler import queue_backtest_target_model
+        from app.scheduler import queue_backtest_options
 
-        target_model = queue_backtest_target_model(target_model)
+        target_model, cadence = queue_backtest_options(target_model, cadence)
 
     job.modify(next_run_time=None)  # Reset, then trigger immediately
     from datetime import datetime, timezone
@@ -654,6 +657,8 @@ async def trigger_job(
     result = {"job": job_name, "status": "triggered", "message": f"Job '{job_name}' triggered for immediate execution"}
     if target_model is not None:
         result["target_model"] = target_model
+    if cadence is not None:
+        result["cadence"] = cadence
     return result
 
 

app/services/alert_service.py

@@ -282,6 +282,7 @@ async def _qualified_setups(db: AsyncSession) -> list[dict]:
         db,
         live_recommendation=True,
         exclude_open_trade_tickers=True,
+        exclude_reentry_lockdown_tickers=True,
     )
     config = await get_activation_config(db)
     return [s for s in setups if setup_qualifies(SimpleNamespace(**s), config)]

app/services/backtest_service.py

@@ -1,7 +1,7 @@
 """Historical backtest (Phase 1): replay the price-derived engine over stored
 OHLCV and measure how the CURRENT config would have performed.
 
-For each ticker we step through history (weekly), and at each as-of date D we
+For each ticker we step through history at the selected entry cadence, and at each as-of date D we
 rebuild the setup using only bars ≤ D (no lookahead), then walk the actual bars
 after D to record the realized outcome. The report contains:
 
@@ -94,12 +94,22 @@ from app.services.scoring_service import (
     compute_technical_from_arrays,
 )
 from app.services.sr_service import detect_gate_target_ladder, detect_sr_levels
+from app.services.trade_policy import REENTRY_LOCKDOWN_SESSIONS
 
 logger = logging.getLogger(__name__)
 
 KEY_REPORT = "backtest_report"
 
-STEP_DAYS = 5          # weekly cadence (≈ 5 trading days)
+WEEKLY_BACKTEST_CADENCE = "weekly"
+DAILY_BACKTEST_CADENCE = "daily"
+DEFAULT_BACKTEST_CADENCE = WEEKLY_BACKTEST_CADENCE
+BACKTEST_CADENCE_SESSIONS = {
+    WEEKLY_BACKTEST_CADENCE: 5,
+    DAILY_BACKTEST_CADENCE: 1,
+}
+# Compatibility alias for research scripts built around the original weekly
+# replay. New code should select a cadence and call ``backtest_step_sessions``.
+STEP_DAYS = BACKTEST_CADENCE_SESSIONS[WEEKLY_BACKTEST_CADENCE]
 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
@@ -156,6 +166,30 @@ def validate_backtest_target_model(value: str) -> str:
     return normalized
 
 
+def validate_backtest_cadence(value: str) -> str:
+    """Validate the supported entry-replay cadences."""
+    normalized = value.strip().lower()
+    if normalized not in BACKTEST_CADENCE_SESSIONS:
+        allowed = ", ".join(BACKTEST_CADENCE_SESSIONS)
+        raise ValueError(
+            f"Unknown backtest cadence {value!r}; expected one of {allowed}"
+        )
+    return normalized
+
+
+def backtest_step_sessions(cadence: str) -> int:
+    return BACKTEST_CADENCE_SESSIONS[validate_backtest_cadence(cadence)]
+
+
+def _ranking_period(as_of: date, cadence: str) -> tuple:
+    """Cross-section key for activation ranks at the selected entry cadence."""
+    cadence = validate_backtest_cadence(cadence)
+    if cadence == DAILY_BACKTEST_CADENCE:
+        return ("date", as_of.toordinal())
+    iso = as_of.isocalendar()
+    return ("week", iso[0], iso[1])
+
+
 # ---------------------------------------------------------------------------
 # RESEARCH / DIAGNOSTIC FALLBACKS (retired experiments)
 #
@@ -455,14 +489,17 @@ def _replay_ticker(
     activation: dict,
     benchmark_closes: dict[date, float] | None = None,
     target_model: str = PRODUCTION_GTL_TARGET_MODEL,
+    cadence: str = DEFAULT_BACKTEST_CADENCE,
 ) -> list[dict]:
-    """Walk one ticker's history weekly, building setups and their realized outcomes."""
+    """Walk one ticker at the selected cadence and resolve each setup outcome."""
+    cadence = validate_backtest_cadence(cadence)
+    step_sessions = backtest_step_sessions(cadence)
     candidates: list[dict] = []
     n = len(records)
     if n < MIN_LOOKBACK + HORIZON:
         return candidates
 
-    for i in range(MIN_LOOKBACK - 1, n - HORIZON, STEP_DAYS):
+    for i in range(MIN_LOOKBACK - 1, n - HORIZON, step_sessions):
         window = records[: i + 1]
         forward = records[i + 1 :]
         forward_bars = [Bar(date=r.date, high=r.high, low=r.low) for r in forward]
@@ -511,6 +548,7 @@ def _replay_ticker(
                 "symbol": symbol,
                 "date": records[i].date.isoformat(),
                 "iso_week": (iso[0], iso[1]),
+                "ranking_period": _ranking_period(records[i].date, cadence),
                 "direction": s["direction"],
                 "entry": s["entry"],
                 "stop": s["stop"],
@@ -967,6 +1005,7 @@ def _replay_and_signals(
     activation: dict,
     benchmark_closes: dict[date, float] | None = None,
     target_model: str = PRODUCTION_GTL_TARGET_MODEL,
+    cadence: str = DEFAULT_BACKTEST_CADENCE,
 ) -> tuple[list[dict], dict]:
     """The CPU-bound per-ticker work, as a top-level (picklable) function so it can
     run in a worker process. Takes primitive column arrays (cheap to pickle),
@@ -986,11 +1025,81 @@ def _replay_and_signals(
             activation,
             benchmark_closes,
             target_model,
+            cadence,
         ),
         _signal_series(bars, benchmark_closes),
     )
 
 
+def _replay_candidates_for_period(
+    symbol: str,
+    columns: tuple,
+    config: dict,
+    activation: dict,
+    benchmark_closes: dict[date, float] | None,
+    start_date: date,
+    cadence: str = DEFAULT_BACKTEST_CADENCE,
+) -> list[dict]:
+    """Slim picklable replay used by local event studies.
+
+    Unlike the full report worker it skips factor-series construction and only
+    evaluates setup dates on or after ``start_date``.
+    """
+    date_ords, opens, highs, lows, closes, volumes = columns
+    bars = [
+        SimpleNamespace(
+            date=date.fromordinal(o), open=op, high=hi, low=lo, close=cl, volume=vo
+        )
+        for o, op, hi, lo, cl, vo in zip(
+            date_ords, opens, highs, lows, closes, volumes
+        )
+    ]
+    cadence = validate_backtest_cadence(cadence)
+    candidates: list[dict] = []
+    for i in range(
+        MIN_LOOKBACK - 1,
+        len(bars) - HORIZON,
+        backtest_step_sessions(cadence),
+    ):
+        if bars[i].date < start_date:
+            continue
+        window = bars[: i + 1]
+        window_closes = [float(r.close) for r in window]
+        window_dates = [r.date for r in window]
+        residual_momentum = _residual_momentum_12_1(
+            window_dates,
+            window_closes,
+            len(window) - 1,
+            benchmark_closes,
+        )
+        vol_6m = _realized_vol_6m(window_closes, len(window) - 1)
+        iso = bars[i].date.isocalendar()
+        for setup in _window_setups(window, config, activation):
+            if setup["direction"] != "long":
+                continue
+            candidates.append({
+                "symbol": symbol,
+                "date": bars[i].date.isoformat(),
+                "iso_week": (iso[0], iso[1]),
+                "ranking_period": _ranking_period(bars[i].date, cadence),
+                "direction": "long",
+                "entry": setup["entry"],
+                "stop": setup["stop"],
+                "target": setup["target"],
+                "rr": setup["rr"],
+                "confidence": setup["confidence"],
+                "primary_prob": setup["primary_prob"],
+                "best_prob": setup["best_prob"],
+                "momentum": setup["momentum"],
+                "residual_momentum": residual_momentum,
+                "vol_6m": vol_6m,
+                "meets_core": setup["meets_core"],
+                "action": setup["action"],
+                "risk_level": setup["risk_level"],
+            })
+    return candidates
+
+
 def _backtest_worker_count() -> int:
     """How many worker processes to replay tickers across. Capped to cpu_count-1
     so a core stays free for the web server; 1 means sequential."""
@@ -1057,14 +1166,17 @@ def _assign_signal_percentiles(
     value_key: str,
     percentile_key: str,
 ) -> None:
-    """Per ISO week, rank candidates by ``value_key`` and attach a 0-100
+    """Per replay period, rank candidates by ``value_key`` and attach a 0-100
     percentile under ``percentile_key`` (100 = strongest). Missing values get
     None and therefore cannot clear a gate based on that signal."""
-    by_week: dict = defaultdict(list)
+    by_period: dict = defaultdict(list)
     for c in candidates:
         if c.get(value_key) is not None:
-            by_week[c["iso_week"]].append(c)
+            # Hand-built/research candidates predating the cadence flag retain
-    for group in by_week.values():
+            # the weekly key as a compatibility fallback.
+            period = c.get("ranking_period") or c["iso_week"]
+            by_period[period].append(c)
+    for group in by_period.values():
         ordered = sorted(group, key=lambda c: c[value_key])
         n = len(ordered)
         for rank, c in enumerate(ordered):
@@ -1074,9 +1186,9 @@ def _assign_signal_percentiles(
 
 
 def _assign_momentum_percentiles(candidates: list[dict]) -> None:
-    """Per ISO week, rank candidates by their ticker's 12-1 momentum and attach a
+    """Per replay period, rank candidates by 12-1 momentum and attach a
     0-100 ``momentum_percentile`` (100 = highest momentum in the universe that
-    week). Candidates whose momentum is unknown (insufficient lookback) get None
+    period). Candidates whose momentum is unknown (insufficient lookback) get None
     and therefore can't clear a momentum gate. Mutates ``candidates``."""
     _assign_signal_percentiles(candidates, "momentum", "momentum_percentile")
 
@@ -1089,7 +1201,7 @@ def _assign_residual_momentum_percentiles(candidates: list[dict]) -> None:
 
 
 def _assign_low_volatility_percentiles(candidates: list[dict]) -> None:
-    """Per ISO week, attach volatility ranks where 100 = lowest 6-month vol."""
+    """Per replay period, attach volatility ranks where 100 = lowest 6-month vol."""
     _assign_signal_percentiles(candidates, "vol_6m", VOL_PERCENTILE_KEY)
     for c in candidates:
         raw = c.get(VOL_PERCENTILE_KEY)
@@ -1293,9 +1405,17 @@ def _simulate_portfolio(
     max_positions: int = SIM_MAX_POSITIONS,
     risk_per_trade: float = SIM_RISK_PER_TRADE,
     atr_trail_multiplier: float = ATR_TRAIL_MULTIPLIER,
+    reentry_cooldown_sessions: int = 0,
+    initial_stop_refresh_fn: (
+        Callable[[str, int, float, dict, Any], float | None] | None
+    ) = None,
+    post_stop_reentry_fn: (
+        Callable[[str, int, dict, Any], dict | None] | None
+    ) = None,
     start_date: date | None = None,
     end_date: date | None = None,
     include_curve: bool = False,
+    include_trades: bool = False,
 ) -> dict | None:
     """Replay the qualified setups as ONE capital-constrained book and report
     portfolio economics from the daily equity curve (return, CAGR, drawdown,
@@ -1309,7 +1429,14 @@ def _simulate_portfolio(
     runs the ATR trail *and* the S/R take-profit together — the trade ends at
     whichever comes first. Stops fill at the worse of stop or open (gaps
     modeled); positions still open at the end are closed at their last mark.
-    Returns None when there is nothing to trade.
+    ``reentry_cooldown_sessions`` blocks a ticker for that many market sessions
+    after an initial-stop loss. Profitable trailing-stop exits do not trigger
+    it. ``initial_stop_refresh_fn`` may supply a lower, point-in-time valid long
+    stop when the active initial stop is touched; the replacement is still
+    checked against the same bar. ``post_stop_reentry_fn`` turns an initial
+    stop-out into a stateful episode and is the only path by which that ticker
+    can re-enter until the callback emits a new candidate. Returns None when
+    there is nothing to trade.
     """
     if qualified_fn is None:
         def _default_qualified(c: dict) -> bool:
@@ -1362,6 +1489,14 @@ def _simulate_portfolio(
     curve: list[tuple[int, float]] = []
     trades: list[dict] = []
     skipped_full = 0
+    skipped_cooldown = 0
+    cooldown_until_index: dict[str, int] = {}
+    stop_refresh_attempts = 0
+    stop_refreshes = 0
+    stop_refresh_same_bar_hits = 0
+    post_stop_states: dict[str, dict] = {}
+    post_stop_events = 0
+    reentry_events: list[dict] = []
     technical_cache: dict[tuple[str, int], float | None] = {}
     atr_cache: dict[tuple[str, int], float | None] = {}
 
@@ -1426,7 +1561,7 @@ def _simulate_portfolio(
             atr_cache[key] = None
         return atr_cache[key]
 
-    def _close_trade(sym: str, fill: float, reason: str) -> None:
+    def _close_trade(sym: str, fill: float, reason: str) -> dict:
         nonlocal cash
         pos = positions.pop(sym)
         proceeds = pos["shares"] * fill
@@ -1434,16 +1569,29 @@ def _simulate_portfolio(
         cash += proceeds - cost
         risk = pos["entry"] - pos["initial_stop"]
         trades.append({
+            "symbol": sym,
+            "entry_ord": pos["entry_ord"],
+            "exit_ord": o,
+            "entry": pos["entry"],
+            "initial_stop": pos["initial_stop"],
+            "active_stop": pos["stop"],
+            "fill": fill,
             "pnl": proceeds - pos["shares"] * pos["entry"] - cost - pos["entry_cost"],
             "r": (fill - pos["entry"]) / risk if risk > 0 else 0.0,
             "hold": pos["bars_held"],
             "reason": reason,
+            "stop_refreshes": pos["stop_refreshes"],
+            "is_reentry": pos["is_reentry"],
+            "reentry_wait_sessions": pos["reentry_wait_sessions"],
+            "transaction_cost": pos["entry_cost"] + cost,
         })
+        return pos
 
     def _marked_equity() -> float:
         return cash + sum(p["shares"] * p["last_close"] for p in positions.values())
 
-    for o in calendar:
+    cooldown_sessions = max(0, int(reentry_cooldown_sessions))
+    for calendar_index, o in enumerate(calendar):
         # 1) exits on today's bars (stop intraday, target intraday, time at close)
         for sym in list(positions):
             pos = positions[sym]
@@ -1460,7 +1608,41 @@ def _simulate_portfolio(
                     if pos["stop"] > pos["initial_stop"] + 1e-9
                     else "stop"
                 )
-                _close_trade(sym, min(pos["stop"], bar.open), reason)
+                survived_refresh = False
+                if reason == "stop" and initial_stop_refresh_fn is not None:
+                    stop_refresh_attempts += 1
+                    refreshed_stop = initial_stop_refresh_fn(
+                        sym, o, float(pos["stop"]), pos, bar
+                    )
+                    if (
+                        refreshed_stop is not None
+                        and 0 < float(refreshed_stop) < pos["stop"] - 1e-9
+                    ):
+                        pos["stop"] = float(refreshed_stop)
+                        pos["stop_refreshes"] += 1
+                        stop_refreshes += 1
+                        if bar.low > pos["stop"]:
+                            survived_refresh = True
+                        else:
+                            stop_refresh_same_bar_hits += 1
+                if not survived_refresh:
+                    fill = min(pos["stop"], bar.open)
+                    closed_pos = _close_trade(sym, fill, reason)
+                    if reason == "stop" and cooldown_sessions:
+                        cooldown_until_index[sym] = calendar_index + cooldown_sessions
+                    if reason == "stop" and post_stop_reentry_fn is not None:
+                        post_stop_events += 1
+                        post_stop_states[sym] = {
+                            "stop_ord": o,
+                            "stop_calendar_index": calendar_index,
+                            "stop_day_high": float(bar.high),
+                            "stop_day_low": float(bar.low),
+                            "stop_day_close": float(bar.close),
+                            "exit_fill": float(fill),
+                            "previous_entry": float(closed_pos["entry"]),
+                            "previous_stop": float(closed_pos["initial_stop"]),
+                            "gate_went_unqualified": False,
+                        }
                     continue
             if exit_policy in ("target", "atr_trail3_target") and pos["target"] and bar.high >= pos["target"]:
                 _close_trade(sym, pos["target"], "target")
@@ -1493,8 +1675,29 @@ def _simulate_portfolio(
 
         # 2) entries at today's close, best momentum first
         equity = _marked_equity()
+        fixed_todays = list(entries_by_ord.get(o, ()))
+        reentry_todays: list[dict] = []
+        if post_stop_reentry_fn is not None:
+            fixed_todays = [
+                candidate
+                for candidate in fixed_todays
+                if candidate["symbol"] not in post_stop_states
+            ]
+            for sym, state in list(post_stop_states.items()):
+                bar = _bar(sym, o)
+                if bar is None:
+                    continue
+                state["sessions_since_stop"] = (
+                    calendar_index - state["stop_calendar_index"]
+                )
+                candidate = post_stop_reentry_fn(sym, o, state, bar)
+                if candidate is None:
+                    continue
+                tagged = dict(candidate)
+                tagged["_post_stop_reentry"] = True
+                reentry_todays.append(tagged)
         todays = sorted(
-            entries_by_ord.get(o, ()),
+            fixed_todays + reentry_todays,
             key=lambda c: c.get(ranking_key) or 0.0,
             reverse=True,
         )
@@ -1502,6 +1705,9 @@ def _simulate_portfolio(
             sym = c["symbol"]
             if sym in positions:
                 continue
+            if calendar_index < cooldown_until_index.get(sym, -1):
+                skipped_cooldown += 1
+                continue
             if len(positions) >= max_positions:
                 skipped_full += 1
                 continue
@@ -1518,9 +1724,23 @@ def _simulate_portfolio(
                 continue
             entry_cost = shares * entry * COST_PER_SIDE
             cash -= shares * entry + entry_cost
+            is_reentry = bool(c.get("_post_stop_reentry"))
+            reentry_wait_sessions: int | None = None
+            if is_reentry:
+                state = post_stop_states.pop(sym, None)
+                if state is not None:
+                    reentry_wait_sessions = int(state["sessions_since_stop"])
+                    reentry_events.append({
+                        "symbol": sym,
+                        "stop_ord": state["stop_ord"],
+                        "reentry_ord": o,
+                        "wait_sessions": reentry_wait_sessions,
+                        "reason": c.get("_reentry_reason"),
+                    })
             positions[sym] = {
                 "shares": shares,
                 "entry": entry,
+                "entry_ord": o,
                 "initial_stop": stop,
                 "stop": stop,
                 "target": float(c["target"]) if c.get("target") else None,
@@ -1528,6 +1748,9 @@ def _simulate_portfolio(
                 "bars_held": 0,
                 "last_close": entry,
                 "highest_close": entry,
+                "stop_refreshes": 0,
+                "is_reentry": is_reentry,
+                "reentry_wait_sessions": reentry_wait_sessions,
             }
             equity = _marked_equity()
 
@@ -1659,6 +1882,42 @@ def _simulate_portfolio(
         result["equity_curve"] = curve_payload
     if benchmark_payload is not None:
         result["benchmark_curve"] = benchmark_payload
+    if cooldown_sessions:
+        result["reentry_cooldown_sessions"] = cooldown_sessions
+        result["skipped_cooldown"] = skipped_cooldown
+    if initial_stop_refresh_fn is not None:
+        result["stop_refresh_attempts"] = stop_refresh_attempts
+        result["stop_refreshes"] = stop_refreshes
+        result["stop_refresh_same_bar_hits"] = stop_refresh_same_bar_hits
+    if post_stop_reentry_fn is not None:
+        result["post_stop_events"] = post_stop_events
+        result["post_stop_reentries"] = len(reentry_events)
+        result["post_stop_states_open_at_end"] = len(post_stop_states)
+        result["reentry_events"] = [
+            {
+                **{
+                    key: value
+                    for key, value in event.items()
+                    if key not in {"stop_ord", "reentry_ord"}
+                },
+                "stop_date": date.fromordinal(event["stop_ord"]).isoformat(),
+                "reentry_date": date.fromordinal(event["reentry_ord"]).isoformat(),
+            }
+            for event in reentry_events
+        ]
+    if include_trades:
+        result["trade_details"] = [
+            {
+                **{
+                    key: value
+                    for key, value in trade.items()
+                    if key not in {"entry_ord", "exit_ord"}
+                },
+                "entry_date": date.fromordinal(trade["entry_ord"]).isoformat(),
+                "exit_date": date.fromordinal(trade["exit_ord"]).isoformat(),
+            }
+            for trade in trades
+        ]
     return result
 
 
@@ -2018,19 +2277,35 @@ PORTFOLIO_MONITOR_STRATEGIES: tuple[dict, ...] = (
         "exit_policy": "hold",
     },
     {
-        "strategy": PRODUCTION_PORTFOLIO_STRATEGY,
+        "strategy": "production_live_no_lockdown",
-        "label": "Production: residual/high-vol 80/20 + 3x ATR trail",
+        "label": "Live setup + 3x ATR trail (no re-entry lockdown)",
         "description": (
-            "The live strategy: production activation gate and Admin exit policy "
+            "Exact live activation, ordering, and Admin exit policy, with only "
-            "as currently configured, 80/20 residual/high-vol rank."
+            "the post-stop re-entry lockdown disabled as the comparison baseline."
         ),
         "entry_variant": "residual80_highvol_blend80_20_fixed10",
         "exit_policy": "atr_trail3",
+        "reentry_lockdown_sessions": 0,
+        "use_live_config": True,
+        "comparison_arm": "live_no_lockdown",
+    },
+    {
+        "strategy": PRODUCTION_PORTFOLIO_STRATEGY,
+        "label": "Production: residual/high-vol 80/20 + 3x ATR trail + 5-session lockdown",
+        "description": (
+            "The live strategy: production activation gate and Admin exit policy "
+            "as currently configured, 80/20 residual/high-vol rank, and a "
+            "five-session re-entry lockdown after an initial-stop exit."
+        ),
+        "entry_variant": "residual80_highvol_blend80_20_fixed10",
+        "exit_policy": "atr_trail3",
+        "reentry_lockdown_sessions": REENTRY_LOCKDOWN_SESSIONS,
         # The production row replays what the platform actually does right now:
         # the live qualification flag (runtime Admin activation settings) and the
         # live Admin exit policy, instead of the frozen research-variant gate.
         "use_live_config": True,
         "is_production": True,
+        "comparison_arm": "live_lockdown_5",
     },
 )
 
@@ -2149,6 +2424,9 @@ def _min_rr_sweep(
     exit_policy = str(strategy["exit_policy"])
     row_hold_days = hold_days
     trail_multiplier = ATR_TRAIL_MULTIPLIER
+    reentry_lockdown_sessions = int(
+        strategy.get("reentry_lockdown_sessions", 0)
+    )
     if strategy.get("use_live_config") and live_exit_policy is not None:
         exit_policy = LIVE_EXIT_MODE_TO_SIM.get(
             str(live_exit_policy.get("mode", "atr_trailing")), "atr_trail3"
@@ -2188,6 +2466,7 @@ def _min_rr_sweep(
             max_positions=int(entry_cfg["max_positions"]),
             risk_per_trade=float(entry_cfg["risk_per_trade"]),
             atr_trail_multiplier=trail_multiplier,
+            reentry_cooldown_sessions=reentry_lockdown_sessions,
             start_date=sweep_start,
         )
         if sim is None:
@@ -2212,6 +2491,7 @@ def _min_rr_sweep(
         "live_qualified_setups": live_qualified,
         "reproduces_production_gate": reproduces,
         "exit_policy": exit_policy,
+        "reentry_lockdown_sessions": reentry_lockdown_sessions,
         "entries_from": sweep_start.isoformat() if sweep_start else None,
         "window": "out-of-sample (test)" if sweep_start else "full history (in-sample)",
         "rows": rows,
@@ -2267,6 +2547,9 @@ def _holdout_evaluation(
     exit_policy = str(strategy["exit_policy"])
     row_hold_days = hold_days
     trail_multiplier = ATR_TRAIL_MULTIPLIER
+    reentry_lockdown_sessions = int(
+        strategy.get("reentry_lockdown_sessions", 0)
+    )
     if strategy.get("use_live_config") and live_exit_policy is not None:
         exit_policy = LIVE_EXIT_MODE_TO_SIM.get(
             str(live_exit_policy.get("mode", "atr_trailing")), "atr_trail3"
@@ -2296,6 +2579,7 @@ def _holdout_evaluation(
             max_positions=int(entry_cfg["max_positions"]),
             risk_per_trade=float(entry_cfg["risk_per_trade"]),
             atr_trail_multiplier=trail_multiplier,
+            reentry_cooldown_sessions=reentry_lockdown_sessions,
             start_date=start,
             end_date=end,
             include_curve=True,
@@ -2307,6 +2591,7 @@ def _holdout_evaluation(
     return {
         "split_date": split.isoformat(),
         "strategy": strategy["strategy"],
+        "reentry_lockdown_sessions": reentry_lockdown_sessions,
         "rows": rows,
         "note": (
             "Train = entries before the split; test = entries on/after it. The two "
@@ -2339,6 +2624,9 @@ def _portfolio_monitor(
         # policy. The overlay opts into this deliberately so only ordering
         # changes relative to the production row.
         use_live = bool(strategy.get("use_live_config"))
+        reentry_lockdown_sessions = int(
+            strategy.get("reentry_lockdown_sessions", 0)
+        )
         exit_policy = str(strategy["exit_policy"])
         row_hold_days = hold_days
         trail_multiplier = ATR_TRAIL_MULTIPLIER
@@ -2367,6 +2655,7 @@ def _portfolio_monitor(
                 max_positions=int(entry_cfg["max_positions"]),
                 risk_per_trade=float(entry_cfg["risk_per_trade"]),
                 atr_trail_multiplier=trail_multiplier,
+                reentry_cooldown_sessions=reentry_lockdown_sessions,
                 start_date=start,
                 include_curve=True,
             )
@@ -2377,10 +2666,12 @@ def _portfolio_monitor(
                 "label": strategy["label"],
                 "description": strategy["description"],
                 "is_production": bool(strategy.get("is_production")),
+                "comparison_arm": strategy.get("comparison_arm"),
                 "entry_variant": strategy["entry_variant"],
                 "ranking_key": ranking_key,
                 "exit_policy": exit_policy,
                 "live_exit_mode": live_exit_mode,
+                "reentry_lockdown_sessions": reentry_lockdown_sessions,
                 "lookback": lookback["lookback"],
                 "lookback_label": lookback["label"],
                 **sim,
@@ -2393,6 +2684,10 @@ def _portfolio_monitor(
                 "label": s["label"],
                 "description": s["description"],
                 "is_production": bool(s.get("is_production")),
+                "comparison_arm": s.get("comparison_arm"),
+                "reentry_lockdown_sessions": int(
+                    s.get("reentry_lockdown_sessions", 0)
+                ),
             }
             for s in strategies
         ],
@@ -2405,7 +2700,48 @@ def _portfolio_monitor(
             "Portfolio monitor runs supported named strategies across cached lookbacks. "
             "The structural overlay appears only in its explicit research arm and changes "
             "ordering, not production qualification. Local snapshot backtests remain the "
-            "research surface for broad variant sweeps."
+            "research surface for broad variant sweeps. The production row applies the "
+            "same five-session post-initial-stop re-entry lockdown as the live setup list."
+        ),
+    }
+
+
+def _production_cadence_comparison(
+    monitor: dict | None,
+    cadence: str,
+) -> dict | None:
+    """Compact full-history live/no-lockdown vs live/5-session comparison."""
+    if not monitor:
+        return None
+    arms: list[dict] = []
+    for row in monitor.get("runs") or []:
+        comparison_arm = row.get("comparison_arm")
+        if not comparison_arm or row.get("lookback") != "all":
+            continue
+        compact = {
+            key: value
+            for key, value in row.items()
+            if key not in {"equity_curve", "benchmark_curve"}
+        }
+        arm_name = (
+            "prod_live_setup"
+            if comparison_arm == "live_no_lockdown"
+            else "cooldown_5"
+        )
+        compact["arm"] = f"{arm_name}_{cadence}"
+        compact["entry_cadence"] = cadence
+        arms.append(compact)
+    if not arms:
+        return None
+    arms.sort(key=lambda row: int(row.get("reentry_lockdown_sessions", 0)))
+    return {
+        "entry_cadence": cadence,
+        "lookback": "all",
+        "arms": arms,
+        "note": (
+            "Both arms use the exact same live gate, ordering, Admin exit policy, "
+            "fees, and candidate cadence. Only the five-session post-stop "
+            "re-entry lockdown changes."
         ),
     }
 
@@ -2618,7 +2954,8 @@ def _build_recommendation(report: dict) -> dict:
     if production_row is not None:
         headline = (
             "Production baseline: residual/high-vol 80/20 entry rank with a "
-            "3x ATR trailing exit and 30-trading-day max hold."
+            "3x ATR trailing exit, 30-trading-day max hold, and 5-session "
+            "re-entry lockdown after an initial stop."
         )
         if (
             production_row.get("cagr_pct") is not None
@@ -2787,9 +3124,11 @@ async def run_backtest(
     progress_cb: Callable[[int, int, str], None] | None = None,
     *,
     target_model: str = PRODUCTION_GTL_TARGET_MODEL,
+    cadence: str = DEFAULT_BACKTEST_CADENCE,
 ) -> dict:
     """Replay every ticker and aggregate the Phase-1 reports for the current config."""
     target_model = validate_backtest_target_model(target_model)
+    cadence = validate_backtest_cadence(cadence)
     config = await get_recommendation_config(db)
     activation = await get_activation_config(db)
 
@@ -2798,7 +3137,9 @@ async def run_backtest(
     total = len(tickers)
 
     candidates: list[dict] = []
-    # collected[signal_name][iso_week] -> list of (signal_value, forward_return)
+    # Signal IC remains a weekly, non-overlapping diagnostic regardless of the
+    # entry cadence. Production activation ranks are assigned from candidates
+    # at their own weekly or exact-date ``ranking_period`` below.
     collected: dict = defaultdict(lambda: defaultdict(list))
 
     # Residual momentum needs a point-in-time benchmark return stream. Best-effort:
@@ -2855,6 +3196,7 @@ async def run_backtest(
                             pool, _replay_and_signals, ticker.symbol, columns, config, activation,
                             benchmark_closes,
                             target_model,
+                            cadence,
                         ))
                 for result in await asyncio.gather(*futures, return_exceptions=True):
                     if isinstance(result, Exception):
@@ -2877,6 +3219,7 @@ async def run_backtest(
                         _replay_and_signals, ticker.symbol, columns, config, activation,
                         benchmark_closes,
                         target_model,
+                        cadence,
                     ))
             except Exception:
                 logger.exception("Backtest replay failed for %s", ticker.symbol)
@@ -2987,13 +3330,19 @@ async def run_backtest(
         "candidates": len(candidates),
         "qualified": len(qualified),
         "params": {
-            "step_days": STEP_DAYS,
+            # Keep step_days for old report consumers; the value counts stored
+            # market sessions rather than calendar days.
+            "step_days": backtest_step_sessions(cadence),
+            "step_sessions": backtest_step_sessions(cadence),
+            "entry_cadence": cadence,
+            "signal_eval_cadence": WEEKLY_BACKTEST_CADENCE,
             "horizon_days": HORIZON,
             "min_lookback": MIN_LOOKBACK,
             "cost_per_side_pct": round(COST_PER_SIDE * 100, 3),
             "target_model": target_model,
             "target_model_label": BACKTEST_TARGET_MODELS[target_model],
             "is_production_target_model": target_model == PRODUCTION_GTL_TARGET_MODEL,
+            "production_reentry_lockdown_sessions": REENTRY_LOCKDOWN_SESSIONS,
         },
         "activation": activation,
         "overall_qualified": _bucket_stats(qualified),
@@ -3055,6 +3404,11 @@ async def run_backtest(
             ),
         },
         "portfolio_monitor": portfolio_monitor_report,
+        "production_cadence_comparison": (
+            _production_cadence_comparison(portfolio_monitor_report, cadence)
+            if target_model == PRODUCTION_GTL_TARGET_MODEL
+            else None
+        ),
         "holdout": holdout_report,
         "min_rr_sweep": min_rr_sweep_report,
         "target_model_diagnostics": _target_model_diagnostics(
@@ -3090,9 +3444,15 @@ async def run_and_store(
     progress_cb: Callable[[int, int, str], None] | None = None,
     *,
     target_model: str = PRODUCTION_GTL_TARGET_MODEL,
+    cadence: str = DEFAULT_BACKTEST_CADENCE,
 ) -> dict:
     """Run the backtest and cache the report in a SystemSetting. Job entrypoint."""
-    report = await run_backtest(db, progress_cb, target_model=target_model)
+    report = await run_backtest(
+        db,
+        progress_cb,
+        target_model=target_model,
+        cadence=cadence,
+    )
     await update_setting(db, KEY_REPORT, json.dumps(report))
     return report
 

app/services/paper_trade_service.py

@@ -20,6 +20,7 @@ from app.services.outcome_service import (
     Bar,
     evaluate_setup_against_bars,
 )
+from app.services.trade_policy import get_reentry_lockdowns
 
 # Exit policy for OPEN paper trades (auto-close). Production defaults to the
 # July 2026 promoted strategy: initial stop + 3x ATR trailing stop, with a max
@@ -318,6 +319,13 @@ async def create_trade(
         raise ValidationError("shares and entry_price must be positive")
 
     ticker = await _get_ticker(db, symbol)
+    remaining_sessions = (await get_reentry_lockdowns(db)).get(ticker.id)
+    if remaining_sessions is not None:
+        suffix = "session" if remaining_sessions == 1 else "sessions"
+        raise ValidationError(
+            f"{ticker.symbol} is in a post-stop re-entry lockdown: "
+            f"{remaining_sessions} market {suffix} remaining"
+        )
     trade = PaperTrade(
         user_id=user_id,
         ticker_id=ticker.id,

app/services/rr_scanner_service.py

@@ -29,6 +29,7 @@ from app.models.trade_setup import TradeSetup
 from app.services.indicator_service import _extract_ohlcv, compute_atr
 from app.services.price_service import query_ohlcv
 from app.services.sr_service import detect_gate_target_ladder
+from app.services.trade_policy import get_reentry_lockdowns
 from app.services.recommendation_service import (
     _risk_level_from_conflicts,
     build_recommendation_snapshot,
@@ -771,6 +772,8 @@ async def get_trade_setups(
     symbol: str | None = None,
     live_recommendation: bool = False,
     exclude_open_trade_tickers: bool = False,
+    exclude_reentry_lockdown_tickers: bool = False,
+    include_reentry_lockdown: bool = False,
 ) -> list[dict]:
     """Get latest stored trade setups, optionally filtered.
 
@@ -794,15 +797,23 @@ async def get_trade_setups(
         stmt = stmt.where(TradeSetup.confidence_score >= min_confidence)
     if recommended_action is not None and not live_recommendation:
         stmt = stmt.where(TradeSetup.recommended_action == recommended_action)
+    excluded_ticker_ids: set[int] = set()
+    reentry_lockdowns: dict[int, int] = {}
     if exclude_open_trade_tickers:
         open_trade_result = await db.execute(
             select(PaperTrade.ticker_id)
             .where(PaperTrade.status == "open")
             .distinct()
         )
-        open_ticker_ids = {ticker_id for ticker_id, in open_trade_result.all()}
+        excluded_ticker_ids.update(
-        if open_ticker_ids:
+            ticker_id for ticker_id, in open_trade_result.all()
-            stmt = stmt.where(~TradeSetup.ticker_id.in_(open_ticker_ids))
+        )
+    if exclude_reentry_lockdown_tickers or include_reentry_lockdown:
+        reentry_lockdowns = await get_reentry_lockdowns(db)
+    if exclude_reentry_lockdown_tickers:
+        excluded_ticker_ids.update(reentry_lockdowns)
+    if excluded_ticker_ids:
+        stmt = stmt.where(~TradeSetup.ticker_id.in_(excluded_ticker_ids))
 
     stmt = stmt.order_by(TradeSetup.detected_at.desc(), TradeSetup.id.desc())
 
@@ -855,6 +866,15 @@ async def get_trade_setups(
             ),
             reverse=True,
         )
+    if include_reentry_lockdown:
+        ticker_by_setup_id = {
+            setup.id: setup.ticker_id for setup, _ in latest_rows
+        }
+        for row in rows_out:
+            ticker_id = ticker_by_setup_id.get(row["id"])
+            row["reentry_lockdown_remaining_sessions"] = (
+                reentry_lockdowns.get(ticker_id) if ticker_id is not None else None
+            )
     return rows_out
 
 

app/services/trade_policy.py

@@ -0,0 +1,120 @@
+"""Shared live/backtest trading-policy constants and availability checks."""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import date, datetime, timezone
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.benchmark_price import BenchmarkPrice
+from app.models.ohlcv import OHLCVRecord
+from app.models.paper_trade import PaperTrade
+from app.services.benchmark_service import BENCHMARK_SYMBOL
+
+# A ticker stopped at its initial stop may qualify again immediately, but the
+# July 2026 event study showed that waiting five market sessions materially
+# improved the production book. The stop session is wait_session=0; the first
+# permitted re-entry is wait_session=5, provided the normal gate still passes.
+REENTRY_LOCKDOWN_SESSIONS = 5
+
+
+async def get_reentry_lockdowns(
+    db: AsyncSession,
+    *,
+    as_of: date | None = None,
+    sessions: int = REENTRY_LOCKDOWN_SESSIONS,
+) -> dict[int, int]:
+    """Return ``{ticker_id: remaining_sessions}`` for active lockdowns.
+
+    SPY is the canonical calendar for the platform's US-equity universe. When
+    the stored benchmark history does not reach an older stop, only that
+    ticker's own OHLCV dates are used as a conservative fallback. Unrelated
+    ticker dates can therefore never shorten a lockdown.
+    """
+    sessions = max(0, int(sessions))
+    if sessions == 0:
+        return {}
+
+    session_cutoff = as_of or datetime.now(timezone.utc).date()
+    stop_result = await db.execute(
+        select(
+            PaperTrade.ticker_id,
+            func.max(PaperTrade.closed_at).label("last_stop_at"),
+        )
+        .where(
+            PaperTrade.status == "closed",
+            PaperTrade.close_reason == "stop",
+            PaperTrade.closed_at.is_not(None),
+        )
+        .group_by(PaperTrade.ticker_id)
+    )
+    stop_dates = {
+        ticker_id: stopped_at.date()
+        for ticker_id, stopped_at in stop_result.all()
+        if stopped_at is not None and stopped_at.date() <= session_cutoff
+    }
+    if not stop_dates:
+        return {}
+
+    benchmark_result = await db.execute(
+        select(BenchmarkPrice.date)
+        .where(
+            BenchmarkPrice.symbol == BENCHMARK_SYMBOL,
+            BenchmarkPrice.date <= session_cutoff,
+        )
+        .order_by(BenchmarkPrice.date.asc())
+    )
+    benchmark_dates = [row[0] for row in benchmark_result.all()]
+
+    lockdowns: dict[int, int] = {}
+    fallback_stops: dict[int, date] = {}
+    first_benchmark_date = benchmark_dates[0] if benchmark_dates else None
+    for ticker_id, stop_date in stop_dates.items():
+        completed = sum(day > stop_date for day in benchmark_dates)
+        if completed >= sessions:
+            continue
+        if first_benchmark_date is not None and first_benchmark_date <= stop_date:
+            lockdowns[ticker_id] = sessions - completed
+        else:
+            # The benchmark table starts after this stop (or is empty), so it
+            # cannot prove how many sessions elapsed. Resolve only this ticker
+            # against its own bars instead of using universe-wide dates.
+            fallback_stops[ticker_id] = stop_date
+
+    if fallback_stops:
+        own_session_result = await db.execute(
+            select(OHLCVRecord.ticker_id, OHLCVRecord.date)
+            .where(
+                OHLCVRecord.ticker_id.in_(fallback_stops),
+                OHLCVRecord.date > min(fallback_stops.values()),
+                OHLCVRecord.date <= session_cutoff,
+            )
+            .distinct()
+        )
+        own_dates: dict[int, set[date]] = defaultdict(set)
+        for ticker_id, market_date in own_session_result.all():
+            own_dates[ticker_id].add(market_date)
+        for ticker_id, stop_date in fallback_stops.items():
+            completed = sum(day > stop_date for day in own_dates[ticker_id])
+            if completed < sessions:
+                lockdowns[ticker_id] = sessions - completed
+
+    return lockdowns
+
+
+async def get_reentry_lockdown_ticker_ids(
+    db: AsyncSession,
+    *,
+    as_of: date | None = None,
+    sessions: int = REENTRY_LOCKDOWN_SESSIONS,
+) -> set[int]:
+    """Compatibility wrapper for callers that only need blocked ticker ids."""
+    return set(
+        await get_reentry_lockdowns(
+            db,
+            as_of=as_of,
+            sessions=sessions,
+        )
+    )

frontend/src/api/admin.ts

@@ -201,9 +201,11 @@ export interface TriggerJobResponse {
   status: 'triggered' | 'busy' | 'blocked' | 'not_found';
   message: string;
   target_model?: BacktestTargetModel;
+  cadence?: BacktestCadence;
 }
 
 export type BacktestTargetModel = 'production_gtl' | 'structural_sr';
+export type BacktestCadence = 'weekly' | 'daily';
 
 export function listJobs() {
   return apiClient.get<JobStatus[]>('admin/jobs').then((r) => r.data);
@@ -219,7 +221,10 @@ export function toggleJob(jobName: string, enabled: boolean) {
     .then((r) => r.data);
 }
 
-export function triggerJob(jobName: string, options?: { target_model?: BacktestTargetModel }) {
+export function triggerJob(
+  jobName: string,
+  options?: { target_model?: BacktestTargetModel; cadence?: BacktestCadence },
+) {
   return apiClient
     .post<TriggerJobResponse>(`admin/jobs/${jobName}/trigger`, options)
     .then((r) => r.data);

frontend/src/components/signals/BacktestPanel.tsx

@@ -2,7 +2,7 @@ import { useMemo, useState } from 'react';
 import { useMutation, useQueryClient } from '@tanstack/react-query';
 import { useBacktestReport } from '../../hooks/useMarketRegime';
 import { triggerJob } from '../../api/admin';
-import type { BacktestTargetModel } from '../../api/admin';
+import type { BacktestCadence, BacktestTargetModel } from '../../api/admin';
 import { Button } from '../ui/Button';
 import { Callout } from '../ui/Callout';
 import { Disclosure } from '../ui/Disclosure';
@@ -145,6 +145,7 @@ export function BacktestPanel() {
   const [selectedStrategy, setSelectedStrategy] = useState('');
   const [selectedLookback, setSelectedLookback] = useState('');
   const [targetModel, setTargetModel] = useState<BacktestTargetModel>('production_gtl');
+  const [cadence, setCadence] = useState<BacktestCadence>('weekly');
 
   const monitor = report?.portfolio_monitor ?? null;
   const activeStrategy =
@@ -161,11 +162,11 @@ export function BacktestPanel() {
   );
 
   const run = useMutation({
-    mutationFn: () => triggerJob('backtest', { target_model: targetModel }),
+    mutationFn: () => triggerJob('backtest', { target_model: targetModel, cadence }),
     onSuccess: (res) => {
       if (res.status === 'triggered') {
         const label = targetModel === 'production_gtl' ? 'Live GTL' : 'Structural S/R comparison';
-        toast.addToast('success', `${label} backtest started — results appear when it finishes.`);
+        toast.addToast('success', `${label} ${cadence} backtest started — results appear when it finishes.`);
         setTimeout(() => queryClient.invalidateQueries({ queryKey: ['backtest-report'] }), 8000);
       } else {
         toast.addToast('info', res.message || 'Could not start backtest');
@@ -180,7 +181,7 @@ export function BacktestPanel() {
         <div className="flex flex-wrap items-start justify-between gap-3">
           <Disclosure summary="How this is measured">
             <p className="max-w-2xl text-xs text-gray-400">
-              The backtest replays the current config weekly through history — at each point the setup is
+              The backtest replays the current config at the selected cadence — at each point the setup is
               rebuilt using only data up to that day (no lookahead) and the following ~30 trading days decide
               its outcome — then simulates one capital-constrained book against the S&P 500. Sentiment and
               fundamentals are held neutral (no point-in-time history). ~6 months is roughly one market regime,
@@ -238,6 +239,56 @@ export function BacktestPanel() {
                 </span>
               </label>
             </fieldset>
+            <fieldset className="grid w-full grid-cols-2 gap-2 sm:w-[34rem]">
+              <legend className="mb-1 text-[11px] font-medium uppercase tracking-wider text-gray-500">
+                Entry cadence
+              </legend>
+              <label
+                className={`cursor-pointer rounded-lg border px-3 py-2 transition-colors focus-within:ring-2 focus-within:ring-blue-400/60 ${
+                  cadence === 'weekly'
+                    ? 'border-blue-400/60 bg-blue-500/10'
+                    : 'border-white/10 bg-white/[0.03] hover:border-white/20'
+                }`}
+              >
+                <input
+                  className="sr-only"
+                  type="radio"
+                  name="backtest-cadence"
+                  value="weekly"
+                  checked={cadence === 'weekly'}
+                  onChange={() => setCadence('weekly')}
+                />
+                <span className="flex items-center justify-between gap-2 text-sm font-medium text-gray-100">
+                  Weekly
+                  <span className="rounded-full border border-blue-400/40 bg-blue-400/10 px-2 py-0.5 text-[9px] font-semibold uppercase tracking-widest text-blue-300">
+                    Default
+                  </span>
+                </span>
+                <span className="mt-1 block text-[11px] leading-4 text-gray-500">
+                  Resource-safe server run at five-session intervals.
+                </span>
+              </label>
+              <label
+                className={`cursor-pointer rounded-lg border px-3 py-2 transition-colors focus-within:ring-2 focus-within:ring-amber-400/60 ${
+                  cadence === 'daily'
+                    ? 'border-amber-400/50 bg-amber-500/10'
+                    : 'border-white/10 bg-white/[0.03] hover:border-white/20'
+                }`}
+              >
+                <input
+                  className="sr-only"
+                  type="radio"
+                  name="backtest-cadence"
+                  value="daily"
+                  checked={cadence === 'daily'}
+                  onChange={() => setCadence('daily')}
+                />
+                <span className="text-sm font-medium text-gray-200">Daily</span>
+                <span className="mt-1 block text-[11px] leading-4 text-amber-300/80">
+                  Research run: roughly 5× the replay work; prefer the offline snapshot runner.
+                </span>
+              </label>
+            </fieldset>
             <Button onClick={() => run.mutate()} loading={run.isPending} className="shrink-0">
               {run.isPending ? 'Starting…' : report ? 'Re-run backtest' : 'Run backtest'}
             </Button>
@@ -257,7 +308,8 @@ export function BacktestPanel() {
           <>
             <p className="text-[11px] text-gray-500">
               Ran {timeAgo(report.generated_at)} · {report.tickers} tickers · {report.candidates} setups
-              ({report.qualified} qualified) · weekly cadence, {report.params.horizon_days}-day horizon
+              ({report.qualified} qualified) · {report.params.entry_cadence ?? 'weekly'} cadence,
+              {' '}{report.params.horizon_days}-day horizon
               {report.params.cost_per_side_pct != null && (
                 <> · net of {report.params.cost_per_side_pct}%/side costs</>
               )}
@@ -321,6 +373,9 @@ export function BacktestPanel() {
                 <p className="text-[11px] text-gray-500">
                   Avg hold {fmtDays(monitorRun.avg_hold_days)} · Best {fmtR(monitorRun.best_trade_r)} / Worst{' '}
                   {fmtR(monitorRun.worst_trade_r)} · Avg P&amp;L per trade {fmtMoney(monitorRun.avg_trade_pnl)}
+                  {monitorRun.reentry_lockdown_sessions ? (
+                    <> · Re-entry lockdown {monitorRun.reentry_lockdown_sessions} market sessions after initial stop</>
+                  ) : null}
                 </p>
 
                 {monitorRun.yearly_returns && monitorRun.yearly_returns.length > 0 && (

frontend/src/components/ticker/RecommendationPanel.tsx

@@ -66,14 +66,19 @@ function entryDrift(setup: TradeSetup, currentPrice?: number) {
   return { pct, r, status };
 }
 
-/**
+type NotActionableState =
- * The only state with no tradeable setup left: price has gone through the stop.
+  | { kind: 'lockdown'; remainingSessions: number }
- * Returns null when there's no live price.
+  | { kind: 'invalidated' }
- */
+  | null;
+
 function notActionableState(setup: TradeSetup, currentPrice?: number) {
+  const remainingSessions = setup.reentry_lockdown_remaining_sessions ?? 0;
+  if (remainingSessions > 0) {
+    return { kind: 'lockdown', remainingSessions } satisfies NotActionableState;
+  }
   if (currentPrice == null) return null;
   if (entryDrift(setup, currentPrice)?.status !== 'invalidated') return null;
-  return { invalidated: true };
+  return { kind: 'invalidated' } satisfies NotActionableState;
 }
 
 function riskClass(risk: TradeSetup['risk_level']) {
@@ -218,9 +223,6 @@ function SetupCard({ setup, action, currentPrice, risk, regime, exitPolicy, sele
   const exitPlan = deriveExitPlan(setup, exitPolicy);
   const honorsTarget = exitPlan?.honorsTarget ?? false;
 
-  // Only price through the stop leaves no tradeable setup.
-  const notActionable = notActionableState(setup, currentPrice) != null;
-
   const createTrade = useCreatePaperTrade();
   const [taking, setTaking] = useState(false);
   const [takeShares, setTakeShares] = useState<number>(sizing?.shares ?? 0);
@@ -268,7 +270,27 @@ function SetupCard({ setup, action, currentPrice, risk, regime, exitPolicy, sele
     );
   };
 
-  if (notActionable) {
+  const inactiveState = notActionableState(setup, currentPrice);
+  if (inactiveState?.kind === 'lockdown') {
+    const remaining = inactiveState.remainingSessions;
+    return (
+      <div data-direction={setup.direction} className="rounded-xl border border-amber-400/20 bg-amber-400/[0.04] p-4">
+        <div className="flex flex-wrap items-center gap-2">
+          <DirTag direction={setup.direction} />
+          <span className="num text-[10px] uppercase tracking-[0.16em] text-amber-300">post-stop lockdown</span>
+          <span className="num ml-auto text-xs text-gray-500">
+            {remaining} market session{remaining === 1 ? '' : 's'} remaining
+          </span>
+        </div>
+        <p className="mt-2 text-[11.5px] leading-relaxed text-gray-400">
+          This setup remains visible for context but cannot be marked as taken. Once the lockdown expires,
+          the scanner recalculates the normal gate before it can become actionable again.
+        </p>
+      </div>
+    );
+  }
+
+  if (inactiveState?.kind === 'invalidated') {
     const dir = setup.direction.toUpperCase();
     return (
       <div data-direction={setup.direction} className="rounded-xl border border-white/[0.07] p-4">
@@ -617,7 +639,21 @@ export function RecommendationPanel({ symbol, longSetup, shortSetup, currentPric
         <div className="min-w-0">
           {preferredInactive ? (
             <span className="text-sm font-semibold text-gray-400">
-              No current setup <span className="font-normal text-gray-500">(last {preferredDirection} bias {recommendationActionLabel(action).toLowerCase()} — invalidated at the stop)</span>
+              {preferredInactive.kind === 'lockdown' ? (
+                <>
+                  Re-entry paused{' '}
+                  <span className="font-normal text-gray-500">
+                    ({preferredInactive.remainingSessions} market session{preferredInactive.remainingSessions === 1 ? '' : 's'} remaining after stop)
+                  </span>
+                </>
+              ) : (
+                <>
+                  No current setup{' '}
+                  <span className="font-normal text-gray-500">
+                    (last {preferredDirection} bias {recommendationActionLabel(action).toLowerCase()} — invalidated at the stop)
+                  </span>
+                </>
+              )}
             </span>
           ) : (() => {
             const reasoning = summary?.reasoning ?? '';

frontend/src/lib/qualification.ts

@@ -43,6 +43,7 @@ export function liveRiskReward(setup: TradeSetup, currentPrice: number): number
  * app/services/qualification.py — keep the two in sync.
  */
 export function qualifiesSetup(setup: TradeSetup, config: ActivationConfig): boolean {
+  if ((setup.reentry_lockdown_remaining_sessions ?? 0) > 0) return false;
   if (setup.rr_ratio < config.min_rr) return false;
   // Live R:R from current price — drops setups whose price has already run
   // toward target (reward consumed) or through the stop.
@@ -79,6 +80,10 @@ export function qualifiesSetup(setup: TradeSetup, config: ActivationConfig): boo
  * qualifiesSetup rule-for-rule (keep the order in sync).
  */
 export function disqualifyReason(setup: TradeSetup, config: ActivationConfig): string | null {
+  const lockdownRemaining = setup.reentry_lockdown_remaining_sessions ?? 0;
+  if (lockdownRemaining > 0) {
+    return `post-stop lockdown · ${lockdownRemaining} session${lockdownRemaining === 1 ? '' : 's'} remaining`;
+  }
   if (setup.rr_ratio < config.min_rr) {
     return `R:R ${setup.rr_ratio.toFixed(1)} below gate ${config.min_rr.toFixed(1)}`;
   }

frontend/src/lib/types.ts

@@ -144,6 +144,7 @@ export interface TradeSetup {
   momentum_percentile?: number | null;
   strategy_rank?: number | null;
   volatility_percentile?: number | null;
+  reentry_lockdown_remaining_sessions?: number | null;
   context_as_of?: TradeSetupContextAsOf | null;
   recommendation_summary?: RecommendationSummary;
 }
@@ -355,15 +356,24 @@ export interface BacktestPortfolioMonitorRun extends BacktestPortfolioPolicy {
   label: string;
   description: string;
   is_production: boolean;
+  comparison_arm?: 'live_no_lockdown' | 'live_lockdown_5' | null;
   entry_variant: string;
   exit_policy: string;
+  reentry_lockdown_sessions?: number;
   lookback: string;
   lookback_label: string;
 }
 
 export interface BacktestPortfolioMonitor {
   production_strategy: string;
-  strategies: { strategy: string; label: string; description: string; is_production: boolean }[];
+  strategies: {
+    strategy: string;
+    label: string;
+    description: string;
+    is_production: boolean;
+    comparison_arm?: 'live_no_lockdown' | 'live_lockdown_5' | null;
+    reentry_lockdown_sessions?: number;
+  }[];
   lookbacks: { lookback: string; label: string }[];
   runs: BacktestPortfolioMonitorRun[];
   note?: string;
@@ -396,12 +406,16 @@ export interface BacktestReport {
   qualified: number;
   params: {
     step_days: number;
+    step_sessions?: number;
+    entry_cadence?: 'weekly' | 'daily';
+    signal_eval_cadence?: 'weekly';
     horizon_days: number;
     min_lookback: number;
     cost_per_side_pct?: number;
     target_model?: 'production_gtl' | 'structural_sr';
     target_model_label?: string;
     is_production_target_model?: boolean;
+    production_reentry_lockdown_sessions?: number;
   };
   overall_qualified: BacktestBucket;
   overall_all: BacktestBucket;

scripts/run_backtest_cadence_comparison.py

@@ -0,0 +1,179 @@
+"""Run the four production cadence/lockdown arms on one offline snapshot.
+
+The command executes the complete backtest once weekly and once daily. Each
+backtest contains two otherwise identical live-policy portfolio arms: no
+post-stop lockdown and the production five-session lockdown. It writes both
+full reports plus one compact four-arm comparison report.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import sys
+from datetime import datetime
+from pathlib import Path
+
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "snapshot",
+        help="SQLite snapshot created by scripts/create_backtest_snapshot.py.",
+    )
+    parser.add_argument(
+        "--out-dir",
+        default="reports",
+        help="Directory for the weekly, daily, and comparison JSON reports.",
+    )
+    parser.add_argument(
+        "--prefix",
+        default=None,
+        help="Output prefix. Defaults to backtest-cadence-<timestamp>.",
+    )
+    parser.add_argument(
+        "--workers",
+        type=int,
+        default=None,
+        help="Override worker count; on a powerful offline PC use CPU count minus one.",
+    )
+    parser.add_argument(
+        "--allow-spawn",
+        action="store_true",
+        help="Enable multiprocessing spawn for the offline Windows run.",
+    )
+    parser.add_argument("--quiet", action="store_true", help="Hide ticker progress.")
+    return parser.parse_args()
+
+
+def _sqlite_url(path: Path) -> str:
+    return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"
+
+
+def _write_json(path: Path, payload: dict) -> None:
+    path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
+
+
+def _comparison_arms(report: dict) -> list[dict]:
+    comparison = report.get("production_cadence_comparison") or {}
+    arms = list(comparison.get("arms") or [])
+    if len(arms) != 2:
+        cadence = (report.get("params") or {}).get("entry_cadence", "unknown")
+        raise RuntimeError(
+            f"Expected two live comparison arms for {cadence}; found {len(arms)}"
+        )
+    return arms
+
+
+def _print_arm(row: dict) -> None:
+    print(
+        f"  {row['arm']}: Sharpe {row.get('sharpe')}, "
+        f"CAGR {row.get('cagr_pct')}%, DD {row.get('max_drawdown_pct')}%, "
+        f"trades {row.get('trades')}, skipped cooldown {row.get('skipped_cooldown', 0)}"
+    )
+
+
+async def _main() -> None:
+    args = _parse_args()
+    snapshot = Path(args.snapshot)
+    if not snapshot.exists():
+        raise SystemExit(f"Snapshot not found: {snapshot}")
+
+    os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
+    if args.allow_spawn:
+        os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
+
+    from app.config import settings
+    from app.services.backtest_service import run_backtest
+
+    if args.workers is not None:
+        settings.backtest_workers = args.workers
+
+    out_dir = Path(args.out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    prefix = args.prefix or f"backtest-cadence-{datetime.now():%Y%m%d-%H%M%S}"
+
+    engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
+    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    reports: dict[str, dict] = {}
+    try:
+        async with Session() as db:
+            for cadence in ("weekly", "daily"):
+                last_progress: tuple[int, int] | None = None
+
+                def progress(done: int, total: int, symbol: str) -> None:
+                    nonlocal last_progress
+                    if args.quiet or last_progress == (done, total):
+                        return
+                    last_progress = (done, total)
+                    label = f" {symbol}" if symbol else ""
+                    print(
+                        f"{cadence} progress: {done}/{total}{label}",
+                        end="\r",
+                    )
+
+                reports[cadence] = await run_backtest(
+                    db,
+                    progress_cb=progress,
+                    target_model="production_gtl",
+                    cadence=cadence,
+                )
+                if not args.quiet:
+                    print("")
+                _write_json(out_dir / f"{prefix}-{cadence}.json", reports[cadence])
+    finally:
+        await engine.dispose()
+
+    arms = [
+        *_comparison_arms(reports["weekly"]),
+        *_comparison_arms(reports["daily"]),
+    ]
+    expected = {
+        "prod_live_setup_weekly",
+        "prod_live_setup_daily",
+        "cooldown_5_weekly",
+        "cooldown_5_daily",
+    }
+    if {row.get("arm") for row in arms} != expected:
+        raise RuntimeError("The generated cadence report does not contain all four arms")
+    arm_order = {
+        "prod_live_setup_weekly": 0,
+        "prod_live_setup_daily": 1,
+        "cooldown_5_weekly": 2,
+        "cooldown_5_daily": 3,
+    }
+    arms.sort(key=lambda row: arm_order[str(row["arm"])])
+
+    comparison = {
+        "generated_at": datetime.now().astimezone().isoformat(),
+        "snapshot": str(snapshot.resolve()),
+        "target_model": "production_gtl",
+        "arms": arms,
+        "full_reports": {
+            cadence: str((out_dir / f"{prefix}-{cadence}.json").resolve())
+            for cadence in ("weekly", "daily")
+        },
+        "note": (
+            "All four arms use the same snapshot, activation settings, target model, "
+            "live Admin exit policy, fees, sizing, and portfolio constraints. Within "
+            "each cadence pair, only the five-session post-stop lockdown differs."
+        ),
+    }
+    comparison_path = out_dir / f"{prefix}-comparison.json"
+    _write_json(comparison_path, comparison)
+
+    print(f"Comparison written: {comparison_path}")
+    for row in arms:
+        _print_arm(row)
+
+
+if __name__ == "__main__":
+    asyncio.run(_main())

scripts/run_backtest_snapshot.py

@@ -32,7 +32,10 @@ def _parse_args() -> argparse.Namespace:
     parser.add_argument(
         "--out",
         default=None,
-        help="JSON report path. Defaults to reports/backtest-<timestamp>.json.",
+        help=(
+            "JSON report path. Defaults to "
+            "reports/backtest-<cadence>-<timestamp>.json."
+        ),
     )
     parser.add_argument(
         "--workers",
@@ -55,6 +58,15 @@ def _parse_args() -> argparse.Namespace:
             "structural_sr is a comparison-only chart-S/R model."
         ),
     )
+    parser.add_argument(
+        "--cadence",
+        choices=("weekly", "daily"),
+        default="weekly",
+        help=(
+            "Entry replay cadence. Weekly is the resource-safe production default; "
+            "daily performs roughly five times as many setup evaluations."
+        ),
+    )
     parser.add_argument(
         "--holdout-split",
         default=None,
@@ -63,9 +75,9 @@ def _parse_args() -> argparse.Namespace:
     return parser.parse_args()
 
 
-def _default_output_path() -> Path:
+def _default_output_path(cadence: str) -> Path:
     stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
-    return Path("reports") / f"backtest-{stamp}.json"
+    return Path("reports") / f"backtest-{cadence}-{stamp}.json"
 
 
 def _pct(value: Any) -> str:
@@ -93,6 +105,7 @@ def _print_summary(report: dict) -> None:
 
     print("")
     print("Backtest summary")
+    print(f"  entry cadence: {(report.get('params') or {}).get('entry_cadence', 'weekly')}")
     print(f"  candidates: {report.get('candidates')}")
     print(f"  qualified: {report.get('qualified')}")
     print(f"  all setups net avg R: {_r(all_setups.get('net_avg_r'))}")
@@ -183,7 +196,7 @@ async def _main() -> None:
     if args.workers is not None:
         settings.backtest_workers = args.workers
 
-    output = Path(args.out) if args.out else _default_output_path()
+    output = Path(args.out) if args.out else _default_output_path(args.cadence)
     output.parent.mkdir(parents=True, exist_ok=True)
 
     engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
@@ -208,6 +221,7 @@ async def _main() -> None:
                 db,
                 progress_cb=progress,
                 target_model=args.target_model,
+                cadence=args.cadence,
             )
     finally:
         await engine.dispose()

scripts/run_gate_protected_stop_study.py

@@ -0,0 +1,422 @@
+"""Targeted offline study of a gate-conditioned initial-stop refresh.
+
+The study replays production entries only for the requested period. Whenever
+an initial stop is touched, it rebuilds that ticker's setup using bars through
+the previous close and recomputes the production momentum gate across the whole
+historical universe. If the gate still passes and the new setup has a lower
+valid stop, the simulator adopts it and checks it against the same day's low.
+
+This is causal: no value from the stop day's eventual close is used to cancel
+an intraday stop. The snapshot is read-only and no live settings are changed.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import bisect
+import json
+import multiprocessing
+import os
+import sys
+from collections import Counter
+from concurrent.futures import ProcessPoolExecutor, as_completed
+from datetime import date, datetime
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+
+def _sqlite_url(path: Path) -> str:
+    return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("snapshot")
+    parser.add_argument("--start-date", default="2024-07-01")
+    parser.add_argument("--workers", type=int, default=6)
+    parser.add_argument("--out", default=None)
+    parser.add_argument("--quiet", action="store_true")
+    return parser.parse_args()
+
+
+def _default_output_path() -> Path:
+    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    return Path("reports") / f"gate-protected-stop-{stamp}.json"
+
+
+class GateStopRefresher:
+    """Point-in-time gate and replacement-stop calculator for stop events."""
+
+    def __init__(
+        self,
+        prices: dict[str, tuple],
+        recommendation_config: dict,
+        activation: dict,
+        benchmark_closes: dict[date, float],
+    ) -> None:
+        from app.services import backtest_service as bt
+
+        self.bt = bt
+        self.prices = prices
+        self.recommendation_config = recommendation_config
+        self.activation = activation
+        self.benchmark_closes = benchmark_closes
+        self.threshold = float(activation.get("min_momentum_percentile", 80.0))
+        self.dates = {
+            symbol: [date.fromordinal(value) for value in columns[0]]
+            for symbol, columns in prices.items()
+        }
+        self.index_of = {
+            symbol: {value: index for index, value in enumerate(columns[0])}
+            for symbol, columns in prices.items()
+        }
+        self.percentile_cache: dict[int, dict[str, float]] = {}
+        self.setup_cache: dict[tuple[str, int], dict | None] = {}
+        self.events: list[dict[str, Any]] = []
+
+    def _momentum_percentiles(self, asof_ord: int) -> dict[str, float]:
+        cached = self.percentile_cache.get(asof_ord)
+        if cached is not None:
+            return cached
+
+        values: dict[str, float] = {}
+        for symbol, columns in self.prices.items():
+            idx = bisect.bisect_right(columns[0], asof_ord) - 1
+            if idx < 252:
+                continue
+            closes = columns[4]
+            value = self.bt._residual_momentum_12_1(
+                self.dates[symbol], closes, idx, self.benchmark_closes
+            )
+            if value is None and closes[idx - 252] > 0:
+                value = closes[idx - 21] / closes[idx - 252] - 1.0
+            if value is not None:
+                values[symbol] = float(value)
+
+        ordered = sorted(values, key=lambda symbol: values[symbol])
+        denominator = len(ordered) - 1
+        percentiles = {
+            symbol: (rank / denominator * 100.0) if denominator > 0 else 100.0
+            for rank, symbol in enumerate(ordered)
+        }
+        self.percentile_cache[asof_ord] = percentiles
+        return percentiles
+
+    def _long_setup(self, symbol: str, asof_idx: int) -> dict | None:
+        columns = self.prices[symbol]
+        asof_ord = columns[0][asof_idx]
+        key = (symbol, asof_ord)
+        if key in self.setup_cache:
+            return self.setup_cache[key]
+
+        records = [
+            SimpleNamespace(
+                date=date.fromordinal(o),
+                open=op,
+                high=high,
+                low=low,
+                close=close,
+                volume=volume,
+            )
+            for o, op, high, low, close, volume in zip(
+                columns[0][: asof_idx + 1],
+                columns[1][: asof_idx + 1],
+                columns[2][: asof_idx + 1],
+                columns[3][: asof_idx + 1],
+                columns[4][: asof_idx + 1],
+                columns[5][: asof_idx + 1],
+            )
+        ]
+        setups = self.bt._window_setups(
+            records, self.recommendation_config, self.activation
+        )
+        setup = next((row for row in setups if row["direction"] == "long"), None)
+        self.setup_cache[key] = setup
+        return setup
+
+    def __call__(
+        self,
+        symbol: str,
+        stop_ord: int,
+        active_stop: float,
+        position: dict,
+        bar: Any,
+    ) -> float | None:
+        columns = self.prices[symbol]
+        stop_idx = self.index_of[symbol].get(stop_ord)
+        if stop_idx is None:
+            stop_idx = bisect.bisect_left(columns[0], stop_ord)
+        asof_idx = stop_idx - 1
+        if asof_idx < self.bt.MIN_LOOKBACK - 1:
+            return None
+
+        asof_ord = columns[0][asof_idx]
+        setup = self._long_setup(symbol, asof_idx)
+        momentum_pct = self._momentum_percentiles(asof_ord).get(symbol)
+        gate_passed = bool(
+            setup is not None
+            and self.bt._momentum_qualifies(
+                {
+                    "meets_core": setup["meets_core"],
+                    "direction": "long",
+                    self.bt.PRODUCTION_PERCENTILE_KEY: momentum_pct,
+                },
+                self.threshold,
+            )
+        )
+        new_stop = float(setup["stop"]) if gate_passed and setup is not None else None
+        lower_stop = bool(new_stop is not None and new_stop < active_stop - 1e-9)
+        original_risk = float(position["entry"] - position["initial_stop"])
+        replacement_risk_r = (
+            (float(position["entry"]) - new_stop) / original_risk
+            if lower_stop and original_risk > 0 and new_stop is not None
+            else None
+        )
+        self.events.append({
+            "symbol": symbol,
+            "stop_date": date.fromordinal(stop_ord).isoformat(),
+            "gate_asof_date": date.fromordinal(asof_ord).isoformat(),
+            "momentum_percentile": round(momentum_pct, 2)
+            if momentum_pct is not None
+            else None,
+            "gate_core_passed": bool(setup and setup["meets_core"]),
+            "gate_passed": gate_passed,
+            "active_stop": round(active_stop, 4),
+            "replacement_stop": round(new_stop, 4) if new_stop is not None else None,
+            "lower_stop": lower_stop,
+            "same_bar_survives": bool(lower_stop and bar.low > new_stop),
+            "replacement_risk_r": round(replacement_risk_r, 3)
+            if replacement_risk_r is not None
+            else None,
+        })
+        return new_stop
+
+
+def _arm(label: str, sim: dict) -> dict:
+    trade_details = sim.pop("trade_details", None)
+    row = {"arm": label, **sim}
+    if trade_details is not None:
+        row["trade_details"] = trade_details
+    return row
+
+
+def _rescued_trade_summary(trades: list[dict]) -> dict:
+    rescued = [trade for trade in trades if trade.get("stop_refreshes", 0) > 0]
+    rs = [float(trade["r"]) for trade in rescued]
+    return {
+        "trades": len(rescued),
+        "wins": sum(value > 0 for value in rs),
+        "win_rate": round(sum(value > 0 for value in rs) / len(rs) * 100.0, 1)
+        if rs
+        else None,
+        "avg_r": round(sum(rs) / len(rs), 3) if rs else None,
+        "total_r": round(sum(rs), 2) if rs else None,
+        "worst_r": round(min(rs), 2) if rs else None,
+        "best_r": round(max(rs), 2) if rs else None,
+        "exit_reasons": dict(Counter(trade["reason"] for trade in rescued)),
+    }
+
+
+async def _main() -> None:
+    args = _parse_args()
+    snapshot = Path(args.snapshot)
+    if not snapshot.exists():
+        raise SystemExit(f"Snapshot not found: {snapshot}")
+    try:
+        start_date = date.fromisoformat(args.start_date)
+    except ValueError as exc:
+        raise SystemExit("--start-date must use YYYY-MM-DD") from exc
+
+    os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
+    os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
+
+    from app.models.ticker import Ticker
+    from app.services import backtest_service as bt
+    from app.services.admin_service import get_activation_config
+    from app.services.paper_trade_service import get_exit_policy
+    from app.services.recommendation_service import get_recommendation_config
+
+    engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
+    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    try:
+        async with Session() as db:
+            recommendation_config = await get_recommendation_config(db)
+            activation = await get_activation_config(db)
+            exit_config = await get_exit_policy(db)
+            benchmark_closes = await bt._load_benchmark_closes_for_backtest(
+                db, days=None, refresh=False
+            )
+            ticker_result = await db.execute(select(Ticker).order_by(Ticker.symbol))
+            symbols = [ticker.symbol for ticker in ticker_result.scalars().all()]
+            prices: dict[str, tuple] = {}
+            for index, symbol in enumerate(symbols, 1):
+                columns = await bt._fetch_columns(db, symbol)
+                if columns is not None:
+                    prices[symbol] = columns
+                if not args.quiet and index % 50 == 0:
+                    print(f"loaded prices: {index}/{len(symbols)}", flush=True)
+    finally:
+        await engine.dispose()
+
+    candidates: list[dict] = []
+    workers = max(1, min(int(args.workers), multiprocessing.cpu_count() - 1))
+    context = multiprocessing.get_context("spawn")
+    with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool:
+        futures = {
+            pool.submit(
+                bt._replay_candidates_for_period,
+                symbol,
+                columns,
+                recommendation_config,
+                activation,
+                benchmark_closes,
+                start_date,
+            ): symbol
+            for symbol, columns in prices.items()
+        }
+        for index, future in enumerate(as_completed(futures), 1):
+            candidates.extend(future.result())
+            if not args.quiet and index % 25 == 0:
+                print(f"replayed tickers: {index}/{len(futures)}", flush=True)
+
+    bt._assign_momentum_percentiles(candidates)
+    bt._assign_residual_momentum_percentiles(candidates)
+    bt._assign_low_volatility_percentiles(candidates)
+    bt._assign_activation_momentum_percentiles(candidates)
+    bt._assign_residual_high_vol_blend(candidates)
+    threshold = float(activation.get("min_momentum_percentile", 80.0))
+    for candidate in candidates:
+        candidate["qualified"] = bt._momentum_qualifies(candidate, threshold)
+
+    strategy = next(
+        row for row in bt.PORTFOLIO_MONITOR_STRATEGIES if row.get("is_production")
+    )
+    entry_config = bt._entry_variant_config(str(strategy["entry_variant"]))
+    if entry_config is None:
+        raise RuntimeError("Production entry configuration missing")
+    exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get(
+        str(exit_config.get("mode", "atr_trailing")), "atr_trail3"
+    )
+    hold_days = int(exit_config.get("hold_days", max(bt.TIME_EXIT_DAYS)))
+    trail_multiplier = float(
+        exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER)
+    )
+    sim_kwargs = {
+        "qualified_fn": None,
+        "ranking_key": str(
+            entry_config.get("ranking_key") or entry_config["percentile_key"]
+        ),
+        "max_positions": int(entry_config["max_positions"]),
+        "risk_per_trade": float(entry_config["risk_per_trade"]),
+        "atr_trail_multiplier": trail_multiplier,
+        "start_date": start_date,
+    }
+
+    baseline = bt._simulate_portfolio(
+        candidates, prices, benchmark_closes, exit_policy, hold_days, **sim_kwargs
+    )
+    cooldown_5 = bt._simulate_portfolio(
+        candidates,
+        prices,
+        benchmark_closes,
+        exit_policy,
+        hold_days,
+        reentry_cooldown_sessions=5,
+        **sim_kwargs,
+    )
+    cooldown_10 = bt._simulate_portfolio(
+        candidates,
+        prices,
+        benchmark_closes,
+        exit_policy,
+        hold_days,
+        reentry_cooldown_sessions=10,
+        **sim_kwargs,
+    )
+    refresher = GateStopRefresher(
+        prices, recommendation_config, activation, benchmark_closes
+    )
+    gate_protected = bt._simulate_portfolio(
+        candidates,
+        prices,
+        benchmark_closes,
+        exit_policy,
+        hold_days,
+        initial_stop_refresh_fn=refresher,
+        include_trades=True,
+        **sim_kwargs,
+    )
+    if any(row is None for row in (baseline, cooldown_5, cooldown_10, gate_protected)):
+        raise RuntimeError("A study arm produced no trades")
+
+    gate_trades = list(gate_protected.get("trade_details") or [])
+    event_counts = Counter()
+    for event in refresher.events:
+        event_counts["stop_touches"] += 1
+        if event["gate_passed"]:
+            event_counts["gate_passed"] += 1
+        if event["lower_stop"]:
+            event_counts["lower_stop"] += 1
+        if event["same_bar_survives"]:
+            event_counts["same_bar_survives"] += 1
+
+    report = {
+        "generated_at": datetime.now().astimezone().isoformat(),
+        "snapshot": str(snapshot.resolve()),
+        "period_start": start_date.isoformat(),
+        "tickers": len(prices),
+        "entry_candidates": len(candidates),
+        "qualified_candidates": sum(bool(row["qualified"]) for row in candidates),
+        "params": {
+            "entry_cadence_days": bt.STEP_DAYS,
+            "setup_stop_atr_multiplier": bt.ATR_MULTIPLIER,
+            "exit_policy": exit_policy,
+            "exit_atr_multiplier": trail_multiplier,
+            "hold_days": hold_days,
+            "momentum_percentile_floor": threshold,
+            "gate_refresh_information_cutoff": "previous close",
+        },
+        "arms": [
+            _arm("baseline", baseline),
+            _arm("cooldown_5", cooldown_5),
+            _arm("cooldown_10", cooldown_10),
+            _arm("gate_protected_stop", gate_protected),
+        ],
+        "gate_stop_events": {
+            **dict(event_counts),
+            "unique_symbols": len({event["symbol"] for event in refresher.events}),
+            "rescued_trade_outcomes": _rescued_trade_summary(gate_trades),
+            "events": refresher.events,
+        },
+        "note": (
+            "The gate-protected arm recalculates the gate at an initial-stop touch "
+            "using only data available through the previous close. It accepts only "
+            "a lower stop from a newly valid long setup and checks that replacement "
+            "against the same bar. It does not cancel stops using the later same-day close."
+        ),
+    }
+    output = Path(args.out) if args.out else _default_output_path()
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+
+    print(f"Report written: {output}")
+    for arm in report["arms"]:
+        print(
+            f"{arm['arm']}: Sharpe {arm['sharpe']}, CAGR {arm['cagr_pct']}%, "
+            f"DD {arm['max_drawdown_pct']}%, trades {arm['trades']}"
+        )
+    print(f"gate stop events: {dict(event_counts)}")
+    print(f"rescued outcomes: {report['gate_stop_events']['rescued_trade_outcomes']}")
+
+
+if __name__ == "__main__":
+    asyncio.run(_main())

scripts/run_post_stop_reentry_study.py

@@ -0,0 +1,541 @@
+"""Offline event study for stateful post-stop re-entry policies.
+
+Initial entries keep the validated weekly production cadence. After an initial
+stop, the affected ticker is evaluated on every subsequent daily close. This
+isolates the exact churn problem without changing the rest of the portfolio.
+All arms retain the hard stop, production position sizing, 3x ATR trail, and
+round-trip transaction costs.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import bisect
+import json
+import multiprocessing
+import os
+import pickle
+import sys
+from collections import Counter
+from concurrent.futures import ProcessPoolExecutor, as_completed
+from datetime import date, datetime
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+RECLAIM_ATR_BUFFER = 0.25
+
+
+def _sqlite_url(path: Path) -> str:
+    return f"sqlite+aiosqlite:///{path.resolve().as_posix()}"
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("snapshot")
+    parser.add_argument("--start-date", default="2024-07-01")
+    parser.add_argument("--workers", type=int, default=6)
+    parser.add_argument("--out", default=None)
+    parser.add_argument(
+        "--candidate-cache",
+        default=None,
+        help="Optional pickle cache for the expensive weekly candidate replay.",
+    )
+    parser.add_argument("--quiet", action="store_true")
+    parser.add_argument(
+        "--cooldowns",
+        type=int,
+        nargs="+",
+        default=None,
+        help=(
+            "Run an immediate baseline plus the given cooldown lengths instead "
+            "of the gate-reset policy study (for example: 3 5 7 10)."
+        ),
+    )
+    return parser.parse_args()
+
+
+def _default_output_path() -> Path:
+    stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    return Path("reports") / f"post-stop-reentry-{stamp}.json"
+
+
+class DailySetupEngine:
+    """Point-in-time daily setup and universe-rank cache."""
+
+    def __init__(
+        self,
+        prices: dict[str, tuple],
+        recommendation_config: dict,
+        activation: dict,
+        benchmark_closes: dict[date, float],
+    ) -> None:
+        from app.services import backtest_service as bt
+
+        self.bt = bt
+        self.prices = prices
+        self.recommendation_config = recommendation_config
+        self.activation = activation
+        self.benchmark_closes = benchmark_closes
+        self.threshold = float(activation.get("min_momentum_percentile", 80.0))
+        self.dates = {
+            symbol: [date.fromordinal(value) for value in columns[0]]
+            for symbol, columns in prices.items()
+        }
+        self.index_of = {
+            symbol: {value: index for index, value in enumerate(columns[0])}
+            for symbol, columns in prices.items()
+        }
+        self.rank_cache: dict[int, dict[str, tuple[float, float]]] = {}
+        self.candidate_cache: dict[tuple[str, int], dict | None] = {}
+        self.atr_cache: dict[tuple[str, int], float | None] = {}
+
+    @staticmethod
+    def _percentiles(values: dict[str, float]) -> dict[str, float]:
+        ordered = sorted(values, key=lambda symbol: values[symbol])
+        denominator = len(ordered) - 1
+        return {
+            symbol: (rank / denominator * 100.0) if denominator > 0 else 100.0
+            for rank, symbol in enumerate(ordered)
+        }
+
+    def _ranks(self, asof_ord: int) -> dict[str, tuple[float, float]]:
+        cached = self.rank_cache.get(asof_ord)
+        if cached is not None:
+            return cached
+
+        momentum_values: dict[str, float] = {}
+        volatility_values: dict[str, float] = {}
+        for symbol, columns in self.prices.items():
+            idx = bisect.bisect_right(columns[0], asof_ord) - 1
+            if idx < 0:
+                continue
+            closes = columns[4]
+            if idx >= 252:
+                momentum = self.bt._residual_momentum_12_1(
+                    self.dates[symbol], closes, idx, self.benchmark_closes
+                )
+                if momentum is None and closes[idx - 252] > 0:
+                    momentum = closes[idx - 21] / closes[idx - 252] - 1.0
+                if momentum is not None:
+                    momentum_values[symbol] = float(momentum)
+            volatility = self.bt._realized_vol_6m(closes, idx)
+            if volatility is not None:
+                volatility_values[symbol] = float(volatility)
+
+        momentum_pct = self._percentiles(momentum_values)
+        volatility_pct = self._percentiles(volatility_values)
+        ranks = {
+            symbol: (momentum_pct[symbol], volatility_pct.get(symbol, 0.0))
+            for symbol in momentum_pct
+        }
+        self.rank_cache[asof_ord] = ranks
+        return ranks
+
+    def atr(self, symbol: str, asof_ord: int) -> float | None:
+        key = (symbol, asof_ord)
+        if key in self.atr_cache:
+            return self.atr_cache[key]
+        columns = self.prices[symbol]
+        idx = self.index_of[symbol].get(asof_ord)
+        if idx is None:
+            idx = bisect.bisect_right(columns[0], asof_ord) - 1
+        if idx < 0:
+            self.atr_cache[key] = None
+            return None
+        try:
+            value = self.bt.compute_atr(
+                columns[2][: idx + 1],
+                columns[3][: idx + 1],
+                columns[4][: idx + 1],
+            )["atr"]
+            result = float(value) if value and value > 0 else None
+        except Exception:
+            result = None
+        self.atr_cache[key] = result
+        return result
+
+    def candidate(self, symbol: str, asof_ord: int) -> dict | None:
+        key = (symbol, asof_ord)
+        if key in self.candidate_cache:
+            cached = self.candidate_cache[key]
+            return dict(cached) if cached is not None else None
+
+        columns = self.prices[symbol]
+        idx = self.index_of[symbol].get(asof_ord)
+        if idx is None or idx < self.bt.MIN_LOOKBACK - 1:
+            self.candidate_cache[key] = None
+            return None
+        records = [
+            SimpleNamespace(
+                date=date.fromordinal(o),
+                open=op,
+                high=high,
+                low=low,
+                close=close,
+                volume=volume,
+            )
+            for o, op, high, low, close, volume in zip(
+                columns[0][: idx + 1],
+                columns[1][: idx + 1],
+                columns[2][: idx + 1],
+                columns[3][: idx + 1],
+                columns[4][: idx + 1],
+                columns[5][: idx + 1],
+            )
+        ]
+        setups = self.bt._window_setups(
+            records, self.recommendation_config, self.activation
+        )
+        setup = next((row for row in setups if row["direction"] == "long"), None)
+        rank = self._ranks(asof_ord).get(symbol)
+        gate_passed = bool(
+            setup is not None
+            and rank is not None
+            and self.bt._momentum_qualifies(
+                {
+                    "meets_core": setup["meets_core"],
+                    "direction": "long",
+                    self.bt.PRODUCTION_PERCENTILE_KEY: rank[0],
+                },
+                self.threshold,
+            )
+        )
+        if not gate_passed or setup is None or rank is None:
+            self.candidate_cache[key] = None
+            return None
+
+        strategy_rank = (
+            rank[0] * self.bt.STRATEGY_RANK_MOMENTUM_WEIGHT
+            + rank[1] * (1.0 - self.bt.STRATEGY_RANK_MOMENTUM_WEIGHT)
+        )
+        candidate = {
+            "symbol": symbol,
+            "date": date.fromordinal(asof_ord).isoformat(),
+            "direction": "long",
+            "entry": float(setup["entry"]),
+            "stop": float(setup["stop"]),
+            "target": float(setup["target"]),
+            "qualified": True,
+            self.bt.PRODUCTION_PERCENTILE_KEY: rank[0],
+            self.bt.RESIDUAL_HIGH_VOL_BLEND_80_20_KEY: strategy_rank,
+        }
+        self.candidate_cache[key] = candidate
+        return dict(candidate)
+
+
+class ReentryPolicy:
+    def __init__(self, name: str, engine: DailySetupEngine) -> None:
+        self.name = name
+        self.engine = engine
+        self.checks = 0
+        self.gate_passes = 0
+        self.emitted = Counter()
+
+    def __call__(
+        self,
+        symbol: str,
+        asof_ord: int,
+        state: dict,
+        bar: Any,
+    ) -> dict | None:
+        self.checks += 1
+        if "reentry_trigger" not in state:
+            stop_atr = self.engine.atr(symbol, state["stop_ord"])
+            state["reentry_trigger"] = (
+                state["stop_day_high"] + RECLAIM_ATR_BUFFER * stop_atr
+                if stop_atr is not None
+                else state["stop_day_high"]
+            )
+
+        candidate = self.engine.candidate(symbol, asof_ord)
+        if candidate is None:
+            state["gate_went_unqualified"] = True
+            return None
+        self.gate_passes += 1
+
+        reason: str | None = None
+        sessions = int(state["sessions_since_stop"])
+        if self.name == "immediate":
+            reason = "gate_still_or_again_qualified"
+        elif self.name.startswith("cooldown_"):
+            cooldown_sessions = int(self.name.removeprefix("cooldown_"))
+            if sessions >= cooldown_sessions:
+                reason = f"{cooldown_sessions}_session_cooldown_complete"
+        elif self.name == "gate_reset":
+            if state["gate_went_unqualified"]:
+                reason = "gate_failed_then_requalified"
+        elif self.name == "gate_reset_or_reclaim":
+            if state["gate_went_unqualified"]:
+                reason = "gate_failed_then_requalified"
+            elif (
+                bar.close > state["reentry_trigger"]
+                and float(candidate["stop"]) > state["previous_stop"]
+            ):
+                reason = "price_reclaim_with_improved_stop"
+        else:
+            raise ValueError(f"Unknown re-entry policy: {self.name}")
+
+        if reason is None:
+            return None
+        emitted = dict(candidate)
+        emitted["_reentry_reason"] = reason
+        self.emitted[reason] += 1
+        return emitted
+
+    def summary(self) -> dict:
+        return {
+            "daily_checks": self.checks,
+            "qualified_checks": self.gate_passes,
+            "emitted_by_reason": dict(self.emitted),
+        }
+
+
+def _trade_summary(trades: list[dict]) -> dict:
+    reentries = [trade for trade in trades if trade.get("is_reentry")]
+    waits = [
+        int(trade["reentry_wait_sessions"])
+        for trade in reentries
+        if trade.get("reentry_wait_sessions") is not None
+    ]
+    return {
+        "transaction_cost": round(
+            sum(float(trade["transaction_cost"]) for trade in trades), 2
+        ),
+        "reentry_trades": len(reentries),
+        "same_day_reentries": sum(wait == 0 for wait in waits),
+        "next_day_reentries": sum(wait == 1 for wait in waits),
+        "reentries_within_5_sessions": sum(wait <= 5 for wait in waits),
+        "avg_reentry_wait_sessions": round(sum(waits) / len(waits), 1)
+        if waits
+        else None,
+        "reentry_win_rate": round(
+            sum(float(trade["pnl"]) > 0 for trade in reentries)
+            / len(reentries)
+            * 100.0,
+            1,
+        )
+        if reentries
+        else None,
+        "reentry_total_pnl": round(
+            sum(float(trade["pnl"]) for trade in reentries), 2
+        ),
+    }
+
+
+async def _main() -> None:
+    args = _parse_args()
+    snapshot = Path(args.snapshot)
+    if not snapshot.exists():
+        raise SystemExit(f"Snapshot not found: {snapshot}")
+    try:
+        start_date = date.fromisoformat(args.start_date)
+    except ValueError as exc:
+        raise SystemExit("--start-date must use YYYY-MM-DD") from exc
+
+    os.environ["BACKTEST_SNAPSHOT_OFFLINE"] = "1"
+    os.environ["BACKTEST_ALLOW_SPAWN"] = "1"
+
+    from app.models.ticker import Ticker
+    from app.services import backtest_service as bt
+    from app.services.admin_service import get_activation_config
+    from app.services.paper_trade_service import get_exit_policy
+    from app.services.recommendation_service import get_recommendation_config
+
+    engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
+    Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
+    try:
+        async with Session() as db:
+            recommendation_config = await get_recommendation_config(db)
+            activation = await get_activation_config(db)
+            exit_config = await get_exit_policy(db)
+            benchmark_closes = await bt._load_benchmark_closes_for_backtest(
+                db, days=None, refresh=False
+            )
+            ticker_result = await db.execute(select(Ticker).order_by(Ticker.symbol))
+            symbols = [ticker.symbol for ticker in ticker_result.scalars().all()]
+            prices: dict[str, tuple] = {}
+            for index, symbol in enumerate(symbols, 1):
+                columns = await bt._fetch_columns(db, symbol)
+                if columns is not None:
+                    prices[symbol] = columns
+                if not args.quiet and index % 50 == 0:
+                    print(f"loaded prices: {index}/{len(symbols)}", flush=True)
+    finally:
+        await engine.dispose()
+
+    cache_path = Path(args.candidate_cache) if args.candidate_cache else None
+    snapshot_stat = snapshot.stat()
+    cache_key = {
+        "snapshot": str(snapshot.resolve()),
+        "snapshot_size": snapshot_stat.st_size,
+        "snapshot_mtime_ns": snapshot_stat.st_mtime_ns,
+        "start_date": start_date.isoformat(),
+    }
+    candidates: list[dict]
+    if cache_path is not None and cache_path.exists():
+        with cache_path.open("rb") as handle:
+            cached_replay = pickle.load(handle)  # noqa: S301 - trusted local cache
+        if cached_replay.get("key") != cache_key:
+            raise SystemExit(f"Candidate cache does not match this run: {cache_path}")
+        candidates = list(cached_replay["candidates"])
+        if not args.quiet:
+            print(f"loaded candidate cache: {cache_path}", flush=True)
+    else:
+        candidates = []
+        workers = max(1, min(int(args.workers), multiprocessing.cpu_count() - 1))
+        context = multiprocessing.get_context("spawn")
+        with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool:
+            futures = {
+                pool.submit(
+                    bt._replay_candidates_for_period,
+                    symbol,
+                    columns,
+                    recommendation_config,
+                    activation,
+                    benchmark_closes,
+                    start_date,
+                ): symbol
+                for symbol, columns in prices.items()
+            }
+            for index, future in enumerate(as_completed(futures), 1):
+                candidates.extend(future.result())
+                if not args.quiet and index % 25 == 0:
+                    print(f"replayed tickers: {index}/{len(futures)}", flush=True)
+        if cache_path is not None:
+            cache_path.parent.mkdir(parents=True, exist_ok=True)
+            with cache_path.open("wb") as handle:
+                pickle.dump(
+                    {"key": cache_key, "candidates": candidates},
+                    handle,
+                    protocol=pickle.HIGHEST_PROTOCOL,
+                )
+            if not args.quiet:
+                print(f"wrote candidate cache: {cache_path}", flush=True)
+
+    bt._assign_momentum_percentiles(candidates)
+    bt._assign_residual_momentum_percentiles(candidates)
+    bt._assign_low_volatility_percentiles(candidates)
+    bt._assign_activation_momentum_percentiles(candidates)
+    bt._assign_residual_high_vol_blend(candidates)
+    threshold = float(activation.get("min_momentum_percentile", 80.0))
+    for candidate in candidates:
+        candidate["qualified"] = bt._momentum_qualifies(candidate, threshold)
+
+    strategy = next(
+        row for row in bt.PORTFOLIO_MONITOR_STRATEGIES if row.get("is_production")
+    )
+    entry_config = bt._entry_variant_config(str(strategy["entry_variant"]))
+    if entry_config is None:
+        raise RuntimeError("Production entry configuration missing")
+    exit_policy = bt.LIVE_EXIT_MODE_TO_SIM.get(
+        str(exit_config.get("mode", "atr_trailing")), "atr_trail3"
+    )
+    hold_days = int(exit_config.get("hold_days", max(bt.TIME_EXIT_DAYS)))
+    trail_multiplier = float(
+        exit_config.get("atr_multiplier", bt.ATR_TRAIL_MULTIPLIER)
+    )
+    sim_kwargs = {
+        "ranking_key": str(
+            entry_config.get("ranking_key") or entry_config["percentile_key"]
+        ),
+        "max_positions": int(entry_config["max_positions"]),
+        "risk_per_trade": float(entry_config["risk_per_trade"]),
+        "atr_trail_multiplier": trail_multiplier,
+        "start_date": start_date,
+        "include_trades": True,
+    }
+
+    daily_engine = DailySetupEngine(
+        prices, recommendation_config, activation, benchmark_closes
+    )
+    if args.cooldowns is None:
+        policy_names = (
+            "immediate",
+            "cooldown_5",
+            "gate_reset",
+            "gate_reset_or_reclaim",
+        )
+    else:
+        cooldowns = sorted(set(args.cooldowns))
+        if any(value < 1 for value in cooldowns):
+            raise SystemExit("--cooldowns values must be positive integers")
+        policy_names = ("immediate", *(f"cooldown_{value}" for value in cooldowns))
+
+    arms: list[dict] = []
+    for policy_name in policy_names:
+        policy = ReentryPolicy(policy_name, daily_engine)
+        sim = bt._simulate_portfolio(
+            candidates,
+            prices,
+            benchmark_closes,
+            exit_policy,
+            hold_days,
+            post_stop_reentry_fn=policy,
+            **sim_kwargs,
+        )
+        if sim is None:
+            raise RuntimeError(f"Policy {policy_name} produced no trades")
+        trades = list(sim.pop("trade_details"))
+        arms.append({
+            "arm": policy_name,
+            **sim,
+            "turnover": _trade_summary(trades),
+            "policy": policy.summary(),
+            "trade_details": trades,
+        })
+
+    output = Path(args.out) if args.out else _default_output_path()
+    report = {
+        "generated_at": datetime.now().astimezone().isoformat(),
+        "snapshot": str(snapshot.resolve()),
+        "period_start": start_date.isoformat(),
+        "tickers": len(prices),
+        "entry_candidates": len(candidates),
+        "qualified_candidates": sum(bool(row["qualified"]) for row in candidates),
+        "params": {
+            "initial_entry_cadence_days": bt.STEP_DAYS,
+            "post_stop_evaluation_cadence_days": 1,
+            "setup_stop_atr_multiplier": bt.ATR_MULTIPLIER,
+            "exit_policy": exit_policy,
+            "exit_atr_multiplier": trail_multiplier,
+            "hold_days": hold_days,
+            "cost_per_side_pct": bt.COST_PER_SIDE * 100.0,
+            "momentum_percentile_floor": threshold,
+            "reclaim_atr_buffer": RECLAIM_ATR_BUFFER,
+            "cooldown_sessions": cooldowns if args.cooldowns is not None else None,
+        },
+        "arms": arms,
+        "note": (
+            "Initial opportunities retain the validated weekly replay cadence. "
+            "Only tickers stopped at their initial stop switch to daily evaluation, "
+            "which isolates next-day/same-episode re-entry churn. A cooldown of N "
+            "sessions permits the first re-entry at wait_sessions=N. Gate reset "
+            "requires at least one unqualified daily close before requalification. "
+            "The reclaim arm alternatively accepts a close above stop-day high + "
+            "0.25 ATR only when the new setup stop is above the prior stop."
+        ),
+    }
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
+
+    print(f"Report written: {output}")
+    for arm in arms:
+        turnover = arm["turnover"]
+        print(
+            f"{arm['arm']}: Sharpe {arm['sharpe']}, CAGR {arm['cagr_pct']}%, "
+            f"DD {arm['max_drawdown_pct']}%, trades {arm['trades']}, "
+            f"reentries {turnover['reentry_trades']}, fees ${turnover['transaction_cost']}"
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(_main())

tests/unit/test_backtest_service.py

@@ -575,6 +575,228 @@ class TestSimulatePortfolio:
         assert sim["trades"] == 1
         assert sim["worst_trade_r"] == pytest.approx(-2.0)  # (90 − 100) / 5
 
+    def test_initial_stop_cooldown_blocks_immediate_reentry(self):
+        closes = [100.0, 94.0, 96.0]
+        prices = {"AAA": _sim_prices(self.ORD, closes)}
+        candidates = [
+            _sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=120.0),
+            _sim_cand("AAA", self.ORD + 1, entry=94.0, stop=89.0, target=110.0),
+        ]
+
+        baseline = bt._simulate_portfolio(candidates, prices, None, "hold", 30)
+        cooldown = bt._simulate_portfolio(
+            candidates,
+            prices,
+            None,
+            "hold",
+            30,
+            reentry_cooldown_sessions=5,
+        )
+
+        assert baseline is not None and baseline["trades"] == 2
+        assert cooldown is not None and cooldown["trades"] == 1
+        assert cooldown["skipped_cooldown"] == 1
+        assert cooldown["reentry_cooldown_sessions"] == 5
+
+    def test_initial_stop_cooldown_unlocks_exactly_after_session_five(self):
+        closes = [100.0, 94.0, 96.0, 96.0, 96.0, 96.0, 97.0, 98.0]
+        prices = {"AAA": _sim_prices(self.ORD, closes)}
+        candidates = [
+            _sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=120.0),
+            # Four completed sessions since the stop: still locked.
+            _sim_cand("AAA", self.ORD + 5, entry=96.0, stop=90.0, target=115.0),
+            # Five completed sessions since the stop: first permitted re-entry.
+            _sim_cand("AAA", self.ORD + 6, entry=97.0, stop=90.0, target=118.0),
+        ]
+
+        sim = bt._simulate_portfolio(
+            candidates,
+            prices,
+            None,
+            "hold",
+            30,
+            reentry_cooldown_sessions=5,
+            include_trades=True,
+        )
+
+        assert sim is not None
+        assert sim["trades"] == 2
+        assert sim["skipped_cooldown"] == 1
+        assert sim["trade_details"][1]["entry_date"] == date.fromordinal(
+            self.ORD + 6
+        ).isoformat()
+
+    def test_production_monitor_applies_live_reentry_lockdown(self, monkeypatch):
+        def fake_simulator(*_args, **kwargs):
+            return {
+                "trades": 0,
+                "applied_reentry_lockdown": kwargs.get("reentry_cooldown_sessions", 0),
+            }
+
+        monkeypatch.setattr(bt, "_simulate_portfolio", fake_simulator)
+        market_ord = date(2026, 7, 1).toordinal()
+        prices = {"AAA": ([market_ord], [], [], [], [], [])}
+
+        monitor = bt._portfolio_monitor([], prices, None, 30)
+        production_rows = [
+            row for row in monitor["runs"] if row["is_production"]
+        ]
+        comparison_rows = [
+            row for row in monitor["runs"] if not row["is_production"]
+        ]
+
+        assert production_rows
+        assert all(
+            row["reentry_lockdown_sessions"] == bt.REENTRY_LOCKDOWN_SESSIONS
+            and row["applied_reentry_lockdown"] == bt.REENTRY_LOCKDOWN_SESSIONS
+            for row in production_rows
+        )
+        assert comparison_rows
+        assert all(
+            row["reentry_lockdown_sessions"] == 0
+            and row["applied_reentry_lockdown"] == 0
+            for row in comparison_rows
+        )
+
+    def test_production_cadence_comparison_names_exact_two_arms(self):
+        monitor = {
+            "runs": [
+                {
+                    "comparison_arm": "live_no_lockdown",
+                    "lookback": "all",
+                    "reentry_lockdown_sessions": 0,
+                    "trades": 10,
+                    "equity_curve": [{"date": "2026-01-01", "value": 1.0}],
+                },
+                {
+                    "comparison_arm": "live_lockdown_5",
+                    "lookback": "all",
+                    "reentry_lockdown_sessions": 5,
+                    "trades": 8,
+                    "benchmark_curve": [{"date": "2026-01-01", "value": 1.0}],
+                },
+            ]
+        }
+
+        comparison = bt._production_cadence_comparison(monitor, "daily")
+
+        assert comparison is not None
+        assert [row["arm"] for row in comparison["arms"]] == [
+            "prod_live_setup_daily",
+            "cooldown_5_daily",
+        ]
+        assert all("equity_curve" not in row for row in comparison["arms"])
+        assert all("benchmark_curve" not in row for row in comparison["arms"])
+
+    def test_initial_stop_can_refresh_lower_and_survive_same_bar(self):
+        closes = [100.0, 94.0, 96.0]
+        prices = {"AAA": _sim_prices(self.ORD, closes)}
+        candidate = _sim_cand(
+            "AAA", self.ORD, entry=100.0, stop=95.0, target=120.0
+        )
+
+        sim = bt._simulate_portfolio(
+            [candidate],
+            prices,
+            None,
+            "hold",
+            2,
+            initial_stop_refresh_fn=lambda *_: 90.0,
+            include_trades=True,
+        )
+
+        assert sim is not None
+        assert sim["stop_refresh_attempts"] == 1
+        assert sim["stop_refreshes"] == 1
+        assert sim["stop_refresh_same_bar_hits"] == 0
+        assert sim["exit_reasons"] == {"time": 1}
+        assert sim["trade_details"][0]["stop_refreshes"] == 1
+
+    def test_refreshed_stop_is_checked_against_same_bar(self):
+        ords = list(range(self.ORD, self.ORD + 2))
+        prices = {
+            "AAA": (
+                ords,
+                [100.0, 94.0],
+                [101.0, 96.0],
+                [99.0, 89.0],
+                [100.0, 94.0],
+                [1, 1],
+            )
+        }
+        candidate = _sim_cand(
+            "AAA", self.ORD, entry=100.0, stop=95.0, target=120.0
+        )
+
+        sim = bt._simulate_portfolio(
+            [candidate],
+            prices,
+            None,
+            "hold",
+            30,
+            initial_stop_refresh_fn=lambda *_: 90.0,
+        )
+
+        assert sim is not None
+        assert sim["stop_refresh_same_bar_hits"] == 1
+        assert sim["worst_trade_r"] == pytest.approx(-2.0)
+
+    def test_post_stop_state_suppresses_same_episode_candidate(self):
+        closes = [100.0, 94.0, 96.0]
+        prices = {"AAA": _sim_prices(self.ORD, closes)}
+        candidates = [
+            _sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=120.0),
+            _sim_cand("AAA", self.ORD + 1, entry=94.0, stop=89.0, target=110.0),
+        ]
+
+        sim = bt._simulate_portfolio(
+            candidates,
+            prices,
+            None,
+            "hold",
+            30,
+            post_stop_reentry_fn=lambda *_: None,
+        )
+
+        assert sim is not None
+        assert sim["trades"] == 1
+        assert sim["post_stop_events"] == 1
+        assert sim["post_stop_reentries"] == 0
+        assert sim["post_stop_states_open_at_end"] == 1
+
+    def test_post_stop_callback_can_reenter_same_day(self):
+        closes = [100.0, 94.0, 96.0]
+        prices = {"AAA": _sim_prices(self.ORD, closes)}
+        initial = _sim_cand(
+            "AAA", self.ORD, entry=100.0, stop=95.0, target=120.0
+        )
+
+        def immediate_reentry(sym, current_ord, _state, bar):
+            return _sim_cand(
+                sym,
+                current_ord,
+                entry=bar.close,
+                stop=bar.close - 5.0,
+                target=bar.close + 15.0,
+            )
+
+        sim = bt._simulate_portfolio(
+            [initial],
+            prices,
+            None,
+            "hold",
+            30,
+            post_stop_reentry_fn=immediate_reentry,
+            include_trades=True,
+        )
+
+        assert sim is not None
+        assert sim["trades"] == 2
+        assert sim["post_stop_reentries"] == 1
+        assert sim["reentry_events"][0]["wait_sessions"] == 0
+        assert sim["trade_details"][1]["is_reentry"] is True
+        assert sim["trade_details"][1]["reentry_wait_sessions"] == 0
+
     def test_sma50_policy_exits_on_close_break(self):
         closes = [100.0] * 56 + [90.0, 91.0]
         prices = {"AAA": _sim_prices(self.ORD, closes)}
@@ -722,6 +944,7 @@ def test_build_recommendation_prefers_production_monitor_headline():
     })
     assert rec["headline"] is not None
     assert "3x ATR trailing exit" in rec["headline"]
+    assert "5-session re-entry lockdown" in rec["headline"]
     assert any(item["topic"] == "production" for item in rec["items"])
 
 
@@ -736,6 +959,15 @@ def test_backtest_target_model_is_small_and_validated():
         bt.validate_backtest_target_model("legacy_range_grid_touch")
 
 
+def test_backtest_cadence_is_small_validated_and_session_based():
+    assert bt.validate_backtest_cadence(" WEEKLY ") == "weekly"
+    assert bt.validate_backtest_cadence("daily") == "daily"
+    assert bt.backtest_step_sessions("weekly") == 5
+    assert bt.backtest_step_sessions("daily") == 1
+    with pytest.raises(ValueError, match="Unknown backtest cadence"):
+        bt.validate_backtest_cadence("monthly")
+
+
 def _flat_window_records():
     return [
         SimpleNamespace(
@@ -829,6 +1061,46 @@ def test_replay_ticker_candidates_carry_gate_fields():
         assert c.get("action") is not None
         assert "risk_level" in c
         assert c["target_model"] == bt.PRODUCTION_GTL_TARGET_MODEL
+        assert c["ranking_period"][0] == "week"
+
+    daily_cands = bt._replay_ticker(
+        "OSC",
+        bars,
+        dict(DEFAULT_RECOMMENDATION_CONFIG),
+        dict(ACTIVATION_DEFAULTS),
+        cadence="daily",
+    )
+    assert len(daily_cands) > len(cands)
+    assert all(c["ranking_period"][0] == "date" for c in daily_cands)
+
+
+def test_daily_replay_uses_exact_date_ranking_periods():
+    candidates = [
+        {
+            "iso_week": (2026, 1),
+            "ranking_period": ("date", date(2026, 1, 5).toordinal()),
+            "momentum": 0.10,
+        },
+        {
+            "iso_week": (2026, 1),
+            "ranking_period": ("date", date(2026, 1, 5).toordinal()),
+            "momentum": 0.20,
+        },
+        {
+            "iso_week": (2026, 1),
+            "ranking_period": ("date", date(2026, 1, 6).toordinal()),
+            "momentum": 0.90,
+        },
+        {
+            "iso_week": (2026, 1),
+            "ranking_period": ("date", date(2026, 1, 6).toordinal()),
+            "momentum": 0.30,
+        },
+    ]
+
+    bt._assign_momentum_percentiles(candidates)
+
+    assert [row["momentum_percentile"] for row in candidates] == [0.0, 100.0, 100.0, 0.0]
 
 
 async def _seed_oscillating_ticker(session, symbol: str, n: int = 160) -> None:
@@ -870,12 +1142,23 @@ async def test_run_backtest_smoke(session):
     assert report["params"]["cost_per_side_pct"] == pytest.approx(bt.COST_PER_SIDE * 100)
     assert report["params"]["target_model"] == bt.PRODUCTION_GTL_TARGET_MODEL
     assert report["params"]["is_production_target_model"] is True
+    assert report["params"]["entry_cadence"] == "weekly"
+    assert report["params"]["step_sessions"] == 5
+    assert (
+        report["params"]["production_reentry_lockdown_sessions"]
+        == bt.REENTRY_LOCKDOWN_SESSIONS
+    )
     assert "net_avg_r" in report["overall_all"]
 
     # ablation baseline reproduces the qualified set exactly, and every row
     # carries the hold-to-horizon grading alongside the target model
     ablation = {r["variant"]: r for r in report["gate_ablation"]}
     assert ablation["all_floors"]["total"] == report["overall_qualified"]["total"]
+
+    daily_report = await bt.run_backtest(session, cadence="daily")
+    assert daily_report["params"]["entry_cadence"] == "daily"
+    assert daily_report["params"]["step_sessions"] == 1
+    assert daily_report["candidates"] > report["candidates"]
     for row in report["gate_ablation"]:
         assert "hold_net_avg_r" in row
 

tests/unit/test_paper_trade_service.py

@@ -48,6 +48,74 @@ async def test_create_and_list_open(session):
     assert row["current_price"] == 110.0  # marked to the latest close
 
 
+async def test_create_trade_enforces_post_stop_lockdown_at_service_boundary(session):
+    blocked_id = await _seed(session, "LOCKQ", close=100.0)
+    released_id = await _seed(session, "FREEQ", close=100.0)
+    today = date.today()
+    market_sessions = [
+        today - timedelta(days=8),
+        today - timedelta(days=7),
+        today - timedelta(days=6),
+        today - timedelta(days=3),
+        today - timedelta(days=2),
+        today - timedelta(days=1),
+    ]
+    for market_date in market_sessions:
+        session.add(BenchmarkPrice(symbol="SPY", date=market_date, close=400.0))
+
+    def stopped_trade(ticker_id: int, closed_on: date) -> PaperTrade:
+        return PaperTrade(
+            user_id=1,
+            ticker_id=ticker_id,
+            direction="long",
+            entry_price=100.0,
+            shares=10.0,
+            stop_loss=95.0,
+            target=115.0,
+            status="closed",
+            opened_at=datetime.combine(
+                closed_on - timedelta(days=1), datetime.min.time(), tzinfo=timezone.utc
+            ),
+            close_price=95.0,
+            closed_at=datetime.combine(
+                closed_on, datetime.min.time(), tzinfo=timezone.utc
+            ),
+            close_reason="stop",
+        )
+
+    session.add_all(
+        [
+            stopped_trade(blocked_id, market_sessions[1]),
+            stopped_trade(released_id, market_sessions[0]),
+        ]
+    )
+    await session.commit()
+
+    with pytest.raises(ValidationError, match="1 market session remaining"):
+        await svc.create_trade(
+            session,
+            1,
+            symbol="LOCKQ",
+            direction="long",
+            entry_price=100.0,
+            shares=10.0,
+            stop_loss=95.0,
+            target=115.0,
+        )
+
+    trade = await svc.create_trade(
+        session,
+        1,
+        symbol="FREEQ",
+        direction="long",
+        entry_price=100.0,
+        shares=10.0,
+        stop_loss=95.0,
+        target=115.0,
+    )
+    assert trade.ticker_id == released_id
+
+
 async def test_close_uses_current_price(session):
     await _seed(session, "AAA", close=112.0)
     trade = await svc.create_trade(session, 1, symbol="AAA", direction="long",

tests/unit/test_rr_scanner_preservation.py

@@ -20,6 +20,7 @@ from hypothesis import given, settings, HealthCheck, strategies as st
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app.models.benchmark_price import BenchmarkPrice
 from app.models.ohlcv import OHLCVRecord
 from app.models.paper_trade import PaperTrade
 from app.models.signal_context_snapshot import SignalContextSnapshot
@@ -607,6 +608,123 @@ async def test_get_trade_setups_can_exclude_tickers_with_open_paper_trades(
     assert [row["symbol"] for row in ticker_rows] == ["OPENQ"]
 
 
+@pytest.mark.asyncio
+async def test_get_trade_setups_applies_five_session_initial_stop_lockdown(
+    db_session: AsyncSession,
+):
+    now = datetime.now(timezone.utc)
+    today = now.date()
+    if await db_session.get(User, 1) is None:
+        db_session.add(
+            User(id=1, username="u", password_hash="x", role="user", has_access=True)
+        )
+        await db_session.flush()
+
+    blocked = Ticker(symbol="STOP4")
+    released = Ticker(symbol="STOP5")
+    trailing = Ticker(symbol="TRAILQ")
+    db_session.add_all([blocked, released, trailing])
+    await db_session.flush()
+
+    # Six SPY sessions D0..D5 form the canonical market calendar. A stop on
+    # D0 has five later sessions and is released; a stop on D1 has only four.
+    market_sessions = [
+        today - timedelta(days=8),
+        today - timedelta(days=7),
+        today - timedelta(days=6),
+        today - timedelta(days=3),
+        today - timedelta(days=2),
+        today - timedelta(days=1),
+    ]
+    for market_date in market_sessions:
+        db_session.add(
+            BenchmarkPrice(
+                symbol="SPY",
+                date=market_date,
+                close=400.0,
+            )
+        )
+    # A bar from an unrelated/scanner-specific calendar must not release the
+    # ticker one session early. The old universe-wide DISTINCT query did.
+    db_session.add(
+        OHLCVRecord(
+            ticker_id=blocked.id,
+            date=today,
+            open=100.0,
+            high=101.0,
+            low=99.0,
+            close=100.0,
+            volume=1_000,
+        )
+    )
+
+    for ticker in (blocked, released, trailing):
+        db_session.add(
+            TradeSetup(
+                ticker_id=ticker.id,
+                direction="long",
+                entry_price=100.0,
+                stop_loss=95.0,
+                target=115.0,
+                rr_ratio=3.0,
+                composite_score=80.0,
+                detected_at=now,
+            )
+        )
+
+    def closed_trade(ticker: Ticker, closed_on: date, reason: str) -> PaperTrade:
+        return PaperTrade(
+            user_id=1,
+            ticker_id=ticker.id,
+            direction="long",
+            entry_price=100.0,
+            shares=10.0,
+            stop_loss=95.0,
+            target=115.0,
+            status="closed",
+            opened_at=datetime.combine(
+                closed_on - timedelta(days=1), datetime.min.time(), tzinfo=timezone.utc
+            ),
+            close_price=95.0,
+            closed_at=datetime.combine(
+                closed_on, datetime.min.time(), tzinfo=timezone.utc
+            ),
+            close_reason=reason,
+        )
+
+    db_session.add_all(
+        [
+            closed_trade(blocked, market_sessions[1], "stop"),
+            closed_trade(released, market_sessions[0], "stop"),
+            closed_trade(trailing, market_sessions[-1], "trailing"),
+        ]
+    )
+    await db_session.flush()
+
+    default_symbols = {
+        row["symbol"] for row in await get_trade_setups(db_session)
+    }
+    assert {"STOP4", "STOP5", "TRAILQ"}.issubset(default_symbols)
+
+    available_symbols = {
+        row["symbol"]
+        for row in await get_trade_setups(
+            db_session,
+            exclude_reentry_lockdown_tickers=True,
+        )
+    }
+    assert "STOP4" not in available_symbols
+    assert {"STOP5", "TRAILQ"}.issubset(available_symbols)
+
+    annotated = await get_trade_setups(
+        db_session,
+        symbol="STOP4",
+        include_reentry_lockdown=True,
+    )
+    assert len(annotated) == 1
+    assert annotated[0]["reentry_lockdown_remaining_sessions"] == 1
+
+
 async def _seed_stale_setup_with_current_scores(db_session: AsyncSession) -> TradeSetup:
     """Stored setup frozen at scan time (conf 82, neutral) vs. current context
     (bullish sentiment, composite 96) that yields live confidence 97.

tests/unit/test_scheduler.py

@@ -3,11 +3,13 @@
 import pytest
 
 from app.scheduler import (
+    _consume_backtest_options,
     _consume_backtest_target_model,
     _parse_frequency,
     _resume_tickers,
     _last_successful,
     configure_scheduler,
+    queue_backtest_options,
     queue_backtest_target_model,
     scheduler,
 )
@@ -24,6 +26,15 @@ def test_manual_backtest_target_model_rejects_removed_research_arms():
         queue_backtest_target_model("production_control")
 
 
+def test_manual_backtest_options_are_one_shot_and_default_back_to_weekly():
+    assert queue_backtest_options("structural_sr", "daily") == (
+        "structural_sr",
+        "daily",
+    )
+    assert _consume_backtest_options() == ("structural_sr", "daily")
+    assert _consume_backtest_options() == ("production_gtl", "weekly")
+
+
 class TestParseFrequency:
     def test_hourly(self):
         assert _parse_frequency("hourly") == {"hours": 1}