Add honest Torch validation gate

.env

@@ -62,7 +62,7 @@ TIME_SERIES_FORECAST_ENABLED=true
 TIME_SERIES_MIN_CANDLES=120
 TIME_SERIES_FORECAST_HORIZON=3
 TIME_SERIES_MIN_EDGE_PERCENT=0.10
-TIME_SERIES_MIN_PROBABILITY_UP=0.52
+TIME_SERIES_MIN_PROBABILITY_UP=0.56
 TIME_SERIES_MIN_CONFIDENCE=0.4
 TIME_SERIES_MAX_ADJUSTMENT=0.08
 TIME_SERIES_LSTM_ENABLED=true

crypto_spot_bot/dashboard.py

@@ -806,12 +806,14 @@ HTML = r"""
       const rec = backtest.recommended || {};
       const replay = backtest.full_replay || {};
       const walk = backtest.walk_forward?.summary || {};
+      const validation = backtest.validation || {};
+      const benchmark = backtest.benchmark?.summary || {};
       return `
         <div class="grid cols-4">
           ${metric('Recommended edge', `${num(rec.edge, 4)}%`, `P(up) ${num((rec.probability || 0) * 100, 1)}%`)}
           ${metric('Entry replay', `${replay.trades || 0} trades`, `${signed(replay.avg_net_percent, 4)}% avg · PF ${num(replay.profit_factor, 3)}`)}
           ${metric('Walk-forward', `${walk.trades || 0} trades`, `${signed(walk.avg_net_percent, 4)}% avg · ${walk.status || ''}`)}
-          ${metric('Retrain guard', retrain.available ? (retrain.accepted ? 'accepted' : 'rejected') : 'no report', retrain.reason || '')}
+          ${metric('Quality gate', validation.status || 'missing', validation.passed ? 'Torch allowed' : 'Torch blocked')}
         </div>
         <div class="grid cols-2" style="margin-top:14px">
           ${panel('Threshold calibration', kvTable([
@@ -832,6 +834,32 @@ HTML = r"""
             ['Profit factor', num(replay.profit_factor, 4)]
           ]))}
         </div>
+        <div class="grid cols-2" style="margin-top:14px">
+          ${panel('Honest validation', kvTable([
+            ['Status', validation.status || 'missing'],
+            ['OOS trades', String(validation.oos_summary?.trades || 0)],
+            ['OOS avg net', signed(validation.oos_summary?.avg_net_percent, 4) + '%'],
+            ['OOS PF', num(validation.oos_summary?.profit_factor, 4)],
+            ['Folds with trades', String(backtest.walk_forward?.folds_with_trades || 0)],
+            ['Benchmark total', signed(benchmark.total_net_percent, 4) + '%'],
+            ['Benchmark trades', String(benchmark.trades || 0)],
+            ['Retrain guard', retrain.available ? (retrain.accepted ? 'accepted' : 'rejected') : 'no report']
+          ]))}
+          ${panel('Gate checks', simpleTable((validation.checks || []).map(row => ({
+            check: row.name,
+            value: row.value,
+            required: row.required,
+            passed: row.passed ? 'yes' : 'no'
+          })), ['check', 'value', 'required', 'passed']))}
+        </div>
+        ${panel('OOS by symbol', simpleTable((backtest.walk_forward?.symbol_breakdown || []).map(row => ({
+          symbol: row.symbol,
+          trades: row.trades,
+          share: num((row.trade_share || 0) * 100, 1) + '%',
+          avg: signed(row.avg_net_percent, 4) + '%',
+          total: signed(row.total_net_percent, 4) + '%',
+          pf: num(row.profit_factor, 3)
+        })), ['symbol', 'trades', 'share', 'avg', 'total', 'pf']))}
         ${panel('Walk-forward folds', simpleTable((backtest.walk_forward?.folds || []).map(row => ({
           fold: row.fold,
           train: row.train_records,

crypto_spot_bot/strategy.py

@@ -665,8 +665,10 @@ def _torch_forecast_entry_signal(
     spread_ok = ticker.spread_percent <= settings.max_spread_percent
     liquidity_ok = ticker.turnover_24h >= settings.min_24h_turnover_usdt
     model_ok = _is_torch_forecast(forecast)
+    quality_gate_ok = forecast.get("quality_gate_passed") is not False
     checks = {
         "torch_model_ok": model_ok,
+        "quality_gate_ok": quality_gate_ok,
         "forecast_usable": bool(forecast.get("usable", False)),
         "forecast_not_blocked": not bool(forecast.get("block_entry", False)),
         "expected_edge_ok": full_edge_ok or probe_edge_ok,
@@ -695,6 +697,8 @@ def _torch_forecast_entry_signal(
         "min_probability_up": min_probability,
         "min_confidence": settings.time_series_min_confidence,
         "skill": skill,
+        "quality_gate": forecast.get("quality_gate", {}),
+        "quality_gate_passed": forecast.get("quality_gate_passed"),
         "spread_percent": round(ticker.spread_percent, 5),
         "turnover_24h": ticker.turnover_24h,
         "checks": checks,

crypto_spot_bot/time_series.py

@@ -154,6 +154,8 @@ class TimeSeriesForecast:
     feature_snapshot: list[dict[str, Any]] = field(default_factory=list)
     horizon_forecasts: dict[str, Any] = field(default_factory=dict)
     candidates: list[dict[str, Any]] = field(default_factory=list)
+    quality_gate_passed: bool | None = None
+    quality_gate: dict[str, Any] = field(default_factory=dict)
 
     def as_dict(self) -> dict[str, Any]:
         return asdict(self)
@@ -164,6 +166,8 @@ class TimeSeriesForecaster:
         self.settings = settings
         self._lstm_artifact_mtime: float | None = None
         self._lstm_artifact: dict[str, Any] = {}
+        self._calibration_mtime: float | None = None
+        self._quality_gate: dict[str, Any] = {}
 
     def forecast(
         self,
@@ -184,6 +188,8 @@ class TimeSeriesForecaster:
             return _empty_forecast(True, "not enough returns for PyTorch forecast")
 
         artifact = self._load_lstm_artifact()
+        quality_gate = self._load_quality_gate()
+        quality_gate_passed = _quality_gate_passed(quality_gate)
         entry = _torch_recurrent_entry(symbol, artifact)
         model = _torch_recurrent_model_name(symbol, artifact)
         clip = _clamp(_float_entry(entry or {}, "clip", 8.0), 1.0, 50.0)
@@ -280,6 +286,8 @@ class TimeSeriesForecaster:
                 feature_snapshot=feature_snapshot,
                 horizon_forecasts=_public_horizon_forecasts(prediction),
                 candidates=[{"model": model, "mae_percent": round(model_mae * 100, 4)}],
+                quality_gate_passed=quality_gate_passed,
+                quality_gate=quality_gate,
             )
 
         direct_horizon = _is_direct_horizon(entry)
@@ -340,6 +348,8 @@ class TimeSeriesForecaster:
             feature_snapshot=feature_snapshot,
             horizon_forecasts={},
             candidates=[{"model": model, "mae_percent": round(model_mae * 100, 4)}],
+            quality_gate_passed=quality_gate_passed,
+            quality_gate=quality_gate,
         )
 
     def _load_lstm_artifact(self) -> dict[str, Any]:
@@ -362,6 +372,25 @@ class TimeSeriesForecaster:
         self._lstm_artifact_mtime = stat.st_mtime
         return self._lstm_artifact
 
+    def _load_quality_gate(self) -> dict[str, Any]:
+        path = self.settings.time_series_lstm_model_path.parent / "torch_threshold_calibration.json"
+        try:
+            stat = path.stat()
+        except OSError:
+            self._calibration_mtime = None
+            self._quality_gate = {}
+            return {}
+        if self._calibration_mtime == stat.st_mtime:
+            return self._quality_gate
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            data = {}
+        validation = data.get("validation") if isinstance(data, dict) else {}
+        self._quality_gate = validation if isinstance(validation, dict) else {}
+        self._calibration_mtime = stat.st_mtime
+        return self._quality_gate
+
 
 def _empty_forecast(enabled: bool, reason: str) -> TimeSeriesForecast:
     return TimeSeriesForecast(
@@ -388,9 +417,25 @@ def _empty_forecast(enabled: bool, reason: str) -> TimeSeriesForecast:
         target_transform="none",
         feature_snapshot=[],
         horizon_forecasts={},
+        candidates=[],
+        quality_gate_passed=None,
+        quality_gate={},
     )
 
 
+def _quality_gate_passed(quality_gate: dict[str, Any]) -> bool | None:
+    if not quality_gate:
+        return None
+    if "passed" in quality_gate:
+        return bool(quality_gate.get("passed"))
+    status = str(quality_gate.get("status", "")).strip().lower()
+    if status in {"pass", "passed", "ok"}:
+        return True
+    if status in {"fail", "failed", "warn"}:
+        return False
+    return None
+
+
 def _log_returns(closes: list[float]) -> list[float]:
     return [math.log(closes[index] / closes[index - 1]) for index in range(1, len(closes))]
 

tests/test_strategy.py

@@ -284,6 +284,40 @@ def test_torch_forecast_blocks_without_valid_torch_model(make_settings, tmp_path
     assert signal.diagnostics["checks"]["torch_model_ok"] is False
 
 
+def test_torch_forecast_blocks_failed_quality_gate(make_settings, tmp_path) -> None:
+    settings = make_settings(
+        tmp_path,
+        strategy_mode="torch_forecast",
+        time_series_min_edge_percent=0.10,
+        time_series_min_probability_up=0.57,
+        max_position_usdt=25,
+        stop_loss_percent=0.04,
+    )
+    strategy = SpotStrategy(settings)
+    ticker = Ticker("BTCUSDT", 105, 104.99, 105.01, 10_000_000, 1000, 1.0)
+
+    signal = strategy.entry_signal(
+        "BTCUSDT",
+        [],
+        ticker,
+        open_positions_for_symbol=0,
+        forecast={
+            "usable": True,
+            "model": "torch_gru",
+            "expected_return_percent": 0.36,
+            "probability_up": 0.66,
+            "skill": 0.22,
+            "block_entry": False,
+            "quality_gate_passed": False,
+            "quality_gate": {"status": "fail"},
+        },
+        account={"equity": 100.0},
+    )
+
+    assert signal.action == "HOLD"
+    assert signal.diagnostics["checks"]["quality_gate_ok"] is False
+
+
 def test_torch_forecast_probe_buys_on_positive_high_probability(make_settings, tmp_path) -> None:
     settings = make_settings(
         tmp_path,

tests/test_time_series.py

@@ -282,6 +282,28 @@ def test_time_series_forecaster_reads_torch_gru_artifact(make_settings, tmp_path
     assert forecast.probability_up > 0.5
 
 
+def test_time_series_forecaster_attaches_quality_gate(make_settings, tmp_path) -> None:
+    artifact_path = tmp_path / "lstm_forecaster.json"
+    _write_torch_gru_artifact(artifact_path, head_bias=0.2)
+    (tmp_path / "torch_threshold_calibration.json").write_text(
+        json.dumps({"validation": {"status": "fail", "passed": False, "checks": []}}),
+        encoding="utf-8",
+    )
+    settings = make_settings(
+        tmp_path,
+        time_series_min_candles=80,
+        time_series_lstm_enabled=True,
+        time_series_lstm_model_path=artifact_path,
+    )
+    returns = [0.00015 if index % 4 else -0.00005 for index in range(140)]
+
+    forecast = TimeSeriesForecaster(settings).forecast(_candles_from_returns(returns), symbol="BTCUSDT")
+
+    assert forecast.usable is True
+    assert forecast.quality_gate_passed is False
+    assert forecast.quality_gate["status"] == "fail"
+
+
 def test_time_series_forecaster_reads_multifeature_direct_horizon_artifact(make_settings, tmp_path) -> None:
     artifact_path = tmp_path / "lstm_forecaster.json"
     _write_multifeature_torch_gru_artifact(artifact_path, head_bias=0.2)

tests/test_torch_candidate_guard.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+from tools.accept_torch_candidate import _decision
+
+
+def _report(*, validation_passed: bool = True, trades: int = 30, total: float = 10.0) -> dict:
+    return {
+        "recommended": {"score": 0.5},
+        "full_replay": {
+            "trades": trades,
+            "avg_net_percent": 0.4,
+            "total_net_percent": total,
+            "profit_factor": 2.0,
+            "max_drawdown_percent": 1.0,
+        },
+        "walk_forward": {"summary": {"trades": trades, "avg_net_percent": 0.3}},
+        "validation": {"passed": validation_passed, "status": "pass" if validation_passed else "fail"},
+    }
+
+
+def test_guard_rejects_candidate_without_honest_validation() -> None:
+    decision = _decision(
+        _report(),
+        _report(validation_passed=False),
+        min_trades=8,
+        min_profit_factor=1.05,
+        min_avg_net_percent=0.0,
+        max_score_regression=0.05,
+    )
+
+    assert decision == {"accepted": False, "reason": "candidate_failed_honest_validation"}
+
+
+def test_guard_accepts_candidate_that_passes_honest_validation() -> None:
+    decision = _decision(
+        _report(total=9.0),
+        _report(total=12.0),
+        min_trades=8,
+        min_profit_factor=1.05,
+        min_avg_net_percent=0.0,
+        max_score_regression=0.05,
+    )
+
+    assert decision == {"accepted": True, "reason": "candidate_passed_guard"}

tools/accept_torch_candidate.py

@@ -64,6 +64,8 @@ def _decision(
     candidate_replay = candidate.get("full_replay") if isinstance(candidate.get("full_replay"), dict) else {}
     candidate_walk = candidate.get("walk_forward") if isinstance(candidate.get("walk_forward"), dict) else {}
     walk_summary = candidate_walk.get("summary") if isinstance(candidate_walk.get("summary"), dict) else {}
+    if not _validation_passed(candidate):
+        return {"accepted": False, "reason": "candidate_failed_honest_validation"}
     if int(candidate_replay.get("trades", 0) or 0) < min_trades:
         return {"accepted": False, "reason": "candidate_has_too_few_full_replay_trades"}
     if float(candidate_replay.get("profit_factor", 0.0) or 0.0) < min_profit_factor:
@@ -77,6 +79,15 @@ def _decision(
     return {"accepted": True, "reason": "candidate_passed_guard"}
 
 
+def _validation_passed(report: dict[str, Any]) -> bool:
+    validation = report.get("validation")
+    if not isinstance(validation, dict):
+        return False
+    if "passed" in validation:
+        return bool(validation.get("passed"))
+    return str(validation.get("status", "")).strip().lower() in {"pass", "passed", "ok"}
+
+
 def _score(report: dict[str, Any]) -> float:
     replay = report.get("full_replay") if isinstance(report.get("full_replay"), dict) else {}
     recommended = report.get("recommended") if isinstance(report.get("recommended"), dict) else {}
@@ -97,6 +108,10 @@ def _summary(report: dict[str, Any]) -> dict[str, Any]:
         "walk_forward_summary": (report.get("walk_forward") or {}).get("summary", {})
         if isinstance(report.get("walk_forward"), dict)
         else {},
+        "benchmark_summary": (report.get("benchmark") or {}).get("summary", {})
+        if isinstance(report.get("benchmark"), dict)
+        else {},
+        "validation": report.get("validation", {}),
     }
 
 

tools/calibrate_torch_thresholds.py

@@ -57,6 +57,8 @@ class ForecastRecord:
     q50_percent: float
     block_entry: bool
     future_net_percent: float
+    benchmark_entry: bool
+    benchmark_exit: bool
 
 
 @dataclass(slots=True)
@@ -138,10 +140,17 @@ def main() -> None:
     for result in results[: min(args.top, len(results))]:
         print(_result_line(result))
 
-    recommended = _choose_recommendation(results, min_trades=args.min_trades)
+    recommended, full_backtest = _choose_replay_recommendation(
+        results,
+        records,
+        min_trades=args.min_trades,
+        min_full_replay_trades=args.min_full_replay_trades,
+        horizon=horizon,
+        round_trip_cost=round_trip_cost,
+        settings=settings,
+    )
     print("\nRECOMMENDED")
     print(_result_line(recommended))
-    full_backtest = _full_backtest(records, recommended, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings)
     print("\nFULL_REPLAY")
     print(_stats_line(full_backtest))
     walk_forward = _walk_forward(
@@ -152,9 +161,32 @@ def main() -> None:
         min_trades=args.min_trades,
         horizon=horizon,
         folds=args.walk_forward_folds,
+        round_trip_cost=round_trip_cost,
+        settings=settings,
+    )
+    benchmark = _benchmark_walk_forward(
+        records,
+        horizon=horizon,
+        folds=args.walk_forward_folds,
+        round_trip_cost=round_trip_cost,
+        settings=settings,
+    )
+    validation = _quality_gate(
+        walk_forward=walk_forward,
+        benchmark=benchmark,
+        min_oos_trades=args.min_oos_trades,
+        min_oos_symbols=args.min_oos_symbols,
+        max_symbol_share=args.max_oos_symbol_share,
+        min_oos_folds=args.min_oos_folds_with_trades,
+        min_profit_factor=args.min_oos_profit_factor,
+        min_benchmark_edge=args.min_benchmark_edge_percent,
     )
     print("\nWALK_FORWARD")
     print(json.dumps(walk_forward["summary"], ensure_ascii=False, sort_keys=True))
+    print("\nBENCHMARK")
+    print(json.dumps(benchmark["summary"], ensure_ascii=False, sort_keys=True))
+    print("\nQUALITY_GATE")
+    print(json.dumps(validation, ensure_ascii=False, sort_keys=True))
     print(
         "env "
         f"TIME_SERIES_MIN_EDGE_PERCENT={recommended.edge:.4f} "
@@ -169,6 +201,8 @@ def main() -> None:
             "recommended": _result_dict(recommended),
             "full_replay": full_backtest,
             "walk_forward": walk_forward,
+            "benchmark": benchmark,
+            "validation": validation,
             "probability_calibration": _probability_calibration(records),
             "top_results": [_result_dict(result) for result in results[: args.top]],
         }
@@ -186,14 +220,21 @@ def _parse_args() -> argparse.Namespace:
     parser.add_argument("--calibration-window", type=int, default=720, help="Tail records used for calibration.")
     parser.add_argument("--horizon", type=int, default=0, help="Forecast horizon to calibrate.")
     parser.add_argument("--min-trades", type=int, default=12, help="Minimum non-overlapping trades for recommendation.")
+    parser.add_argument("--min-full-replay-trades", type=int, default=8, help="Prefer recommendations with at least this many full replay trades.")
     parser.add_argument("--edge-grid", default="0.00,0.02,0.04,0.05,0.06,0.08,0.10", help="Percent edge thresholds.")
     parser.add_argument("--probability-grid", default="0.55,0.56,0.57,0.58,0.59,0.60,0.62,0.64,0.66,0.68,0.70", help="P(up) thresholds.")
-    parser.add_argument("--confidence-grid", default="0.50,0.56,0.60,0.64,0.68,0.72", help="Confidence thresholds.")
+    parser.add_argument("--confidence-grid", default="0.40,0.50,0.56,0.60,0.64,0.68,0.72", help="Confidence thresholds.")
     parser.add_argument("--top", type=int, default=15, help="How many top results to print and save.")
     parser.add_argument("--output", default="", help="Optional JSON output path.")
     parser.add_argument("--batch-size", type=int, default=256, help="Torch inference batch size.")
     parser.add_argument("--threads", type=int, default=0, help="Torch CPU threads; 0 keeps torch default.")
     parser.add_argument("--walk-forward-folds", type=int, default=4, help="Threshold walk-forward folds.")
+    parser.add_argument("--min-oos-trades", type=int, default=30, help="Minimum out-of-sample walk-forward trades for a valid model.")
+    parser.add_argument("--min-oos-symbols", type=int, default=2, help="Minimum symbols with out-of-sample trades.")
+    parser.add_argument("--max-oos-symbol-share", type=float, default=0.75, help="Reject if one symbol contributes more than this share of out-of-sample trades.")
+    parser.add_argument("--min-oos-folds-with-trades", type=int, default=2, help="Minimum walk-forward folds that must produce trades.")
+    parser.add_argument("--min-oos-profit-factor", type=float, default=1.10, help="Minimum out-of-sample profit factor.")
+    parser.add_argument("--min-benchmark-edge-percent", type=float, default=0.0, help="Required total-net percent advantage over the benchmark.")
     return parser.parse_args()
 
 
@@ -237,6 +278,7 @@ def _forecast_records(
     batched_records = _batch_forecast_records(
         symbol=symbol,
         candles=candles,
+        trend_candles=trend_candles,
         feature_rows=feature_rows,
         closes=closes,
         entry=entry,
@@ -287,6 +329,8 @@ def _forecast_records(
                 q50_percent=q50_percent,
                 block_entry=False,
                 future_net_percent=future_net_percent,
+                benchmark_entry=_benchmark_entry_signal(candles, trend_candles, index),
+                benchmark_exit=_benchmark_exit_signal(candles, index),
             )
         )
     return records
@@ -296,6 +340,7 @@ def _batch_forecast_records(
     *,
     symbol: str,
     candles: list[Candle],
+    trend_candles: list[Candle],
     feature_rows: list[list[float]],
     closes: list[float],
     entry: dict[str, Any],
@@ -380,6 +425,8 @@ def _batch_forecast_records(
                         q50_percent=q50_percent,
                         block_entry=False,
                         future_net_percent=future_net_percent,
+                        benchmark_entry=_benchmark_entry_signal(candles, trend_candles, index),
+                        benchmark_exit=_benchmark_exit_signal(candles, index),
                     )
                 )
     return records
@@ -515,6 +562,7 @@ def _full_backtest(
     horizon: int,
     round_trip_cost: float,
     settings: Any,
+    detail_limit: int = 50,
 ) -> dict[str, Any]:
     positions: dict[str, dict[str, Any]] = {}
     trades: list[float] = []
@@ -595,7 +643,92 @@ def _full_backtest(
                 "entry_expected_percent": round(float(position["expected_percent"]), 4),
             }
         )
-    return {**_stats(trades), "trades_detail": rows[-50:]}
+    return {
+        **_stats(trades),
+        "trades_detail": _limited_rows(rows, detail_limit),
+        "symbol_breakdown": _symbol_breakdown(rows),
+    }
+
+
+def _benchmark_backtest(
+    records: list[ForecastRecord],
+    *,
+    horizon: int,
+    round_trip_cost: float,
+    settings: Any,
+    detail_limit: int = 50,
+) -> dict[str, Any]:
+    positions: dict[str, dict[str, Any]] = {}
+    trades: list[float] = []
+    rows: list[dict[str, Any]] = []
+    max_hold = max(12, horizon * 8)
+    stop_loss_percent = max(0.003, min(0.08, float(settings.stop_loss_percent))) * 100.0
+    atr_multiplier = max(0.5, min(10.0, float(settings.atr_trailing_multiplier)))
+    for record in sorted(records, key=lambda item: (item.timestamp, item.symbol)):
+        position = positions.get(record.symbol)
+        if position is not None:
+            position["highest"] = max(position["highest"], record.close)
+            net_percent = _net_percent(position["entry_price"], record.close, round_trip_cost)
+            held = record.index - int(position["entry_index"])
+            atr_stop = (
+                record.close <= position["highest"] - record.atr * atr_multiplier
+                if record.atr > 0 and position["highest"] > position["entry_price"]
+                else False
+            )
+            exit_reason = ""
+            if net_percent <= -stop_loss_percent:
+                exit_reason = "stop_loss"
+            elif atr_stop:
+                exit_reason = "atr_trailing_stop"
+            elif record.benchmark_exit:
+                exit_reason = "benchmark_exit"
+            elif held >= max_hold:
+                exit_reason = "max_hold"
+            if exit_reason:
+                trades.append(net_percent)
+                rows.append(
+                    {
+                        "symbol": record.symbol,
+                        "entry_timestamp": position["timestamp"],
+                        "exit_timestamp": record.timestamp,
+                        "net_percent": round(net_percent, 4),
+                        "reason": exit_reason,
+                        "held_bars": held,
+                    }
+                )
+                positions.pop(record.symbol, None)
+                continue
+
+        if record.symbol in positions:
+            continue
+        if record.benchmark_entry:
+            positions[record.symbol] = {
+                "entry_price": record.close,
+                "entry_index": record.index,
+                "timestamp": record.timestamp,
+                "highest": record.close,
+            }
+    for symbol, position in list(positions.items()):
+        tail = next((record for record in reversed(records) if record.symbol == symbol), None)
+        if tail is None:
+            continue
+        net_percent = _net_percent(position["entry_price"], tail.close, round_trip_cost)
+        trades.append(net_percent)
+        rows.append(
+            {
+                "symbol": symbol,
+                "entry_timestamp": position["timestamp"],
+                "exit_timestamp": tail.timestamp,
+                "net_percent": round(net_percent, 4),
+                "reason": "end_of_replay",
+                "held_bars": tail.index - int(position["entry_index"]),
+            }
+        )
+    return {
+        **_stats(trades),
+        "trades_detail": _limited_rows(rows, detail_limit),
+        "symbol_breakdown": _symbol_breakdown(rows),
+    }
 
 
 def _walk_forward(
@@ -607,6 +740,8 @@ def _walk_forward(
     min_trades: int,
     horizon: int,
     folds: int,
+    round_trip_cost: float,
+    settings: Any,
 ) -> dict[str, Any]:
     ordered = sorted(records, key=lambda item: item.timestamp)
     if folds < 2 or len(ordered) < folds * 20:
@@ -615,6 +750,7 @@ def _walk_forward(
     fold_size = max(1, len(timestamps) // folds)
     rows = []
     all_test_trades: list[float] = []
+    all_test_rows: list[dict[str, Any]] = []
     for fold in range(1, folds):
         test_start = timestamps[fold * fold_size]
         test_end = timestamps[(fold + 1) * fold_size - 1] if fold < folds - 1 else timestamps[-1]
@@ -631,20 +767,124 @@ def _walk_forward(
         if not train_results:
             continue
         selected = _choose_recommendation(train_results, min_trades=max(4, min_trades // 2))
-        test_trades = _selected_trades(test, selected.edge, selected.probability, selected.confidence, horizon)
+        test_backtest = _full_backtest(
+            test,
+            selected,
+            horizon=horizon,
+            round_trip_cost=round_trip_cost,
+            settings=settings,
+            detail_limit=0,
+        )
+        test_rows = test_backtest.get("trades_detail", [])
+        test_trades = [float(row.get("net_percent", 0.0) or 0.0) for row in test_rows if isinstance(row, dict)]
         all_test_trades.extend(test_trades)
+        all_test_rows.extend(test_rows)
         rows.append(
             {
                 "fold": fold,
                 "train_records": len(train),
                 "test_records": len(test),
                 "thresholds": _result_dict(selected),
-                "test": _stats(test_trades),
+                "test": {key: value for key, value in test_backtest.items() if key != "trades_detail"},
             }
         )
     summary = _stats(all_test_trades)
     summary["status"] = "ok" if summary["trades"] >= min_trades and summary["avg_net_percent"] > 0 else "warn"
-    return {"summary": summary, "folds": rows}
+    return {
+        "summary": summary,
+        "symbol_breakdown": _symbol_breakdown(all_test_rows),
+        "folds_with_trades": sum(1 for row in rows if int((row.get("test") or {}).get("trades", 0) or 0) > 0),
+        "folds": rows,
+    }
+
+
+def _benchmark_walk_forward(
+    records: list[ForecastRecord],
+    *,
+    horizon: int,
+    folds: int,
+    round_trip_cost: float,
+    settings: Any,
+) -> dict[str, Any]:
+    ordered = sorted(records, key=lambda item: item.timestamp)
+    if folds < 2 or len(ordered) < folds * 20:
+        return {"summary": {"status": "insufficient"}, "symbol_breakdown": [], "folds_with_trades": 0, "folds": []}
+    timestamps = sorted({record.timestamp for record in ordered})
+    fold_size = max(1, len(timestamps) // folds)
+    rows = []
+    all_test_rows: list[dict[str, Any]] = []
+    for fold in range(1, folds):
+        test_start = timestamps[fold * fold_size]
+        test_end = timestamps[(fold + 1) * fold_size - 1] if fold < folds - 1 else timestamps[-1]
+        test = [record for record in ordered if test_start <= record.timestamp <= test_end]
+        test_backtest = _benchmark_backtest(
+            test,
+            horizon=horizon,
+            round_trip_cost=round_trip_cost,
+            settings=settings,
+            detail_limit=0,
+        )
+        test_rows = test_backtest.get("trades_detail", [])
+        all_test_rows.extend(test_rows)
+        rows.append(
+            {
+                "fold": fold,
+                "test_records": len(test),
+                "test": {key: value for key, value in test_backtest.items() if key != "trades_detail"},
+            }
+        )
+    trades = [float(row.get("net_percent", 0.0) or 0.0) for row in all_test_rows if isinstance(row, dict)]
+    summary = _stats(trades)
+    summary["status"] = "ok" if summary["trades"] > 0 else "no_trades"
+    return {
+        "name": "trend_macd_baseline",
+        "summary": summary,
+        "symbol_breakdown": _symbol_breakdown(all_test_rows),
+        "folds_with_trades": sum(1 for row in rows if int((row.get("test") or {}).get("trades", 0) or 0) > 0),
+        "folds": rows,
+    }
+
+
+def _quality_gate(
+    *,
+    walk_forward: dict[str, Any],
+    benchmark: dict[str, Any],
+    min_oos_trades: int,
+    min_oos_symbols: int,
+    max_symbol_share: float,
+    min_oos_folds: int,
+    min_profit_factor: float,
+    min_benchmark_edge: float,
+) -> dict[str, Any]:
+    summary = walk_forward.get("summary") if isinstance(walk_forward.get("summary"), dict) else {}
+    benchmark_summary = benchmark.get("summary") if isinstance(benchmark.get("summary"), dict) else {}
+    breakdown = walk_forward.get("symbol_breakdown") if isinstance(walk_forward.get("symbol_breakdown"), list) else []
+    symbols_with_trades = sum(1 for row in breakdown if int(row.get("trades", 0) or 0) > 0)
+    max_share = max((float(row.get("trade_share", 0.0) or 0.0) for row in breakdown), default=0.0)
+    oos_total = float(summary.get("total_net_percent", 0.0) or 0.0)
+    benchmark_total = float(benchmark_summary.get("total_net_percent", 0.0) or 0.0)
+    checks = [
+        _gate_check("oos_trades", int(summary.get("trades", 0) or 0), min_oos_trades, int(summary.get("trades", 0) or 0) >= min_oos_trades),
+        _gate_check("oos_symbols", symbols_with_trades, min_oos_symbols, symbols_with_trades >= min_oos_symbols),
+        _gate_check("max_symbol_share", round(max_share, 4), max_symbol_share, max_share <= max_symbol_share if breakdown else False),
+        _gate_check("folds_with_trades", int(walk_forward.get("folds_with_trades", 0) or 0), min_oos_folds, int(walk_forward.get("folds_with_trades", 0) or 0) >= min_oos_folds),
+        _gate_check("oos_avg_net_positive", float(summary.get("avg_net_percent", 0.0) or 0.0), "> 0", float(summary.get("avg_net_percent", 0.0) or 0.0) > 0),
+        _gate_check("oos_profit_factor", float(summary.get("profit_factor", 0.0) or 0.0), min_profit_factor, float(summary.get("profit_factor", 0.0) or 0.0) >= min_profit_factor),
+        _gate_check("beats_benchmark_total", round(oos_total - benchmark_total, 4), min_benchmark_edge, (oos_total - benchmark_total) > min_benchmark_edge),
+    ]
+    passed = all(bool(row["passed"]) for row in checks)
+    return {
+        "status": "pass" if passed else "fail",
+        "passed": passed,
+        "checks": checks,
+        "oos_summary": summary,
+        "benchmark_summary": benchmark_summary,
+        "symbol_breakdown": breakdown,
+    }
+
+
+def _gate_check(name: str, value: Any, required: Any, passed: bool) -> dict[str, Any]:
+    return {"name": name, "value": value, "required": required, "passed": bool(passed)}
 
 
 def _probability_calibration(records: list[ForecastRecord]) -> dict[str, Any]:
@@ -691,12 +931,89 @@ def _candidate_allows(record: ForecastRecord, edge: float, probability: float, c
     )
 
 
+def _benchmark_entry_signal(candles: list[Candle], trend_candles: list[Candle], index: int) -> bool:
+    if index <= 0 or index >= len(candles):
+        return False
+    previous = candles[index - 1]
+    current = candles[index]
+    rsi = current.rsi_14
+    return bool(
+        _daily_trend_ok(trend_candles, current.timestamp)
+        and _macd_crossed_up(previous, current)
+        and current.ema_50 is not None
+        and current.close > current.ema_50
+        and rsi is not None
+        and 45.0 <= rsi <= 65.0
+    )
+
+
+def _benchmark_exit_signal(candles: list[Candle], index: int) -> bool:
+    if index <= 0 or index >= len(candles):
+        return False
+    previous = candles[index - 1]
+    current = candles[index]
+    return bool(_macd_crossed_down(previous, current) or (current.ema_50 is not None and current.close < current.ema_50))
+
+
+def _daily_trend_ok(trend_candles: list[Candle], timestamp: int) -> bool:
+    for candle in reversed(trend_candles):
+        if candle.timestamp > timestamp:
+            continue
+        return bool(
+            candle.ema_50 is not None
+            and candle.ema_200 is not None
+            and candle.close > candle.ema_200
+            and candle.ema_50 > candle.ema_200
+        )
+    return False
+
+
+def _macd_crossed_up(previous: Candle, current: Candle) -> bool:
+    if None in (previous.macd, previous.macd_signal, current.macd, current.macd_signal):
+        return False
+    return bool(previous.macd <= previous.macd_signal and current.macd > current.macd_signal)
+
+
+def _macd_crossed_down(previous: Candle, current: Candle) -> bool:
+    if None in (previous.macd, previous.macd_signal, current.macd, current.macd_signal):
+        return False
+    return bool(previous.macd >= previous.macd_signal and current.macd < current.macd_signal)
+
+
 def _net_percent(entry_price: float, exit_price: float, round_trip_cost: float) -> float:
     if entry_price <= 0 or exit_price <= 0:
         return 0.0
     return (math.exp(math.log(exit_price / entry_price) - round_trip_cost) - 1.0) * 100.0
 
 
+def _limited_rows(rows: list[dict[str, Any]], detail_limit: int) -> list[dict[str, Any]]:
+    if detail_limit <= 0:
+        return rows
+    return rows[-detail_limit:]
+
+
+def _symbol_breakdown(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    by_symbol: dict[str, list[float]] = {}
+    for row in rows:
+        symbol = str(row.get("symbol", ""))
+        if not symbol:
+            continue
+        by_symbol.setdefault(symbol, []).append(float(row.get("net_percent", 0.0) or 0.0))
+    total_trades = sum(len(values) for values in by_symbol.values())
+    result = []
+    for symbol in sorted(by_symbol):
+        values = by_symbol[symbol]
+        stats = _stats(values)
+        result.append(
+            {
+                "symbol": symbol,
+                **stats,
+                "trade_share": round(len(values) / total_trades, 4) if total_trades else 0.0,
+            }
+        )
+    return result
+
+
 def _stats(values: list[float]) -> dict[str, Any]:
     wins = sum(1 for value in values if value > 0)
     total = sum(values)
@@ -799,6 +1116,51 @@ def _choose_recommendation(results: list[CalibrationResult], *, min_trades: int)
     return viable[0] if viable else results[0]
 
 
+def _choose_replay_recommendation(
+    results: list[CalibrationResult],
+    records: list[ForecastRecord],
+    *,
+    min_trades: int,
+    min_full_replay_trades: int,
+    horizon: int,
+    round_trip_cost: float,
+    settings: Any,
+) -> tuple[CalibrationResult, dict[str, Any]]:
+    fallback = _choose_recommendation(results, min_trades=min_trades)
+    fallback_replay = _full_backtest(records, fallback, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings)
+    if min_full_replay_trades <= 0:
+        return fallback, fallback_replay
+
+    viable: list[tuple[CalibrationResult, dict[str, Any]]] = []
+    for result in results:
+        if result.trades < min(4, min_trades):
+            continue
+        if result.average_net_percent <= 0 or result.total_net_percent <= 0 or result.profit_factor < 1.05:
+            continue
+        replay = _full_backtest(records, result, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings)
+        if int(replay.get("trades", 0) or 0) < min_full_replay_trades:
+            continue
+        if float(replay.get("avg_net_percent", 0.0) or 0.0) <= 0:
+            continue
+        if float(replay.get("profit_factor", 0.0) or 0.0) < 1.05:
+            continue
+        viable.append((result, replay))
+
+    if not viable:
+        return fallback, fallback_replay
+    viable.sort(
+        key=lambda item: (
+            item[0].score,
+            float(item[1].get("avg_net_percent", 0.0) or 0.0),
+            float(item[1].get("total_net_percent", 0.0) or 0.0),
+            int(item[1].get("trades", 0) or 0),
+            item[0].confidence,
+        ),
+        reverse=True,
+    )
+    return viable[0]
+
+
 def _forecast_confidence(expected_return: float, probability_up: float, skill: float, min_edge: float) -> float:
     expected_return = max(0.0, expected_return)
     skill = max(0.0, skill)

tools/run_retrain_until_replay8.ps1

@@ -18,6 +18,7 @@ $RepoRoot = (Resolve-Path (Join-Path $PSScriptRoot "..")).Path
 $RuntimeDir = Join-Path $RepoRoot "runtime"
 $LoopLog = Join-Path $RuntimeDir "torch_retrain_until_replay8.log"
 $GuardReport = Join-Path $RuntimeDir "torch_retrain_guard.json"
+$ActiveCalibration = Join-Path $RuntimeDir "torch_threshold_calibration.json"
 $Runner = Join-Path $RepoRoot "tools\run_torch_retrain.ps1"
 New-Item -ItemType Directory -Force -Path $RuntimeDir | Out-Null
 
@@ -45,7 +46,8 @@ function Read-GuardSummary {
         return [pscustomobject]@{
             Accepted = $false
             Reason = "guard_report_missing"
-            ReplayTrades = 0
+            CandidateReplayTrades = 0
+            CurrentReplayTrades = 0
             WalkForwardTrades = 0
         }
     }
@@ -54,7 +56,8 @@ function Read-GuardSummary {
         return [pscustomobject]@{
             Accepted = [bool]$payload.accepted
             Reason = [string]$payload.reason
-            ReplayTrades = ConvertTo-IntOrZero $payload.candidate.full_replay.trades
+            CandidateReplayTrades = ConvertTo-IntOrZero $payload.candidate.full_replay.trades
+            CurrentReplayTrades = ConvertTo-IntOrZero $payload.current.full_replay.trades
             WalkForwardTrades = ConvertTo-IntOrZero $payload.candidate.walk_forward_summary.trades
         }
     }
@@ -62,14 +65,47 @@ function Read-GuardSummary {
         return [pscustomobject]@{
             Accepted = $false
             Reason = "guard_report_unreadable"
-            ReplayTrades = 0
+            CandidateReplayTrades = 0
+            CurrentReplayTrades = 0
             WalkForwardTrades = 0
         }
     }
 }
 
+function Read-ActiveReplayTrades {
+    if (-not (Test-Path $ActiveCalibration)) {
+        return 0
+    }
+    try {
+        $payload = Get-Content -Raw -LiteralPath $ActiveCalibration | ConvertFrom-Json
+        return ConvertTo-IntOrZero $payload.full_replay.trades
+    }
+    catch {
+        return 0
+    }
+}
+
+function Read-ActiveValidationPassed {
+    if (-not (Test-Path $ActiveCalibration)) {
+        return $false
+    }
+    try {
+        $payload = Get-Content -Raw -LiteralPath $ActiveCalibration | ConvertFrom-Json
+        return [bool]$payload.validation.passed
+    }
+    catch {
+        return $false
+    }
+}
+
 $attempt = 0
 while ($true) {
+    $activeReplayTrades = Read-ActiveReplayTrades
+    if (Read-ActiveValidationPassed) {
+        Write-LoopLog "Stop condition reached: active calibration passed honest validation with full_replay.trades=$activeReplayTrades."
+        exit 0
+    }
+
     $attempt += 1
     if ($SeedStart -gt 0) {
         $attemptSeed = $SeedStart + $attempt - 1
@@ -98,10 +134,10 @@ while ($true) {
     & powershell.exe @runnerArgs 2>&1 | Tee-Object -FilePath $LoopLog -Append
     $runnerExit = $LASTEXITCODE
     $summary = Read-GuardSummary
-    Write-LoopLog "Attempt $attempt finished; runner_exit=$runnerExit accepted=$($summary.Accepted) reason=$($summary.Reason) full_replay.trades=$($summary.ReplayTrades) walk_forward.trades=$($summary.WalkForwardTrades)."
+    Write-LoopLog "Attempt $attempt finished; runner_exit=$runnerExit accepted=$($summary.Accepted) reason=$($summary.Reason) candidate_full_replay.trades=$($summary.CandidateReplayTrades) current_full_replay.trades=$($summary.CurrentReplayTrades) walk_forward.trades=$($summary.WalkForwardTrades)."
 
-    if ($summary.ReplayTrades -ge $MinReplayTrades) {
-        Write-LoopLog "Stop condition reached: full_replay.trades=$($summary.ReplayTrades) >= $MinReplayTrades."
+    if ($summary.Accepted -and (Read-ActiveValidationPassed)) {
+        Write-LoopLog "Stop condition reached: accepted candidate passed honest validation with full_replay.trades=$($summary.CandidateReplayTrades)."
         exit 0
     }