Prefer replay-qualified Torch calibration

tools/calibrate_torch_thresholds.py

@@ -138,10 +138,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(
@@ -186,9 +193,10 @@ 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.")
@@ -799,6 +807,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)