feat: add daily reentry policy matrix

app/services/backtest_service.py

@@ -1405,6 +1405,7 @@ def _simulate_portfolio(
     max_positions: int = SIM_MAX_POSITIONS,
     risk_per_trade: float = SIM_RISK_PER_TRADE,
     atr_trail_multiplier: float = ATR_TRAIL_MULTIPLIER,
+    cost_per_side: float = COST_PER_SIDE,
     reentry_cooldown_sessions: int = 0,
     initial_stop_refresh_fn: (
         Callable[[str, int, float, dict, Any], float | None] | None
@@ -1436,8 +1437,12 @@ def _simulate_portfolio(
     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.
+    there is nothing to trade. ``cost_per_side`` is charged on entry and exit
+    and therefore changes both cash availability and subsequent position sizing.
     """
+    cost_rate = float(cost_per_side)
+    if not 0.0 <= cost_rate < 1.0:
+        raise ValueError("cost_per_side must be between 0 (inclusive) and 1")
     if qualified_fn is None:
         def _default_qualified(c: dict) -> bool:
             return bool(c.get("qualified"))
@@ -1565,7 +1570,7 @@ def _simulate_portfolio(
         nonlocal cash
         pos = positions.pop(sym)
         proceeds = pos["shares"] * fill
-        cost = proceeds * COST_PER_SIDE
+        cost = proceeds * cost_rate
         cash += proceeds - cost
         risk = pos["entry"] - pos["initial_stop"]
         trades.append({
@@ -1641,6 +1646,7 @@ def _simulate_portfolio(
                             "exit_fill": float(fill),
                             "previous_entry": float(closed_pos["entry"]),
                             "previous_stop": float(closed_pos["initial_stop"]),
+                            "previous_rank": closed_pos["entry_rank"],
                             "gate_went_unqualified": False,
                         }
                     continue
@@ -1677,7 +1683,9 @@ def _simulate_portfolio(
         equity = _marked_equity()
         fixed_todays = list(entries_by_ord.get(o, ()))
         reentry_todays: list[dict] = []
-        if post_stop_reentry_fn is not None:
+        if post_stop_reentry_fn is not None and (
+            end_ord is None or o < end_ord
+        ):
             fixed_todays = [
                 candidate
                 for candidate in fixed_todays
@@ -1718,11 +1726,11 @@ def _simulate_portfolio(
             shares = min(
                 (equity * risk_per_trade) / risk_ps,
                 (equity * SIM_NOTIONAL_CAP) / entry,
-                max(cash, 0.0) / (entry * (1.0 + COST_PER_SIDE)),
+                max(cash, 0.0) / (entry * (1.0 + cost_rate)),
             )
             if shares * entry < 1.0:  # can't fund a meaningful position
                 continue
-            entry_cost = shares * entry * COST_PER_SIDE
+            entry_cost = shares * entry * cost_rate
             cash -= shares * entry + entry_cost
             is_reentry = bool(c.get("_post_stop_reentry"))
             reentry_wait_sessions: int | None = None
@@ -1748,6 +1756,9 @@ def _simulate_portfolio(
                 "bars_held": 0,
                 "last_close": entry,
                 "highest_close": entry,
+                "entry_rank": (
+                    float(c[ranking_key]) if c.get(ranking_key) is not None else None
+                ),
                 "stop_refreshes": 0,
                 "is_reentry": is_reentry,
                 "reentry_wait_sessions": reentry_wait_sessions,
@@ -1856,6 +1867,7 @@ def _simulate_portfolio(
 
     result = {
         "starting_capital": SIM_STARTING_CAPITAL,
+        "cost_per_side_pct": round(cost_rate * 100.0, 3),
         "final_equity": round(final_equity, 2),
         "total_return_pct": round(total_return_pct, 1),
         "cagr_pct": round(cagr_pct, 1) if cagr_pct is not None else None,

scripts/run_daily_reentry_matrix.py

@@ -0,0 +1,597 @@
+"""Run the full daily post-stop re-entry study from one candidate replay.
+
+The expensive point-in-time setup replay and cross-sectional ranking happen
+once.  Every policy, lookback, transaction-cost, capacity, and holdout arm then
+uses that same qualified daily candidate set, so differences come only from the
+portfolio/re-entry rules being compared.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import copy
+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 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))
+
+POLICY_NAMES = (
+    "immediate",
+    "next_session",
+    "cooldown_5",
+    "gate_reset",
+    "gate_reset_improved",
+    "two_session_confirmation",
+)
+CACHE_VERSION = "daily-reentry-matrix-v1"
+
+
+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", help="SQLite backtest snapshot.")
+    parser.add_argument(
+        "--start-date",
+        default=None,
+        help="Optional earliest replay/simulation date (YYYY-MM-DD).",
+    )
+    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. It stores only the ranked, production-qualified "
+            "daily candidates, not the much larger raw replay."
+        ),
+    )
+    parser.add_argument(
+        "--policies",
+        nargs="+",
+        choices=POLICY_NAMES,
+        default=list(POLICY_NAMES),
+    )
+    parser.add_argument("--base-cost-per-side-pct", type=float, default=0.1)
+    parser.add_argument("--base-capacity", type=int, default=10)
+    parser.add_argument(
+        "--costs-per-side-pct",
+        type=float,
+        nargs="+",
+        default=[0.1, 0.2, 0.3],
+    )
+    parser.add_argument(
+        "--capacities", type=int, nargs="+", default=[5, 10, 15]
+    )
+    parser.add_argument(
+        "--holdout-split",
+        default="2025-01-01",
+        help="Train/test split date (YYYY-MM-DD), or 'none' to disable.",
+    )
+    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"daily-reentry-matrix-{stamp}.json"
+
+
+class PrecomputedDailyEngine:
+    """Exact date/symbol lookup over the already-ranked production gate."""
+
+    def __init__(self, qualified_candidates: list[dict]) -> None:
+        self.by_key = {
+            (row["symbol"], date.fromisoformat(row["date"]).toordinal()): row
+            for row in qualified_candidates
+        }
+
+    def candidate(self, symbol: str, asof_ord: int) -> dict | None:
+        row = self.by_key.get((symbol, asof_ord))
+        return dict(row) if row is not None else None
+
+
+class ReentryPolicy:
+    """Stateful policy evaluated after every initial-stop exit."""
+
+    def __init__(
+        self,
+        name: str,
+        engine: PrecomputedDailyEngine,
+        ranking_key: str,
+    ) -> None:
+        if name not in POLICY_NAMES:
+            raise ValueError(f"Unknown re-entry policy: {name}")
+        self.name = name
+        self.engine = engine
+        self.ranking_key = ranking_key
+        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
+        sessions = int(state["sessions_since_stop"])
+        candidate = self.engine.candidate(symbol, asof_ord)
+        if candidate is None:
+            state["gate_went_unqualified"] = True
+            state["qualified_streak"] = 0
+            return None
+        self.gate_passes += 1
+
+        # Two-session confirmation means two complete post-stop closes. The
+        # stop day's close (sessions=0) deliberately does not count.
+        if self.name == "two_session_confirmation" and sessions == 0:
+            state["qualified_streak"] = 0
+            return None
+        state["qualified_streak"] = int(state.get("qualified_streak", 0)) + 1
+
+        reason: str | None = None
+        if self.name == "immediate":
+            reason = "gate_still_or_again_qualified"
+        elif self.name == "next_session":
+            if sessions >= 1:
+                reason = "stop_day_block_complete"
+        elif self.name == "cooldown_5":
+            if sessions >= 5:
+                reason = "5_session_cooldown_complete"
+        elif self.name == "gate_reset":
+            if state["gate_went_unqualified"]:
+                reason = "gate_failed_then_requalified"
+        elif self.name == "gate_reset_improved":
+            previous_rank = state.get("previous_rank")
+            current_rank = candidate.get(self.ranking_key)
+            rank_not_weaker = (
+                current_rank is not None
+                and (
+                    previous_rank is None
+                    or float(current_rank) >= float(previous_rank)
+                )
+            )
+            if (
+                state["gate_went_unqualified"]
+                and float(candidate["stop"]) > float(state["previous_stop"])
+                and rank_not_weaker
+            ):
+                reason = "gate_reset_with_improved_stop_and_rank"
+        elif self.name == "two_session_confirmation":
+            if state["qualified_streak"] >= 2:
+                reason = "two_qualified_post_stop_closes"
+
+        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_candidates_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_session_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
+        ),
+    }
+
+
+def _parse_optional_date(value: str | None, option: str) -> date | None:
+    if value is None or value.strip().lower() == "none":
+        return None
+    try:
+        return date.fromisoformat(value)
+    except ValueError as exc:
+        raise SystemExit(f"{option} must use YYYY-MM-DD or 'none'") from exc
+
+
+def _max_date(left: date | None, right: date | None) -> date | None:
+    if left is None:
+        return right
+    if right is None:
+        return left
+    return max(left, right)
+
+
+async def _main() -> None:
+    args = _parse_args()
+    snapshot = Path(args.snapshot)
+    if not snapshot.exists():
+        raise SystemExit(f"Snapshot not found: {snapshot}")
+    requested_start = _parse_optional_date(args.start_date, "--start-date")
+    holdout_split = _parse_optional_date(args.holdout_split, "--holdout-split")
+    if args.workers < 1:
+        raise SystemExit("--workers must be positive")
+    if args.base_capacity < 1 or any(value < 1 for value in args.capacities):
+        raise SystemExit("capacities must be positive")
+    all_costs = sorted(
+        set([args.base_cost_per_side_pct, *args.costs_per_side_pct])
+    )
+    if any(value < 0 or value >= 100 for value in all_costs):
+        raise SystemExit("cost percentages must be in [0, 100)")
+    all_capacities = sorted(set([args.base_capacity, *args.capacities]))
+    policies = tuple(dict.fromkeys(args.policies))
+
+    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
+
+    db_engine = create_async_engine(_sqlite_url(snapshot), pool_pre_ping=True)
+    Session = async_sessionmaker(
+        db_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 db_engine.dispose()
+
+    replay_start = requested_start or date(1900, 1, 1)
+    snapshot_stat = snapshot.stat()
+    cache_key = {
+        "version": CACHE_VERSION,
+        "snapshot": str(snapshot.resolve()),
+        "snapshot_size": snapshot_stat.st_size,
+        "snapshot_mtime_ns": snapshot_stat.st_mtime_ns,
+        "start_date": replay_start.isoformat(),
+        "cadence": "daily",
+        "target_model": "production_gtl",
+    }
+    cache_path = Path(args.candidate_cache) if args.candidate_cache else None
+    qualified_candidates: list[dict] | None = None
+    entry_candidate_count = 0
+    if cache_path is not None and cache_path.exists():
+        with cache_path.open("rb") as handle:
+            cached = pickle.load(handle)  # noqa: S301 - trusted local cache
+        if cached.get("key") == cache_key:
+            qualified_candidates = list(cached["qualified_candidates"])
+            entry_candidate_count = int(cached["entry_candidate_count"])
+            if not args.quiet:
+                print(f"loaded qualified candidate cache: {cache_path}", flush=True)
+        elif not args.quiet:
+            print(f"candidate cache mismatch; rebuilding: {cache_path}", flush=True)
+
+    if qualified_candidates is None:
+        candidates: list[dict] = []
+        workers = max(1, min(int(args.workers), multiprocessing.cpu_count() - 1))
+        context = bt._mp_context() or 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,
+                    replay_start,
+                    "daily",
+                ): 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"daily replay: {index}/{len(futures)} tickers", flush=True)
+
+        entry_candidate_count = len(candidates)
+        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)
+        qualified_candidates = [
+            candidate
+            for candidate in candidates
+            if candidate["qualified"] and candidate.get("direction") == "long"
+        ]
+        del candidates
+        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,
+                        "entry_candidate_count": entry_candidate_count,
+                        "qualified_candidates": qualified_candidates,
+                    },
+                    handle,
+                    protocol=pickle.HIGHEST_PROTOCOL,
+                )
+            if not args.quiet:
+                print(f"wrote qualified candidate cache: {cache_path}", flush=True)
+
+    if not qualified_candidates:
+        raise RuntimeError("Daily replay produced no production-qualified candidates")
+
+    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")
+    ranking_key = str(
+        entry_config.get("ranking_key") or entry_config["percentile_key"]
+    )
+    threshold = float(activation.get("min_momentum_percentile", 80.0))
+    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)
+    )
+    qualified_symbols = {row["symbol"] for row in qualified_candidates}
+    simulation_prices = {
+        symbol: columns
+        for symbol, columns in prices.items()
+        if symbol in qualified_symbols
+    }
+    daily_engine = PrecomputedDailyEngine(qualified_candidates)
+    latest_ord = max(
+        date.fromisoformat(row["date"]).toordinal()
+        for row in qualified_candidates
+    )
+    latest_date = date.fromordinal(latest_ord)
+    if holdout_split is not None and not (
+        (requested_start or date.min) < holdout_split <= latest_date
+    ):
+        raise SystemExit(
+            f"--holdout-split must be after the start and no later than {latest_date}"
+        )
+
+    run_cache: dict[tuple, dict] = {}
+    completed_sims = 0
+
+    def run_policy(
+        policy_name: str,
+        *,
+        start_date: date | None,
+        end_date: date | None,
+        cost_pct: float,
+        capacity: int,
+    ) -> dict:
+        nonlocal completed_sims
+        key = (policy_name, start_date, end_date, float(cost_pct), int(capacity))
+        if key in run_cache:
+            return copy.deepcopy(run_cache[key])
+        policy = ReentryPolicy(policy_name, daily_engine, ranking_key)
+        sim = bt._simulate_portfolio(
+            qualified_candidates,
+            simulation_prices,
+            benchmark_closes,
+            exit_policy,
+            hold_days,
+            ranking_key=ranking_key,
+            max_positions=capacity,
+            risk_per_trade=float(entry_config["risk_per_trade"]),
+            atr_trail_multiplier=trail_multiplier,
+            cost_per_side=cost_pct / 100.0,
+            start_date=start_date,
+            end_date=end_date,
+            post_stop_reentry_fn=policy,
+            include_trades=True,
+        )
+        if sim is None:
+            raise RuntimeError(f"Policy {policy_name} produced no trades")
+        trades = list(sim.pop("trade_details"))
+        events = list(sim.pop("reentry_events", []))
+        row = {
+            **sim,
+            "turnover": _trade_summary(trades),
+            "policy": policy.summary(),
+            "reentry_events": events,
+        }
+        run_cache[key] = row
+        completed_sims += 1
+        if not args.quiet:
+            print(
+                f"portfolio simulations: {completed_sims} "
+                f"({policy_name}, cost={cost_pct}%, capacity={capacity})",
+                flush=True,
+            )
+        return copy.deepcopy(row)
+
+    primary: list[dict] = []
+    for lookback in bt.PORTFOLIO_MONITOR_LOOKBACKS:
+        lookback_start = bt._lookback_start(latest_ord, lookback["days"])
+        sim_start = _max_date(requested_start, lookback_start)
+        for policy_name in policies:
+            row = run_policy(
+                policy_name,
+                start_date=sim_start,
+                end_date=None,
+                cost_pct=args.base_cost_per_side_pct,
+                capacity=args.base_capacity,
+            )
+            if lookback["lookback"] != "all":
+                row.pop("reentry_events", None)
+            primary.append({
+                "arm": policy_name,
+                "lookback": lookback["lookback"],
+                "lookback_label": lookback["label"],
+                "capacity": args.base_capacity,
+                **row,
+            })
+
+    robustness: list[dict] = []
+    for cost_pct in all_costs:
+        for capacity in all_capacities:
+            for policy_name in policies:
+                row = run_policy(
+                    policy_name,
+                    start_date=requested_start,
+                    end_date=None,
+                    cost_pct=cost_pct,
+                    capacity=capacity,
+                )
+                row.pop("reentry_events", None)
+                robustness.append({
+                    "arm": policy_name,
+                    "lookback": "all",
+                    "cost_per_side_pct_requested": cost_pct,
+                    "capacity": capacity,
+                    **row,
+                })
+
+    holdout: list[dict] = []
+    if holdout_split is not None:
+        for segment, segment_start, segment_end in (
+            ("train", requested_start, holdout_split),
+            ("test", _max_date(requested_start, holdout_split), None),
+        ):
+            for policy_name in policies:
+                row = run_policy(
+                    policy_name,
+                    start_date=segment_start,
+                    end_date=segment_end,
+                    cost_pct=args.base_cost_per_side_pct,
+                    capacity=args.base_capacity,
+                )
+                row.pop("reentry_events", None)
+                holdout.append({
+                    "arm": policy_name,
+                    "segment": segment,
+                    "split_date": holdout_split.isoformat(),
+                    "capacity": args.base_capacity,
+                    **row,
+                })
+
+    output = Path(args.out) if args.out else _default_output_path()
+    report = {
+        "generated_at": datetime.now().astimezone().isoformat(),
+        "snapshot": str(snapshot.resolve()),
+        "period_start_requested": (
+            requested_start.isoformat() if requested_start else None
+        ),
+        "last_eligible_candidate_date": latest_date.isoformat(),
+        "tickers_loaded": len(prices),
+        "tickers_qualified": len(qualified_symbols),
+        "entry_candidates": entry_candidate_count,
+        "qualified_candidates": len(qualified_candidates),
+        "params": {
+            "entry_cadence": "daily",
+            "target_model": "production_gtl",
+            "policies": list(policies),
+            "base_cost_per_side_pct": args.base_cost_per_side_pct,
+            "base_capacity": args.base_capacity,
+            "robustness_costs_per_side_pct": all_costs,
+            "robustness_capacities": all_capacities,
+            "holdout_split": (
+                holdout_split.isoformat() if holdout_split else None
+            ),
+            "setup_stop_atr_multiplier": bt.ATR_MULTIPLIER,
+            "exit_policy": exit_policy,
+            "exit_atr_multiplier": trail_multiplier,
+            "hold_days": hold_days,
+            "risk_per_trade": float(entry_config["risk_per_trade"]),
+            "momentum_percentile_floor": threshold,
+            "ranking_key": ranking_key,
+        },
+        "primary_lookback_matrix": primary,
+        "cost_capacity_robustness": robustness,
+        "holdout": holdout,
+        "portfolio_simulations_executed": completed_sims,
+        "note": (
+            "The point-in-time daily setup replay and universe ranking are executed "
+            "once. All arms reuse the identical production-qualified candidate set. "
+            "Immediate is the daily no-lockdown baseline; next_session blocks only "
+            "the stop day; cooldown_5 permits re-entry at wait_sessions=5; gate_reset "
+            "requires an unqualified close before requalification; "
+            "gate_reset_improved additionally requires a higher stop and a non-weaker "
+            "production rank; two_session_confirmation requires two consecutive "
+            "qualified post-stop closes and excludes the stop day's close. Transaction "
+            "costs alter cash and position sizing, not just reported P&L."
+        ),
+    }
+    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 row in primary:
+        if row["lookback"] == "all":
+            print(
+                f"{row['arm']}: Sharpe {row['sharpe']}, CAGR {row['cagr_pct']}%, "
+                f"DD {row['max_drawdown_pct']}%, trades {row['trades']}, "
+                f"reentries {row['turnover']['reentry_trades']}, "
+                f"fees ${row['turnover']['transaction_cost']}"
+            )
+
+
+if __name__ == "__main__":
+    asyncio.run(_main())

tests/unit/test_backtest_service.py

@@ -543,6 +543,7 @@ class TestSimulatePortfolio:
         sim = bt._simulate_portfolio([cand], prices, None, "hold", 3)
         assert sim is not None
         assert sim["trades"] == 1
+        assert sim["cost_per_side_pct"] == pytest.approx(0.1)
         # 20 shares (1% risk / $5 stop distance), exit at the day-3 close 106:
         # pnl = 2120 − 2000 − 2.00 entry cost − 2.12 exit cost = 115.88
         assert sim["final_equity"] == pytest.approx(10_115.88, abs=0.01)
@@ -556,6 +557,29 @@ class TestSimulatePortfolio:
             {"year": 2025, "return_pct": pytest.approx(1.2, abs=0.05)}
         ]
 
+    def test_cost_parameter_changes_cash_and_position_path(self):
+        closes = [100.0, 102.0, 104.0, 106.0]
+        prices = {"AAA": _sim_prices(self.ORD, closes)}
+        cand = _sim_cand("AAA", self.ORD, entry=100.0, stop=95.0, target=130.0)
+
+        free = bt._simulate_portfolio(
+            [cand], prices, None, "hold", 3, cost_per_side=0.0
+        )
+        stressed = bt._simulate_portfolio(
+            [cand], prices, None, "hold", 3, cost_per_side=0.002
+        )
+
+        assert free is not None and stressed is not None
+        assert free["final_equity"] == pytest.approx(10_120.0, abs=0.01)
+        assert stressed["cost_per_side_pct"] == pytest.approx(0.2)
+        assert stressed["final_equity"] == pytest.approx(10_111.76, abs=0.01)
+
+    def test_cost_parameter_rejects_invalid_rate(self):
+        with pytest.raises(ValueError, match="cost_per_side"):
+            bt._simulate_portfolio(
+                [], {}, None, "hold", 3, cost_per_side=-0.001
+            )
+
     def test_target_policy_exits_at_target(self):
         closes = [100.0, 102.0, 104.0, 106.0, 108.0, 110.0]
         prices = {"AAA": _sim_prices(self.ORD, closes)}
@@ -626,6 +650,35 @@ class TestSimulatePortfolio:
             self.ORD + 6
         ).isoformat()
 
+    def test_post_stop_reentry_cannot_cross_holdout_end(self):
+        prices = {"AAA": _sim_prices(self.ORD, [100.0, 94.0, 96.0, 98.0])}
+        candidate = _sim_cand(
+            "AAA", self.ORD, entry=100.0, stop=95.0, target=120.0
+        )
+        callback_dates: list[int] = []
+
+        def reenter_after_split(symbol, asof_ord, _state, _bar):
+            callback_dates.append(asof_ord)
+            if asof_ord < self.ORD + 2:
+                return None
+            return _sim_cand(
+                symbol, asof_ord, entry=96.0, stop=90.0, target=115.0
+            )
+
+        sim = bt._simulate_portfolio(
+            [candidate],
+            prices,
+            None,
+            "hold",
+            3,
+            end_date=date.fromordinal(self.ORD + 2),
+            post_stop_reentry_fn=reenter_after_split,
+        )
+
+        assert sim is not None
+        assert sim["trades"] == 1
+        assert callback_dates == [self.ORD + 1]
+
     def test_production_monitor_applies_live_reentry_lockdown(self, monkeypatch):
         def fake_simulator(*_args, **kwargs):
             return {

tests/unit/test_daily_reentry_matrix.py

@@ -0,0 +1,97 @@
+from datetime import date
+
+from scripts.run_daily_reentry_matrix import PrecomputedDailyEngine, ReentryPolicy
+
+
+RANKING_KEY = "strategy_rank"
+ORD = date(2025, 1, 6).toordinal()
+
+
+def _candidate(day_ord: int, *, stop: float = 91.0, rank: float = 81.0) -> dict:
+    return {
+        "qualified": True,
+        "direction": "long",
+        "symbol": "AAA",
+        "date": date.fromordinal(day_ord).isoformat(),
+        "entry": 100.0,
+        "stop": stop,
+        "target": 120.0,
+        RANKING_KEY: rank,
+    }
+
+
+def _state(sessions: int = 0) -> dict:
+    return {
+        "sessions_since_stop": sessions,
+        "previous_stop": 90.0,
+        "previous_rank": 80.0,
+        "gate_went_unqualified": False,
+    }
+
+
+def _call(policy: ReentryPolicy, day_ord: int, state: dict, sessions: int):
+    state["sessions_since_stop"] = sessions
+    return policy("AAA", day_ord, state, object())
+
+
+def test_next_session_blocks_only_stop_day():
+    engine = PrecomputedDailyEngine([_candidate(ORD), _candidate(ORD + 1)])
+    policy = ReentryPolicy("next_session", engine, RANKING_KEY)
+    state = _state()
+
+    assert _call(policy, ORD, state, 0) is None
+    assert _call(policy, ORD + 1, state, 1) is not None
+
+
+def test_five_session_cooldown_unlocks_at_exact_boundary():
+    engine = PrecomputedDailyEngine([_candidate(ORD + 4), _candidate(ORD + 5)])
+    policy = ReentryPolicy("cooldown_5", engine, RANKING_KEY)
+    state = _state()
+
+    assert _call(policy, ORD + 4, state, 4) is None
+    assert _call(policy, ORD + 5, state, 5) is not None
+
+
+def test_gate_reset_requires_failure_before_requalification():
+    engine = PrecomputedDailyEngine([_candidate(ORD), _candidate(ORD + 2)])
+    policy = ReentryPolicy("gate_reset", engine, RANKING_KEY)
+    state = _state()
+
+    assert _call(policy, ORD, state, 0) is None
+    assert _call(policy, ORD + 1, state, 1) is None
+    emitted = _call(policy, ORD + 2, state, 2)
+    assert emitted is not None
+    assert emitted["_reentry_reason"] == "gate_failed_then_requalified"
+
+
+def test_improved_gate_reset_requires_better_stop_and_non_weaker_rank():
+    engine = PrecomputedDailyEngine([
+        _candidate(ORD + 1, stop=89.0, rank=82.0),
+        _candidate(ORD + 2, stop=92.0, rank=79.0),
+        _candidate(ORD + 3, stop=92.0, rank=81.0),
+    ])
+    policy = ReentryPolicy("gate_reset_improved", engine, RANKING_KEY)
+    state = _state()
+
+    assert _call(policy, ORD, state, 0) is None
+    assert _call(policy, ORD + 1, state, 1) is None
+    assert _call(policy, ORD + 2, state, 2) is None
+    emitted = _call(policy, ORD + 3, state, 3)
+    assert emitted is not None
+    assert emitted["_reentry_reason"] == "gate_reset_with_improved_stop_and_rank"
+
+
+def test_two_session_confirmation_excludes_stop_day_close():
+    engine = PrecomputedDailyEngine([
+        _candidate(ORD),
+        _candidate(ORD + 1),
+        _candidate(ORD + 2),
+    ])
+    policy = ReentryPolicy("two_session_confirmation", engine, RANKING_KEY)
+    state = _state()
+
+    assert _call(policy, ORD, state, 0) is None
+    assert _call(policy, ORD + 1, state, 1) is None
+    emitted = _call(policy, ORD + 2, state, 2)
+    assert emitted is not None
+    assert emitted["_reentry_reason"] == "two_qualified_post_stop_closes"