Blame for batteryswap_example/submit.py

"""
Adaptive submission generator.

Held-out splits (fleet not in the packaged train table): the foreign train pool
POOLED with the split's own closed-window pairs, refit on a cadence as the pool
grows. Phase 24b had switched the own-history half off after the half-fleet
holdout called static-vs-rolling a tie - the instrument later retired, and its
small target fleet distorts exactly this comparison. Degrades to the plain
static foreign fit (the prior champion) if the build or the refits run long;
never to the pickle, which is far below it. Set BATTERYSWAP_HELDOUT_POOL=0 to
restore the pure static behaviour.

Train-fleet splits: refit at every scenario origin on the split's own
self-masked history, identical to the walk-forward procedure validated
locally (and to how the local all-48 metric is produced).

Label integrity: the runtime dataset ships each split's eol_times.csv, which
contains ground truth beyond the scenario origins. This module NEVER uses it
directly for planning. The split's OWN training pairs are restricted to origins
whose full 42-day window CLOSED before the scenario origin, so every label used
was observable at planning time; an assertion enforces it. (This mirrors
iterate_scenarios' own legitimate use of eol_times to drop already-dead devices.)
Pairs from the shipped TRAIN table are a different fleet whose labels were all
known when the submission was built - those are pooled UNMASKED (closed windows
only), except for devices that also belong to the target split's fleet, which
stay under the time mask so a train-split run cannot read its own future.

Falls back to the static pickled planner per split if anything goes wrong.
"""

from pathlib import Path
import os
import pickle
import time
import traceback

import numpy as np
import pandas as pd

from batteryswap_public.utils import load_dataset, iterate_scenarios

from batteryswap_example.features import (
    daily_voltage, daily_all, weekly_betas, master_curve, physics_features,
    features_at_origin, WINDOW_DAYS, BETA_DEFAULT)
from batteryswap_example.models_cv import SklearnRUL, LogBagFeats, plan_tsroute
from batteryswap_example import models_cv

TRAIN_TABLE = Path(__file__).parent / 'data' / 'train_features.parquet'
GRID_STEP_DAYS = 4
# Held-out in-fleet pooling. The static branch (phase 24b) fits ONE model on
# the 82 foreign train deaths and discards the target split's own history
# entirely. But that split ships its own eol_times.csv, and every death with
# eol <= T - and every training pair whose 42d window CLOSED before T - was
# observable at planning time (iterate_scenarios reads the same file for the
# same reason). On a ~2x fleet those in-fleet pairs outnumber the foreign 82
# by the late origins, and they are the only signal that can correct a
# fleet-level shift the foreign fit never saw.
#
# Phase 24b measured static-vs-rolling as a tie and switched this off - on the
# half-fleet holdout, the instrument later retired precisely because its small
# target fleet distorts exactly this comparison.
#
# NOT LOCALLY MEASURABLE: the panel and all-48 run the ROLLING branch (the
# train fleet is its own fleet), so the static branch never executes locally
# and this change is byte-identical on every local metric by construction.
# It is gated on the phase-27 plumbing rehearsal instead, and on the fail-safe
# below - never on a local score, because there cannot be one.
#
# Runtime is the reason it was affordable to skip: build_split_table at the
# native 4-day grid was ~17 min for 461 devices, so ~35 for a 2x fleet, twice
# over, against a 30-minute hard limit. HELDOUT_GRID_STEP_DAYS coarsens the
# grid for this path only; the pairs it drops are pairs we do not have at all
# today. If the build overruns HELDOUT_BUILD_BUDGET_S we degrade to the plain
# static foreign fit - i.e. to the CURRENT CHAMPION - and never to the pickle
# fallback, which is far below it. Set HELDOUT_POOL=0 to disable.
# Phase 42 PUBLIC verdict: pooling was a wash with the wrong sign (1665.89 vs
# 1661.45; 3x training deaths bought zero recall). Default OFF so the phase-48
# submission isolates cert-first; env re-enables for experiments.
HELDOUT_POOL = os.environ.get('BATTERYSWAP_HELDOUT_POOL', '0') != '0'
HELDOUT_GRID_STEP_DAYS = int(os.environ.get('BATTERYSWAP_HELDOUT_GRID', '16'))
HELDOUT_BUILD_BUDGET_S = float(os.environ.get('BATTERYSWAP_HELDOUT_BUDGET_S', '360'))
# Refit cadence. The rehearsal showed the table build is cheap (7s at grid 16d
# on 461 devices) and the PER-ORIGIN REFITS dominate (~4.5s each x 48). At a
# ~2x held-out fleet that is ~8 min/split, ~16 for both, against a 20-min
# budget - too close for comfort when the real fleet size is unobservable from
# here. Reusing one fit across REFIT_EVERY consecutive origins halves the cost
# for almost no information: consecutive origins are days apart and the pool
# grows by a handful of closed pairs between them. 1 = refit at every origin.
# MEASURED (Docker, 2x fleet of 922 devices/split, both splits): at every-2
# origins a split costs ~10 min and the pair lands at 20m11s - over budget, and
# the guard correctly degraded private from s_31 onward, which silently throws
# away the pooling benefit on a third of that split. Refits are ~23 s each at
# this fleet size and dominate everything (table build is 22 s). Every-4 halves
# them again: ~5.5 min/split, ~11 min for the pair, comfortably inside 20.
HELDOUT_REFIT_EVERY = int(os.environ.get('BATTERYSWAP_HELDOUT_REFIT_EVERY', '4'))
MODEL_KIND = 'logbag'
# Phase 56 frontier (2026-08-22 purge-CV campaign, ~250 candidates across 11
# lanes): logbag + within-origin fleet-rank features (r_vlast, r_tau) with a
# gbm rank-blend at w=0.30, value-preserving (blended ORDER mapped back onto
# the primary's own p42 values, so count/pricing/timing machinery see the
# identical p42 distribution - ordering-only, swaps flat 9.23 vs 9.21).
# THE SCORE: 1630.3 vs champion 1972.0 (-341.7 +/- 103.8, 37W/11L, wins all
# three blocks). Components independently significant: rank features -225.6
# (linear, clean), gbm order increment -116.1 (non-monotone response; carries
# the accepted flexible-model purge-flattery risk - monotonic-constrained
# partner destroys the increment, so the public score is the arbiter).
# Confirmed local optimum by 5 lanes (hyperparams, features, weights, neural
# rankers, physics blends all subsumed or worse head-to-head).
# DEFAULT OFF (phase 56 final decomposition): the frontier's purge gain was
# a hidden per-scenario appetite dial - corr(count delta, cost gain) 0.76;
# at matched counts the ordering is worth only -37..-47 +/- 43 (n.s.).
# Era-hazard leaks through CALIBRATION in both-sides purge fits; deployment
# (past-only) cannot replicate it and public refuted it (1877.08). The dials
# below remain for explicit A/B only.
FRONTIER = os.environ.get('BATTERYSWAP_FRONTIER', '0') != '0'
# PUBLIC REFUTATION of the UNANCHORED frontier (698401a: 1877.08 vs 1661.45,
# swaps 517 vs 580): under deployment's past-only fits, logbag_rank's
# sum(p42) collapses vs plain logbag as the pool grows (measured ratio 1.02
# at s_5 -> 0.78 at s_45; the purge CV's both-sides fits hid it - swaps read
# flat there). The ordering signal was never publicly tested; the count
# channel drowned it. ANCHOR fixes this by construction: plain logbag is the
# VALUE DONOR (count/pricing/timing = the measured-580-swap calibration,
# byte-identical to the old champion's machinery) and the frontier supplies
# ONLY the rank order. Ordering-only in deployment, not just in CV.
# ANCHOR MODES (purge decomposition, phase 56 addendum 2): the frontier's
# -341.7 CV gain is mostly the VALUE SHAPE, not the order - value-anchoring
# to plain logbag keeps only the order and scores 1934.8 on the purge
# (-37.2 +/- 44.1 vs old champion, n.s.). 'scale' keeps logbag_rank's value
# SHAPE (the gain carrier) but rescales sum(p42) to the plain-logbag donor's
# sum, so the swap count is the measured champion calibration by construction
# while shape and order stay the frontier's. 'value' = donor values (order
# only). 'off' = unanchored (publicly refuted 1877.08).
FRONTIER_ANCHOR = os.environ.get('BATTERYSWAP_FRONTIER_ANCHOR', 'scale').lower()
if FRONTIER_ANCHOR in ('1', 'true'):
    FRONTIER_ANCHOR = 'value'
if FRONTIER_ANCHOR in ('0', 'false', 'none'):
    FRONTIER_ANCHOR = 'off'
# Blend weight dial for public A/B: 0.30 = full frontier (purge 1630.3);
# 0 = plain logbag_rank, no gbm (purge 1746.4 - the clean linear component,
# fallback if the gbm increment turns out to be purge flattery on public);
# BATTERYSWAP_FRONTIER=0 = legacy plain logbag (purge 1972.0, the 1661.45
# public config).
FRONTIER_W2 = float(os.environ.get('BATTERYSWAP_FRONTIER_W2', '0.30'))


class FrontierRUL:
    """logbag_rank + gbm value-preserving rank blend (the phase-56 frontier).

    Faithful deployment of cv.run_all48_oof(kind='logbag_rank',
    blend=('gbm', 0.30)): primary = LogBagFeats(['r_vlast', 'r_tau']) (bagged
    logistic on FEATURES + within-origin fleet-rank transforms), partner =
    SklearnRUL('gbm') on FEATURES; the blended rank ORDER is mapped back onto
    the primary's own sorted p42 values, and the DayQuantiles timing columns
    stay with their device (exactly what the CV blend branch does).
    """

    def fit(self, train):
        t = train.copy()
        t['r_vlast'] = t.groupby('origin')['v_last'].rank(pct=True)
        t['r_tau'] = t.groupby('origin')['tau_projected'].rank(pct=True)
        self.primary = LogBagFeats(['r_vlast', 'r_tau']).fit(t)
        self.partner = SklearnRUL('gbm').fit(t)
        # donor: the old champion's calibration owns the swap count
        self.donor = (SklearnRUL(MODEL_KIND).fit(train)
                      if FRONTIER_ANCHOR != 'off' else None)
        return self

    def predict(self, rows):
        r = rows.copy()
        # prediction rows are one fleet snapshot at one origin: the
        # within-origin rank is a rank over the fleet
        r['r_vlast'] = r['v_last'].rank(pct=True)
        r['r_tau'] = r['tau_projected'].rank(pct=True)
        pred = self.primary.predict(r)
        rk = pred['p42'].rank(pct=True)
        if FRONTIER_W2 > 0:
            p2 = self.partner.predict(r)
            rk = (1 - FRONTIER_W2) * rk + FRONTIER_W2 * p2['p42'].rank(pct=True)
        if FRONTIER_ANCHOR == 'value':
            base = self.donor.predict(rows)      # donor values, frontier order
        else:
            base = pred                          # frontier values (and shape)
        vals = np.sort(base['p42'].values)[::-1]
        out = base.copy()
        out['p42'] = pd.Series(
            vals, index=rk.sort_values(ascending=False).index
        ).reindex(out.index)
        if FRONTIER_ANCHOR == 'scale':
            # calibration correction: frontier shape, donor COUNT - fixes the
            # measured past-only sum(p42) collapse (0.78-0.80x at late origins)
            dsum, psum = self.donor.predict(rows)['p42'].sum(), out['p42'].sum()
            if psum > 0:
                out['p42'] *= dsum / psum
        return out


# CAMPAIGN-2 FRONTIER (2026-08-23, building-GroupKFold v5 instrument):
# ft_vmin7r_wnr20 = logbag on FEATURES + vmin7_rank (fleet pct-rank of the
# 7d-min voltage, computed by features_at_origin over whatever fleet it
# ranks - own-fleet semantics in deployment) with a 0.20 value-preserving
# WienerFHT order blend. THE SCORE 1688.4 vs champion 1760.2 (-71.8 +/-
# 23.5, 3.05 sigma, Wilcoxon p=.037, swaps 9.58 flat, count-clean by every
# control; components additive: wiener -50.3, vmin7_rank -23.2; reproduced
# independently by three lanes). Worst-case reference-shift bound: with
# vmin7_rank recomputed WITHIN each ~5-building test fleet the stack still
# reads -60.8 +/- 26.3 (2.31 sigma); the scored fleets are full ~922-device
# populations, much closer to the shared-reference case. Value-preserving
# means the blended rank ORDER is mapped back onto the primary's own sorted
# p42 values: count/pricing/timing machinery see the primary's exact p42
# distribution (the phase-56/58 shipping bar). Both components degrade
# softly: missing wnr_* columns drop the wiener source (pure vm7 logbag),
# any fit error falls back to the legacy champion via _fit_model.
FT2 = os.environ.get('BATTERYSWAP_FT2', '1') != '0'
FT2_W = float(os.environ.get('BATTERYSWAP_FT2_W', '0.30'))
# Scale hardening (phase 62 synthetic scaling study): vmin7_rank as a
# FIXED-reference percentile (train ECDF of v_min_7) instead of the scored
# fleet's live rank. Live-rank ft2's edge decays with fleet size and
# inverts at 4x (per-device 4.506 vs champ 4.438 at 1844 devices); the
# fixed form is scale-neutral (4.443) and best at 2x (3.848). Cost on the
# standard bgkf instrument: +2.1 +/- 14.4 (free at the mean, slightly
# sign-worse on small fleets). Docker rehearsal measured the real splits
# at 922 devices (= the 2x regime, where live-rank is fine), so this ships
# OFF - flip to 1 if the scored fleets turn out much larger than 922.
FT2_FIXEDREF = os.environ.get('BATTERYSWAP_FT2_FIXEDREF', '0') != '0'
# Gap-approach third order source (2026-08-23 late campaign): depth-2 HGB
# on FEATURES + vmin7_rank + four gap-approach columns (gap_ratio,
# rel_close, dwell_flat, gap_log - the plateau-vs-fast-crosser shape the
# autopsy exposed), entering the ORDER blend at this weight (primary takes
# 1 - FT2_W - FT2_GAPW). Value-preserving as ever: counts untouched.
# bgkf at the shipped operating point: 1568.6 (-32.5 +/- 25.2 vs 1601.2);
# BELOW the formal 2-sigma bar for new mechanisms - shipped on the user's
# call: six-of-six sign consistency across weights/bases and the count-flat
# order-blend risk class (never publicly refuted). 0 disables.
FT2_GAPW = float(os.environ.get('BATTERYSWAP_FT2_GAPW', '0.20'))
GAP_SRC = ['v_corr_last', 'v_min_30', 'v_min_90', 'slope_lastk',
           'days_below_245']


def _gap_features(frame):
    """The four gap-approach columns, derived from columns both the train
    table and features_at_origin provide. None if any source is missing."""
    if any(c not in frame.columns for c in GAP_SRC):
        return None
    gap = (frame['v_corr_last'] - 2.40).clip(lower=0.0)
    g30 = (frame['v_min_30'] - 2.40).clip(lower=1e-4)
    g90 = (frame['v_min_90'] - 2.40).clip(lower=1e-4)
    out = pd.DataFrame(index=frame.index)
    out['gap_ratio'] = (g30 / g90).clip(0, 2)
    out['rel_close'] = (-frame['slope_lastk'] / gap.clip(lower=0.005)).clip(-1, 1)
    out['dwell_flat'] = (frame['days_below_245'].fillna(0).clip(lower=0)
                         * out['gap_ratio'])
    out['gap_log'] = np.log(gap.clip(lower=1e-3))
    return out


_HZ_BUCKETS = [(0, 7), (7, 14), (14, 21), (21, 28), (28, 42), (42, 63), (63, 84)]


class _DiscHaz:
    """Discrete-time hazard HGB on person-period (pair, bucket) rows from
    the censored pool of whatever pair table fit() receives (durations
    censored at its last origin); order source P(die<=42d) = 1 - prod(1-h_k).
    Censoring is what the plain classifiers can't see - this model's edge is
    cold-start origins, which is the only place the warmup gate lets it act."""

    def fit(self, train):
        pool = models_cv.censored_pairs(train, train['origin'].max())
        self.cols = list(models_cv.FEATURES)
        dur = pool['duration'].values
        ev = pool['event'].values.astype(bool)
        Xb = pool[self.cols].values.astype(float)
        Xs, ys, mids = [], [], []
        for lo, hi in _HZ_BUCKETS:
            died = ev & (dur > lo) & (dur <= hi)
            use = (dur >= hi) | died
            Xs.append(Xb[use])
            ys.append(died[use])
            mids.append(np.full(int(use.sum()), (lo + hi) / 2.0))
        from sklearn.ensemble import HistGradientBoostingClassifier
        self.clf = HistGradientBoostingClassifier(
            max_depth=3, learning_rate=0.06, max_iter=250,
            l2_regularization=1.0, random_state=0, early_stopping=False,
        ).fit(np.column_stack([np.vstack(Xs), np.concatenate(mids)]),
              np.concatenate(ys))
        return self

    def predict(self, rows):
        Xb = rows[self.cols].values.astype(float)
        n = len(rows)
        surv = np.ones(n)
        for lo, hi in _HZ_BUCKETS[:5]:          # buckets covering 0-42d
            X = np.column_stack([Xb, np.full(n, (lo + hi) / 2.0)])
            surv = surv * (1.0 - self.clf.predict_proba(X)[:, 1])
        return pd.DataFrame({'p42': 1.0 - surv}, index=rows.index)


class Ft2RUL:
    """Faithful deployment of cv.run_bgkf(kind='logbag', variant='vm7',
    blend=('wiener', 0.20)): primary = SklearnRUL('logbag') fit with
    FEATURES + vmin7_rank (the CV sets mc.FEATURES globally; here the swap
    is scoped to this class's fits and predicts), partner = WienerFHT
    (fit under the same feature list, exactly as the CV's global swap did),
    blended rank order mapped onto the primary's own sorted p42 values,
    timing columns staying with their device."""

    def _swap(self):
        class _S:
            def __enter__(s):
                s.old = models_cv.FEATURES
                if 'vmin7_rank' not in models_cv.FEATURES:
                    models_cv.FEATURES = models_cv.FEATURES + ['vmin7_rank']
                return s
            def __exit__(s, *a):
                models_cv.FEATURES = s.old
        return _S()

    def fit(self, train):
        with self._swap():
            self.primary = SklearnRUL(MODEL_KIND).fit(train)
            self.wiener = (models_cv.WienerFHT().fit(train)
                           if FT2_W > 0 and 'wnr_n' in train.columns else None)
        self._ecdf = (np.sort(train['v_min_7'].dropna().to_numpy())
                      if FT2_FIXEDREF and 'v_min_7' in train.columns else None)
        self.gap = None
        g = _gap_features(train) if FT2_GAPW > 0 else None
        if g is not None and 'vmin7_rank' in train.columns:
            from sklearn.ensemble import HistGradientBoostingClassifier
            self._gap_cols = (list(models_cv.FEATURES) + ['vmin7_rank']
                              + list(g.columns))
            X = pd.concat([train[list(models_cv.FEATURES) + ['vmin7_rank']],
                           g], axis=1)
            self.gap = HistGradientBoostingClassifier(
                max_depth=2, learning_rate=0.06, max_iter=300,
                l2_regularization=1.0, random_state=0
            ).fit(X[self._gap_cols], train['dies42'])
        self.hazard, self.dev_eol = None, None
        if ('eol' not in train.columns and 'days_to_eol' in train.columns
                and 'origin' in train.columns):
            # the cached parquet ships days_to_eol but not eol (cv.load_table
            # adds it at load time); the derivation is exact (verified 0 diff
            # on all 8206 dying rows)
            train = train.copy()
            train['eol'] = train['origin'] + pd.to_timedelta(
                train['days_to_eol'], unit='D')
        if WARMUP_RATE > 0 and 'eol' in train.columns and 'device_id' in train.columns:
            try:
                self._gate_thr = WARMUP_RATE * train['device_id'].nunique()
                self.dev_eol = (train[['device_id', 'eol']].dropna()
                                .drop_duplicates('device_id')
                                .set_index('device_id')['eol'])
                with self._swap():
                    self.hazard = _DiscHaz().fit(train)
            except Exception:
                # warmup gate is an overlay: a broken hazard fit degrades to
                # the plain (previous-head) behavior, never to a crash
                traceback.print_exc()
                self.hazard = None
        return self

    def predict(self, rows):
        if self._ecdf is not None and 'v_min_7' in rows.columns:
            rows = rows.copy()
            v = rows['v_min_7'].to_numpy(dtype=float)
            pct = np.searchsorted(self._ecdf, v, side='right') / len(self._ecdf)
            rows['vmin7_rank'] = np.where(np.isfinite(v), pct, np.nan)
        if 'vmin7_rank' not in rows.columns:
            rows = rows.copy()
            rows['vmin7_rank'] = (rows['v_min_7'].rank(pct=True)
                                  if 'v_min_7' in rows.columns else np.nan)
        gate = False
        if (self.hazard is not None and self.dev_eol is not None
                and 'origin' in rows.columns):
            # T3 warmup gate: TRAIN-ledger deaths recorded <= this plan origin
            seen = int((self.dev_eol <= rows['origin'].max()).sum())
            gate = seen < self._gate_thr
        g = _gap_features(rows) if self.gap is not None else None
        with self._swap():
            pred = self.primary.predict(rows)
            if gate:
                try:
                    rk = self.hazard.predict(rows)['p42'].rank(pct=True)
                    models_cv.plan_tsroute.alpha_boost = WARMUP_ALPHA
                    vals = np.sort(pred['p42'].values)[::-1]
                    out = pred.copy()
                    out['p42'] = pd.Series(
                        vals, index=rk.sort_values(ascending=False).index
                    ).reindex(out.index)
                    return out
                except Exception:
                    traceback.print_exc()
                    models_cv.plan_tsroute.alpha_boost = 1.0
            rk = pred['p42'].rank(pct=True)
            w3 = FT2_GAPW if g is not None else 0.0
            # round the derived primary weight: 1-0.30-0.20 is 0.4999...94 in
            # floating point, and a 1e-17 weight difference can flip an exact
            # rank tie (measured: one flipped tie moved a cell by 539)
            w1 = round(1.0 - FT2_W - w3, 12)
            if self.wiener is not None and 'wnr_n' in rows.columns:
                # inside the swap: the wiener's DayQuantiles was fit under
                # the same swapped feature list (exactly the CV's global set)
                p2 = self.wiener.predict(rows)
                rk = w1 * rk + FT2_W * p2['p42'].rank(pct=True)
            elif w3 > 0:
                rk = (1 - w3) * rk
            if w3 > 0:
                X = pd.concat([rows, g], axis=1)
                p3 = pd.Series(
                    self.gap.predict_proba(X[self._gap_cols])[:, 1],
                    index=rows.index)
                rk = rk + w3 * p3.rank(pct=True)
        vals = np.sort(pred['p42'].values)[::-1]
        out = pred.copy()
        out['p42'] = pd.Series(
            vals, index=rk.sort_values(ascending=False).index
        ).reindex(out.index)
        return out


def _fit_model(train):
    if FT2:
        try:
            return Ft2RUL().fit(train)
        except Exception:
            # deadline-day discipline: a broken frontier fit degrades to the
            # measured champion, never to a crash
            traceback.print_exc()
    return (FrontierRUL() if FRONTIER else SklearnRUL(MODEL_KIND)).fit(train)
# Inclusion appetite, tuned by local CV: alpha sweep 0.7-1.3 on dev has a clear
# minimum at 0.8 (beats every alternative on 11-12 of 16 origins), confirmed on
# the test block (0.8: 1398 vs 1.15: 1516). Component curves show why: past 0.8
# each extra inclusion converts late->early at a worsening exchange rate, and
# late saturates ~1100 regardless - the remaining misses are voltage-silent
# deaths that no inclusion appetite can catch.
ALPHA = float(os.environ.get('BATTERYSWAP_ALPHA', '1.4'))
# Phase 26 adopted ALPHA_STATIC=1.5 from the half-fleet holdout ("the static
# foreign fit undercounts sum(p42)"). PUBLIC REFUTED IT (2026-08-20,
# submission 6ea03da: 2188.78 vs 1661.45, battery_swap +68%, early +444,
# late +13 - pure over-inclusion, zero recall gain): on the real held-out
# fleet the probability scale transfers, exactly as the phase-20 fleet
# holdout found (sum(p42) 15.8 vs 9.1 true - if anything it OVER-counts).
# The half-fleet construction manufactures the undercount; do not trust it
# for appetite decisions. Late cost is saturated by voltage-silent deaths
# that no inclusion appetite can catch (phases 1/6). Default back to 0.8.
# 2026-08-23 night (user call): flat alpha 1.4 on the CURRENT ordering for
# a public A/B - bgkf 1498.3 (-43.0 vs 1541.3, count-coupled: swaps
# 2.27->2.97/fleet = the podium's 15-18/scen operating point; NOT
# per-origin significant, 19W/24L - the mean is carried by death-heavy
# origins, which is exactly what the public read arbitrates). The 2189
# refutation above was alpha 1.5 on the OLD ordering; untested on ft2.
# Both knobs at 1.4 (static path governs held-out/public fleets).
# Rehearsal /tmp/subtest2: 520->688 swaps (+32.3%), 43/48 scenarios.
# Revert to the measured-1575.31 config: both defaults back to 1.0.
ALPHA_STATIC = float(os.environ.get('BATTERYSWAP_ALPHA_STATIC', '1.4'))
# Break-even top-up (phase 11): additionally include every FRESH battery
# (imputed EOL beyond the window) whose best-day expected cost beats parking,
# gated to fire only where a wrong inclusion costs <= EC_PMAX. Empty at
# dev-like prices; widens near a split's data end, where false swaps are
# nearly free but misses still cost 10/day. Panel 1702.0 vs 1715.8, selected
# in 63% of random half-splits, never beaten in a decided blocked split.
EC_MARGIN = float(os.environ.get('BATTERYSWAP_EC_MARGIN', '0.8'))
EC_PMAX = float(os.environ.get('BATTERYSWAP_EC_PMAX', '25'))
# Phase 14: let the day search split a building across days by best-day
# bucket (7d) instead of forcing one visit day per building. Panel 1679.2 vs
# 1702.0, selected in 100% of random AND blocked half-splits.
SPLIT_DAYS = int(os.environ.get('BATTERYSWAP_SPLIT_DAYS', '3'))
# T2 planner lane (2026-08-23): weight the timing curves x2 inside
# optimize_days' objective and split buildings at 3d (not 7d) buckets -
# together they refuse cheap trip-merges that drag a swap off its
# timing-optimal day (late -3.5, early -1.2, bought with +0.8 mobility).
# bgkf -4.11 +/- 1.41 (2.9sigma, 25W/4L, Wilcoxon 8.8e-6), count-flat with
# battery sets cell-identical; dose-response plateau on both dials.
# TIMING_WEIGHT=1 + SPLIT_DAYS=7 restores the old plan path exactly.
TIMING_WEIGHT = float(os.environ.get('BATTERYSWAP_TIMING_WEIGHT', '2.0'))
# Phase 17: cold-start inclusion floor. While the training set has seen fewer
# than FLOOR_GATE unique deaths, sum(p42) undercounts prevalence ~4x and the
# count rule parks visibly-dying batteries (warm-up block s_0-s_7: 46% of the
# all-48 late cost, catch rate 0.00-0.17 at s_0-s_2 with ranking AUC 0.97).
# Floor the count estimate with max(deaths recorded in the last 42d, fresh
# devices that crossed 2.45 within the last 42d) x FLOOR_SCALE. Inert once
# the gate closes (s_8+ locally): the 34-origin panel is byte-identical to
# the champion. Warm-up CV: 4256 -> 2380 at k=2.5, winning 8/8 origins.
FLOOR_SCALE = float(os.environ.get('BATTERYSWAP_FLOOR', '2.5'))
FLOOR_GATE = int(os.environ.get('BATTERYSWAP_FLOOR_GATE', '20'))
# T3 warmup gate (2026-08-23 night, user CV-first directive): while the
# TRAINING fleet's ledger shows < WARMUP_GATE recorded deaths at the plan
# origin (plan-time observable, FLOOR_GATE family; the full train ledger
# crosses 35 at s_15), the ORDER switches to a censored-pool discrete-time
# hazard HGB and the appetite is boosted x WARMUP_ALPHA via a consume-once
# plan_tsroute.alpha_boost. bgkf: gated alpha 1.4 alone -125.4 (4.0sigma,
# width bowl clean 6..48); + gated hazard order -143.3 (3.8sigma, 16W/0L,
# Wilcoxon <.001) = CV 1398.0 vs the 1541.3 head. Scale harness: the gate
# narrows as fleets grow (8/4/0 gate-active scenarios at 1x/2x/4x) so the
# downside is bounded - on fleets past the gate this is byte-identical to
# the previous head. WARMUP_GATE=0 disables both channels exactly.
# The gate threshold is a RATE (deaths per training-fleet device), anchored
# at 35 deaths / 461 devices: a fixed count fires wider on the CV's 4/5-size
# fold ledgers than on the full deployment ledger (measured: fixed 35 ->
# 22 changed origins, 15W/7L, -111.6; rate form matches the width-16 bowl
# optimum on both). WARMUP_RATE=0 disables the gate entirely.
# PUBLICLY REFUTED 2026-08-23: gate ON scored 1594.89 vs 1575.31 gate-off
# (late -55 but early +60, ops +14; public cold-scenario trade ~1.4:1
# against vs 6:1 on bgkf warmup). Default OFF; head = 60d65da config.
WARMUP_RATE = float(os.environ.get('BATTERYSWAP_WARMUP_RATE', '0.0'))
WARMUP_ALPHA = float(os.environ.get('BATTERYSWAP_WARMUP_ALPHA', '1.4'))
# Phase 28: cliff-gated appetite for the static held-out path. The static
# foreign fit undercounts exactly where a knee is already visible in the raw
# hourly readings (>=35% of the last 7 days' readings below 2.45 V, >=20
# readings, corrected voltage < 2.46): scale those rows' inclusion count and
# ranking by CLIFF_SCALE while leaving p42 - and hence every timing curve -
# untouched. Adjudication (exp_x28.py): real-label static holdout
# -54.5/scenario (4/5 fleet splits), barrier-event resampler -66.8 with the
# sign held in 120/120 draws (correlated per-quarter-shock variant -66.1,
# also 120/120), and beats a plain alpha raise on mean, variance and tail
# (alpha=2.0: -55 but sd 298 and worst +421; alpha=2.5 turns NEGATIVE).
# The native rolling path rejected it (panel +27): static branch ONLY.
# CLIFF_SCALE=1.0 disables (boost becomes the identity).
# PUBLIC (2026-08-20, 6ea03da at ALPHA_STATIC=1.5 + CLIFF_SCALE=2.2):
# 2188.78 - the stacked appetite over-included with zero recall gain. The
# cliff boost was adjudicated ON TOP of the 1.5 baseline whose own premise
# public refuted, so its holdout gain is unproven at appetite 0.8 on the
# real fleet. Disabled by default until (if ever) a public look at
# ALPHA_STATIC=0.8 + CLIFF_SCALE=2.2 isolates its solo effect.
CLIFF_SCALE = float(os.environ.get('BATTERYSWAP_CLIFF_SCALE', '1.0'))
CLIFF_FRAC = 0.35
CLIFF_V = 2.46
CLIFF_MIN_N = 20
# Phase 46: certainty-first ordering. The label process (reverse-engineered,
# 81/82 exact) stamps EOL ~3d after the gated daily-median voltage first
# sustains < 2.40 V for 3 consecutive gated readings. A device whose crossing
# is FULLY CONFIRMED in the past yet is still in the fleet is a label-lag /
# wobble case: locally 54 such flags, 51 die in-window, 3 die by +90d, ZERO
# never die. Boosting them to the top of the ordering (rank only - count and
# timing untouched) is -37.9 on the all-48 cost harness, 3W/0L/45T. The
# confirmation depth is load-bearing: k=1 (+291) and k=2 (+200) admit
# recoveries; k=3 - the label rule's own sustain - is the unique safe point.
# Past-only by construction: the flag requires all 3 readings <= T.
# PUBLIC REFUTED (2026-08-22, submission 9438aac: 1704.64 vs 1661.45, +43.19).
# Inversion: +18 swaps at ~111 early-days each, late +15.6. On train every
# sustained-crossing device dies by +90d (54/54), so the boost was loss-free
# locally; the public fleet contains crossed-but-NEVER-LABELED devices
# (ghosts), each force-swapped repeatedly and priced against end_time+30.
# Third consecutive locally-validated intervention refuted by public
# (ALPHA_STATIC=1.5, cliff boost, cert-first). Default OFF.
CERT_FIRST = os.environ.get('BATTERYSWAP_CERT_FIRST', '0') != '0'
CERT_V = 2.40
CERT_SUS = 3


def cert_confirm_days(daily):
    """Per device: the day of the CERT_SUS-th consecutive gated daily-median
    reading below CERT_V (the confirmation day), NaT-free Series. A device is
    certainty-flagged at origin T iff its confirmation day <= T, which uses
    only readings at or before T."""
    out = {}
    for dev, g in daily.sort_values(['device_id', 'day']).groupby(
            'device_id', sort=False):
        b = (g['v'] < CERT_V).to_numpy()
        for i in range(len(b) - CERT_SUS + 1):
            if b[i:i + CERT_SUS].all():
                out[dev] = g['day'].iloc[i + CERT_SUS - 1]
                break
    return pd.Series(out, dtype='datetime64[ns]')


def build_split_table(locations, eol_times, scen_starts, daily, dall, betas,
                      log=print, step_days=GRID_STEP_DAYS, deadline=None):
    """(device, origin) training pairs from this split's own history.

    deadline: wall-clock epoch after which the build aborts with TimeoutError.
    The held-out path passes one so an overrun degrades to the static foreign
    fit (the current champion) instead of eating the whole split's budget.
    """
    t0 = dall['day'].min() + pd.Timedelta(days=120)
    grid = pd.date_range(t0, max(scen_starts), freq=f'{step_days}D')

    rows = []
    for origin in grid:
        if deadline is not None and time.time() > deadline:
            raise TimeoutError(
                f'split-table build exceeded budget at {len(rows)}/{len(grid)} origins')
        alive = eol_times.isna() | (eol_times > origin)
        f = features_at_origin(daily, locations, origin).reindex(eol_times.index[alive])

        past = betas[betas['week_end'] <= origin]['beta']
        beta = float(past.median()) if len(past) >= 20 else BETA_DEFAULT
        dead = eol_times[eol_times.notna() & (eol_times <= origin)]
        curve = master_curve(dall[dall['day'] <= origin], dead, beta)
        f = f.join(physics_features(dall, origin, beta, curve, match=False))

        f['origin'] = origin
        eol = eol_times.reindex(f.index)
        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
        rows.append(f.reset_index().rename(columns={'index': 'device_id'}))
    table = pd.concat(rows, ignore_index=True)
    log(f'  split table: {len(table)} pairs over {len(grid)} origins, '
        f'{table.dies42.sum()} positive')
    return table


def cliff_day_counts(timeseries):
    """Per (device, day) raw-reading counts for the phase-28 cliff mask:
    total readings and readings below 2.45 V. Deliberately ungated and
    uncorrected - the mask is a telemetry-level knee detector; the official
    smoothing/gating is exactly what it must see through."""
    raw = timeseries.reset_index()
    raw['device_id'] = raw['device_id'].astype(str)
    raw['day'] = raw['end_time'].dt.normalize()
    raw['b245'] = raw['voltage'] < 2.45
    return raw.groupby(['device_id', 'day']).agg(
        n=('voltage', 'size'), n245=('b245', 'sum'))


def cliff_boost(pred, feats, day_counts, T):
    """Attach rank_score / p42_sum = clip(CLIFF_SCALE * p42, 0, 1) on cliff
    rows (plan_tsroute reads both columns natively; raw p42 stays untouched,
    so cost_curves timing is unaffected). Returns the number of boosted rows.
    Window convention matches the validated harness (exp_x28/features.py):
    calendar days [T-6d, T] inclusive."""
    days = day_counts.index.get_level_values('day')
    tn = T.normalize()
    win = day_counts[(days >= tn - pd.Timedelta(days=6)) & (days <= tn)]
    agg = win.groupby('device_id').sum()
    key = pred.index.astype(str)
    n7 = agg['n'].reindex(key).fillna(0.0).to_numpy(float)
    n245 = agg['n245'].reindex(key).fillna(0.0).to_numpy(float)
    frac = n245 / np.clip(n7, 1.0, None)
    v = feats['v_corr_last'].fillna(feats['v_last']).fillna(3.0).to_numpy(float)
    cliff = (frac >= CLIFF_FRAC) & (n7 >= CLIFF_MIN_N) & (v < CLIFF_V)
    boosted = pred['p42'].to_numpy(float).copy()
    boosted[cliff] = np.clip(boosted[cliff] * CLIFF_SCALE, 0.0, 1.0)
    pred['rank_score'] = boosted
    pred['p42_sum'] = float(boosted.sum())
    return int(cliff.sum())


def load_best_effort(split_path, log):
    """load_dataset with the consistency check demoted to a logged warning:
    a benign set mismatch in a held-out split must not cost the whole split
    (dress rehearsal: an assertion here killed adaptive AND the fallback,
    which re-raises identically - one bad split -> empty submission)."""
    from batteryswap_public.utils import check_dataset
    locations, timeseries, eol_times, scenarios = load_dataset(
        split_path, check=False)
    try:
        check_dataset(locations, timeseries, eol_times, scenarios)
    except Exception as e:
        log(f'  WARNING check_dataset: {e}')
    return locations, timeseries, eol_times, scenarios


def adaptive_split_plans(split, split_path, log=print, deadline=None):
    locations, timeseries, eol_times, scenarios = load_best_effort(split_path, log)
    scen_starts = [pd.Timestamp(s['start_time']) for s in scenarios]
    daily = daily_voltage(timeseries)
    ft2_daily = None
    if FT2:
        # per-scenario Wiener (drift, variance) for the ft2 order blend -
        # computed from THIS split's own past-only telemetry, same estimator
        # the training cache used. Any failure just drops the wiener source.
        try:
            from batteryswap_example.fht_features import (
                daily_corrected, wiener_at_origin)
            ft2_daily = daily_corrected(timeseries)
        except Exception as e:
            log(f'  WARNING ft2 wiener daily build failed ({e}); '
                f'order blend drops the wiener source')
    cert_days = cert_confirm_days(daily) if CERT_FIRST else pd.Series(dtype='datetime64[ns]')
    dall = daily_all(timeseries)
    betas = weekly_betas(dall)

    fleet = set(locations['battery'])
    base = None
    if TRAIN_TABLE.exists():
        base = pd.read_parquet(TRAIN_TABLE)
        # open-window rows (origins within 42d of the train data end) carry
        # unverifiable "survives" labels; filter explicitly - and BEFORE the
        # column intersect below, which drops window_closed
        base = base[base['window_closed']]
    own_fleet = (float(locations['battery'].isin(set(base['device_id'])).mean())
                 if base is not None else 1.0)

    static_model, static_uniq, cliff_counts = None, 0, None
    pool = None
    if base is not None and own_fleet < 0.5:
        # PHASE 24 (static fit): on a held-out split, per-origin refits with
        # the target's own masked history are a statistical tie against ONE
        # fit on the full foreign train pool (holdout harness: static 1057.7
        # vs rolling 1087.0 over 16 origins, 6W/7L/3T) - the refit machinery
        # is runtime cost with no measured benefit. A single fit also nearly
        # eliminates the time-budget fallback risk. Own-fleet rows are
        # excluded: same fleet means same answer key.
        foreign = base[~base['device_id'].isin(fleet)]
        static_model = _fit_model(foreign)
        static_uniq = int(foreign.loc[foreign['dies42'] == 1, 'device_id'].nunique())
        log(f'  static foreign fit: {len(foreign)} pairs, {static_uniq} unique '
            f'deaths (own-fleet frac {own_fleet:.2f})')
        cliff_counts = cliff_day_counts(timeseries) if CLIFF_SCALE != 1.0 else None

        # In-fleet pooling for the held-out path. Same label discipline as the
        # rolling branch: own pairs enter only where the 42d window closed
        # before T (asserted below), foreign pairs are a different fleet whose
        # labels were all known when this submission was built. static_model
        # stays live as the degradation target.
        if HELDOUT_POOL:
            t_build = time.time()
            try:
                own = build_split_table(
                    locations, eol_times, scen_starts, daily, dall, betas,
                    log=log, step_days=HELDOUT_GRID_STEP_DAYS,
                    deadline=t_build + HELDOUT_BUILD_BUDGET_S)
                own['foreign'] = False
                f2 = foreign[np.intersect1d(foreign.columns, own.columns)].copy()
                f2['foreign'] = True
                pool = pd.concat([own, f2], ignore_index=True)
                own_uniq = int(own.loc[own['dies42'] == 1, 'device_id'].nunique())
                log(f'  in-fleet pooling ON: +{len(own)} own pairs, {own_uniq} '
                    f'own deaths vs {static_uniq} foreign '
                    f'({time.time()-t_build:.0f}s, grid {HELDOUT_GRID_STEP_DAYS}d)')
            except Exception as e:
                # degrade to the CURRENT CHAMPION, never to the pickle
                log(f'  WARNING in-fleet pooling skipped ({e}); '
                    f'staying on the static foreign fit')
    else:
        # target split IS (mostly) the train fleet: keep the rolling
        # walk-forward path - its own labels are the answer key and must stay
        # under the T-42d mask
        table = build_split_table(locations, eol_times, scen_starts,
                                  daily, dall, betas, log=log)
        table['foreign'] = False
        pooled = [table]
        if base is not None:
            b = base[np.intersect1d(base.columns, table.columns)].copy()
            b['foreign'] = True
            pooled.append(b)
            log(f'  pooled with train table: +{len(b)} closed-window pairs')
        pool = pd.concat(pooled, ignore_index=True)
        # a train-table row for a device in THIS split's fleet is not foreign -
        # same fleet means same answer key - so it stays under the time mask.
        # Makes the unmasked pooling inert when the target split IS train.
        pool['foreign'] &= ~pool['device_id'].isin(fleet)

    plans = []
    fitted = {}
    for si, (scenario, locs, cut, eol) in enumerate(iterate_scenarios(
            locations, timeseries, eol_times, scenarios)):
        T = pd.Timestamp(scenario['start_time'])

        if deadline is not None and time.time() > deadline:
            # abort into the per-split fallback instead of hitting the hard
            # runtime kill with a partial submission
            raise TimeoutError('split deadline exceeded mid-adaptive')

        # cost_curves prices with module-level penalty constants; the
        # evaluator charges whatever THIS scenario's settings say. Constant
        # 0.5/10.0 across all train scenarios, but wire them through so a
        # held-out scenario with different penalties is priced correctly.
        stg = scenario['settings']
        models_cv.EARLY = float(stg.early_replacement_penalty_daily)
        models_cv.LATE = float(stg.late_replacement_penalty_daily)
        if int(stg.planning_window_days) != WINDOW_DAYS:
            # the 42-day grid is baked into features/labels/quantiles and
            # cannot be rewired at runtime - flag it loudly and plan anyway
            log(f'  WARNING {scenario["name"]}: planning_window_days='
                f'{stg.planning_window_days} != {WINDOW_DAYS}')

        # Mid-loop degradation: if the per-origin refits are eating the split
        # budget, drop to the static foreign fit for the REMAINING origins
        # rather than letting the loop-level deadline raise, which lands in the
        # pickle fallback - far below the static champion.
        if (pool is not None and static_model is not None and deadline is not None
                and time.time() > deadline - 4 * 60):
            log(f'  {scenario["name"]}: refit budget low - degrading to the '
                f'static foreign fit for the remaining origins')
            pool = None

        if pool is None:
            # no in-fleet pairs (pooling off, build overran, or budget low)
            model, train_uniq = static_model, static_uniq
        else:
            # Cadence applies to the HELD-OUT path only (static_model set).
            # The train path keeps its per-T key so the all-48 checksum is
            # untouched. Reusing a bucket's fit at a later origin only ever
            # uses FEWER closed pairs than were available - conservative, never
            # a leak.
            fkey = (si // HELDOUT_REFIT_EVERY) if static_model is not None else T
            if fkey not in fitted:
                # LABEL INTEGRITY: own pairs only where the window closed
                # before T (the split's eol_times is its answer key). Foreign
                # train-fleet pairs are used in full: masking them was CV
                # hygiene misapplied to deployment, starving early origins to
                # 3-20 unique training deaths out of the 82 available (the
                # cold-start mechanism). This is the assertion that keeps the
                # held-out path honest now that it reads the split's own
                # eol_times: every own label used was observable at T.
                closed = pool['origin'] <= T - pd.Timedelta(days=WINDOW_DAYS)
                train = pool[closed | pool['foreign']]
                own_pairs = train[~train['foreign']]
                assert (own_pairs['origin'] + pd.Timedelta(days=WINDOW_DAYS) <= T).all()
                uniq = int(train.loc[train['dies42'] == 1, 'device_id'].nunique())
                fitted[fkey] = (_fit_model(train), uniq)
            model, train_uniq = fitted[fkey]

        feats = features_at_origin(daily, locs, T)
        past = betas[betas['week_end'] <= T]['beta']
        beta = float(past.median()) if len(past) >= 20 else BETA_DEFAULT
        dead = eol_times[eol_times.notna() & (eol_times <= T)]
        curve = master_curve(dall[dall['day'] <= T], dead, beta)
        feats = feats.join(physics_features(dall, T, beta, curve, match=False))
        if ft2_daily is not None:
            try:
                feats = feats.join(wiener_at_origin(ft2_daily, T))
            except Exception as e:
                log(f'  WARNING {scenario["name"]}: ft2 wiener params failed '
                    f'({e}); wiener source dropped this scenario')
        feats = feats.reindex(locs['battery'])
        # plan-origin timestamp for the T3 warmup gate (the cv table carries
        # this column natively; the split pipeline adds it here)
        feats['origin'] = T

        pred = model.predict(feats)
        if static_model is not None and cliff_counts is not None:
            # phase 28: static-path-only appetite boost on hourly-knee rows;
            # the alpha-pairing sweep confirmed ALPHA_STATIC=1.5 stays optimal
            # UNDER the boost (baseline of every holdout comparison was 1.5,
            # so the gain is additive to phase 26, not double-counted)
            n_cliff = cliff_boost(pred, feats, cliff_counts, T)
            if n_cliff:
                log(f'  {scenario["name"]}: cliff boost x{CLIFF_SCALE} on '
                    f'{n_cliff} knee devices')
        if CERT_FIRST and len(cert_days):
            ch = cert_days.reindex(pred.index)
            cert = (ch.notna() & (ch <= T)).to_numpy()
            if cert.any():
                # rank only: stacks on top of whatever rank_score exists
                # (cliff, if enabled); p42 and p42_sum are untouched, so the
                # inclusion count and every timing curve stay byte-identical
                base_rs = (pred['rank_score'] if 'rank_score' in pred.columns
                           else pred['p42']).astype(float).to_numpy().copy()
                base_rs[cert] += 2.0
                pred['rank_score'] = base_rs
                log(f'  {scenario["name"]}: cert-first boost on '
                    f'{int(cert.sum())} confirmed-crossing devices')
        plan_tsroute.alpha = ALPHA_STATIC if static_model is not None else ALPHA
        plan_tsroute.ec_margin = EC_MARGIN
        plan_tsroute.ec_pmax = EC_PMAX
        plan_tsroute.split_days = SPLIT_DAYS
        plan_tsroute.timing_weight = TIMING_WEIGHT
        plan_tsroute.floor_scale = FLOOR_SCALE
        plan_tsroute.recent_deaths = 0.0
        if FLOOR_SCALE > 0 and train_uniq < FLOOR_GATE:
            # both signals are observable at T: recorded EOLs (the harness
            # itself drops eol <= T devices) and the split's own voltage zone
            recent = int(((eol_times > T - pd.Timedelta(days=WINDOW_DAYS))
                          & (eol_times <= T)).sum())
            zone = int(((feats['days_below_245'] >= 0)
                        & (feats['days_below_245'] <= WINDOW_DAYS)
                        & (feats['staleness'] <= 7)).sum())
            plan_tsroute.recent_deaths = float(max(recent, zone))
            log(f'  {scenario["name"]}: cold-start floor active '
                f'(train deaths {train_uniq}, recent {recent}, zone {zone})')
        # NOTE (phase 20): a foreign-fleet floor branch (fire when training
        # positives are mostly another fleet's) was built here and REMOVED
        # the same day: the fleet-holdout harness showed the probability
        # scale transfers across device subsets (half-fleet training:
        # sum(p42) 15.8 vs 9.1 true deaths - if anything it OVER-counts),
        # so the floor could only add over-inclusion (+15, 0 wins).
        plan = plan_tsroute(pred, None, locs, T, 0.05,
                            travel_costs=scenario['travel_costs'],
                            settings=scenario['settings'])
        plan = repark_for_window(plan, T, scenario['settings'], log)
        plan['split'] = split
        plan['scenario'] = scenario['name']
        plans.append(plan)
        log(f'  {scenario["name"]}: planned '
            f'{(plan.day <= T + pd.Timedelta(days=WINDOW_DAYS)).sum()} swaps')
    return plans


def repark_for_window(plan, T, settings, log=print):
    """Move parked rows past the ACTUAL window end.

    plan_tsroute (and the pickled planner) park non-included batteries at
    day 43, relying on the evaluator cutting rows past the window. That
    convention hard-wires 42 into the OUTPUT: with planning_window_days > 42
    the evaluator would EXECUTE every parked row as a real swap and charge
    early penalties on the whole fleet (measured on a 56d evaluator: 23582
    vs 1087 - a 20x blow-up). Inert at W <= 42, where day 43 is already
    outside the window (checksum-verified byte-identical).
    """
    W = int(settings.planning_window_days)
    if W <= WINDOW_DAYS:
        return plan
    parked_day = (T + pd.Timedelta(days=WINDOW_DAYS + 1)).normalize()
    mask = plan['day'] == parked_day
    if not mask.any():
        return plan
    log(f'  window={W}d > {WINDOW_DAYS}d: re-parking {int(mask.sum())} rows')
    plan = plan.copy()
    plan.loc[mask, 'day'] = (T + pd.Timedelta(days=W + 1)).normalize()
    return plan.sort_values('day', kind='stable').reset_index(drop=True)


def fallback_split_plans(split, split_path, log=print):
    log(f'  FALLBACK to static pickle for split={split}')
    with open(Path(__file__).parent / 'planners' / 'best.pickle', 'rb') as f:
        planner = pickle.load(f)
    locations, timeseries, eol_times, scenarios = load_best_effort(split_path, log)
    plans = []
    for scenario, locs, cut, eol in iterate_scenarios(
            locations, timeseries, eol_times, scenarios):
        plan = planner.plan(cut, locs, scenario['travel_costs'], scenario['settings'])
        plan = repark_for_window(plan, pd.Timestamp(scenario['start_time']),
                                 scenario['settings'], log)
        plan['split'] = split
        plan['scenario'] = scenario['name']
        plans.append(plan)
    return plans


def last_resort_split_plans(split, split_path, log=print):
    """Defer-everything plans from raw files only (no load_dataset, no
    model, no pickle): every battery alive at the scenario origin is placed
    one day past the window, which the evaluator cuts - misses become
    emergencies, but the split scores finitely and validly."""
    import json
    log(f'  LAST RESORT (defer plans) for split={split}')
    devices = pd.read_csv(split_path / 'devices.csv')
    eol = pd.read_csv(split_path / 'eol_times.csv')
    eol['end_time'] = pd.to_datetime(eol['end_time'])
    eol = eol.set_index('device_id')['end_time']
    scenarios = json.load(open(split_path / 'scenarios.json'))
    plans = []
    for s in scenarios:
        T = pd.Timestamp(s['start_time'])
        e = eol.reindex(devices['device_id'])
        alive = devices['device_id'][e.isna().values | (e > T).values]
        W = max(int(s['settings']['planning_window_days'])
                if isinstance(s.get('settings'), dict) else WINDOW_DAYS,
                WINDOW_DAYS)
        plans.append(pd.DataFrame({
            'day': (T + pd.Timedelta(days=W + 1)).normalize(),
            'battery': alive.values,
            'split': split,
            'scenario': s['name'],
        }))
    return plans


def main():
    dataset_path = Path(os.environ.get('BATTERYSWAP_DATASET_PATH', '/tmp/data'))
    splits = os.environ.get('BATTERYSWAP_SPLITS', 'public,private').split(',')
    started = time.time()

    # leave headroom inside the 30-minute runtime limit: if the adaptive path has
    # eaten the budget, remaining splits use the fast static-pickle fallback
    time_budget = float(os.environ.get('BATTERYSWAP_TIME_BUDGET_S', 20 * 60))

    all_plans = []
    for split in splits:
        log = lambda msg: print(f'[{time.time()-started:6.0f}s] {msg}', flush=True)
        log(f'split={split}')
        try:
            if time.time() - started > time_budget:
                raise TimeoutError(f'over time budget ({time_budget:.0f}s)')
            deadline = started + time_budget + 3 * 60  # hard stop for this split
            all_plans.extend(adaptive_split_plans(split, dataset_path / split,
                                                  log=log, deadline=deadline))
        except Exception:
            traceback.print_exc()
            try:
                all_plans.extend(
                    fallback_split_plans(split, dataset_path / split, log=log))
            except Exception:
                traceback.print_exc()
                try:
                    all_plans.extend(
                        last_resort_split_plans(split, dataset_path / split, log=log))
                except Exception:
                    # partial submission beats no submission: never let one
                    # split's failure destroy the other splits' plans
                    traceback.print_exc()

    submission = pd.concat(all_plans)[['day', 'battery', 'split', 'scenario']]
    submission.to_csv('submission.csv', index=False)
    print(f'submission.csv written: {len(submission)} rows '
          f'in {time.time()-started:.0f}s')
    assert os.path.exists('submission.csv')


if __name__ == '__main__':
    main()