Logistic RUL model with adaptive top-sum planner (walk-forward CV selected)

- features.py: cached (device, origin) feature table; leak-free per-origin slicing
- models_cv.py: walk-forward CV harness; logistic/GBM/AFT/rank models,
threshold/topsum/batching planners; dev s_8-s_23, test s_24-s_41
- planner.py: final TopSumPlanner (logistic p42, N = 0.8 * sum(p42),
q05 day placement, building batching); refit on all closed windows
- harness.py: baseline comparisons (defer-all, lowest-voltage-k, oracle)
- best.pickle trained inside the competition Docker image (sklearn 1.7.2)
to avoid version skew; verified via script.py + metric in the image

Test-block walk-forward mean total_cost 1368 (voltage-15 2034, defer 2512,
oracle 186); train split via official metric: 2150 vs 8673 for the example.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

batteryswap_example/features.py

@@ -0,0 +1,117 @@
+"""
+Precompute the (device, origin) feature/label table used by every model.
+
+One row per (alive device, scenario origin). Features use only data at or before
+the origin; no fleet-wide statistics are computed here, so slicing at an origin
+cannot leak the future. Labels use the full eol_times and are for
+training/evaluation only - the planner never reads them.
+
+Cached to dataset/train/features.parquet (rebuild with --force).
+"""
+
+from pathlib import Path
+import os
+import sys
+
+import numpy as np
+import pandas as pd
+
+from batteryswap_public.utils import load_dataset
+
+MIN_TEMP = 10
+MAX_TEMP = 30
+WINDOW_DAYS = 42
+
+
+def daily_voltage(timeseries: pd.DataFrame) -> pd.DataFrame:
+    """In-band daily median voltage per device: (device_id, day, v)."""
+    df = timeseries.reset_index()
+    # device_id is categorical in the parquet; observed=False groupbys would emit
+    # the full device x day cartesian product with NaN voltages
+    df['device_id'] = df['device_id'].astype(str)
+    df = df[(df['temperature'] > MIN_TEMP) & (df['temperature'] < MAX_TEMP)]
+    df['day'] = df['end_time'].dt.normalize()
+    out = df.groupby(['device_id', 'day'])['voltage'].median().rename('v').reset_index()
+    return out.dropna(subset=['v']).sort_values('day')
+
+
+def slope(days: np.ndarray, values: np.ndarray) -> float:
+    """Per-day slope; NaN when under 2 points."""
+    if len(values) < 2:
+        return np.nan
+    x = (days - days[-1]) / np.timedelta64(1, 'D')
+    return float(np.polyfit(x.astype(float), values, 1)[0])
+
+
+def features_at_origin(daily: pd.DataFrame, devices: pd.DataFrame,
+                       origin: pd.Timestamp) -> pd.DataFrame:
+    """Feature rows for every device with any reading at or before origin."""
+    h = daily[daily['day'] <= origin]
+
+    g = h.groupby('device_id')
+    last = g.tail(1).set_index('device_id')
+    feats = pd.DataFrame({'v_last': last['v']})
+    feats['staleness'] = (origin - last['day']).dt.days
+
+    # per-device historical reference: median over first 90 observed days
+    first_day = g['day'].min()
+    early = h[h['day'] <= h['device_id'].map(first_day) + pd.Timedelta(days=90)]
+    feats['v_ref'] = early.groupby('device_id')['v'].median()
+    feats['v_drop'] = feats['v_ref'] - feats['v_last']
+
+    for w in (30, 90):
+        hw = h[h['day'] > origin - pd.Timedelta(days=w)]
+        gw = hw.groupby('device_id')
+        feats[f'slope_{w}'] = gw.apply(
+            lambda u: slope(u['day'].values, u['v'].values), include_groups=False)
+        feats[f'v_min_{w}'] = gw['v'].min()
+        feats[f'n_{w}'] = gw['v'].size()
+
+    install = devices.set_index('battery')['start_time'].dt.tz_localize(None)
+    feats['age_days'] = (origin - install.reindex(feats.index)).dt.days
+
+    return feats
+
+
+def build(dataset_path: Path, out_path: Path) -> pd.DataFrame:
+    locations, timeseries, eol_times, scenarios = load_dataset(dataset_path / 'train')
+    daily = daily_voltage(timeseries)
+
+    rows = []
+    for idx, sc in enumerate(scenarios):
+        origin = pd.Timestamp(sc['start_time'])
+        alive = eol_times.isna() | (eol_times > origin)
+        alive_ids = eol_times.index[alive]
+
+        f = features_at_origin(daily, locations, origin)
+        f = f.reindex(alive_ids)  # NaN feature rows kept: planner must handle them
+        f['origin_idx'] = idx
+        f['origin'] = origin
+
+        eol = eol_times.reindex(alive_ids)
+        f['eol'] = eol.values
+        f['dies42'] = ((eol.notna()) & (eol <= origin + pd.Timedelta(days=WINDOW_DAYS))).astype(int).values
+        f['days_to_eol'] = ((eol - origin).dt.total_seconds() / 86400).values
+        f['window_closed'] = origin + pd.Timedelta(days=WINDOW_DAYS) <= pd.Timestamp('2026-07-31')
+
+        rows.append(f.reset_index().rename(columns={'index': 'device_id'}))
+        print(f'{sc["name"]}: {len(f)} alive, {f["dies42"].sum()} die in window', file=sys.stderr)
+
+    table = pd.concat(rows, ignore_index=True)
+    table.to_parquet(out_path)
+    return table
+
+
+def load(dataset_path: Path = Path('dataset'), force: bool = False) -> pd.DataFrame:
+    out_path = dataset_path / 'train' / 'features.parquet'
+    if out_path.exists() and not force:
+        return pd.read_parquet(out_path)
+    return build(dataset_path, out_path)
+
+
+if __name__ == '__main__':
+    force = '--force' in sys.argv
+    dataset_path = Path(os.environ.get('BATTERYSWAP_DATASET_PATH', 'dataset'))
+    t = load(dataset_path, force=force)
+    print(t.groupby('origin_idx')['dies42'].agg(['size', 'sum']).T)
+    print(t.describe().round(3).to_string())

batteryswap_example/harness.py

@@ -0,0 +1,247 @@
+"""
+Baseline planners and a forward-split evaluation harness.
+
+The harness fits nothing (these baselines have no parameters), but it establishes
+the reference points any real model has to beat:
+
+    current    - the shipped OrderedPlanner
+    defer      - park every battery past the horizon (costs nothing but emergencies)
+    voltage-k  - swap the k lowest-voltage batteries
+    oracle     - knows the true EOL, swaps on the exact day
+
+Scored on origins s_24..s_41, which sit after the training block (origins whose
+42-day window closes before s_24) and before the origins whose windows run past
+the end of the data.
+"""
+
+import os
+from pathlib import Path
+from typing import Optional
+
+import numpy
+import pandas
+import structlog
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+from batteryswap_public.interfaces import Planner
+from batteryswap_public.utils import load_dataset, iterate_scenarios
+from batteryswap_public.evaluate import evaluate_plan, check_plan_valid
+
+log = structlog.get_logger()
+
+# Origins whose 42-day window closes before s_24 are the training block.
+TEST_ORIGINS = range(24, 42)
+
+# Temperature band for trusting a voltage reading, as in utils.smooth_series
+MIN_TEMP = 10
+MAX_TEMP = 30
+
+
+def infer_start_time(battery_data: pandas.DataFrame) -> pandas.Timestamp:
+    """
+    Recover the planning start from the truncated timeseries.
+
+    The Planner interface does not receive start_time - iterate_scenarios cuts the
+    series at it, so the last observation is the best available estimate.
+    """
+    return battery_data.reset_index()['end_time'].max().normalize()
+
+
+def voltage_feature(battery_data: pandas.DataFrame) -> pandas.Series:
+    """
+    Last daily-median voltage per device, using only in-band readings.
+
+    Daily median rejects within-day noise; the temperature band rejects the part of
+    the voltage swing that reflects ambient rather than state-of-charge. No recency
+    limit - a stale low reading still predicts EOL far better than no reading, and
+    the 95th percentile of staleness is ~148 days.
+
+    Deliberately uses no fleet-wide statistic, so slicing history at an origin can
+    never leak information from after it.
+    """
+    df = battery_data.reset_index()
+    df = df[(df['temperature'] > MIN_TEMP) & (df['temperature'] < MAX_TEMP)]
+    if df.empty:
+        return pandas.Series(dtype=float, name='voltage')
+
+    df['day'] = df['end_time'].dt.normalize()
+    last_day = df.groupby('device_id')['day'].transform('max')
+    latest = df[df['day'] == last_day]
+    return latest.groupby('device_id')['voltage'].median().rename('voltage')
+
+
+def _assemble(selected_days: pandas.Series, locations, start_time, window_days):
+    """
+    Build a valid plan: scheduled batteries on their days, everything else parked
+    one day past the horizon where evaluate_plan discards it for free.
+    """
+    parked_day = start_time + pandas.Timedelta(days=window_days + 1)
+    scheduled = set(selected_days.index)
+    parked = [b for b in locations['battery'] if b not in scheduled]
+
+    plan = pandas.concat([
+        pandas.DataFrame({'day': selected_days.values, 'battery': selected_days.index}),
+        pandas.DataFrame({'day': parked_day, 'battery': parked}),
+    ], ignore_index=True)
+    plan = plan.sort_values('day', kind='stable').reset_index(drop=True)
+
+    check_plan_valid(plan, locations, start_time=start_time)
+    return plan
+
+
+class DeferAllPlanner(Planner):
+    """Park everything. Pays only forced emergency visits."""
+
+    def plan(self, battery_data, locations, travel_costs, settings):
+        start_time = infer_start_time(battery_data)
+        empty = pandas.Series(dtype='datetime64[ns]')
+        return _assemble(empty, locations, start_time, settings.planning_window_days)
+
+
+class LowestVoltagePlanner(Planner):
+    """
+    Swap the k batteries with the lowest last in-band voltage, worst first, one per
+    day from the start of the window.
+
+    k is deliberately allowed to exceed the expected number of deaths: missing a
+    battery that dies costs an emergency visit, while including a healthy one costs
+    only early_replacement_penalty_daily per day.
+    """
+
+    def __init__(self, k: int = 10, feature_cache: Optional[dict] = None):
+        self.k = k
+        self.feature_cache = feature_cache if feature_cache is not None else {}
+
+    def _voltage(self, battery_data, start_time):
+        key = pandas.Timestamp(start_time)
+        if key not in self.feature_cache:
+            self.feature_cache[key] = voltage_feature(battery_data)
+        return self.feature_cache[key]
+
+    def plan(self, battery_data, locations, travel_costs, settings):
+        start_time = infer_start_time(battery_data)
+        voltage = self._voltage(battery_data, start_time)
+
+        # Restrict to batteries still in the fleet; unmeasured ones sort last.
+        v = voltage.reindex(locations['battery']).fillna(numpy.inf)
+        worst = v.nsmallest(self.k, keep='first')
+
+        days = start_time + pandas.to_timedelta(numpy.arange(len(worst)), unit='D')
+        selected = pandas.Series(days, index=worst.index)
+        return _assemble(selected, locations, start_time, settings.planning_window_days)
+
+
+class OraclePlanner(Planner):
+    """Upper bound: knows the true EOL and swaps on exactly that day."""
+
+    def __init__(self, eol_times: pandas.Series):
+        self.eol_times = eol_times
+
+    def plan(self, battery_data, locations, travel_costs, settings):
+        start_time = infer_start_time(battery_data)
+        window = pandas.Timedelta(days=settings.planning_window_days)
+        eol = self.eol_times.reindex(locations['battery']).dropna()
+        doomed = eol[(eol > start_time) & (eol <= start_time + window)].sort_values()
+        return _assemble(doomed, locations, start_time, settings.planning_window_days)
+
+
+def check_feature_parity(battery_data, cache_key, feature_cache):
+    """
+    Guard against train/serve skew: the harness prefills the same cache the planner
+    reads, so assert the cached value matches a fresh online computation.
+    """
+    cached = feature_cache[cache_key]
+    online = voltage_feature(battery_data)
+    pandas.testing.assert_series_equal(cached.sort_index(), online.sort_index())
+
+
+def load_shipped_planner(path='batteryswap_example/planners/best.pickle'):
+    """
+    best.pickle was written by running train.py as a script, so its classes are
+    recorded under __main__. Register them there before unpickling.
+    """
+    import pickle
+    import __main__
+    from batteryswap_example import train as train_module
+    for name in ('OrderedPlanner', 'DummyRULModel'):
+        setattr(__main__, name, getattr(train_module, name))
+    with open(path, 'rb') as f:
+        return pickle.load(f)
+
+
+def run(dataset_path: Path, origins=TEST_ORIGINS, k_values=(5, 10, 15, 20, 30)):
+    locations, timeseries, eol_times, scenarios = load_dataset(dataset_path / 'train')
+    log.info('harness-loaded', devices=len(locations), origins=list(origins))
+
+    shipped = load_shipped_planner()
+
+    feature_cache: dict = {}
+    policies = {'current': shipped, 'defer': DeferAllPlanner(),
+                'oracle': OraclePlanner(eol_times)}
+    for k in k_values:
+        policies[f'voltage-{k}'] = LowestVoltagePlanner(k, feature_cache=feature_cache)
+
+    records = []
+    parity_checked = False
+    gen = iterate_scenarios(locations, timeseries, eol_times, scenarios)
+    for index, (scenario, locs, cut, eol) in enumerate(gen):
+        if index not in origins:
+            continue
+        start_time = pandas.Timestamp(scenario['start_time'])
+        inferred = infer_start_time(cut)
+        assert inferred == start_time, (scenario['name'], inferred, start_time)
+
+        for name, planner in policies.items():
+            plan = planner.plan(cut, locs, scenario['travel_costs'], scenario['settings'])
+            _, _, overall = evaluate_plan(
+                plan, locs, scenario['travel_costs'], scenario['settings'],
+                eol_times=eol, start_time=start_time, verbose=0)
+            records.append(dict(overall, policy=name, origin=scenario['name']))
+
+        if not parity_checked:
+            check_feature_parity(cut, start_time, feature_cache)
+            parity_checked = True
+            log.info('feature-parity-ok', origin=scenario['name'])
+
+        log.info('origin-done', origin=scenario['name'])
+
+    return pandas.DataFrame(records)
+
+
+def report(df: pandas.DataFrame):
+    order = ['total_cost', 'late_swap', 'early_swap', 'daily_limit', 'weekly_limit',
+             'overtime', 'travel', 'battery_swap', 'building_change', 'room_change']
+    mean = df.groupby('policy')[order].mean().sort_values('total_cost')
+
+    defer = mean.loc['defer', 'total_cost']
+    per_origin = df.pivot(index='origin', columns='policy', values='total_cost')
+    beats = (per_origin.lt(per_origin['defer'], axis=0)).sum()
+    spread = per_origin.std()
+
+    summary = mean.copy()
+    summary['vs_defer_%'] = (100 * (mean['total_cost'] / defer - 1)).round(1)
+    summary['origins_beating_defer'] = beats.reindex(summary.index).astype(str) + f'/{len(per_origin)}'
+    summary['std_across_origins'] = spread.reindex(summary.index).round(1)
+
+    print(f'\nForward split: scored on {len(per_origin)} origins '
+          f'({per_origin.index[0]}..{per_origin.index[-1]})\n')
+    print(summary.round(2).to_string())
+    return summary
+
+
+class Config(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix='', cli_parse_args=True,
+                                      cli_ignore_unknown_args=True)
+    dataset_path: Optional[Path] = None
+
+
+def main():
+    cfg = Config()
+    dataset_path = cfg.dataset_path or Path(
+        os.environ.get('BATTERYSWAP_DATASET_PATH', 'dataset'))
+    df = run(Path(dataset_path))
+    report(df)
+
+
+if __name__ == '__main__':
+    main()

batteryswap_example/models_cv.py

@@ -0,0 +1,399 @@
+"""
+Model comparison under walk-forward CV.
+
+Protocol
+--------
+- Training pairs at origin T: all (device, origin') rows whose 42-day window closed
+  before T (origin' <= T - 42d). Window closed => label fully observed, and there is
+  a built-in embargo: no training window overlaps the scored window.
+- Walk-forward: the model is refit at every scored origin on exactly the pairs that
+  would have been observable then.
+- DEV block s_8..s_23: hyperparameters (p_star, day quantile, planner) are tuned here.
+- TEST block s_24..s_41: scored once with dev-chosen settings.
+
+Models predict, per alive device:
+  p42   - probability of EOL within the 42-day window
+  day_q - quantile of days-to-EOL conditional on dying in the window
+
+Planners turn (p42, day_q) into a plan:
+  onepd  - worst-first, one per day (control, shape of the shipped planner)
+  sched  - swap each included battery on its predicted day
+  batch  - sched + move each building's swaps to the group's earliest day and
+           cap building-groups per day (travel is expensive, early days are cheap)
+"""
+
+from pathlib import Path
+import itertools
+import os
+import sys
+import warnings
+
+import numpy as np
+import pandas as pd
+
+from sklearn.impute import SimpleImputer
+from sklearn.preprocessing import StandardScaler
+from sklearn.pipeline import make_pipeline
+from sklearn.linear_model import LogisticRegression
+from sklearn.ensemble import HistGradientBoostingClassifier, GradientBoostingRegressor
+from sklearn.metrics import roc_auc_score
+
+from batteryswap_public.utils import load_dataset, iterate_scenarios
+from batteryswap_public.evaluate import evaluate_plan, check_plan_valid
+
+from batteryswap_example.features import load as load_features, WINDOW_DAYS
+
+FEATURES = ['v_last', 'v_drop', 'slope_30', 'slope_90', 'v_min_30',
+            'staleness', 'age_days']
+DEV_ORIGINS = range(8, 24)
+TEST_ORIGINS = range(24, 42)
+DAY_QUANTILES = (0.05, 0.20, 0.50)
+
+
+# --------------------------------------------------------------------------- models
+
+class DayQuantiles:
+    """Quantiles of days-to-EOL, fit on training pairs that died in-window."""
+
+    def __init__(self):
+        self.models = {}
+        self.fallback = {}
+
+    def fit(self, X, days):
+        for q in DAY_QUANTILES:
+            self.fallback[q] = float(np.quantile(days, q))
+            if len(days) >= 40:
+                m = make_pipeline(
+                    SimpleImputer(strategy='median'),
+                    GradientBoostingRegressor(loss='quantile', alpha=q,
+                                              n_estimators=100, max_depth=2,
+                                              random_state=0))
+                m.fit(X, days)
+                self.models[q] = m
+        return self
+
+    def predict(self, X):
+        out = {}
+        for q in DAY_QUANTILES:
+            if q in self.models:
+                out[f'day_q{q}'] = np.clip(self.models[q].predict(X), 0, WINDOW_DAYS)
+            else:
+                out[f'day_q{q}'] = np.full(len(X), self.fallback[q])
+        return pd.DataFrame(out, index=X.index)
+
+
+class SklearnRUL:
+    """p42 from a sklearn classifier + DayQuantiles for timing."""
+
+    def __init__(self, kind):
+        self.kind = kind
+
+    def fit(self, train):
+        X, y = train[FEATURES], train['dies42']
+        if self.kind == 'logistic':
+            self.clf = make_pipeline(SimpleImputer(strategy='median'),
+                                     StandardScaler(),
+                                     LogisticRegression(max_iter=2000, C=1.0))
+        elif self.kind == 'gbm':
+            self.clf = HistGradientBoostingClassifier(
+                max_iter=200, max_depth=3, learning_rate=0.1,
+                min_samples_leaf=20, random_state=0)
+        self.clf.fit(X, y)
+
+        dying = train[train['dies42'] == 1]
+        self.days = DayQuantiles().fit(dying[FEATURES], dying['days_to_eol'])
+        return self
+
+    def predict(self, rows):
+        X = rows[FEATURES]
+        out = self.days.predict(X)
+        out['p42'] = self.clf.predict_proba(X)[:, 1]
+        return out
+
+
+class RankVoltage:
+    """Control: score by v_last only; p42 is a rank in [0,1], day from DayQuantiles."""
+
+    def fit(self, train):
+        dying = train[train['dies42'] == 1]
+        self.days = DayQuantiles().fit(dying[FEATURES], dying['days_to_eol'])
+        return self
+
+    def predict(self, rows):
+        out = self.days.predict(rows[FEATURES])
+        v = rows['v_last'].fillna(np.inf)
+        out['p42'] = 1.0 - v.rank(pct=True).values  # lowest voltage -> highest score
+        return out
+
+
+class WeibullAFT:
+    """lifelines Weibull AFT on censored pairs; survival curve gives p42 and timing."""
+
+    def fit(self, train_censored):
+        from lifelines import WeibullAFTFitter
+        df = train_censored[FEATURES + ['duration', 'event']].copy()
+        self.imputer = SimpleImputer(strategy='median').fit(df[FEATURES])
+        df[FEATURES] = self.imputer.transform(df[FEATURES])
+        self.aft = WeibullAFTFitter(penalizer=0.01)
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            self.aft.fit(df, duration_col='duration', event_col='event')
+        return self
+
+    def predict(self, rows):
+        X = rows[FEATURES].copy()
+        X[FEATURES] = self.imputer.transform(X[FEATURES])
+        times = np.arange(1, WINDOW_DAYS + 1)
+        surv = self.aft.predict_survival_function(X, times=times)  # times x devices
+        cdf = 1.0 - surv.T  # devices x times
+        p42 = cdf.iloc[:, -1]
+        out = pd.DataFrame({'p42': p42.values}, index=rows.index)
+
+        # per-day death mass and expected residual life given surviving the window,
+        # for the expected-cost planner
+        pmf = np.diff(np.column_stack([np.zeros(len(cdf)), cdf.values]), axis=1)
+        for t in times:
+            out[f'pd_{t}'] = pmf[:, t - 1]
+        far = np.arange(WINDOW_DAYS, 400)
+        surv_far = self.aft.predict_survival_function(X, times=far).T.values
+        s42 = np.maximum(surv_far[:, 0], 1e-12)
+        out['e_far'] = WINDOW_DAYS + np.trapezoid(surv_far, far, axis=1) / s42
+        # quantiles of death time conditional on dying within the window
+        cond = cdf.div(p42.replace(0, np.nan), axis=0).clip(upper=1.0)
+        for q in DAY_QUANTILES:
+            day = (cond.values >= q).argmax(axis=1) + 1.0
+            day[~(cond.values[:, -1] >= q)] = WINDOW_DAYS
+            out[f'day_q{q}'] = np.where(p42.values > 0, day, WINDOW_DAYS)
+        return out
+
+
+MODELS = {
+    'rank-v': RankVoltage,
+    'logistic': lambda: SklearnRUL('logistic'),
+    'gbm': lambda: SklearnRUL('gbm'),
+    'aft': WeibullAFT,
+}
+
+
+# ------------------------------------------------------------------------- planners
+
+def assemble(selected_days, locations, start_time):
+    parked_day = start_time + pd.Timedelta(days=WINDOW_DAYS + 1)
+    scheduled = set(selected_days.index)
+    parked = [b for b in locations['battery'] if b not in scheduled]
+    plan = pd.concat([
+        pd.DataFrame({'day': selected_days.values, 'battery': selected_days.index}),
+        pd.DataFrame({'day': parked_day, 'battery': parked}),
+    ], ignore_index=True).sort_values('day', kind='stable').reset_index(drop=True)
+    check_plan_valid(plan, locations, start_time=start_time)
+    return plan
+
+
+def plan_onepd(pred, included, locations, start_time, day_q):
+    order = pred.loc[included, 'p42'].sort_values(ascending=False)
+    days = start_time + pd.to_timedelta(np.arange(len(order)), unit='D')
+    return assemble(pd.Series(days, index=order.index), locations, start_time)
+
+
+def plan_sched(pred, included, locations, start_time, day_q):
+    offs = pred.loc[included, f'day_q{day_q}'].round().clip(0, WINDOW_DAYS)
+    days = start_time + pd.to_timedelta(offs.values, unit='D')
+    return assemble(pd.Series(days, index=offs.index), locations, start_time)
+
+
+def plan_batch(pred, included, locations, start_time, day_q, max_groups_per_day=2):
+    offs = pred.loc[included, f'day_q{day_q}'].round().clip(0, WINDOW_DAYS)
+    bldg = locations.set_index('battery')['building'].reindex(offs.index)
+
+    # one visit per building at that group's earliest predicted day
+    group_day = offs.groupby(bldg).min().sort_values()
+
+    # cap building-groups per day; push overflow to the next free day
+    used = {}
+    final = {}
+    for building, day in group_day.items():
+        d = int(day)
+        while used.get(d, 0) >= max_groups_per_day and d < WINDOW_DAYS:
+            d += 1
+        used[d] = used.get(d, 0) + 1
+        final[building] = d
+
+    offs = bldg.map(final).astype(float)
+    days = start_time + pd.to_timedelta(offs.values, unit='D')
+    return assemble(pd.Series(days, index=offs.index), locations, start_time)
+
+
+def _batch_days(offs, locations, start_time, max_groups_per_day=2):
+    """Shared batching pass: one visit per building at the group's earliest day."""
+    bldg = locations.set_index('battery')['building'].reindex(offs.index)
+    group_day = offs.groupby(bldg).min().sort_values()
+    used, final = {}, {}
+    for building, day in group_day.items():
+        d = int(day)
+        while used.get(d, 0) >= max_groups_per_day and d < WINDOW_DAYS:
+            d += 1
+        used[d] = used.get(d, 0) + 1
+        final[building] = d
+    offs = bldg.map(final).astype(float)
+    days = start_time + pd.to_timedelta(offs.values, unit='D')
+    return assemble(pd.Series(days, index=offs.index), locations, start_time)
+
+
+EARLY, LATE = 0.5, 10.0
+EMERGENCY_DAY = WINDOW_DAYS + 3  # emergencies run just after the plan ends
+
+
+def plan_expcost(pred, included, locations, start_time, day_q):
+    """
+    Include/park and swap-day both chosen by expected cost under the model's
+    death distribution. `included` and `day_q` are ignored.
+    """
+    t = np.arange(1, WINDOW_DAYS + 1)
+    pmf = pred[[f'pd_{i}' for i in t]].values          # devices x 42
+    p_surv = np.clip(1.0 - pmf.sum(axis=1), 0.0, 1.0)
+    e_far = pred['e_far'].values
+
+    d = np.arange(0, WINDOW_DAYS + 1)
+    delta = d[None, :, None] - t[None, None, :]        # 1 x 43 x 42
+    swap_penalty = np.where(delta >= 0, LATE * delta, EARLY * (-delta))
+    cost = (pmf[:, None, :] * swap_penalty).sum(axis=2)          # devices x 43
+    cost += p_surv[:, None] * EARLY * np.maximum(e_far[:, None] - d[None, :], 0)
+
+    cost_park = (pmf * (LATE * (EMERGENCY_DAY - t))[None, :]).sum(axis=1)
+
+    best_day = cost.argmin(axis=1)
+    include = cost[np.arange(len(cost)), best_day] < cost_park
+    offs = pd.Series(best_day[include].astype(float), index=pred.index[include])
+    if offs.empty:
+        return assemble(pd.Series(dtype='datetime64[ns]'), locations, start_time)
+    return _batch_days(offs, locations, start_time)
+
+
+def plan_topsum(pred, included, locations, start_time, day_q):
+    """
+    Adaptive inclusion: take the top-N batteries by p42 with N = alpha * sum(p42).
+    sum(p42) tracks the expected death count even when the probability scale
+    drifts between refits, unlike a fixed threshold. alpha arrives via p_star.
+    """
+    alpha = plan_topsum.alpha
+    n = int(round(alpha * pred['p42'].sum()))
+    chosen = pred['p42'].nlargest(n).index
+    offs = pred.loc[chosen, f'day_q{day_q}'].round().clip(0, WINDOW_DAYS)
+    if offs.empty:
+        return assemble(pd.Series(dtype='datetime64[ns]'), locations, start_time)
+    return _batch_days(offs, locations, start_time)
+
+
+plan_topsum.alpha = 1.2
+
+PLANNERS = {'onepd': plan_onepd, 'sched': plan_sched, 'batch': plan_batch,
+            'expcost': plan_expcost, 'topsum': plan_topsum}
+
+
+# ------------------------------------------------------------------------ harness
+
+def training_pairs(table, origin_time):
+    closed = table['origin'] <= origin_time - pd.Timedelta(days=WINDOW_DAYS)
+    return table[closed]
+
+
+def censored_pairs(table, origin_time):
+    """All prior pairs with duration censored at origin_time (for AFT)."""
+    prior = table[table['origin'] < origin_time].copy()
+    eol = prior['eol']
+    end = eol.where(eol.notna() & (eol <= origin_time), origin_time)
+    prior['duration'] = ((end - prior['origin']).dt.total_seconds() / 86400).clip(lower=0.5)
+    prior['event'] = (eol.notna() & (eol <= origin_time)).astype(int)
+    return prior
+
+
+def walk_forward(table, scenario_ctx, origins, configs):
+    """configs: list of dicts(model, planner, p_star, day_q, name)."""
+    fitted = {}
+    records = []
+    for idx in origins:
+        ctx = scenario_ctx[idx]
+        origin_time = ctx['start_time']
+        rows = table[table['origin_idx'] == idx].set_index('device_id')
+
+        preds = {}
+        for model_name in {c['model'] for c in configs}:
+            if (model_name, idx) not in fitted:
+                factory = MODELS[model_name]
+                if model_name == 'aft':
+                    train = censored_pairs(table, origin_time)
+                else:
+                    train = training_pairs(table, origin_time)
+                fitted[(model_name, idx)] = factory().fit(train)
+            preds[model_name] = fitted[(model_name, idx)].predict(rows)
+
+        for cfg in configs:
+            pred = preds[cfg['model']]
+            if cfg['planner'] == 'topsum':
+                PLANNERS['topsum'].alpha = cfg['p_star']  # p_star carries alpha
+            included = pred.index[pred['p42'] >= cfg['p_star']]
+            plan = PLANNERS[cfg['planner']](pred, included, ctx['locs'],
+                                            origin_time, cfg['day_q'])
+            _, _, overall = evaluate_plan(
+                plan, ctx['locs'], ctx['travel_costs'], ctx['settings'],
+                eol_times=ctx['eol'], start_time=origin_time, verbose=0)
+            records.append(dict(overall, origin=idx, **cfg))
+
+        auc = {m: roc_auc_score(rows['dies42'], p['p42'])
+               for m, p in preds.items() if rows['dies42'].sum() > 0}
+        print(f"s_{idx}: " + ' '.join(f'{m}={a:.3f}' for m, a in auc.items()),
+              file=sys.stderr)
+    return pd.DataFrame(records)
+
+
+def load_context(dataset_path):
+    locations, timeseries, eol_times, scenarios = load_dataset(dataset_path / 'train')
+    ctx = {}
+    gen = iterate_scenarios(locations, timeseries, eol_times, scenarios)
+    for idx, (scenario, locs, cut, eol) in enumerate(gen):
+        ctx[idx] = dict(start_time=pd.Timestamp(scenario['start_time']),
+                        locs=locs, eol=eol,
+                        travel_costs=scenario['travel_costs'],
+                        settings=scenario['settings'])
+    return ctx
+
+
+def summarize(df, label):
+    df['name'] = (df['model'] + '/' + df['planner'] + '/p' +
+                  df['p_star'].astype(str) + '/q' + df['day_q'].astype(str))
+    mean = df.groupby('name')['total_cost'].agg(['mean', 'std']).sort_values('mean')
+    print(f'\n=== {label} ===')
+    print(mean.round(1).to_string())
+    return mean
+
+
+def main():
+    dataset_path = Path(os.environ.get('BATTERYSWAP_DATASET_PATH', 'dataset'))
+    table = load_features(dataset_path)
+    ctx = load_context(dataset_path)
+
+    phase = sys.argv[1] if len(sys.argv) > 1 else 'dev'
+
+    if phase == 'dev':
+        configs = []
+        for model in MODELS:
+            for planner in PLANNERS:
+                for p_star in (0.1, 0.2, 0.35):
+                    for day_q in (0.05, 0.20):
+                        configs.append(dict(model=model, planner=planner,
+                                            p_star=p_star, day_q=day_q))
+        df = walk_forward(table, ctx, DEV_ORIGINS, configs)
+        df.to_csv('/tmp/dev_results.csv', index=False)
+        summarize(df, f'DEV s_8..s_23 ({len(configs)} configs)')
+    else:
+        # phase 'dev2' or 'test': configs from /tmp/<phase>_configs.json
+        import json
+        configs = json.loads(Path(f'/tmp/{phase}_configs.json').read_text())
+        origins = DEV_ORIGINS if phase.startswith('dev') else TEST_ORIGINS
+        df = walk_forward(table, ctx, origins, configs)
+        df.to_csv(f'/tmp/{phase}_results.csv', index=False)
+        summarize(df, f'{phase.upper()} origins {list(origins)[0]}..{list(origins)[-1]}')
+
+
+if __name__ == '__main__':
+    main()

batteryswap_example/planner.py

@@ -0,0 +1,102 @@
+"""
+Final planner: logistic RUL model + adaptive top-sum inclusion + building batching.
+
+Selected by walk-forward CV (see models_cv.py):
+  dev block  s_8..s_23  - hyperparameters chosen (alpha=0.8, day_q=0.05, batching)
+  test block s_24..s_41 - mean total_cost 1368 vs voltage-15 2034, defer-all 2512,
+                          oracle 186
+
+Everything the planner needs at plan() time is computed from the arguments the
+competition passes it: the truncated timeseries, the locations table (which carries
+each device's install date), and the scenario settings.
+"""
+
+import pickle
+
+import numpy as np
+import pandas as pd
+
+from batteryswap_public.interfaces import Planner
+
+from batteryswap_example.features import daily_voltage, features_at_origin, load as load_features
+from batteryswap_example.models_cv import SklearnRUL, FEATURES
+
+
+class TopSumPlanner(Planner):
+
+    def __init__(self, model: SklearnRUL, alpha: float = 0.8, day_q: float = 0.05,
+                 max_groups_per_day: int = 2):
+        self.model = model
+        self.alpha = alpha
+        self.day_q = day_q
+        self.max_groups_per_day = max_groups_per_day
+
+    def plan(self, battery_data, locations, travel_costs, settings):
+        window = int(settings.planning_window_days)
+        origin = battery_data.reset_index()['end_time'].max().normalize()
+
+        daily = daily_voltage(battery_data)
+        feats = features_at_origin(daily, locations, origin)
+        feats = feats.reindex(locations['battery'])
+
+        pred = self.model.predict(feats)
+
+        # adaptive inclusion: expected death count tracks prevalence even when the
+        # probability scale drifts between refits, unlike a fixed threshold
+        n = int(round(self.alpha * pred['p42'].sum()))
+        chosen = pred['p42'].nlargest(n).index
+
+        offs = pred.loc[chosen, f'day_q{self.day_q}'].round().clip(0, window)
+
+        # batching: one visit per building at the group's earliest predicted day,
+        # capped groups per day (travel is expensive, early days are cheap)
+        if len(offs):
+            bldg = locations.set_index('battery')['building'].reindex(offs.index)
+            group_day = offs.groupby(bldg).min().sort_values()
+            used, final = {}, {}
+            for building, day in group_day.items():
+                d = int(day)
+                while used.get(d, 0) >= self.max_groups_per_day and d < window:
+                    d += 1
+                used[d] = used.get(d, 0) + 1
+                final[building] = d
+            offs = bldg.map(final).astype(float)
+
+        parked_day = origin + pd.Timedelta(days=window + 1)
+        scheduled = set(offs.index)
+        parked = [b for b in locations['battery'] if b not in scheduled]
+        plan = pd.concat([
+            pd.DataFrame({'day': origin + pd.to_timedelta(offs.values, unit='D'),
+                          'battery': offs.index}),
+            pd.DataFrame({'day': parked_day, 'battery': parked}),
+        ], ignore_index=True)
+        return plan.sort_values('day', kind='stable').reset_index(drop=True)
+
+
+def train_full(dataset_path='dataset'):
+    """Refit the selected model on every fully observed (device, origin) pair."""
+    from pathlib import Path
+    table = load_features(Path(dataset_path))
+    train = table[table['window_closed']]
+    model = SklearnRUL('logistic').fit(train)
+    print(f'refit on {len(train)} pairs, {train.dies42.sum()} positive, '
+          f'origins s_0..s_{train.origin_idx.max()}')
+    return TopSumPlanner(model)
+
+
+def main():
+    # When run as a script/-m, classes defined here would pickle under __main__ and
+    # be unloadable by script.py. Rebuild the instance from the imported module so
+    # the pickle records the importable path.
+    import os
+    import batteryswap_example.planner as this_module
+    dataset_path = os.environ.get('BATTERYSWAP_DATASET_PATH', 'dataset')
+    planner = this_module.train_full(dataset_path)
+    path = 'batteryswap_example/planners/best.pickle'
+    with open(path, 'wb') as f:
+        pickle.dump(planner, f)
+    print('planner-save', path)
+
+
+if __name__ == '__main__':
+    main()

batteryswap_example/planners/best.pickle

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:66b857b72fcb7f476bbfc14077480f894ea15dfcd0043ca90bf1056cbc677de8
-size 589
+oid sha256:ba4a9792d820d91eab6cf328c8c8329dc3921d837dc864c23bdd7dbae0b1bc59
+size 237194