Thibaut Forest
Revert "HEAD: order re-weight .50/.20/.30 for the alpha-1.4 band (user call, public A/B) - bgkf 1476.0; revert FT2_W .30 / FT2_GAPW .20"
435304e | """ | |
| 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() | |