Submission 006: official scenario interface and causal degradation model

README.md

@@ -1,73 +1,47 @@
-# BatterySwapAI 2026 — V4
+# BatterySwapAI 2026 — V05 (official scenario interface)
 
-Candidate. V1 remains the best confirmed public result (1576.57) until a better public score is recorded.
+Candidate. Not submitted. V04 remains the best public result (1407.63), but it was produced by a
+pipeline that read sensor rows timestamped after the scenario start.
 
-- Building-held-out score: **1384.08**
-- Held-out late-swap / early-swap: **582.92 / 584.49**
-- Held-out labour penalties (overtime + daily + weekly): **151.28**
-- Event ranking, building-held-out: **AP 0.5258**, AUC 0.9640
+- Building-held-out score, official cut: **1598.26**
+- Late / early: **754.79 / 618.06**
+- Event ranking: AP **0.4170**, AUC 0.9586
 - Mean scheduled batteries: **15.75**
-- Tests: **34/34 passed**
-- Artifact SHA-256: `5cb8bf244aa8755fde47bc937761a71d240c6277862e7d533e44233b03171fdf`
-- Base: `versions/v3-generalization-fix` at `9de40de38cf75b0977ae452b1f86bda18af9892d`
+- Tests: **44/44 passed**
+- Artifact SHA-256: `63d71091d7f46e8fb11798b7c2be936cc7e0022954824a4eb98c07dc1bf9b0b3`
+- Base: `research/v4-first-place` at `5ade97b8ab7b33e10a00f9d182a7ebcde69bc89c`
 
 ## What changed
 
-**Scheduling.** The evaluator resets the day clock and *then* adds the trip home, so every
-return leg is billed again against the next working day's overtime and 24 h limit. On train
-plans that accounted for 96% of daily-limit cost and 25% of overtime. The scheduler now costs a
-day as carry-in plus its own hours, and routes weight the final leg by the overtime factor.
+**Official scenario input.** `iterate_scenarios` yields a `cut` truncated at
+`end_time <= scenario.start_time`, and `make_submissions` passes exactly that to the planner.
+Our pipeline discarded it and rebuilt features from the whole split, so the cutoff day's bucket
+carried that day's later hourly readings — 7,274 to 9,847 rows per scenario. Inference now
+consumes the cut, and `scenario_history` truncates defensively at the same boundary.
 
-**Selection.** Candidates are ranked against a learned survivor horizon — the median remaining
-life of batteries that outlive the window, about 177 days — instead of V3's 42-day window, which
-under-priced a wasted swap sevenfold. Workload is a fixed 3.8% of fleet size inside a wide 8–24
-rail, so it cannot drift when calibration moves.
-
-**Event model.** Four views of the same 82 observed failures are ranked inside each scenario and
-mapped back onto the calibrated scale, so only the ordering changes: the base ensemble, a
-device-balanced refit, a room/building context model, and a discrete-time hazard model over
-7 intervals. Precision at the operating point is what the score is made of — see below.
-
-## Why precision is the target
-
-At the operating point, 9.7 of 16.5 scheduled batteries were false positives, and they cost
-590 per scenario in wasted asset life — 41% of the whole score. True positives cost 63.
-The optimal replacement quantile is `1/(21p)` for a battery with failure probability `p`, so
-q=0.05 is right only when `p=1`; delaying low-probability jobs was measured and is worth ~10,
-because delay saves 0.5/day but risks 10/day. Precision is the only real lever, and it is
-limited by having 82 failures to learn from, not by the feature set.
+**Degradation dynamics.** Threshold-crossing forecasts trained on every observed sensor
+transition plus biweekly voltage bins back to one year, blended into the event ranking as a 0.25
+residual nudge. Worth −76.0 against the same blend under the official cut.
 
 ## Validation
 
-Buildings are held out, so no battery, room or building is shared between fit and score.
-Blending happens over complete scenarios in validation exactly as it does at inference.
-
-The 48 shipped scenarios all use a handful of base locations, which understates difficulty and
-leaves too little power to separate candidates. Pairing every date with every real base location
-gives 768 cases and roughly four times the power. Measured there, on identical predictions:
-
-| planner + model | mean | vs V1 | t |
-|---|---|---|---|
-| V1 planner + base ensemble | 1672.16 | — | — |
-| V3 planner + base ensemble (band 16:20) | 1647.90 | −24.3 | −1.7 |
-| V4 planner + base ensemble | 1519.89 | −152.3 | −9.3 |
-| **V4 planner + blended model** | **1460.04** | **−212.1** | **−13.0** |
-
-V3's shipped configuration was never significantly better than V1.
+Historical families re-baselined on identical building-held-out folds and the official cut:
 
-Under a ±30% calibration shift the total moves 7.8 points and the workload does not move at all.
-Widening or narrowing the 8–24 rail changes nothing, because the fraction never reaches it.
+| generation | public | AP | total | late | early |
+|---|---|---|---|---|---|
+| **V05** | — | 0.4170 | **1598.26** | 754.79 | 618.06 |
+| V04 | 1407.63 | 0.4099 | 1674.27 | 831.88 | 623.12 |
+| V03 | 1700.12 | 0.3947 | 1757.11 | 657.29 | 826.50 |
+| V01 | 1576.57 | 0.3972 | 1766.77 | 892.08 | 598.99 |
 
-Measured and rejected: physics extrapolation to the 2.380 V failure threshold, exact pricing of
-the censored horizon, late-window insurance placement, ranking by predicted RUL, marginal
-add/remove passes, device weighting on its own, and five- and six-way blends.
+Removing the post-cutoff readings costs about 290 points: the same V04 model scores 1384.08 under
+the old pipeline and 1674.27 under the official cut.
 
 ## Artifact
 
-Trained under the evaluator runtime: scikit-learn `1.7.2`, numpy `2.2.6`, pandas `2.3.3`.
-`submission_artifacts/planner.json` records the exact policy that runs.
-54 s and 1.5 GB per split; deterministic and byte-identical offline.
+Trained and serialised under scikit-learn `1.7.2`, numpy `2.2.6`, pandas `2.3.3`.
+215 s and 1.68 GB per split; deterministic and byte-identical with `--network none`.
 
-## How to run
+## How to reproduce
 
 See `REPRODUCIBILITY.md`.

REPRODUCIBILITY.md

@@ -1,4 +1,4 @@
-# Reproduce V4
+# Reproduce V05
 
 Dataset revision:
 
@@ -27,10 +27,10 @@ pytest
 Expected checks:
 
 ```text
-34/34 tests pass
-building-held-out score: 1384.08
+44/44 tests pass
+building-held-out score: 1598.26
 mean scheduled batteries: 15.75
-artifact SHA-256: 5cb8bf244aa8755fde47bc937761a71d240c6277862e7d533e44233b03171fdf
+artifact SHA-256: 63d71091d7f46e8fb11798b7c2be936cc7e0022954824a4eb98c07dc1bf9b0b3
 ```
 
 Train the artifact under the evaluator runtime, not a local interpreter:
@@ -53,4 +53,10 @@ docker run --rm -e BATTERYSWAP_SPLITS=train -v "$(pwd)/data/raw":/tmp/data:ro ba
 ```
 
 The Docker run must produce all 19,890 required rows under the evaluator runtime.
-Measured: 54 s per split, 1.52 GB peak RSS, byte-identical output across repeated runs.
+Measured: 215 s per split, 1.68 GB peak RSS, byte-identical output across repeated runs.
+
+## Artifact reproducibility
+
+Retraining reproduces the model but not the exact bytes: independently trained artifacts differ
+in serialisation while agreeing to 5e-11 in predicted risk and producing byte-identical plans.
+Verify a rebuild by comparing plans, not SHAs.

script.py

@@ -4,14 +4,13 @@ import os
 import sys
 from pathlib import Path
 
-# Makes the local src package available when the evaluator runs script.py directly.
 sys.path.insert(0, str(Path(__file__).resolve().parent / "src"))
 
 import joblib
 import pandas as pd
 from batteryswap_public.utils import iterate_scenarios, load_dataset
 
-from batteryswapai.competition_features import build_daily_features, scenario_snapshot
+from batteryswapai.competition_features import scenario_history_snapshot
 
 
 def main() -> None:
@@ -27,16 +26,15 @@ def main() -> None:
     plans = []
     for split in splits:
         locations, timeseries, hidden_eol_times, scenarios = load_dataset(dataset_path / split)
-        daily = build_daily_features(timeseries)
 
-        for scenario, locs, _, _ in iterate_scenarios(
+        for scenario, locs, visible_history, _ in iterate_scenarios(
             locations,
             timeseries,
             hidden_eol_times,
             scenarios,
         ):
-            snapshot = scenario_snapshot(
-                daily,
+            snapshot = scenario_history_snapshot(
+                visible_history,
                 locs,
                 scenario["name"],
                 scenario["start_time"],
@@ -62,4 +60,4 @@ def main() -> None:
 
 
 if __name__ == "__main__":
-    main()
+    main()

scripts/train_submission.py

@@ -13,6 +13,8 @@ from batteryswap_public.evaluate import evaluate_plan
 from batteryswap_public.utils import iterate_scenarios, load_dataset
 
 from batteryswapai.competition_features import (
+    build_trajectory_matrix,
+    scenario_history_snapshot,
     attach_training_targets,
     build_daily_features,
     scenario_snapshot,
@@ -37,6 +39,8 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--expected-gain-margin", type=float, default=10.0)
     parser.add_argument("--expected-service-cost-hours", type=float, default=2.0)
     parser.add_argument("--scheduled-fraction", type=float, default=0.038)
+    parser.add_argument("--weekly-guard-fraction", type=float, default=0.95)
+    parser.add_argument("--hard-limit-penalty-multiplier", type=float, default=1.5)
     parser.add_argument("--minimum-scheduled-batteries", type=int, default=8)
     parser.add_argument("--maximum-scheduled-batteries", type=int, default=24)
     parser.add_argument("--calibration-folds", type=int, default=5)
@@ -50,8 +54,10 @@ def main() -> None:
     daily = build_daily_features(timeseries)
 
     snapshots = []
-    for scenario, locs, _, _ in iterate_scenarios(locations, timeseries, eol_times, scenarios):
-        snapshot = scenario_snapshot(daily, locs, scenario["name"], scenario["start_time"])
+    for scenario, locs, visible, _ in iterate_scenarios(locations, timeseries, eol_times, scenarios):
+        snapshot = scenario_history_snapshot(
+            visible, locs, scenario["name"], scenario["start_time"]
+        )
         snapshot = attach_training_targets(
             snapshot,
             eol_times,
@@ -65,6 +71,8 @@ def main() -> None:
         quantile=args.quantile,
         dataset_revision=DATASET_REVISION,
         calibration_folds=args.calibration_folds,
+        trajectory=build_trajectory_matrix(daily),
+        eol_times=eol_times,
     )
     planner = CompetitionPlanner(
         model,
@@ -76,6 +84,8 @@ def main() -> None:
             expected_gain_margin=args.expected_gain_margin,
             expected_service_cost_hours=args.expected_service_cost_hours,
             scheduled_fraction=args.scheduled_fraction,
+            capacity_weekly_limit_fraction=args.weekly_guard_fraction,
+            capacity_limit_penalty_multiplier=args.hard_limit_penalty_multiplier,
             minimum_scheduled_batteries=args.minimum_scheduled_batteries,
             maximum_scheduled_batteries=args.maximum_scheduled_batteries,
             capacity_lookahead_days=21,
@@ -94,11 +104,15 @@ def main() -> None:
         "expected_gain_margin": args.expected_gain_margin,
         "expected_service_cost_hours": args.expected_service_cost_hours,
         "scheduled_fraction": args.scheduled_fraction,
+        "weekly_guard_fraction": args.weekly_guard_fraction,
+        "hard_limit_penalty_multiplier": args.hard_limit_penalty_multiplier,
         "minimum_scheduled_batteries": args.minimum_scheduled_batteries,
         "maximum_scheduled_batteries": args.maximum_scheduled_batteries,
         "calibration_folds": args.calibration_folds,
         "planner_policy": dataclasses.asdict(planner.policy),
         "survivor_rul_head": model.survivor_rul_estimator is not None,
+        "trajectory_weight": model.trajectory_weight,
+        "forecast_heads": sorted(model.forecasters or {}),
         "training_rows": len(training),
         "training_devices": int(training["battery"].nunique()),
         "features": model.feature_columns,
@@ -119,10 +133,12 @@ def main() -> None:
 
     if not args.skip_evaluation:
         scores = []
-        for scenario, locs, _, not_dead in iterate_scenarios(
+        for scenario, locs, visible, not_dead in iterate_scenarios(
             locations, timeseries, eol_times, scenarios
         ):
-            snapshot = scenario_snapshot(daily, locs, scenario["name"], scenario["start_time"])
+            snapshot = scenario_history_snapshot(
+                visible, locs, scenario["name"], scenario["start_time"]
+            )
             plan = planner.plan_snapshot(
                 snapshot,
                 locs,

scripts/validate_submission.py

@@ -12,11 +12,17 @@ from sklearn.metrics import average_precision_score, roc_auc_score
 from sklearn.model_selection import GroupKFold
 
 from batteryswapai.competition_features import (
+    build_trajectory_matrix,
+    scenario_history_snapshot,
     attach_training_targets,
     build_daily_features,
     scenario_snapshot,
 )
-from batteryswapai.competition_model import blend_risk_components, fit_event_time_model
+from batteryswapai.competition_model import (
+    blend_risk_components,
+    blend_trajectory,
+    fit_event_time_model,
+)
 from batteryswapai.competition_planner import CompetitionPlanner, PlannerPolicy
 from train_submission import DATASET_REVISION
 
@@ -90,10 +96,12 @@ def main() -> None:
 
     snapshots = []
     scenario_inputs = {}
-    for scenario, locs, _, not_dead in iterate_scenarios(
+    for scenario, locs, visible, not_dead in iterate_scenarios(
         locations, timeseries, eol_times, scenarios
     ):
-        snapshot = scenario_snapshot(daily, locs, scenario["name"], scenario["start_time"])
+        snapshot = scenario_history_snapshot(
+            visible, locs, scenario["name"], scenario["start_time"]
+        )
         snapshot = attach_training_targets(
             snapshot,
             eol_times,
@@ -107,6 +115,8 @@ def main() -> None:
     oof_rul = np.full(len(training), np.nan)
     oof_survivor = np.full(len(training), np.nan)
     oof_components: dict[str, np.ndarray] = {}
+    oof_residual = np.full(len(training), np.nan)
+    oof_residual_weight = np.zeros(len(training))
     blend_weights = (1.0, 0.0, 0.0, 0.0)
     if args.predictions_csv is not None:
         cached = pd.read_csv(args.predictions_csv).set_index(["scenario", "battery"])
@@ -130,6 +140,8 @@ def main() -> None:
                 dataset_revision=DATASET_REVISION,
                 random_state=2026 + fold_number,
                 calibration_folds=args.inner_calibration_folds,
+                trajectory=build_trajectory_matrix(daily),
+                eol_times=eol_times,
             )
             valid = training.iloc[valid_index]
             for name, values in model.predict_risk_components(valid).items():
@@ -137,14 +149,22 @@ def main() -> None:
                     name, np.full(len(training), np.nan)
                 )[valid_index] = values
             blend_weights = model.blend_weights
+            residual, residual_weight = model._residual_risk(valid)
+            if residual is not None:
+                oof_residual[valid_index] = residual
+                oof_residual_weight[valid_index] = residual_weight
             oof_rul[valid_index] = model.predict_rul(valid)
             survivor = model.predict_survivor_rul(valid)
             if survivor is not None:
                 oof_survivor[valid_index] = survivor
 
+    scenario_groups = training["scenario"].astype(str).to_numpy()
     if oof_components:
-        oof_risk[:] = blend_risk_components(
-            oof_components, blend_weights, training["scenario"].astype(str).to_numpy()
+        oof_risk[:] = blend_risk_components(oof_components, blend_weights, scenario_groups)
+    # Mirror inference: the trajectory nudge is applied over complete scenarios.
+    if np.any(oof_residual_weight > 0.0):
+        oof_risk[:] = blend_trajectory(
+            oof_risk, np.nan_to_num(oof_residual), oof_residual_weight, scenario_groups
         )
 
     due = (

src/batteryswapai/competition_features.py

@@ -9,6 +9,11 @@ import pandas as pd
 ROLLING_WINDOWS_DAYS = (3, 7, 14, 30, 60, 90, 180, 365)
 
 CONTEXT_LOW_VOLTAGE = 2.45
+TRAJECTORY_LAGS = tuple(range(0, 365, 14))
+TRAJECTORY_COLUMNS = tuple(f"traj_lag{lag}" for lag in TRAJECTORY_LAGS)
+TRAJECTORY_COVERAGE_COLUMN = "traj_coverage"
+MIN_TRAJECTORY_COVERAGE = 0.5
+TRAJECTORY_REFERENCE_OFFSET = 0
 CONTEXT_COLUMNS = (
     "room_voltage_mean",
     "room_voltage_min",
@@ -40,6 +45,9 @@ NON_FEATURE_COLUMNS = {
     "scenario_month_cos",
     # Scenario-level context, used only by the auxiliary context model.
     *CONTEXT_COLUMNS,
+    # Trajectory history, used only by the degradation models.
+    *TRAJECTORY_COLUMNS,
+    TRAJECTORY_COVERAGE_COLUMN,
 }
 
 
@@ -132,6 +140,103 @@ def build_daily_features(
     return daily
 
 
+def scenario_history(history: pd.DataFrame, scenario_time) -> pd.DataFrame:
+    """Restrict raw observations to what the scenario can see, exactly as iterate_scenarios does."""
+
+    start = pd.Timestamp(scenario_time)
+    frame = history
+    if "end_time" not in frame.columns:
+        frame = frame.reset_index()
+    visible = frame[pd.to_datetime(frame["end_time"]) <= start]
+    return visible.set_index(["device_id", "end_time"])
+
+
+def scenario_history_snapshot(
+    history: pd.DataFrame,
+    locations: pd.DataFrame,
+    scenario_name: str,
+    scenario_time,
+    windows: Iterable[int] = ROLLING_WINDOWS_DAYS,
+) -> pd.DataFrame:
+    """Features for one scenario, built only from that scenario's visible history."""
+
+    visible = scenario_history(history, scenario_time)
+    return scenario_snapshot(
+        build_daily_features(visible, windows=windows), locations, scenario_name, scenario_time
+    )
+
+
+def build_trajectory_matrix(daily_features: pd.DataFrame) -> dict:
+    """Dense device-by-day smoothed voltage grid, built once per split and cached."""
+
+    cached = daily_features.attrs.get("_trajectory_matrix")
+    if cached is not None:
+        return cached
+
+    frame = daily_features[["device_id", "day", "stable_voltage_median"]].dropna()
+    frame = frame.sort_values(["device_id", "day"])
+    smooth = frame.groupby("device_id", observed=True)["stable_voltage_median"].transform(
+        lambda values: values.rolling(7, min_periods=3).median()
+    )
+    frame = frame.assign(smooth=smooth).dropna(subset=["smooth"])
+    if frame.empty:
+        empty = {"index": {}, "origin": pd.Timestamp("2000-01-01"),
+                 "grid": np.zeros((0, 1), dtype=np.float32),
+                 "first": np.zeros(0, dtype=np.int64),
+                 "observed_last": np.zeros(0, dtype=np.int64)}
+        daily_features.attrs["_trajectory_matrix"] = empty
+        return empty
+    devices = sorted(frame["device_id"].astype(str).unique())
+    index = {device: position for position, device in enumerate(devices)}
+    origin = frame["day"].min().normalize()
+    span = int((frame["day"].max().normalize() - origin).days) + 1
+    grid = np.full((len(devices), span), np.nan, dtype=np.float32)
+    rows = frame["device_id"].astype(str).map(index).to_numpy(dtype=np.int64)
+    columns = (frame["day"] - origin).dt.days.to_numpy(dtype=np.int64)
+    grid[rows, columns] = frame["smooth"].to_numpy()
+
+    # Carry the last reading forward to the end of the grid. Stopping at the device's
+    # final observation would make availability depend on whether it reported again later.
+    first = np.full(len(devices), np.iinfo(np.int64).max, dtype=np.int64)
+    observed_last = np.full(len(devices), -1, dtype=np.int64)
+    for position in range(len(devices)):
+        seen = np.flatnonzero(np.isfinite(grid[position]))
+        if seen.size == 0:
+            continue
+        low = int(seen[0])
+        segment = grid[position, low:]
+        carry = np.maximum.accumulate(
+            np.where(np.isfinite(segment), np.arange(segment.size), 0)
+        )
+        grid[position, low:] = segment[carry]
+        first[position] = low
+        observed_last[position] = int(seen[-1])
+
+    built = {"index": index, "origin": origin, "grid": grid, "first": first,
+             "observed_last": observed_last}
+    daily_features.attrs["_trajectory_matrix"] = built
+    return built
+
+
+def trajectory_bins(matrix: dict, batteries, cutoff) -> np.ndarray:
+    """Voltage at the cutoff and every 14 days before it; never reads past the cutoff."""
+
+    grid = matrix["grid"]
+    first = matrix["first"]
+    rows = np.array([matrix["index"].get(str(name), -1) for name in batteries], dtype=np.int64)
+    cut = int((pd.Timestamp(cutoff).normalize() - matrix["origin"]).days)
+    lags = np.asarray(TRAJECTORY_LAGS, dtype=np.int64) + TRAJECTORY_REFERENCE_OFFSET
+    columns = cut - lags[None, :]
+    out = np.full((len(rows), lags.size), np.nan, dtype=float)
+    known = rows >= 0
+    if known.any():
+        safe = np.clip(columns, 0, grid.shape[1] - 1)
+        values = grid[rows[known][:, None], safe]
+        usable = (columns >= 0) & (columns >= first[rows[known]][:, None])
+        out[known] = np.where(usable, values, np.nan)
+    return out
+
+
 def scenario_snapshot(
     daily_features: pd.DataFrame,
     locations: pd.DataFrame,
@@ -178,7 +283,14 @@ def scenario_snapshot(
     snapshot["room_battery_count"] = snapshot.groupby("room", observed=True)["battery"].transform(
         "size"
     )
-    return add_context_columns(snapshot)
+    snapshot = add_context_columns(snapshot)
+    bins = trajectory_bins(
+        build_trajectory_matrix(daily_features), snapshot["battery"].astype(str), start
+    )
+    for position, column in enumerate(TRAJECTORY_COLUMNS):
+        snapshot[column] = bins[:, position]
+    snapshot[TRAJECTORY_COVERAGE_COLUMN] = np.isfinite(bins).mean(axis=1)
+    return snapshot
 
 
 def add_context_columns(snapshot: pd.DataFrame) -> pd.DataFrame:

src/batteryswapai/competition_model.py

@@ -16,6 +16,11 @@ from scipy.stats import rankdata
 
 from .competition_features import (
     CONTEXT_COLUMNS,
+    MIN_TRAJECTORY_COVERAGE,
+    TRAJECTORY_COLUMNS,
+    TRAJECTORY_REFERENCE_OFFSET,
+    TRAJECTORY_COVERAGE_COLUMN,
+    TRAJECTORY_LAGS,
     add_context_columns,
     numeric_feature_columns,
 )
@@ -81,6 +86,169 @@ def _derived_features(base: pd.DataFrame, kind: str) -> pd.DataFrame:
 
 
 HAZARD_EDGES = (0, 14, 28, 42, 70, 120, 200, 365)
+FORECAST_HORIZONS = (30, 60, 90)
+FORECAST_THRESHOLDS = (2.50, 2.45)
+TRAJECTORY_BLEND_WEIGHT = 0.25
+
+
+def _trajectory_features(bins: np.ndarray) -> pd.DataFrame:
+    """Shape of the degradation trajectory: levels, drops, steps and summaries."""
+
+    current = bins[:, 0]
+    highest = np.nanmax(bins, axis=1)
+    lowest = np.nanmin(bins, axis=1)
+    spread = np.nanstd(bins, axis=1)
+    columns: dict[str, np.ndarray] = {}
+    for position, lag in enumerate(TRAJECTORY_LAGS):
+        columns[f"v_lag{lag}"] = bins[:, position]
+    for position, lag in enumerate(TRAJECTORY_LAGS[1:], start=1):
+        columns[f"drop{lag}"] = current - bins[:, position]
+        columns[f"ndrop{lag}"] = (current - bins[:, position]) / np.where(
+            spread > 1e-6, spread, np.nan
+        )
+    steps = np.diff(bins, axis=1)
+    for position in range(steps.shape[1]):
+        columns[f"step{position}"] = -steps[:, position]
+    columns.update(
+        own_max=highest,
+        own_min=lowest,
+        own_std=spread,
+        cur_minus_max=current - highest,
+        cur_over_max=current / highest,
+        pos=(current - lowest) / np.where(highest - lowest > 1e-6, highest - lowest, np.nan),
+        fb50=np.nanmean(bins < 2.50, axis=1),
+        fb45=np.nanmean(bins < 2.45, axis=1),
+        mono=np.nanmean(steps < 0, axis=1),
+    )
+    return pd.DataFrame(columns)
+
+
+def _snapshot_bins(snapshot: pd.DataFrame) -> np.ndarray | None:
+    if not set(TRAJECTORY_COLUMNS).issubset(snapshot.columns):
+        return None
+    return snapshot.reindex(columns=list(TRAJECTORY_COLUMNS)).to_numpy(dtype=float)
+
+
+def _dense_transitions(matrix: dict, buildings: dict[str, str]):
+    """Every observed degradation transition, not just the terminal failures."""
+
+    grid = matrix["grid"]
+    first = matrix["first"]
+    devices = list(matrix["index"])
+    rows, cuts = [], []
+    for name, position in matrix["index"].items():
+        begin = int(first[position])
+        if begin >= grid.shape[1]:
+            continue
+        end = int(np.flatnonzero(np.isfinite(grid[position]))[-1])
+        start = begin + 180
+        for cut in range(start, end + 1, 14):
+            rows.append(position)
+            cuts.append(cut)
+    rows = np.asarray(rows, dtype=np.int64)
+    cuts = np.asarray(cuts, dtype=np.int64)
+    lags = np.asarray(TRAJECTORY_LAGS, dtype=np.int64) + TRAJECTORY_REFERENCE_OFFSET
+    columns = cuts[:, None] - lags[None, :]
+    safe = np.clip(columns, 0, grid.shape[1] - 1)
+    values = grid[rows[:, None], safe]
+    usable = (columns >= 0) & (columns >= first[rows][:, None])
+    bins = np.where(usable, values, np.nan)
+    names = np.array([devices[position] for position in rows])
+    where = np.array([buildings.get(name, "") for name in names])
+    return rows, cuts, bins, names, where
+
+
+def _forecast_targets(matrix: dict, rows, cuts, eol_day: dict[str, float], names):
+    """Threshold crossings and future voltage, read only from observed history."""
+
+    grid = matrix["grid"]
+    last = matrix["observed_last"]
+    ends_at = np.array([eol_day.get(name, np.nan) for name in names], dtype=float)
+    targets: dict[tuple, np.ndarray] = {}
+    for horizon in FORECAST_HORIZONS:
+        finish = cuts + horizon
+        reachable = finish <= last[rows]
+        died = np.isfinite(ends_at) & (ends_at <= finish)
+        lowest = np.full(len(rows), np.nan)
+        for position in range(len(rows)):
+            begin = cuts[position] + 1
+            stop = min(int(finish[position]), int(last[rows[position]]))
+            if stop >= begin:
+                lowest[position] = np.nanmin(grid[rows[position], begin : stop + 1])
+        for threshold in FORECAST_THRESHOLDS:
+            label = np.where(
+                died, 1.0, np.where(reachable, (lowest <= threshold).astype(float), np.nan)
+            )
+            targets[(horizon, threshold)] = np.where(~reachable & ~died, np.nan, label)
+        future = np.where(reachable, grid[rows, np.clip(finish, 0, grid.shape[1] - 1)], np.nan)
+        targets[(horizon, "voltage")] = np.where(died, 2.30, future)
+    return targets
+
+
+def _forecaster(random_state: int, regression: bool):
+    common = dict(learning_rate=0.06, max_iter=200, max_leaf_nodes=15, min_samples_leaf=40,
+                  l2_regularization=5.0, early_stopping=True, validation_fraction=0.1,
+                  n_iter_no_change=15, random_state=random_state)
+    if regression:
+        return HistGradientBoostingRegressor(**common)
+    return HistGradientBoostingClassifier(loss="log_loss", **common)
+
+
+def _fit_forecasters(matrix, buildings, eol_day, random_state,
+                     snapshot_features, snapshot_buildings):
+    """Learn degradation dynamics from dense transitions; score snapshots by held-out building."""
+
+    rows, cuts, bins, names, where = _dense_transitions(matrix, buildings)
+    if len(rows) < 500:
+        return None, None
+    features = _trajectory_features(bins)
+    targets = _forecast_targets(matrix, rows, cuts, eol_day, names)
+    unique = np.array(sorted(set(where) - {""}))
+    if unique.size < 3:
+        return None, None
+    assignment = {name: position % 3 for position, name in enumerate(unique)}
+    dense_fold = np.array([assignment.get(name, -1) for name in where])
+    snap_fold = np.array([assignment.get(str(name), -1) for name in snapshot_buildings])
+
+    fitted: dict[str, object] = {}
+    holdout: dict[str, np.ndarray] = {}
+    for (horizon, kind), values in targets.items():
+        regression = kind == "voltage"
+        name = f"v{horizon}" if regression else f"p{horizon}_{kind}"
+        usable = np.isfinite(values)
+        if usable.sum() < 200:
+            continue
+        estimator = _forecaster(random_state, regression)
+        estimator.fit(features[usable], values[usable])
+        fitted[name] = estimator
+        out = np.full(len(snapshot_features), np.nan)
+        for group in range(3):
+            train = usable & (dense_fold != group)
+            target_rows = np.flatnonzero(snap_fold == group)
+            if train.sum() < 100 or target_rows.size == 0:
+                continue
+            fold_model = _forecaster(random_state + group + 1, regression)
+            fold_model.fit(features[train], values[train])
+            block = snapshot_features.iloc[target_rows]
+            out[target_rows] = (
+                fold_model.predict(block) if regression
+                else fold_model.predict_proba(block)[:, 1]
+            )
+        holdout[name] = out
+    if not fitted:
+        return None, None
+    return fitted, pd.DataFrame(holdout)
+
+
+def _forecast_features(forecasters, features: pd.DataFrame) -> pd.DataFrame:
+    out: dict[str, np.ndarray] = {}
+    for name, estimator in forecasters.items():
+        if name.startswith("v"):
+            out[name] = estimator.predict(features)
+        else:
+            out[name] = estimator.predict_proba(features)[:, 1]
+    return pd.DataFrame(out, index=features.index)
+
 
 
 def _context_features(snapshot: pd.DataFrame) -> pd.DataFrame:
@@ -280,6 +448,10 @@ class EventTimeModel:
     context_columns: list[str] | None = None
     hazard_estimator: object | None = None
     blend_weights: tuple[float, float, float, float] = (1.0, 0.0, 0.0, 0.0)
+    forecasters: dict | None = None
+    residual_model: object | None = None
+    residual_columns: list[str] | None = None
+    trajectory_weight: float = 0.0
 
     def _feature_sets(
         self, snapshot: pd.DataFrame
@@ -324,15 +496,43 @@ class EventTimeModel:
             )
         return components
 
+    def _residual_risk(self, snapshot: pd.DataFrame):
+        """Degradation-dynamics view; None when the trajectory history is unusable."""
+
+        if self.residual_model is None or self.forecasters is None:
+            return None, None
+        bins = _snapshot_bins(snapshot)
+        if bins is None:
+            return None, None
+        trajectory = _trajectory_features(bins)
+        forecast = _forecast_features(self.forecasters, trajectory)
+        frame = pd.concat([forecast, trajectory], axis=1)
+        frame = frame.reindex(columns=self.residual_columns).astype(float)
+        risk = self.residual_model.predict_proba(frame)[:, 1]
+        if TRAJECTORY_COVERAGE_COLUMN in snapshot.columns:
+            coverage = pd.to_numeric(
+                snapshot[TRAJECTORY_COVERAGE_COLUMN], errors="coerce"
+            ).to_numpy(dtype=float)
+        else:
+            coverage = np.isfinite(bins).mean(axis=1)
+        coverage = np.where(np.isfinite(coverage), coverage, 0.0)
+        # A battery without enough history keeps its blend position rather than moving on noise.
+        weight = np.where(coverage >= MIN_TRAJECTORY_COVERAGE, self.trajectory_weight, 0.0)
+        return risk, weight
+
     def predict_event_risk(self, snapshot: pd.DataFrame) -> np.ndarray:
         groups = (
             snapshot["scenario"].astype(str).to_numpy()
             if "scenario" in snapshot.columns
             else None
         )
-        return blend_risk_components(
+        primary = blend_risk_components(
             self.predict_risk_components(snapshot), self.blend_weights, groups
         )
+        risk, weight = self._residual_risk(snapshot)
+        if risk is None or not np.any(weight > 0.0):
+            return primary
+        return blend_trajectory(primary, risk, weight, groups)
 
     def predict_rul(self, snapshot: pd.DataFrame) -> np.ndarray:
         base = _clean(snapshot, self.feature_columns)
@@ -372,6 +572,26 @@ def blend_risk_components(
     return _map_to_scale(blended, primary, groups)
 
 
+def blend_trajectory(primary, residual, weight, groups):
+    """Nudge the ranking by the degradation view, keeping the calibrated scale."""
+
+    primary = np.asarray(primary, dtype=float)
+    residual = np.asarray(residual, dtype=float)
+    weight = np.asarray(weight, dtype=float)
+    residual = np.where(np.isfinite(residual), residual, np.nanmin(residual[np.isfinite(residual)])
+                        if np.isfinite(residual).any() else 0.0)
+    blended = np.empty(len(primary), dtype=float)
+    keys = [None] if groups is None else list(pd.unique(groups))
+    for key in keys:
+        mask = slice(None) if key is None else (groups == key)
+        size = len(primary[mask])
+        first = rankdata(primary[mask]) / max(size, 1)
+        second = rankdata(residual[mask]) / max(size, 1)
+        local = weight[mask]
+        blended[mask] = (1.0 - local) * first + local * second
+    return _map_to_scale(blended, primary, groups)
+
+
 def _map_to_scale(blended: np.ndarray, reference: np.ndarray, groups) -> np.ndarray:
     """Keep the calibrated marginal distribution and take only the new ordering."""
 
@@ -386,6 +606,38 @@ def _map_to_scale(blended: np.ndarray, reference: np.ndarray, groups) -> np.ndar
     return mapped
 
 
+def _fit_trajectory_stack(work, trajectory, eol_times, event_target, random_state):
+    """Dense degradation forecasting plus the temporal-bin view, as one residual ranker."""
+
+    bins = _snapshot_bins(work)
+    if bins is None:
+        return None, None, None, 0.0
+    snapshot_features = _trajectory_features(bins)
+    buildings = dict(zip(work["battery"].astype(str), work["building"].astype(str)))
+    eol_day: dict[str, float] = {}
+    if eol_times is not None:
+        origin = trajectory["origin"]
+        for name, when in eol_times.items():
+            if pd.notna(when):
+                eol_day[str(name)] = float((pd.Timestamp(when).normalize() - origin).days)
+    forecasters, holdout = _fit_forecasters(
+        trajectory, buildings, eol_day, random_state,
+        snapshot_features, work["building"].astype(str).to_numpy(),
+    )
+    if forecasters is None:
+        return None, None, None, 0.0
+    frame = pd.concat([holdout.reset_index(drop=True), snapshot_features.reset_index(drop=True)], axis=1)
+    columns = list(frame.columns)
+    residual = HistGradientBoostingClassifier(
+        loss="log_loss", learning_rate=0.05, max_iter=300, max_leaf_nodes=15,
+        min_samples_leaf=30, l2_regularization=5.0, early_stopping=True,
+        validation_fraction=0.12, n_iter_no_change=25, class_weight={0: 1.0, 1: 5.0},
+        random_state=random_state,
+    )
+    residual.fit(frame.astype(float), event_target)
+    return forecasters, residual, columns, TRAJECTORY_BLEND_WEIGHT
+
+
 def _fit_survivor_estimator(
     base: pd.DataFrame,
     target_rul: pd.Series,
@@ -428,6 +680,8 @@ def fit_event_time_model(
     calibration_folds: int = 5,
     device_balanced: bool = False,
     with_auxiliary: bool = True,
+    trajectory: dict | None = None,
+    eol_times: pd.Series | None = None,
     blend_weights: tuple[float, float, float, float] = (0.55, 0.15, 0.15, 0.15),
     max_iter: int | None = None,
 ) -> EventTimeModel:
@@ -517,6 +771,10 @@ def fit_event_time_model(
     )
 
     balanced = None
+    forecasters = None
+    residual_model = None
+    residual_columns = None
+    trajectory_weight = 0.0
     context_estimators = None
     context_imputer = None
     context_columns = None
@@ -545,6 +803,10 @@ def fit_event_time_model(
             base, target_rul, work["event_observed"].astype(bool), random_state
         )
         weights_used = tuple(float(value) for value in blend_weights)
+        if trajectory is not None:
+            forecasters, residual_model, residual_columns, trajectory_weight = _fit_trajectory_stack(
+                work, trajectory, eol_times, event_target, random_state
+            )
 
     return EventTimeModel(
         estimators,
@@ -564,4 +826,8 @@ def fit_event_time_model(
         context_columns,
         hazard_estimator,
         weights_used,
+        forecasters,
+        residual_model,
+        residual_columns,
+        trajectory_weight,
     )

submission_artifacts/planner.joblib

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:5cb8bf244aa8755fde47bc937761a71d240c6277862e7d533e44233b03171fdf
-size 12590691
+oid sha256:63d71091d7f46e8fb11798b7c2be936cc7e0022954824a4eb98c07dc1bf9b0b3
+size 15508289

submission_artifacts/planner.json

@@ -8,6 +8,8 @@
   "expected_gain_margin": 10.0,
   "expected_service_cost_hours": 2.0,
   "scheduled_fraction": 0.038,
+  "weekly_guard_fraction": 0.95,
+  "hard_limit_penalty_multiplier": 1.5,
   "minimum_scheduled_batteries": 8,
   "maximum_scheduled_batteries": 24,
   "calibration_folds": 5,
@@ -19,8 +21,8 @@
     "capacity_lookback_days": 14,
     "capacity_lookahead_days": 21,
     "capacity_daily_limit_fraction": 1.0,
-    "capacity_weekly_limit_fraction": 1.0,
-    "capacity_limit_penalty_multiplier": 1.0,
+    "capacity_weekly_limit_fraction": 0.95,
+    "capacity_limit_penalty_multiplier": 1.5,
     "capacity_operational_cost_weight": 1.5,
     "use_expected_cost": true,
     "risk_calibration_scale": 1.5,
@@ -40,6 +42,18 @@
     "selection_key": "expected_gain"
   },
   "survivor_rul_head": true,
+  "trajectory_weight": 0.25,
+  "forecast_heads": [
+    "p30_2.45",
+    "p30_2.5",
+    "p60_2.45",
+    "p60_2.5",
+    "p90_2.45",
+    "p90_2.5",
+    "v30",
+    "v60",
+    "v90"
+  ],
   "training_rows": 19890,
   "training_devices": 458,
   "features": [
@@ -121,7 +135,7 @@
     "building_battery_count",
     "room_battery_count"
   ],
-  "artifact_sha256": "5cb8bf244aa8755fde47bc937761a71d240c6277862e7d533e44233b03171fdf",
+  "artifact_sha256": "63d71091d7f46e8fb11798b7c2be936cc7e0022954824a4eb98c07dc1bf9b0b3",
   "runtime_versions": {
     "batteryswap_public": "0.3.4",
     "fastparquet": "2026.5.0",
@@ -132,15 +146,15 @@
   },
   "validation_report": "artifacts/submission_cv_final.json",
   "optimistic_train_score": {
-    "overtime": 57.55209027777776,
-    "building_change": 10.625,
-    "weekly_limit": 45.833333333333336,
-    "late_swap": 72.5,
-    "room_change": 6.770833333333333,
-    "early_swap": 381.5833333333333,
-    "daily_limit": 16.666666666666668,
-    "travel": 36.408777777777786,
-    "battery_swap": 4.020833333333333,
-    "total_cost": 631.9608680555556
+    "daily_limit": 12.5,
+    "weekly_limit": 43.75,
+    "late_swap": 75.83333333333333,
+    "battery_swap": 4.026041666666667,
+    "room_change": 6.791666666666667,
+    "overtime": 57.756944444444464,
+    "early_swap": 386.8645833333333,
+    "building_change": 10.520833333333334,
+    "travel": 36.906100694444454,
+    "total_cost": 634.9495034722223
   }
 }

tests/test_competition_features.py

@@ -4,6 +4,8 @@ import pandas as pd
 from batteryswapai.competition_features import (
     build_daily_features,
     numeric_feature_columns,
+    scenario_history,
+    scenario_history_snapshot,
     scenario_snapshot,
 )
 
@@ -36,10 +38,11 @@ def test_scenario_snapshot_never_uses_future_sensor_rows():
         }
     )
 
-    daily = build_daily_features(raw, windows=(3, 7, 14, 30, 60, 90, 180, 365))
-    snapshot = scenario_snapshot(daily, locations, "s_0", "2026-01-02")
-    assert snapshot.loc[0, "day"] == pd.Timestamp("2026-01-02")
-    assert np.isclose(snapshot.loc[0, "voltage_median"], 2.85)
+    # The 2026-01-02 readings are timestamped after the 00:00 scenario start, so the
+    # evaluator's cut excludes them and the freshest usable bucket is the previous day.
+    snapshot = scenario_history_snapshot(raw, locations, "s_0", "2026-01-02")
+    assert snapshot.loc[0, "day"] == pd.Timestamp("2026-01-01")
+    assert np.isclose(snapshot.loc[0, "voltage_median"], 3.05)
 
 
 def test_temperature_filter_falls_back_when_no_stable_reading_exists():
@@ -73,3 +76,133 @@ def test_model_features_exclude_dataset_boundary_and_calendar_fields():
     assert "censor_proxy_rul_days" not in columns
     assert "scenario_month_sin" not in columns
     assert "scenario_month_cos" not in columns
+
+
+def _voltage_history(days: int = 420, start: str = "2025-01-01") -> pd.DataFrame:
+    stamps = pd.date_range(start, periods=days, freq="D")
+    rows = []
+    for device, offset in (("d_a", 0.0), ("d_b", 0.12)):
+        rows.append(
+            pd.DataFrame(
+                {
+                    "device_id": device,
+                    "day": stamps,
+                    "stable_voltage_median": np.linspace(2.90, 2.42, days) + offset,
+                }
+            )
+        )
+    return pd.concat(rows, ignore_index=True)
+
+
+def test_trajectory_bins_never_read_past_the_cutoff():
+    """The bins must be identical whether or not future rows exist in the frame."""
+
+    from batteryswapai.competition_features import build_trajectory_matrix, trajectory_bins
+
+    history = _voltage_history()
+    cutoff = pd.Timestamp("2025-10-01")
+
+    full = build_trajectory_matrix(history.copy())
+    truncated = build_trajectory_matrix(history[history["day"] <= cutoff].copy())
+
+    from_full = trajectory_bins(full, ["d_a", "d_b"], cutoff)
+    from_truncated = trajectory_bins(truncated, ["d_a", "d_b"], cutoff)
+    assert np.allclose(from_full, from_truncated, equal_nan=True)
+    assert np.isfinite(from_full).any()
+
+
+def test_trajectory_bins_are_missing_without_history():
+    from batteryswapai.competition_features import (
+        MIN_TRAJECTORY_COVERAGE,
+        build_trajectory_matrix,
+        trajectory_bins,
+    )
+
+    matrix = build_trajectory_matrix(_voltage_history(days=40, start="2025-01-01").copy())
+    early = trajectory_bins(matrix, ["d_a"], pd.Timestamp("2025-01-20"))
+    assert np.isfinite(early).mean() < MIN_TRAJECTORY_COVERAGE
+    unknown = trajectory_bins(matrix, ["d_missing"], pd.Timestamp("2025-01-20"))
+    assert not np.isfinite(unknown).any()
+
+
+def test_scenario_history_matches_the_evaluator_cut():
+    """scenario_history must reproduce iterate_scenarios' truncation exactly."""
+
+    stamps = pd.date_range("2025-05-01", "2025-06-30", freq="6h")
+    frame = pd.DataFrame(
+        {
+            "device_id": "d_a",
+            "end_time": stamps,
+            "voltage": np.linspace(2.9, 2.5, len(stamps)),
+            "temperature": 20.0,
+        }
+    ).set_index(["device_id", "end_time"])
+    cutoff = pd.Timestamp("2025-06-01")
+
+    visible = scenario_history(frame, cutoff).reset_index()
+    assert visible["end_time"].max() == cutoff          # endpoint is inclusive
+    assert not (visible["end_time"] > cutoff).any()
+    expected = frame.reset_index()
+    expected = expected[expected["end_time"] <= cutoff]
+    assert len(visible) == len(expected)
+
+
+def test_features_ignore_readings_taken_after_the_scenario_start():
+    cutoff = pd.Timestamp("2025-06-01")
+    stamps = pd.date_range("2025-01-01", "2025-06-30", freq="6h")
+    frame = pd.DataFrame(
+        {
+            "device_id": "d_a",
+            "end_time": stamps,
+            "voltage": np.linspace(2.90, 2.40, len(stamps)),
+            "temperature": 20.0,
+        }
+    ).set_index(["device_id", "end_time"])
+    locations = pd.DataFrame(
+        {
+            "battery": ["d_a"],
+            "building": ["b_1"],
+            "room": ["r_1"],
+            "start_time": [pd.Timestamp("2024-01-01")],
+            "end_time": [pd.Timestamp("2025-06-30")],
+        }
+    )
+    flat = frame.reset_index()
+    already_cut = flat[flat["end_time"] <= cutoff].set_index(["device_id", "end_time"])
+
+    full = scenario_history_snapshot(frame, locations, "s", cutoff)
+    trimmed = scenario_history_snapshot(already_cut, locations, "s", cutoff)
+    columns = [c for c in full.columns if pd.api.types.is_numeric_dtype(full[c])]
+    assert np.allclose(
+        full[columns].astype(float).to_numpy(),
+        trimmed[columns].astype(float).to_numpy(),
+        equal_nan=True,
+    )
+
+
+def test_trajectory_availability_does_not_depend_on_later_readings():
+    """A sensor gap must look the same whether or not the device reports again later."""
+
+    from batteryswapai.competition_features import build_trajectory_matrix, trajectory_bins
+
+    cutoff = pd.Timestamp("2025-06-01")
+    early = pd.date_range("2024-01-01", "2025-03-01", freq="D")
+    later = pd.date_range("2025-08-01", "2025-09-01", freq="D")   # resumes AFTER the cutoff
+
+    def frame(days):
+        return pd.DataFrame(
+            {
+                "device_id": "d_a",
+                "day": days,
+                "stable_voltage_median": np.linspace(2.90, 2.60, len(days)),
+            }
+        )
+
+    without_future = build_trajectory_matrix(frame(early).copy())
+    with_future = build_trajectory_matrix(
+        pd.concat([frame(early), frame(later)], ignore_index=True).copy()
+    )
+    a = trajectory_bins(without_future, ["d_a"], cutoff)
+    b = trajectory_bins(with_future, ["d_a"], cutoff)
+    assert np.allclose(a, b, equal_nan=True)
+    assert np.array_equal(np.isfinite(a), np.isfinite(b))

tests/test_competition_model.py

@@ -69,3 +69,53 @@ def test_context_columns_describe_one_scenario_and_are_not_model_features():
     assert not set(CONTEXT_COLUMNS) & set(numeric_feature_columns(snapshot))
     # the auxiliary model reads the precomputed columns rather than regrouping
     assert list(_context_features(snapshot).columns) == list(CONTEXT_COLUMNS)
+
+
+def test_trajectory_blend_leaves_short_history_batteries_where_they_were():
+    import numpy as np
+    from batteryswapai.competition_model import blend_trajectory
+
+    groups = np.array(["s_0"] * 4)
+    primary = np.array([0.40, 0.30, 0.20, 0.10])
+    residual = np.array([0.10, 0.20, 0.30, 0.40])   # exactly reversed
+
+    unchanged = blend_trajectory(primary, residual, np.zeros(4), groups)
+    assert np.allclose(unchanged, primary)
+
+    nudged = blend_trajectory(primary, residual, np.full(4, 0.25), groups)
+    assert sorted(nudged) == sorted(primary)         # calibrated scale is preserved
+    # a quarter weight is a nudge, not a takeover; a dominant weight does reorder
+    strong = blend_trajectory(primary, residual, np.full(4, 0.9), groups)
+    assert sorted(strong) == sorted(primary)
+    assert not np.allclose(strong, primary)
+
+    # a battery with no usable history keeps its own position while the others move
+    partial = blend_trajectory(primary, residual, np.array([0.0, 0.25, 0.25, 0.0]), groups)
+    assert sorted(partial) == sorted(primary)
+
+
+def test_trajectory_blend_is_deterministic_and_scenario_local():
+    import numpy as np
+    from batteryswapai.competition_model import blend_trajectory
+
+    groups = np.array(["s_0", "s_0", "s_1", "s_1"])
+    primary = np.array([0.9, 0.1, 0.02, 0.01])
+    residual = np.array([0.1, 0.9, 0.01, 0.02])
+    first = blend_trajectory(primary, residual, np.full(4, 0.25), groups)
+    second = blend_trajectory(primary, residual, np.full(4, 0.25), groups)
+    assert np.array_equal(first, second)
+    # scenario 0 values never leak into scenario 1
+    assert sorted(first[:2]) == sorted(primary[:2])
+    assert sorted(first[2:]) == sorted(primary[2:])
+
+
+def test_model_without_a_trajectory_stack_returns_the_blend_unchanged():
+    import numpy as np
+    import pandas as pd
+    from batteryswapai.competition_model import EventTimeModel
+
+    model = EventTimeModel.__new__(EventTimeModel)
+    model.residual_model = None
+    model.forecasters = None
+    risk, weight = model._residual_risk(pd.DataFrame({"battery": ["d_a"]}))
+    assert risk is None and weight is None

tests/test_competition_planner.py

@@ -1,3 +1,4 @@
+from pathlib import Path
 from types import SimpleNamespace
 
 import pytest
@@ -554,3 +555,56 @@ def test_planner_runs_with_an_artifact_that_has_no_survivor_head():
     planner = CompetitionPlanner(DummyEventTimeModel(), PlannerPolicy(use_expected_cost=True))
     plan = planner.plan_snapshot(snapshot, locations, travel, settings, "2025-09-01")
     check_plan_valid(plan, locations, start_time=pd.Timestamp("2025-09-01"))
+
+
+def test_weekly_guard_band_reserves_headroom_below_the_hard_limit():
+    """The evaluator charges the weekly penalty at >= 24h, so the planner keeps a margin."""
+
+    policy = PlannerPolicy(capacity_weekly_limit_fraction=0.95,
+                           capacity_limit_penalty_multiplier=1.5)
+    settings = EvaluationSettings(base_location="b_base", base_room="r_base")
+    guarded = float(settings.worker_limit_weekly_hours) * policy.capacity_weekly_limit_fraction
+    assert guarded < float(settings.worker_limit_weekly_hours)
+    assert guarded == pytest.approx(22.8)
+    assert policy.capacity_limit_penalty_multiplier > 1.0
+
+
+def _train_split_available() -> bool:
+    from pathlib import Path
+
+    return Path("data/raw/train/scenarios.json").exists() and Path(
+        "submission_artifacts/planner.joblib"
+    ).exists()
+
+
+@pytest.mark.skipif(not _train_split_available(), reason="train split or artifact absent")
+def test_official_interface_matches_our_direct_scenario_path():
+    """make_submissions -> Planner.plan must reproduce our iterate_scenarios path exactly."""
+
+    import joblib
+    from batteryswap_public.utils import iterate_scenarios, load_dataset
+    from batteryswapai.competition_features import scenario_history_snapshot
+
+    locations, timeseries, eol_times, scenarios = load_dataset(Path("data/raw/train"))
+    subset = scenarios[:3]
+    planner = joblib.load("submission_artifacts/planner.joblib")
+
+    for scenario, locs, cut, _ in iterate_scenarios(locations, timeseries, eol_times, subset):
+        start = pd.Timestamp(scenario["start_time"])
+        rows = cut.reset_index()
+        # the planner is handed exactly the scenario-visible history
+        assert rows["end_time"].max() == start
+        assert not (rows["end_time"] > start).any()
+
+        official = planner.plan(cut, locs, scenario["travel_costs"], scenario["settings"])
+        snapshot = scenario_history_snapshot(cut, locs, scenario["name"], start)
+        direct = planner.plan_snapshot(
+            snapshot, locs, scenario["travel_costs"], scenario["settings"], start
+        )
+        pd.testing.assert_frame_equal(official, direct)
+
+        # ranking is deterministic across repeated calls
+        again = planner.plan_snapshot(
+            snapshot, locs, scenario["travel_costs"], scenario["settings"], start
+        )
+        pd.testing.assert_frame_equal(direct, again)