Sync from GitHub (tests passed)

app/features.py

@@ -39,6 +39,42 @@ logger = logging.getLogger(__name__)
 # Data Loading
 # =============================================================================
 
+def _normalize_bar_calendar_dates(raw_dates: pd.Series, utc_ts: pd.Series) -> pd.Series:
+    """Normalize bar dates to tz-naive midnight preserving explicit non-UTC exchange offsets."""
+    utc_norm = utc_ts.dt.tz_localize(None).dt.normalize()
+    if raw_dates.empty:
+        return utc_norm
+
+    needs_offset_scan = False
+    for raw_val in raw_dates:
+        if isinstance(raw_val, datetime):
+            offset = raw_val.utcoffset()
+            if offset is not None and offset != timedelta(0):
+                needs_offset_scan = True
+                break
+        elif isinstance(raw_val, str):
+            stripped = raw_val.strip()
+            if not stripped.endswith(("Z", "z", "+00:00", "-00:00")):
+                needs_offset_scan = True
+                break
+
+    if not needs_offset_scan:
+        return utc_norm
+
+    normalized = []
+    for raw_val, utc_val in zip(raw_dates, utc_ts):
+        if isinstance(raw_val, datetime):
+            normalized.append(raw_val.date())
+        elif isinstance(raw_val, str) and not raw_val.strip().endswith(
+            ("Z", "z", "+00:00", "-00:00")
+        ):
+            parsed = pd.to_datetime(raw_val, errors="coerce")
+            normalized.append(parsed.date() if pd.notna(parsed) else utc_val.date())
+        else:
+            normalized.append(utc_val.date())
+    return pd.to_datetime(normalized)
+
+
 def _rows_to_price_df(rows: list, symbol: str) -> pd.DataFrame:
     """Convert raw PriceBar rows into a validated, date-indexed DataFrame."""
     if not rows:
@@ -59,8 +95,73 @@ def _rows_to_price_df(rows: list, symbol: str) -> pd.DataFrame:
         df = df.loc[~invalid_close].copy()
     if df.empty:
         return pd.DataFrame()
-    df["date"] = pd.to_datetime(df["date"]).dt.tz_localize(None)
+
+    raw_ts = pd.to_datetime(df["date"], errors="coerce", utc=True)
+    invalid_dates = raw_ts.isna()
+    if invalid_dates.any():
+        logger.warning(
+            "Excluded %s rows with invalid date for %s", int(invalid_dates.sum()), symbol
+        )
+        df = df.loc[~invalid_dates].copy()
+        raw_ts = raw_ts.loc[~invalid_dates]
+        if df.empty:
+            return pd.DataFrame()
+
+    cal_dates = _normalize_bar_calendar_dates(df["date"], raw_ts)
+    df["_raw_ts"] = raw_ts.to_numpy()
+    df["date"] = cal_dates.to_numpy()
+
+    # Drop leaked synthetic test fixture rows (100/105/95/102/1000) when mixed with
+    # real market series whose price scale, volume scale, or calendar day differs.
+    fixture_mask = (
+        np.isclose(df["open"].astype(float), 100.0, rtol=1e-6, atol=1e-6)
+        & np.isclose(df["high"].astype(float), 105.0, rtol=1e-6, atol=1e-6)
+        & np.isclose(df["low"].astype(float), 95.0, rtol=1e-6, atol=1e-6)
+        & np.isclose(df["close"].astype(float), 102.0, rtol=1e-6, atol=1e-6)
+        & np.isclose(df["volume"].fillna(0.0).astype(float), 1000.0, rtol=1e-6, atol=1e-6)
+    )
+    if fixture_mask.any() and (~fixture_mask).any():
+        non_fixture = df.loc[~fixture_mask]
+        non_fixture_median = float(non_fixture["close"].astype(float).median())
+        non_fixture_vols = non_fixture["volume"].dropna().astype(float)
+        non_fixture_vol_median = (
+            float(non_fixture_vols.median()) if not non_fixture_vols.empty else None
+        )
+        price_mismatch = abs(non_fixture_median - 102.0) > 25.0
+        vol_mismatch = (
+            non_fixture_vol_median is not None
+            and (non_fixture_vol_median < 100.0 or non_fixture_vol_median > 10000.0)
+        )
+        if price_mismatch or vol_mismatch:
+            logger.warning(
+                "Excluded %s leaked synthetic fixture rows for %s "
+                "(non-fixture close median=%.2f, volume median=%s)",
+                int(fixture_mask.sum()),
+                symbol,
+                non_fixture_median,
+                f"{non_fixture_vol_median:.1f}" if non_fixture_vol_median is not None else "n/a",
+            )
+            df = non_fixture.copy()
+        else:
+            colliding_fixture = fixture_mask & df["date"].isin(set(non_fixture["date"]))
+            if colliding_fixture.any():
+                logger.warning(
+                    "Excluded %s colliding synthetic fixture rows for %s",
+                    int(colliding_fixture.sum()),
+                    symbol,
+                )
+                df = df.loc[~colliding_fixture].copy()
+        if df.empty:
+            return pd.DataFrame()
+
+    df = df.sort_values("_raw_ts").drop(columns=["_raw_ts"])
     df = df.set_index("date")
+    if df.index.duplicated().any():
+        dup_count = int(df.index.duplicated(keep="last").sum())
+        logger.warning(
+            "Deduplicated %s same-calendar-day price bars for %s", dup_count, symbol
+        )
+        df = df[~df.index.duplicated(keep="last")]
     return df
 
 

deep_learning/data/feature_store.py

@@ -336,6 +336,71 @@ def _build_calendar_features(index: pd.DatetimeIndex) -> pd.DataFrame:
 # Price / technical features (reuses existing helpers)
 # ---------------------------------------------------------------------------
 
+def _sanitize_close_spikes(
+    close_series: pd.Series,
+    symbol: str = "target",
+    *,
+    max_abs_log_jump: float = 0.35,
+    confirm_bars: int = 2,
+    extreme_log_jump: float = 1.0,
+    extreme_confirm_bars: int = 3,
+) -> pd.Series:
+    """Replace transient corrupt price spikes causally without flatlining persistent shifts."""
+    close = close_series.astype(float).where(lambda s: s > 0.0).ffill()
+    if len(close) <= 1:
+        return close
+
+    close_vals = close.to_numpy(dtype=float, copy=False)
+    valid = np.isfinite(close_vals) & (close_vals > 0.0)
+    log_vals = np.zeros_like(close_vals)
+    log_vals[valid] = np.log(close_vals[valid])
+    has_jumps = bool((valid[1:] & valid[:-1] & (np.abs(np.diff(log_vals)) > max_abs_log_jump)).any())
+    if not has_jumps:
+        return close
+
+    close_vals = close_vals.copy()
+    repaired = 0
+    pending_anchor: float | None = None
+    pending_run = 0
+    for idx in range(1, len(close_vals)):
+        prev_val = close_vals[idx - 1]
+        cur_val = close_vals[idx]
+        if not (np.isfinite(prev_val) and prev_val > 0 and np.isfinite(cur_val) and cur_val > 0):
+            pending_anchor = None
+            pending_run = 0
+            continue
+        log_jump = abs(np.log(cur_val / prev_val))
+        if log_jump > max_abs_log_jump:
+            needed = extreme_confirm_bars if log_jump > extreme_log_jump else confirm_bars
+            if (
+                pending_anchor is not None
+                and abs(np.log(cur_val / pending_anchor)) <= max_abs_log_jump
+            ):
+                pending_run += 1
+                pending_anchor = cur_val
+            else:
+                pending_anchor = cur_val
+                pending_run = 1
+            if pending_run >= needed:
+                # Persistent shift confirmed purely from past/current bars; resume true level.
+                pending_anchor = None
+                pending_run = 0
+            else:
+                close_vals[idx] = prev_val
+                repaired += 1
+        else:
+            pending_anchor = None
+            pending_run = 0
+    if repaired:
+        logger.warning(
+            "Repaired %d corrupt single-day close spike(s) (> %.2f log jump) for %s",
+            repaired,
+            max_abs_log_jump,
+            symbol,
+        )
+    return pd.Series(close_vals, index=close_series.index, name=close_series.name)
+
+
 def _build_price_features(
     session,
     symbol: str,
@@ -355,6 +420,11 @@ def _build_price_features(
     if price_df.empty:
         return pd.DataFrame(), pd.DataFrame()
 
+    sanitized_close = _sanitize_close_spikes(price_df["close"], symbol=symbol)
+    if not sanitized_close.equals(price_df["close"]):
+        price_df = price_df.copy()
+        price_df["close"] = sanitized_close
+
     features = generate_symbol_features(price_df, symbol)
     return price_df, features
 
@@ -624,7 +694,7 @@ def build_tft_dataframe(
     calendar_features = _build_calendar_features(target_index)
 
     # ---- 6. Target: daily log-return path plus explicit weekly helpers ----
-    close = price_df["close"].astype(float)
+    close = _sanitize_close_spikes(price_df["close"], symbol=target_symbol)
     log_close = np.log(close)
 
     target_1d = log_close.shift(-cfg.forecast.auxiliary_horizon_days) - log_close
@@ -642,7 +712,13 @@ def build_tft_dataframe(
     observed_return_1d = log_close.diff().fillna(0.0)
     observed_return_1d.name = "observed_return_1d"
 
-    realized_vol_20d = target_1d.rolling(20, min_periods=10).std()
+    realized_vol_20d = (
+        observed_return_1d.rolling(20, min_periods=5)
+        .std()
+        .replace(0, np.nan)
+        .ffill()
+        .fillna(0.01)
+    )
     realized_vol_20d.name = "realized_vol_20d"
 
     material_move_5d = (

deep_learning/data/validation.py

@@ -39,6 +39,12 @@ def validate_weekly_target_contract(
         for col in ("target", "target_1d_log_return", "target_5d_log_return"):
             if df[col].isna().any():
                 raise ValueError(f"{col} contains NaN after feature-store construction")
+        valid_1d = df["target_1d_log_return"].dropna()
+        max_abs_1d = float(valid_1d.abs().max()) if not valid_1d.empty else 0.0
+        if max_abs_1d > 0.50:
+            raise ValueError(
+                f"Corrupt single-day log return detected in training frame: max_abs={max_abs_1d:.4f} > 0.50"
+            )
 
     comparable = df[["target", "target_1d_log_return"]].dropna()
     if not comparable.empty and not np.allclose(

deep_learning/inference/predictor.py

@@ -144,6 +144,8 @@ class TFTPredictor:
         self._direction_sign_multiplier = 1
         self._daily_sign_multiplier = 1
         self._weekly_sign_threshold = 0.0
+        self._weekly_magnitude_scale = 1.0
+        self._weekly_median_cap: Optional[float] = None
         self._weekly_direction_model: dict = {}
         self._weekly_interval_scale = 1.0
         self._weekly_interval_conditioning: dict = {}
@@ -286,6 +288,35 @@ class TFTPredictor:
                 "Incompatible TFT checkpoint: invalid validation-fitted weekly sign threshold. Retraining required."
             )
         self._weekly_sign_threshold = weekly_sign_threshold
+        magnitude_cal = metadata.get("magnitude_calibration") or {}
+        weekly_magnitude_scale = float(
+            direction_calibration.get(
+                "weekly_magnitude_scale",
+                magnitude_cal.get("scale", 1.0),
+            )
+            or 1.0
+        )
+        if not np.isfinite(weekly_magnitude_scale) or weekly_magnitude_scale <= 0.0 or weekly_magnitude_scale > 5.0:
+            raise IncompatibleTFTCheckpointError(
+                "Incompatible TFT checkpoint: invalid validation-fitted weekly magnitude scale. Retraining required."
+            )
+        self._weekly_magnitude_scale = weekly_magnitude_scale
+        raw_cap = config.get(
+            "weekly_median_cap",
+            (metadata.get("target_scale_audit") or {}).get(
+                "weekly_median_cap",
+                metadata.get("weekly_median_cap"),
+            ),
+        )
+        if raw_cap is not None:
+            cap_val = float(raw_cap)
+            if not np.isfinite(cap_val) or cap_val <= 0.0:
+                raise IncompatibleTFTCheckpointError(
+                    "Incompatible TFT checkpoint: invalid training weekly_median_cap. Retraining required."
+                )
+            self._weekly_median_cap = cap_val
+        else:
+            self._weekly_median_cap = None
         weekly_direction_model = metadata.get("weekly_direction_model") or {}
         if weekly_direction_model.get("enabled"):
             required_fields = {"feature_names", "mean", "scale", "coef", "intercept"}
@@ -561,6 +592,14 @@ class TFTPredictor:
                     threshold=float(self._weekly_direction_model.get("decision_threshold", 0.50)),
                     horizon=self.cfg.forecast.primary_horizon_days,
                 )[0]
+            if abs(self._weekly_magnitude_scale - 1.0) > 1e-6:
+                from deep_learning.training.metrics import apply_weekly_magnitude_scale_np
+
+                pred_for_format = apply_weekly_magnitude_scale_np(
+                    pred_for_format[None, ...],
+                    self._weekly_magnitude_scale,
+                    quantiles=tuple(self.cfg.model.quantiles),
+                )[0]
             if self._weekly_interval_conditioning:
                 from deep_learning.training.metrics import apply_weekly_interval_scale_np
 
@@ -593,6 +632,20 @@ class TFTPredictor:
                     self._weekly_interval_scale,
                     quantiles=tuple(self.cfg.model.quantiles),
                 )[0]
+            if self._weekly_median_cap is not None and self._weekly_median_cap > 0.0:
+                from deep_learning.training.metrics import apply_weekly_median_cap_np
+
+                effective_horizon = min(
+                    int(self.cfg.forecast.primary_horizon_days),
+                    int(pred_for_format.shape[0]),
+                )
+                if effective_horizon > 0:
+                    pred_for_format = apply_weekly_median_cap_np(
+                        pred_for_format[None, ...],
+                        weekly_median_cap=self._weekly_median_cap,
+                        quantiles=tuple(self.cfg.model.quantiles),
+                        horizon=effective_horizon,
+                    )[0][0]
 
         except IncompatibleTFTCheckpointError as exc:
             logger.warning("TFT checkpoint incompatible: %s", exc)
@@ -638,6 +691,8 @@ class TFTPredictor:
             "direction_sign_multiplier": self._direction_sign_multiplier,
             "daily_sign_multiplier": self._daily_sign_multiplier,
             "weekly_sign_threshold": self._weekly_sign_threshold,
+            "weekly_magnitude_scale": self._weekly_magnitude_scale,
+            "weekly_median_cap": self._weekly_median_cap,
             "weekly_direction_model_enabled": bool(self._weekly_direction_model),
             "weekly_interval_scale": self._weekly_interval_scale,
             "weekly_interval_conditioning_enabled": bool(self._weekly_interval_conditioning),

deep_learning/models/tft_copper.py

@@ -111,11 +111,16 @@ def _weekly_saturation_loss(
     # one per sample.  This creates deployment margin without letting extreme
     # initialization outliers dominate the objective again.
     moderate_violation_barrier = torch.tanh(above_cap_excess).pow(2)
+    # Keep a sublinear sqrt restoring penalty on moderate excess above the cap
+    # so gradients remain strong at 1.2x-5x violations while staying sublinear
+    # on extreme initialization outliers.
+    sqrt_excess_penalty = torch.sqrt(1.0 + above_cap_excess) - 1.0
 
     return (
         0.10 * _robust_squared(near_cap_excess).mean()
         + 2.0 * _robust_squared(above_cap_excess).mean()
         + moderate_violation_barrier.mean()
+        + 1.5 * sqrt_excess_penalty.mean()
     )
 
 
@@ -123,20 +128,23 @@ def _weekly_scale_losses(
     pred_weekly_median: torch.Tensor,
     actual_weekly: torch.Tensor,
     eps: float = 1e-8,
+    raw_pred_weekly_median: Optional[torch.Tensor] = None,
 ) -> dict[str, torch.Tensor]:
     """
     Scale diagnostics used by WeeklyASROPFLoss.
 
     Hyperopt promotes on weekly_magnitude_ratio, so the loss must penalize the
     same failure mode directly instead of relying only on slowly growing
-    log-ratio pressure.
+    log-ratio pressure. When raw_pred_weekly_median is provided, also apply
+    scale penalties before cap clipping so gradients never zero out at the cap.
     """
+    actual_ref = actual_weekly.clamp(min=-0.25, max=0.25)
     pred_abs_median = pred_weekly_median.abs().median() + eps
-    actual_abs_median = actual_weekly.abs().median() + eps
+    actual_abs_median = actual_ref.abs().median() + eps
     magnitude_ratio = pred_abs_median / actual_abs_median
 
     pred_abs_mean = pred_weekly_median.abs().mean() + eps
-    actual_abs_mean = actual_weekly.abs().mean() + eps
+    actual_abs_mean = actual_ref.abs().mean() + eps
     mean_magnitude_ratio = pred_abs_mean / actual_abs_mean
 
     def _bounded_scale_loss(ratio: torch.Tensor) -> torch.Tensor:
@@ -150,23 +158,23 @@ def _weekly_scale_losses(
             + 4.0 * structural_explosion.pow(2)
         )
 
-    # The median ratio remains available as the production diagnostic, but a
-    # batch median is an order statistic: its active element can change after
-    # an otherwise negligible floating-point perturbation.  Feeding that
-    # gradient into the optimizer made otherwise identical CPU runs follow
-    # different paths.  Use the smooth mean ratio for optimization while
-    # preserving the median ratio used by hyperopt and the quality gate.
-    magnitude_loss = _bounded_scale_loss(mean_magnitude_ratio)
-
-    pred_std = pred_weekly_median.std(unbiased=False) + eps
-    actual_std = actual_weekly.std(unbiased=False) + eps
-    # A batch can temporarily collapse to almost one weekly value.  The raw
-    # log-ratio then reaches ~15 and its 1/pred_std derivative overwhelms the
-    # rest of the objective, making cross-run CPU round-off choose a different
-    # basin.  Keep the dispersion signal in a broad, pre-specified [0.25, 4]
-    # ratio band and use a smooth-L1 penalty; this still discourages material
-    # under/over-dispersion without allowing one pathological batch to steer
-    # the whole epoch.
+    bounded_mag_loss = _bounded_scale_loss(mean_magnitude_ratio)
+    if raw_pred_weekly_median is not None:
+        raw_abs_mean = raw_pred_weekly_median.abs().mean() + eps
+        raw_ratio = raw_abs_mean / actual_abs_mean
+        raw_mag_loss = (
+            _bounded_scale_loss(raw_ratio)
+            + torch.relu(raw_ratio - 1.20).pow(2)
+            + torch.relu(0.75 - raw_ratio).pow(2)
+        )
+        magnitude_loss = 0.5 * bounded_mag_loss + 0.5 * raw_mag_loss
+        effective_pred_weekly_median = 0.5 * pred_weekly_median + 0.5 * raw_pred_weekly_median
+    else:
+        magnitude_loss = bounded_mag_loss
+        effective_pred_weekly_median = pred_weekly_median
+
+    pred_std = effective_pred_weekly_median.std(unbiased=False) + eps
+    actual_std = actual_ref.std(unbiased=False) + eps
     dispersion_log_error = torch.log(
         (pred_std / actual_std).clamp(min=0.25, max=4.0)
     )
@@ -176,16 +184,10 @@ def _weekly_scale_losses(
         beta=0.5,
     )
 
-    model_mae = torch.mean(torch.abs(pred_weekly_median - actual_weekly))
+    model_mae = torch.mean(torch.abs(effective_pred_weekly_median - actual_ref))
     zero_mae = actual_abs_mean
     naive_relative_loss = torch.relu((model_mae / zero_mae) - 1.0)
-    mean_gap = (pred_weekly_median.mean() - actual_weekly.mean()) / zero_mae
-    median_gap = (
-        (pred_weekly_median.median() - actual_weekly.median())
-        / actual_abs_median
-    )
-    # As above, keep the median gap as a diagnostic but avoid its
-    # order-statistic gradient in the train objective.
+    mean_gap = (effective_pred_weekly_median.mean() - actual_ref.mean()) / zero_mae
     bias_loss = 1.50 * torch.abs(mean_gap)
 
     return {
@@ -286,20 +288,25 @@ def _weekly_interval_undercoverage_loss(
     upper = pred_weekly_quantiles[:, q90_idx]
     width = torch.clamp(upper - lower, min=0.0)
 
-    actual_scale = actual_weekly.abs().mean().clamp_min(eps)
+    actual_ref = actual_weekly.clamp(min=-0.25, max=0.25)
+    actual_scale = actual_ref.abs().mean().clamp_min(eps)
     miss_loss = (
-        torch.relu(lower - actual_weekly)
-        + torch.relu(actual_weekly - upper)
+        torch.relu(lower - actual_ref)
+        + torch.relu(actual_ref - upper)
     ).mean() / actual_scale
 
-    actual_std = actual_weekly.std(unbiased=False).clamp_min(eps)
+    actual_std = actual_ref.std(unbiased=False).clamp_min(eps)
     width_ratio = width.mean() / (2.56 * actual_std + eps)
     target_width = torch.as_tensor(
         float(target_width_ratio),
         device=pred_weekly_quantiles.device,
         dtype=pred_weekly_quantiles.dtype,
     )
-    width_scale_loss = (width_ratio - target_width).pow(2)
+    width_scale_loss = F.smooth_l1_loss(
+        width_ratio.clamp(max=4.0),
+        target_width.clamp(max=4.0),
+        beta=0.25,
+    )
     return miss_loss + width_scale_loss
 
 
@@ -440,6 +447,11 @@ try:
             self.sharpe_eps = sharpe_eps
             self.debug_mode = debug_mode
             self.median_idx = len(quantiles) // 2
+            self.register_buffer(
+                "_pinball_q_buf",
+                torch.tensor(quantiles, dtype=torch.float32).view(1, -1),
+                persistent=False,
+            )
             self.reset_component_accumulators()
 
         def reset_component_accumulators(self) -> None:
@@ -539,7 +551,7 @@ try:
             }
 
         def _pinball(self, pred: torch.Tensor, actual: torch.Tensor) -> torch.Tensor:
-            q = torch.tensor(self.quantiles, device=pred.device, dtype=pred.dtype).view(1, -1)
+            q = self._pinball_q_buf.to(device=pred.device, dtype=pred.dtype)
             err = actual.unsqueeze(-1) - pred
             return torch.maximum(q * err, (q - 1.0) * err).mean()
 
@@ -592,7 +604,20 @@ try:
             pred_weekly_quantiles = ordered_pred.sum(dim=1)
             actual_weekly = y_actual.sum(dim=1)
 
-            weekly_q_loss = self._pinball(pred_weekly_quantiles, actual_weekly)
+            weekly_q_loss_bounded = self._pinball(pred_weekly_quantiles, actual_weekly)
+            if self.weekly_median_cap is not None and float(self.weekly_median_cap) > 0.0:
+                # Non-median quantile spreads from enforce_monotonic_quantiles depend only on
+                # non-median channels (untouched by _bound_weekly_median_path), so the raw
+                # ordered weekly sum equals pred_weekly_quantiles shifted by the median diff.
+                median_shift_sum = (raw_median_path - median_path).sum(dim=1, keepdim=True)
+                weekly_q_loss_raw = self._pinball(
+                    pred_weekly_quantiles + median_shift_sum,
+                    actual_weekly,
+                )
+                weekly_q_loss = 0.7 * weekly_q_loss_bounded + 0.3 * weekly_q_loss_raw
+            else:
+                weekly_q_loss = weekly_q_loss_bounded
+
             t1_q_loss = super().loss(ordered_pred[:, 0:1, :], y_actual[:, 0:1])
             t1_directional_loss = _directional_sign_loss(
                 median_path[:, 0],
@@ -603,7 +628,12 @@ try:
 
             pred_weekly_median = median_path.sum(dim=1)
             eps = self.sharpe_eps
-            scale_losses = _weekly_scale_losses(pred_weekly_median, actual_weekly, eps=eps)
+            scale_losses = _weekly_scale_losses(
+                pred_weekly_median,
+                actual_weekly,
+                eps=eps,
+                raw_pred_weekly_median=raw_pred_weekly_median,
+            )
             dispersion_loss = scale_losses["dispersion_loss"]
             magnitude_loss = scale_losses["magnitude_loss"]
             naive_relative_loss = scale_losses["naive_relative_loss"]

deep_learning/training/hyperopt.py

@@ -173,7 +173,23 @@ def _select_preflight_safe_trial(study, fold_diagnostics: list[dict]):
         if candidate["preflight_passed"]
     ]
     if eligible:
-        selected = min(eligible, key=lambda trial: float(trial.value))
+        gate_safe = [
+            trial
+            for trial in eligible
+            if 0.65 <= float(
+                trial.user_attrs.get(
+                    "avg_weekly_magnitude_ratio",
+                    diagnostics_by_trial.get(int(trial.number), {}).get("avg_weekly_magnitude_ratio", 1.0),
+                )
+            ) <= 1.35
+            and float(
+                trial.user_attrs.get(
+                    "avg_weekly_raw_magnitude_ratio",
+                    diagnostics_by_trial.get(int(trial.number), {}).get("avg_weekly_raw_magnitude_ratio", 1.0),
+                )
+            ) <= 3.0
+        ]
+        selected = min(gate_safe or eligible, key=lambda trial: float(trial.value))
         mode = "lowest_objective_preflight_pass"
     elif finite_trials:
         selected = min(finite_trials, key=lambda trial: float(trial.value))
@@ -337,6 +353,8 @@ def _build_prune_diagnostics(study) -> tuple[dict[str, int], list[dict]]:
         "avg_raw_median_sort_gap",
         "avg_median_sort_gap",
         "avg_weekly_magnitude_ratio",
+        "avg_weekly_raw_magnitude_ratio",
+        "avg_weekly_median_bound_rate",
         "avg_weekly_pi80_coverage",
         "avg_weekly_pred_positive_rate",
         "avg_weekly_actual_positive_rate",
@@ -574,7 +592,10 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
     import torch
     from deep_learning.data.dataset import build_cv_folds, create_dataloaders
     from deep_learning.models.tft_copper import create_tft_model
-    from deep_learning.training.callbacks import CurriculumLossScheduler
+    from deep_learning.training.callbacks import (
+        CurriculumLossScheduler,
+        WeeklyLossComponentLogger,
+    )
     from deep_learning.training.metrics import (
         apply_weekly_median_cap_np,
         cumulative_horizon,
@@ -608,6 +629,8 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
     fold_raw_median_gap_list: list[float] = []
     fold_weekly_objectives: list[float] = []
     fold_weekly_mr_list: list[float] = []
+    fold_weekly_raw_mr_list: list[float] = []
+    fold_weekly_bound_rate_list: list[float] = []
     fold_weekly_pi80_coverage_list: list[float] = []
     fold_weekly_pred_positive_rate_list: list[float] = []
     fold_weekly_actual_positive_rate_list: list[float] = []
@@ -677,6 +700,7 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
             return float("inf")
 
         callbacks = [
+            WeeklyLossComponentLogger(),
             EarlyStopping(
                 monitor="val_loss",
                 patience=trial_cfg.training.early_stopping_patience,
@@ -821,6 +845,8 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
                 horizon=fold_cfg.forecast.primary_horizon_days,
             )
             fold_weekly_mr = float(metrics.get("weekly_magnitude_ratio", 1.0))
+            fold_weekly_raw_mr = float(metrics.get("weekly_raw_magnitude_ratio", fold_weekly_mr))
+            fold_weekly_bound_rate = float(metrics.get("weekly_median_bound_applied_rate", 0.0))
             fold_weekly_pi80_coverage = float(metrics.get("weekly_pi80_coverage", 0.0))
             fold_weekly_pred_positive_rate = float(metrics.get("weekly_pred_positive_rate", 0.5))
             fold_weekly_actual_positive_rate = float(metrics.get("weekly_actual_positive_rate", 0.5))
@@ -894,6 +920,14 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
             )
             width_penalty = max(0.0, fold_weekly_pi80_width_ratio - 1.5)
             tail_width_penalty = max(0.0, fold_weekly_pi96_width_ratio - 3.0)
+            raw_magnitude_penalty = (
+                0.40 * max(0.0, fold_weekly_raw_mr - 1.8)
+                + 2.0 * max(0.0, fold_weekly_raw_mr - 3.0)
+            )
+            bounded_magnitude_band_penalty = 1.50 * (
+                max(0.0, fold_weekly_mr - 1.30) + max(0.0, 0.70 - fold_weekly_mr)
+            )
+            bound_rate_penalty = 0.35 * max(0.0, fold_weekly_bound_rate - 0.30)
             fold_weekly_objective = (
                 0.35 * weekly_pinball
                 + 0.15 * (1.0 - float(metrics.get("weekly_directional_accuracy", 0.5)))
@@ -905,6 +939,9 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
                 + 0.10 * interval_score_penalty
                 + 0.05 * interval_score_96_penalty
                 + 0.25 * fold_weekly_sorted_crossing
+                + raw_magnitude_penalty
+                + bounded_magnitude_band_penalty
+                + bound_rate_penalty
             )
         except Exception as exc:
             logger.warning(
@@ -921,6 +958,8 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
         fold_raw_median_gap_list.append(fold_raw_median_gap)
         fold_weekly_objectives.append(fold_weekly_objective)
         fold_weekly_mr_list.append(fold_weekly_mr)
+        fold_weekly_raw_mr_list.append(fold_weekly_raw_mr)
+        fold_weekly_bound_rate_list.append(fold_weekly_bound_rate)
         fold_weekly_pi80_coverage_list.append(fold_weekly_pi80_coverage)
         fold_weekly_pred_positive_rate_list.append(fold_weekly_pred_positive_rate)
         fold_weekly_actual_positive_rate_list.append(fold_weekly_actual_positive_rate)
@@ -1078,6 +1117,12 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
         float(np.mean(fold_raw_median_gap_list)) if fold_raw_median_gap_list else 0.0
     )
     avg_weekly_mr = float(np.mean(fold_weekly_mr_list)) if fold_weekly_mr_list else 1.0
+    avg_weekly_raw_mr = (
+        float(np.mean(fold_weekly_raw_mr_list)) if fold_weekly_raw_mr_list else avg_weekly_mr
+    )
+    avg_weekly_bound_rate = (
+        float(np.mean(fold_weekly_bound_rate_list)) if fold_weekly_bound_rate_list else 0.0
+    )
     avg_weekly_pi80_coverage = (
         float(np.mean(fold_weekly_pi80_coverage_list))
         if fold_weekly_pi80_coverage_list
@@ -1145,6 +1190,8 @@ def _objective(trial, base_cfg: TFTASROConfig, master_data: tuple) -> float:
     trial.set_user_attr("avg_raw_median_sort_gap", round(avg_raw_median_gap, 4))
     trial.set_user_attr("avg_median_sort_gap", round(avg_median_gap, 4))
     trial.set_user_attr("avg_weekly_magnitude_ratio", round(avg_weekly_mr, 4))
+    trial.set_user_attr("avg_weekly_raw_magnitude_ratio", round(avg_weekly_raw_mr, 4))
+    trial.set_user_attr("avg_weekly_median_bound_rate", round(avg_weekly_bound_rate, 4))
     trial.set_user_attr("avg_weekly_pi80_coverage", round(avg_weekly_pi80_coverage, 4))
     trial.set_user_attr("avg_weekly_pred_positive_rate", round(avg_weekly_pred_positive_rate, 4))
     trial.set_user_attr("avg_weekly_actual_positive_rate", round(avg_weekly_actual_positive_rate, 4))

deep_learning/training/metrics.py

@@ -954,6 +954,150 @@ def apply_weekly_sign_correction_np(
     return arr
 
 
+def apply_weekly_magnitude_scale_np(
+    pred: np.ndarray,
+    scale: float = 1.0,
+    *,
+    quantiles: tuple[float, ...] = (0.02, 0.10, 0.25, 0.50, 0.75, 0.90, 0.98),
+) -> np.ndarray:
+    """Scale the median return path while shifting quantiles uniformly to preserve order."""
+    arr = np.asarray(pred, dtype=np.float64).copy()
+    if arr.ndim != 3:
+        raise ValueError(f"Expected [n,horizon,q] predictions, got {arr.shape}")
+    if arr.shape[-1] != len(quantiles):
+        raise ValueError(
+            f"Quantile dim mismatch: prediction has {arr.shape[-1]}, config has {len(quantiles)}"
+        )
+    s = float(scale)
+    if not np.isfinite(s) or s <= 0.0:
+        raise ValueError(f"Weekly magnitude scale must be finite and positive, got {scale!r}")
+    if abs(s - 1.0) <= 1e-6 or arr.shape[0] == 0:
+        return arr
+
+    median_idx = len(quantiles) // 2
+    median = arr[..., median_idx : median_idx + 1]
+    return arr + (s - 1.0) * median
+
+
+def fit_weekly_magnitude_scale(
+    y_actual_path: np.ndarray,
+    y_pred_quantiles_path: np.ndarray,
+    *,
+    horizon: int = 5,
+    quantiles: tuple[float, ...] = (0.02, 0.10, 0.25, 0.50, 0.75, 0.90, 0.98),
+    weekly_median_cap: float | None = None,
+    target_ratio: float = 1.0,
+    min_ratio_band: float = 0.85,
+    max_ratio_band: float = 1.15,
+    min_scale: float = 0.15,
+    max_scale: float = 1.35,
+    min_samples: int = 10,
+) -> dict[str, float | int | bool | str]:
+    """Fit a validation-only median magnitude scale multiplier.
+
+    Rescales over- or under-confident raw median paths using chronological
+    validation targets so validation predictions do not saturate against
+    weekly_median_cap and test/live magnitude ratios remain inside gate bounds.
+    """
+    actual_path = np.asarray(y_actual_path, dtype=np.float64)
+    pred_path = np.asarray(y_pred_quantiles_path, dtype=np.float64)
+    if actual_path.ndim < 2 or pred_path.ndim != 3:
+        raise ValueError("Magnitude calibration requires [n,horizon] targets and [n,horizon,q] predictions")
+    if actual_path.shape[1] < horizon or pred_path.shape[1] < horizon:
+        raise ValueError(f"Magnitude calibration requires a {horizon}-step prediction path")
+
+    n = min(len(actual_path), len(pred_path))
+    median_idx = len(quantiles) // 2
+    actual_path = actual_path[:n]
+    pred_path = pred_path[:n]
+    weekly_actual = cumulative_horizon(actual_path, horizon=horizon)
+
+    raw_ordered = monotonic_quantiles_np(pred_path, median_idx=median_idx)
+    raw_weekly_pred = raw_ordered[:, :horizon, median_idx].sum(axis=1)
+    raw_ratio = magnitude_ratio(weekly_actual, raw_weekly_pred)
+
+    if weekly_median_cap is not None and float(weekly_median_cap) > 0.0:
+        bounded_path, _ = apply_weekly_median_cap_np(
+            pred_path,
+            weekly_median_cap=weekly_median_cap,
+            quantiles=quantiles,
+            horizon=horizon,
+        )
+        bounded_ordered = monotonic_quantiles_np(bounded_path, median_idx=median_idx)
+        bounded_weekly_pred = bounded_ordered[:, :horizon, median_idx].sum(axis=1)
+        bounded_ratio = magnitude_ratio(weekly_actual, bounded_weekly_pred)
+    else:
+        bounded_ratio = raw_ratio
+
+    selected_scale = 1.0
+    applied = False
+    raw_ratio_after = raw_ratio
+    bounded_ratio_after = bounded_ratio
+
+    if (
+        n >= int(min_samples)
+        and raw_ratio > 1e-6
+        and (
+            bounded_ratio < float(min_ratio_band)
+            or bounded_ratio > float(max_ratio_band)
+            or raw_ratio > 1.50
+        )
+    ):
+        candidate_scale = float(
+            np.clip(float(target_ratio) / raw_ratio, float(min_scale), float(max_scale))
+        )
+        if abs(candidate_scale - 1.0) > 1e-3:
+            scaled_path = apply_weekly_magnitude_scale_np(
+                pred_path,
+                candidate_scale,
+                quantiles=quantiles,
+            )
+            scaled_raw_ordered = monotonic_quantiles_np(scaled_path, median_idx=median_idx)
+            scaled_raw_weekly = scaled_raw_ordered[:, :horizon, median_idx].sum(axis=1)
+            candidate_raw_ratio = magnitude_ratio(weekly_actual, scaled_raw_weekly)
+            if weekly_median_cap is not None and float(weekly_median_cap) > 0.0:
+                scaled_bounded_path, _ = apply_weekly_median_cap_np(
+                    scaled_path,
+                    weekly_median_cap=weekly_median_cap,
+                    quantiles=quantiles,
+                    horizon=horizon,
+                )
+                scaled_bounded_ordered = monotonic_quantiles_np(
+                    scaled_bounded_path,
+                    median_idx=median_idx,
+                )
+                candidate_bounded_ratio = magnitude_ratio(
+                    weekly_actual,
+                    scaled_bounded_ordered[:, :horizon, median_idx].sum(axis=1),
+                )
+            else:
+                candidate_bounded_ratio = candidate_raw_ratio
+
+            improves_bounded = abs(candidate_bounded_ratio - float(target_ratio)) < abs(
+                bounded_ratio - float(target_ratio)
+            )
+            improves_raw = raw_ratio > 1.50 and candidate_raw_ratio < raw_ratio
+            if (improves_bounded or improves_raw) and (
+                0.65 <= candidate_bounded_ratio <= 1.35 or improves_bounded
+            ):
+                selected_scale = candidate_scale
+                applied = True
+                raw_ratio_after = candidate_raw_ratio
+                bounded_ratio_after = candidate_bounded_ratio
+
+    return {
+        "fit_split": "validation",
+        "sample_count": int(n),
+        "applied": bool(applied),
+        "scale": float(selected_scale),
+        "weekly_magnitude_scale": float(selected_scale),
+        "validation_raw_magnitude_ratio_before": float(raw_ratio),
+        "validation_magnitude_ratio_before": float(bounded_ratio),
+        "validation_raw_magnitude_ratio_after": float(raw_ratio_after),
+        "validation_magnitude_ratio_after": float(bounded_ratio_after),
+    }
+
+
 def prediction_interval_coverage(
     y_actual: np.ndarray,
     lower: np.ndarray,

deep_learning/training/trainer.py

@@ -847,6 +847,14 @@ def train_tft_model(
         "fit_split": "validation",
         "sample_count": 0,
         "direction_sign_multiplier": 1,
+        "weekly_magnitude_scale": 1.0,
+    }
+    magnitude_calibration = {
+        "fit_split": "validation",
+        "sample_count": 0,
+        "applied": False,
+        "scale": 1.0,
+        "weekly_magnitude_scale": 1.0,
     }
     interval_calibration = {
         "fit_split": "validation",
@@ -860,11 +868,13 @@ def train_tft_model(
     val_pred_np = None
     try:
         from deep_learning.training.metrics import (
+            apply_weekly_magnitude_scale_np,
             apply_weekly_sign_correction_np,
             cumulative_horizon,
             directional_accuracy,
             fit_direction_sign_calibration,
             fit_weekly_interval_scale,
+            fit_weekly_magnitude_scale,
         )
 
         val_pred_np, val_actual_path = _predict_quantiles_to_np(
@@ -943,6 +953,20 @@ def train_tft_model(
                         weekly_direction_model["validation_pred_positive_rate"] = candidate_rate
                 else:
                     weekly_direction_model["reason"] = "validation_origin_count_mismatch"
+            magnitude_calibration = fit_weekly_magnitude_scale(
+                val_actual_path,
+                oriented_val_pred_np,
+                horizon=cfg.forecast.primary_horizon_days,
+                quantiles=tuple(cfg.model.quantiles),
+                weekly_median_cap=cfg.weekly_loss.weekly_median_cap,
+            )
+            weekly_magnitude_scale = float(magnitude_calibration.get("scale", 1.0))
+            direction_calibration["weekly_magnitude_scale"] = weekly_magnitude_scale
+            oriented_val_pred_np = apply_weekly_magnitude_scale_np(
+                oriented_val_pred_np,
+                weekly_magnitude_scale,
+                quantiles=tuple(cfg.model.quantiles),
+            )
             if WEEKLY_INTERVAL_CONDITIONING_FEATURE in master_df.columns:
                 validation_interval_conditioning_values = _weekly_interval_conditioning_values(
                     master_df,
@@ -975,8 +999,11 @@ def train_tft_model(
     # OOS replay showed that they can reverse the held-out direction.
     daily_sign_multiplier = 1
     weekly_sign_threshold = float(direction_calibration.get("weekly_sign_threshold", 0.0))
+    weekly_magnitude_scale = float(magnitude_calibration.get("scale", 1.0))
+    direction_calibration.setdefault("weekly_magnitude_scale", weekly_magnitude_scale)
     weekly_interval_scale = float(interval_calibration.get("weekly_interval_scale", 1.0))
     logger.info("Validation-only direction calibration: %s", direction_calibration)
+    logger.info("Validation-only magnitude calibration: %s", magnitude_calibration)
     logger.info("Validation-only weekly interval calibration: %s", interval_calibration)
 
     # ---- 8. Evaluate on the promoted best checkpoint ----
@@ -987,6 +1014,7 @@ def train_tft_model(
     if test_dl is not None:
         import torch
         from deep_learning.training.metrics import (
+            apply_weekly_magnitude_scale_np,
             apply_weekly_sign_correction_np,
         )
 
@@ -1025,6 +1053,11 @@ def train_tft_model(
                 threshold=float(weekly_direction_model.get("decision_threshold", 0.50)),
                 horizon=cfg.forecast.primary_horizon_days,
             )
+        pred_np = apply_weekly_magnitude_scale_np(
+            pred_np,
+            weekly_magnitude_scale,
+            quantiles=tuple(cfg.model.quantiles),
+        )
         test_interval_conditioning_values = None
         if interval_calibration.get("weekly_interval_conditioning_enabled"):
             test_interval_conditioning_values = _weekly_interval_conditioning_values(
@@ -1057,6 +1090,7 @@ def train_tft_model(
         direction_sign_multiplier=direction_sign_multiplier,
         daily_sign_multiplier=daily_sign_multiplier,
         weekly_sign_threshold=weekly_sign_threshold,
+        weekly_magnitude_scale=weekly_magnitude_scale,
         weekly_direction_model=weekly_direction_model,
         validation_start_time=train_cutoff - 1,
         validation_end_time=val_cutoff - cfg.forecast.primary_horizon_days,
@@ -1119,6 +1153,7 @@ def train_tft_model(
         "data_snapshot": _build_data_snapshot_metadata(master_df),
         "runtime_environment": _runtime_environment_metadata(),
         "direction_calibration": direction_calibration,
+        "magnitude_calibration": magnitude_calibration,
         "weekly_direction_model": weekly_direction_model,
         "interval_calibration": interval_calibration,
         "checkpoint_selection": checkpoint_selection,
@@ -1190,6 +1225,7 @@ def _write_conformal_calibration_artifact(
     direction_sign_multiplier: int = 1,
     daily_sign_multiplier: int = 1,
     weekly_sign_threshold: float = 0.0,
+    weekly_magnitude_scale: float = 1.0,
     weekly_direction_model: Optional[dict] = None,
     validation_start_time: Optional[int] = None,
     validation_end_time: Optional[int] = None,
@@ -1211,6 +1247,7 @@ def _write_conformal_calibration_artifact(
             rolling_conformal_adjustment,
         )
         from deep_learning.training.metrics import (
+            apply_weekly_magnitude_scale_np,
             apply_weekly_sign_correction_np,
             apply_weekly_median_cap_np,
             apply_weekly_interval_scale_np,
@@ -1255,6 +1292,11 @@ def _write_conformal_calibration_artifact(
                 threshold=float(weekly_direction_model.get("decision_threshold", 0.50)),
                 horizon=cfg.forecast.primary_horizon_days,
             )
+        pred_np = apply_weekly_magnitude_scale_np(
+            pred_np,
+            float(weekly_magnitude_scale),
+            quantiles=tuple(cfg.model.quantiles),
+        )
         interval_calibration = weekly_interval_calibration or {}
         effective_interval_scale = float(
             interval_calibration.get("weekly_interval_scale", weekly_interval_scale)