Fix remote training and model validation pipeline

tools/train_torch_recurrent_forecaster.py

@@ -125,6 +125,9 @@ def main() -> None:
     decision_horizon = args.horizon if args.horizon > 0 else max(1, settings.time_series_forecast_horizon)
     target_horizons = _horizons(args.horizons, decision_horizon)
     feature_names = _feature_names_arg(args.features)
+    if args.pooled:
+        feature_names.extend(f"symbol_is_{symbol}" for symbol in symbols)
+    ensemble_seeds = _ints(args.ensemble_seeds) or [args.seed]
     round_trip_cost = max(0.0, 2.0 * (float(settings.taker_fee_rate) + float(settings.slippage_rate)))
     _progress(
         f"training started: symbols={len(symbols)} interval={interval} "
@@ -151,9 +154,82 @@ def main() -> None:
         "feature_names": feature_names,
         "feature_count": len(feature_names),
         "device": str(device),
+        "ensemble_seeds": ensemble_seeds,
+        "selection_folds": args.selection_folds,
         "symbols": {},
     }
 
+    if args.pooled:
+        artifact["version"] = 5
+        artifact["pooled_multi_asset"] = True
+        artifact["symbol_embedding"] = "learned_one_hot_projection"
+        artifact["symbols"] = _train_pooled_symbols(
+            client=client,
+            symbols=symbols,
+            interval=interval,
+            limit=args.limit,
+            validation_window=args.validation_window,
+            holdout_window=args.holdout_window,
+            target_horizons=target_horizons,
+            decision_horizon=decision_horizon,
+            feature_names=feature_names,
+            round_trip_cost=round_trip_cost,
+            context_symbols=_strings(args.context_symbols),
+            architectures=_strings(args.architectures),
+            lookbacks=_ints(args.lookbacks),
+            hidden_sizes=_ints(args.hidden_sizes),
+            layers_values=_ints(args.layers),
+            dropouts=_floats(args.dropouts),
+            epochs=args.epochs,
+            patience=args.patience,
+            batch_size=args.batch_size,
+            learning_rate=args.learning_rate,
+            weight_decay=args.weight_decay,
+            clip=args.clip,
+            attention_pooling=args.attention_pooling,
+            context_norm=args.context_norm,
+            device=device,
+            seeds=ensemble_seeds,
+            selection_folds=args.selection_folds,
+        )
+    else:
+        artifact["symbols"] = _train_independent_symbols(
+            client=client,
+            symbols=symbols,
+            interval=interval,
+            args=args,
+            target_horizons=target_horizons,
+            decision_horizon=decision_horizon,
+            feature_names=feature_names,
+            round_trip_cost=round_trip_cost,
+            device=device,
+            ensemble_seeds=ensemble_seeds,
+        )
+
+    for symbol, result in artifact["symbols"].items():
+        _progress(
+            f"{symbol}: model={result['model']} lookback={result['lookback']} "
+            f"features={result['input_size']} hidden={result['hidden_size']} "
+            f"layers={result['num_layers']} horizons={','.join(map(str, result['target_horizons']))} "
+            f"mae={result['validation_mae_percent']:.5f}% "
+            f"baseline={result['baseline_mae_percent']:.5f}% "
+            f"skill={result['skill']:.4f} dir={result['directional_accuracy']:.3f} "
+            f"p_brier={result['probability_brier']:.4f}"
+        )
+
+    output.parent.mkdir(parents=True, exist_ok=True)
+    tmp_output = output.with_name(f"{output.name}.tmp")
+    tmp_output.write_text(json.dumps(artifact, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    tmp_output.replace(output)
+    _progress(f"saved {output}")
+
+
+def _train_independent_symbols(
+    *, client: BybitClient, symbols: list[str], interval: str, args: argparse.Namespace,
+    target_horizons: list[int], decision_horizon: int, feature_names: list[str],
+    round_trip_cost: float, device: torch.device, ensemble_seeds: list[int],
+) -> dict[str, Any]:
+    results: dict[str, Any] = {}
     total_symbols = len(symbols)
     for index, symbol in enumerate(symbols, start=1):
         _progress(f"{symbol}: training started ({index}/{total_symbols})")
@@ -183,27 +259,221 @@ def main() -> None:
             attention_pooling=args.attention_pooling,
             context_norm=args.context_norm,
             device=device,
-            seed=args.seed,
+            seeds=ensemble_seeds,
+            selection_folds=args.selection_folds,
         )
         if result is None:
             _progress(f"{symbol}: skipped, not enough candles or train/validation samples")
             continue
-        artifact["symbols"][symbol] = result
-        _progress(
-            f"{symbol}: model={result['model']} lookback={result['lookback']} "
-            f"features={result['input_size']} hidden={result['hidden_size']} "
-            f"layers={result['num_layers']} horizons={','.join(map(str, result['target_horizons']))} "
-            f"mae={result['validation_mae_percent']:.5f}% "
-            f"baseline={result['baseline_mae_percent']:.5f}% "
-            f"skill={result['skill']:.4f} dir={result['directional_accuracy']:.3f} "
-            f"p_brier={result['probability_brier']:.4f}"
-        )
+        results[symbol] = result
+    return results
 
-    output.parent.mkdir(parents=True, exist_ok=True)
-    tmp_output = output.with_name(f"{output.name}.tmp")
-    tmp_output.write_text(json.dumps(artifact, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
-    tmp_output.replace(output)
-    _progress(f"saved {output}")
+
+def _train_pooled_symbols(
+    *, client: BybitClient, symbols: list[str], interval: str, limit: int,
+    validation_window: int, holdout_window: int, target_horizons: list[int],
+    decision_horizon: int, feature_names: list[str], round_trip_cost: float,
+    context_symbols: list[str], architectures: list[str], lookbacks: list[int],
+    hidden_sizes: list[int], layers_values: list[int], dropouts: list[float],
+    epochs: int, patience: int, batch_size: int, learning_rate: float,
+    weight_decay: float, clip: float, attention_pooling: bool, context_norm: bool,
+    device: torch.device, seeds: list[int], selection_folds: int,
+) -> dict[str, Any]:
+    market_candles: dict[str, list[Candle]] = {}
+    for symbol in sorted({item.upper() for item in symbols + context_symbols}):
+        rows = _historical_klines(client, symbol, interval, limit)
+        add_indicators(rows)
+        market_candles[symbol] = rows
+        _progress(f"{symbol}: pooled data loaded ({len(rows)} candles)")
+    trend_by_symbol: dict[str, list[Candle]] = {}
+    for symbol in symbols:
+        rows = _historical_klines(client, symbol, "D", min(max(260, limit // 24 + 260), 1000))
+        add_indicators(rows)
+        trend_by_symbol[symbol] = rows
+
+    best: dict[str, Any] | None = None
+    best_prepared: dict[str, PreparedData] = {}
+    for lookback in lookbacks:
+        prepared_by_symbol: dict[str, PreparedData] = {}
+        for symbol in symbols:
+            prepared = _prepare_data(
+                candles=market_candles[symbol],
+                feature_names=feature_names,
+                lookback=lookback,
+                target_horizons=target_horizons,
+                decision_horizon=decision_horizon,
+                round_trip_cost=round_trip_cost,
+                market_candles=market_candles,
+                trend_candles=trend_by_symbol[symbol],
+                validation_window=validation_window,
+                holdout_window=holdout_window,
+                clip=clip,
+                device=device,
+            )
+            if prepared is not None:
+                prepared_by_symbol[symbol] = prepared
+        if len(prepared_by_symbol) < 2:
+            continue
+        for architecture in architectures:
+            if architecture not in {"lstm", "gru"}:
+                continue
+            for hidden_size in hidden_sizes:
+                for num_layers in layers_values:
+                    for dropout in dropouts:
+                        if num_layers <= 1 and dropout != 0.0:
+                            continue
+                        _progress(
+                            f"pooled: fitting {architecture} lookback={lookback} hidden={hidden_size} "
+                            f"layers={num_layers} dropout={dropout} symbols={len(prepared_by_symbol)}"
+                        )
+                        members = [
+                            _fit_pooled_candidate(
+                                prepared_by_symbol=prepared_by_symbol,
+                                architecture=architecture,
+                                input_size=len(feature_names),
+                                output_size=len(target_horizons) * len(OUTPUT_LAYOUT),
+                                hidden_size=hidden_size,
+                                num_layers=num_layers,
+                                dropout=dropout,
+                                epochs=epochs,
+                                patience=patience,
+                                batch_size=batch_size,
+                                learning_rate=learning_rate,
+                                weight_decay=weight_decay,
+                                clip=clip,
+                                attention_pooling=attention_pooling,
+                                context_norm=context_norm,
+                                device=device,
+                                seed=member_seed,
+                                selection_folds=selection_folds,
+                            )
+                            for member_seed in seeds
+                        ]
+                        candidate = _ensemble_candidate(members, seeds)
+                        candidate.update(
+                            model=f"torch_{architecture}", architecture=architecture,
+                            lookback=lookback, hidden_size=hidden_size, num_layers=num_layers,
+                            dropout=dropout if num_layers > 1 else 0.0,
+                            attention_pooling=attention_pooling, context_norm=context_norm,
+                            input_size=len(feature_names), output_size=len(target_horizons) * len(OUTPUT_LAYOUT),
+                        )
+                        if best is None or _candidate_score(candidate) < _candidate_score(best):
+                            best = candidate
+                            best_prepared = prepared_by_symbol
+    if best is None:
+        return {}
+
+    results: dict[str, Any] = {}
+    symbol_metrics = best.get("symbol_metrics", {})
+    common = {key: value for key, value in best.items() if key != "symbol_metrics"}
+    for symbol, prepared in best_prepared.items():
+        metrics = symbol_metrics.get(symbol, {}) if isinstance(symbol_metrics, dict) else {}
+        baseline = sum(abs(row[prepared.decision_horizon_index]) for row in prepared.validation_targets) / len(prepared.validation_targets)
+        validation_mae = float(metrics.get("validation_mae", baseline))
+        results[symbol] = {
+            **common, **metrics,
+            "pooled_multi_asset": True,
+            "target_horizon": prepared.decision_horizon,
+            "target_horizons": prepared.target_horizons,
+            "direct_horizon": True,
+            "target_transform": "net_return_over_volatility",
+            "round_trip_cost": round(round_trip_cost, 10),
+            "output_layout": list(OUTPUT_LAYOUT),
+            "feature_names": feature_names,
+            "feature_means": prepared.feature_means,
+            "feature_scales": prepared.feature_scales,
+            "target_means": prepared.target_means,
+            "target_scales": prepared.target_scales,
+            "target_mean": prepared.target_means[prepared.decision_horizon_index],
+            "target_scale": prepared.target_scales[prepared.decision_horizon_index],
+            "clip": clip,
+            "validation_mae_percent": validation_mae * 100,
+            "baseline_mae_percent": baseline * 100,
+            "validation_skill": (baseline - validation_mae) / baseline if baseline > 0 else 0.0,
+            # Runtime and calibration may use validation skill. Untouched
+            # holdout skill is report-only and must never gate individual entries.
+            "skill": (baseline - validation_mae) / baseline if baseline > 0 else 0.0,
+            "train_samples": prepared.train_samples,
+            "validation_samples": prepared.validation_samples,
+            "holdout_samples": prepared.holdout_samples,
+            "holdout_start_timestamp": prepared.holdout_start_timestamp,
+        }
+    return results
+
+
+def _fit_pooled_candidate(
+    *, prepared_by_symbol: dict[str, PreparedData], architecture: str, input_size: int,
+    output_size: int, hidden_size: int, num_layers: int, dropout: float, epochs: int,
+    patience: int, batch_size: int, learning_rate: float, weight_decay: float,
+    clip: float, attention_pooling: bool, context_norm: bool, device: torch.device,
+    seed: int, selection_folds: int,
+) -> dict[str, Any]:
+    _seed(seed)
+    model = RecurrentReturnModel(
+        architecture=architecture, input_size=input_size, hidden_size=hidden_size,
+        num_layers=num_layers, dropout=dropout, output_size=output_size,
+        attention_pooling=attention_pooling, context_norm=context_norm,
+    ).to(device)
+    optimizer = torch.optim.AdamW(model.parameters(), lr=learning_rate, weight_decay=weight_decay)
+    loader = DataLoader(
+        TensorDataset(
+            torch.cat([row.train_x for row in prepared_by_symbol.values()]),
+            torch.cat([row.train_y for row in prepared_by_symbol.values()]),
+            torch.cat([row.train_up for row in prepared_by_symbol.values()]),
+        ),
+        batch_size=max(1, batch_size), shuffle=True,
+        generator=torch.Generator(device="cpu").manual_seed(seed),
+    )
+    best_state: dict[str, torch.Tensor] | None = None
+    best_score = math.inf
+    stale = 0
+    best_epoch = 0
+    for epoch in range(1, max(1, epochs) + 1):
+        model.train()
+        for batch_x, batch_y, batch_up in loader:
+            optimizer.zero_grad(set_to_none=True)
+            loss = _forecast_loss(model(batch_x), batch_y, batch_up, len(next(iter(prepared_by_symbol.values())).target_horizons))
+            loss.backward()
+            nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
+            optimizer.step()
+        symbol_rows = {
+            symbol: _validation_metrics(model, prepared, clip)
+            for symbol, prepared in prepared_by_symbol.items()
+        }
+        score = sum(float(row["validation_mae"]) for row in symbol_rows.values()) / len(symbol_rows)
+        if score + 1e-12 < best_score:
+            best_score = score
+            best_epoch = epoch
+            best_state = {key: value.detach().cpu().clone() for key, value in model.state_dict().items()}
+            stale = 0
+        else:
+            stale += 1
+            if stale >= max(1, patience):
+                break
+    if best_state:
+        model.load_state_dict(best_state)
+    per_symbol: dict[str, dict[str, Any]] = {}
+    for symbol, prepared in prepared_by_symbol.items():
+        metrics = _validation_metrics(model, prepared, clip)
+        metrics.update(_validation_stability_metrics(model, prepared, clip, selection_folds))
+        metrics.update(_holdout_metrics(model, prepared, clip))
+        per_symbol[symbol] = metrics
+    aggregate: dict[str, Any] = {"symbol_metrics": per_symbol}
+    for name in (
+        "validation_mae", "directional_accuracy", "buy_precision", "probability_brier",
+        "holdout_skill", "validation_fold_mae_std", "validation_trade_mean",
+        "validation_trade_win_rate",
+    ):
+        values = [float(row[name]) for row in per_symbol.values() if isinstance(row.get(name), (int, float))]
+        aggregate[name] = sum(values) / len(values) if values else 0.0
+    aggregate.update(
+        best_epoch=best_epoch, epochs_trained=best_epoch + stale,
+        state_dict=_export_recurrent_state(model),
+        head_weight=_round_nested(model.head.weight.detach().cpu().tolist()),
+        head_bias=_round_list(model.head.bias.detach().cpu().tolist()),
+        **_export_context_state(model),
+    )
+    return aggregate
 
 
 def _progress(message: str) -> None:
@@ -236,6 +506,9 @@ def _parse_args() -> argparse.Namespace:
     parser.add_argument("--attention-pooling", action=argparse.BooleanOptionalAction, default=True, help="Use exportable attention pooling over recurrent states.")
     parser.add_argument("--context-norm", action=argparse.BooleanOptionalAction, default=True, help="Use exportable LayerNorm before the forecast head.")
     parser.add_argument("--seed", type=int, default=7, help="Random seed.")
+    parser.add_argument("--ensemble-seeds", default="7,19,43", help="Comma-separated seeds averaged at inference time.")
+    parser.add_argument("--selection-folds", type=int, default=3, help="Validation slices used to penalize unstable candidates.")
+    parser.add_argument("--pooled", action=argparse.BooleanOptionalAction, default=True, help="Train shared multi-asset recurrent weights with learned symbol one-hot projection.")
     parser.add_argument("--threads", type=int, default=0, help="Torch CPU threads; 0 keeps torch default.")
     parser.add_argument("--device", default="auto", help="auto, cpu, cuda, or mps.")
     parser.add_argument("--output", default="", help="Output JSON path. Defaults to TIME_SERIES_LSTM_MODEL_PATH.")
@@ -277,7 +550,8 @@ def _train_symbol(
     attention_pooling: bool,
     context_norm: bool,
     device: torch.device,
-    seed: int,
+    seeds: list[int],
+    selection_folds: int,
 ) -> dict[str, Any] | None:
     candles = _historical_klines(client, symbol, interval, limit)
     add_indicators(candles)
@@ -339,25 +613,30 @@ def _train_symbol(
                             f"lookback={lookback} hidden={hidden_size} "
                             f"layers={num_layers} dropout={dropout}"
                         )
-                        candidate = _fit_candidate(
-                            prepared=prepared,
-                            architecture=architecture,
-                            input_size=len(feature_names),
-                            output_size=len(target_horizons) * len(OUTPUT_LAYOUT),
-                            hidden_size=hidden_size,
-                            num_layers=num_layers,
-                            dropout=dropout,
-                            epochs=epochs,
-                            patience=patience,
-                            batch_size=batch_size,
-                            learning_rate=learning_rate,
-                            weight_decay=weight_decay,
-                            clip=clip,
-                            attention_pooling=attention_pooling,
-                            context_norm=context_norm,
-                            device=device,
-                            seed=seed,
-                        )
+                        members = [
+                            _fit_candidate(
+                                prepared=prepared,
+                                architecture=architecture,
+                                input_size=len(feature_names),
+                                output_size=len(target_horizons) * len(OUTPUT_LAYOUT),
+                                hidden_size=hidden_size,
+                                num_layers=num_layers,
+                                dropout=dropout,
+                                epochs=epochs,
+                                patience=patience,
+                                batch_size=batch_size,
+                                learning_rate=learning_rate,
+                                weight_decay=weight_decay,
+                                clip=clip,
+                                attention_pooling=attention_pooling,
+                                context_norm=context_norm,
+                                device=device,
+                                seed=member_seed,
+                                selection_folds=selection_folds,
+                            )
+                            for member_seed in seeds
+                        ]
+                        candidate = _ensemble_candidate(members, seeds)
                         validation_mae = float(candidate["validation_mae"])
                         skill = (baseline_mae - validation_mae) / baseline_mae if baseline_mae > 0 else 0.0
                         row = {
@@ -408,7 +687,7 @@ def _train_symbol(
     if best is None:
         return None
     best["validation_skill"] = best.get("skill", 0.0)
-    best["skill"] = best.get("holdout_skill", 0.0)
+    best["skill"] = best["validation_skill"]
     best.pop("validation_mae", None)
     return best
 
@@ -484,7 +763,7 @@ def _prepare_data(
     if len(train_samples) < 24 or len(validation_samples) < 8 or len(holdout_samples) < 16:
         return None
 
-    feature_means, feature_scales = _feature_stats(train_samples, len(feature_names))
+    feature_means, feature_scales = _feature_stats(train_samples, feature_names)
     target_means, target_scales = _target_stats(train_samples, len(target_horizons))
     decision_horizon = decision_horizon if decision_horizon in target_horizons else min(
         target_horizons,
@@ -545,7 +824,8 @@ def _prepare_data(
     )
 
 
-def _feature_stats(samples: list[TrainingSample], input_size: int) -> tuple[list[float], list[float]]:
+def _feature_stats(samples: list[TrainingSample], feature_names: list[str]) -> tuple[list[float], list[float]]:
+    input_size = len(feature_names)
     columns = [[] for _ in range(input_size)]
     for sample in samples:
         window = sample.window
@@ -554,7 +834,11 @@ def _feature_stats(samples: list[TrainingSample], input_size: int) -> tuple[list
                 columns[index].append(float(row[index] if index < len(row) else 0.0))
     means: list[float] = []
     scales: list[float] = []
-    for values in columns:
+    for index, values in enumerate(columns):
+        if feature_names[index].startswith("symbol_is_"):
+            means.append(0.0)
+            scales.append(1.0)
+            continue
         if not values:
             means.append(0.0)
             scales.append(1.0)
@@ -641,6 +925,7 @@ def _fit_candidate(
     context_norm: bool,
     device: torch.device,
     seed: int,
+    selection_folds: int,
 ) -> dict[str, Any]:
     _seed(seed)
     model = RecurrentReturnModel(
@@ -676,6 +961,7 @@ def _fit_candidate(
             optimizer.step()
 
         metrics = _validation_metrics(model, prepared, clip)
+        metrics.update(_validation_stability_metrics(model, prepared, clip, selection_folds))
         if metrics["validation_mae"] + 1e-12 < best_metrics["validation_mae"]:
             best_metrics = metrics
             best_epoch = epoch
@@ -701,6 +987,66 @@ def _fit_candidate(
     }
 
 
+def _ensemble_candidate(members: list[dict[str, Any]], seeds: list[int]) -> dict[str, Any]:
+    if not members:
+        raise ValueError("ensemble requires at least one member")
+    result = dict(members[0])
+    metric_names = (
+        "validation_mae",
+        "directional_accuracy",
+        "buy_precision",
+        "probability_brier",
+        "holdout_mae",
+        "holdout_baseline_mae",
+        "holdout_skill",
+        "holdout_directional_accuracy",
+        "holdout_buy_precision",
+        "holdout_probability_brier",
+        "validation_fold_mae_std",
+        "validation_fold_mae_worst",
+        "validation_trade_mean",
+        "validation_trade_win_rate",
+    )
+    for name in metric_names:
+        values = [float(member[name]) for member in members if isinstance(member.get(name), (int, float))]
+        if values:
+            result[name] = sum(values) / len(values)
+    export_names = (
+        "state_dict",
+        "head_weight",
+        "head_bias",
+        "attention_weight",
+        "attention_bias",
+        "context_norm_weight",
+        "context_norm_bias",
+    )
+    result["ensemble_members"] = [
+        {name: member[name] for name in export_names if name in member}
+        | {"seed": seeds[index] if index < len(seeds) else index}
+        for index, member in enumerate(members)
+    ]
+    result["ensemble_size"] = len(members)
+    symbol_names = sorted(
+        {
+            symbol
+            for member in members
+            for symbol in (member.get("symbol_metrics") or {})
+        }
+    )
+    if symbol_names:
+        result["symbol_metrics"] = {}
+        for symbol in symbol_names:
+            rows = [member.get("symbol_metrics", {}).get(symbol, {}) for member in members]
+            keys = {key for row in rows if isinstance(row, dict) for key in row}
+            averaged: dict[str, float] = {}
+            for key in keys:
+                values = [float(row[key]) for row in rows if isinstance(row.get(key), (int, float))]
+                if values:
+                    averaged[key] = sum(values) / len(values)
+            result["symbol_metrics"][symbol] = averaged
+    return result
+
+
 def _validation_metrics(model: nn.Module, prepared: PreparedData, clip: float) -> dict[str, float]:
     return _evaluation_metrics(
         model,
@@ -712,6 +1058,39 @@ def _validation_metrics(model: nn.Module, prepared: PreparedData, clip: float) -
     )
 
 
+def _validation_stability_metrics(
+    model: nn.Module,
+    prepared: PreparedData,
+    clip: float,
+    folds: int,
+) -> dict[str, float]:
+    fold_count = max(1, min(int(folds), len(prepared.validation_targets)))
+    fold_size = max(1, len(prepared.validation_targets) // fold_count)
+    maes: list[float] = []
+    for fold in range(fold_count):
+        start = fold * fold_size
+        end = len(prepared.validation_targets) if fold == fold_count - 1 else min(len(prepared.validation_targets), start + fold_size)
+        if end <= start:
+            continue
+        metrics = _evaluation_metrics(
+            model,
+            values=prepared.validation_x[start:end],
+            targets=prepared.validation_targets[start:end],
+            volatility_scales=prepared.validation_volatility_scales[start:end],
+            prepared=prepared,
+            clip=clip,
+        )
+        maes.append(float(metrics["validation_mae"]))
+    if not maes:
+        return {"validation_fold_mae_std": 0.0, "validation_fold_mae_worst": math.inf}
+    mean = sum(maes) / len(maes)
+    variance = sum((value - mean) ** 2 for value in maes) / len(maes)
+    return {
+        "validation_fold_mae_std": math.sqrt(variance),
+        "validation_fold_mae_worst": max(maes),
+    }
+
+
 def _holdout_metrics(model: nn.Module, prepared: PreparedData, clip: float) -> dict[str, Any]:
     metrics = _evaluation_metrics(
         model,
@@ -790,6 +1169,13 @@ def _evaluation_metrics(
         if prediction > 0
     ]
     buy_wins = [actual for actual in buy_predictions if actual > 0]
+    ranked = sorted(
+        zip(decision_predictions, [row[decision] for row in probabilities], decision_targets),
+        key=lambda item: item[0] * max(0.0, item[1] - 0.5),
+        reverse=True,
+    )
+    selected = ranked[: max(8, len(ranked) // 5)]
+    selected_targets = [row[2] for row in selected]
     by_horizon = {}
     baseline_by_horizon = {}
     for horizon_index, horizon in enumerate(prepared.target_horizons):
@@ -815,6 +1201,12 @@ def _evaluation_metrics(
         "directional_accuracy": len(correct) / len(non_zero) if non_zero else 0.0,
         "buy_precision": len(buy_wins) / len(buy_predictions) if buy_predictions else 0.0,
         "probability_brier": sum(probability_errors) / len(probability_errors) if probability_errors else 1.0,
+        "validation_trade_mean": sum(selected_targets) / len(selected_targets) if selected_targets else 0.0,
+        "validation_trade_win_rate": (
+            sum(1 for value in selected_targets if value > 0) / len(selected_targets)
+            if selected_targets
+            else 0.0
+        ),
     }
 
 
@@ -824,9 +1216,13 @@ def _candidate_score(row: dict[str, Any]) -> float:
     directional = float(row.get("directional_accuracy", 0.0))
     buy_precision = float(row.get("buy_precision", 0.0))
     probability_brier = float(row.get("probability_brier", 1.0))
-    return mae * (1.0 - max(0.0, skill) * 0.05) * (1.0 - max(0.0, directional - 0.5) * 0.03) * (
+    fold_std = max(0.0, float(row.get("validation_fold_mae_std", 0.0)))
+    stability_penalty = 1.0 + min(1.0, fold_std / max(mae, 1e-9)) * 0.25
+    trade_mean = float(row.get("validation_trade_mean", 0.0))
+    trade_penalty = max(0.0, -trade_mean) * 2.0 - max(0.0, trade_mean) * 0.5
+    return mae * stability_penalty * (1.0 - max(0.0, skill) * 0.05) * (1.0 - max(0.0, directional - 0.5) * 0.03) * (
         1.0 - max(0.0, buy_precision - 0.5) * 0.02
-    ) * (1.0 + max(0.0, probability_brier - 0.25) * 0.02)
+    ) * (1.0 + max(0.0, probability_brier - 0.25) * 0.02) + trade_penalty
 
 
 def _forecast_loss(outputs: torch.Tensor, targets: torch.Tensor, up_targets: torch.Tensor, horizon_count: int) -> torch.Tensor:
@@ -842,7 +1238,19 @@ def _forecast_loss(outputs: torch.Tensor, targets: torch.Tensor, up_targets: tor
     probabilities = torch.sigmoid(logits)
     pt = probabilities * up_targets + (1.0 - probabilities) * (1.0 - up_targets)
     focal = ((1.0 - pt) ** 2.0 * bce).mean()
-    return mean_loss + 0.35 * sum(quantile_losses) / len(quantile_losses) + 0.15 * focal
+    soft_long = torch.sigmoid(values[:, :, 0] * 2.0) * probabilities
+    after_cost_utility = -(soft_long * targets).mean()
+    prediction_centered = values[:, :, 0] - values[:, :, 0].mean(dim=0, keepdim=True)
+    target_centered = targets - targets.mean(dim=0, keepdim=True)
+    cosine = nn.functional.cosine_similarity(prediction_centered, target_centered, dim=0).mean()
+    ranking_loss = 1.0 - cosine
+    return (
+        mean_loss
+        + 0.35 * sum(quantile_losses) / len(quantile_losses)
+        + 0.15 * focal
+        + 0.10 * after_cost_utility
+        + 0.05 * ranking_loss
+    )
 
 
 def _export_recurrent_state(model: RecurrentReturnModel) -> dict[str, Any]: