all48oof defaults to the oof arm only (THE SCORE); 'cv all48oof self' re-enables the validity-check arm and the flattery readout

batteryswap_example/cv.py

@@ -86,6 +86,24 @@ def load_table(dataset_path=Path('dataset')):
     # within-origin fleet ranks: leak-free (data <= origin only)
     t['r_vlast'] = t.groupby('origin_idx')['v_last'].rank(pct=True)
     t['r_tau'] = t.groupby('origin_idx')['tau_projected'].rank(pct=True)
+    # aux usage-dynamics and Wiener-FHT columns (same merges the old
+    # harness __main__ did): cached parquets, data <= origin only; inert
+    # unless a variant/model asks for the columns
+    try:
+        from batteryswap_example.aux_features import load as load_aux
+        aux = load_aux(dataset_path)
+        dup = [c for c in aux.columns if c in t.columns
+               and c not in ('device_id', 'origin_idx')]
+        t = t.merge(aux.drop(columns=dup),
+                    on=['device_id', 'origin_idx'], how='left')
+    except FileNotFoundError:
+        pass
+    try:
+        from batteryswap_example.fht_features import load as load_fht
+        t = t.merge(load_fht(dataset_path),
+                    on=['device_id', 'origin_idx'], how='left')
+    except FileNotFoundError:
+        pass
     return t
 
 
@@ -383,9 +401,35 @@ def _blind3(t):
     return ['v_last', 'v_min_7', 'n_30']
 
 
+def _ranks(t):
+    # phase-8 logbag_rank, STRICT form: the within-origin rank transforms
+    # REPLACE the two level features (the registry's logbag_rank ADDS them).
+    # Ranks are the cross-fleet transfer device; the original lost the old
+    # scenario CV by less than that harness's label-luck floor.
+    f = [c for c in BASE_FEATURES if c not in ('v_last', 'tau_projected')]
+    return f + ['r_vlast', 'r_tau']
+
+
+def _vrel_bldg(t):
+    # The untested arm of the vrel family: BUILDING-relative voltage. The
+    # phase-44 refutation covers only a fleet-UNIFORM offset (cannot reorder
+    # a linear model); a per-building offset does reorder across buildings,
+    # which is exactly what the building-OOF fit can no longer absorb.
+    # Deployment-legal: computed from the target fleet's own telemetry.
+    if 'vrel_bldg' not in t.columns:
+        dev_b = (pd.read_csv(Path('dataset') / 'train' / 'devices.csv')
+                 .drop_duplicates('device_id')
+                 .set_index('device_id')['building_id'])
+        b = t['device_id'].map(dev_b)
+        t['vrel_bldg'] = (t['v_last'] - t.groupby([t['origin_idx'], b])
+                          ['v_last'].transform('median'))
+    return BASE_FEATURES + ['vrel_bldg']
+
+
 VARIANTS = {'base': _base, 'zone': _zone, 'vcorr': _vcorr, 'vrel': _vrel,
             'vrel_only': _vrel_only, 'vmin7_only': _vmin7_only,
-            'v_last_only': _v_last_only, 'blind4': _blind4, 'blind3': _blind3}
+            'v_last_only': _v_last_only, 'blind4': _blind4, 'blind3': _blind3,
+            'ranks': _ranks, 'vrel_bldg': _vrel_bldg}
 
 ABLATION = ('base', 'vrel_only', 'vmin7_only', 'vrel', 'v_last_only')
 
@@ -565,7 +609,8 @@ def ablation_report(df, ns, label, ref='base'):
 # --------------------------------------------------------------------------
 
 def run_all48_oof(n_folds=5, seed=0, origins=range(48),
-                  dataset_path=Path('dataset')):
+                  dataset_path=Path('dataset'), kind=None, variant='base',
+                  arms=('self', 'oof'), blend=None):
     """The walk-forward cost harness with BUILDING-CLEAN predictions.
 
     all48 itself is temporal, not fold-based, and its PLANNING must stay
@@ -582,9 +627,19 @@ def run_all48_oof(n_folds=5, seed=0, origins=range(48),
     the cold-start floor off, so the paired per-origin delta isolates the
     self-fleet fit effect. Levels may differ slightly from canonical all48
     (floor off, no warm-start state); read the DELTA, not the level.
+
+    Candidate re-tests (phase 52 addendum sweep): `kind` swaps the model
+    family, `variant` swaps the feature set (VARIANTS key), `blend`=(kind2,
+    w2) rank-averages a second family in at weight w2 and then maps the
+    blended ORDER back onto the primary family's own p42 values - value-
+    preserving, so the count/pricing machinery sees the identical p42
+    distribution and only WHO holds each rank changes (the phase-51 lesson:
+    a post-hoc blend whose beliefs disagree with the count machinery fails
+    for that reason alone).
     """
+    kind = kind or CHAMPION['model']
     t = load_table(dataset_path)
-    mc.FEATURES = list(BASE_FEATURES)
+    mc.FEATURES = list(VARIANTS[variant](t))
     ctx = mc.load_context(dataset_path)
     dev_bldg = (pd.read_csv(dataset_path / 'train' / 'devices.csv')
                 .drop_duplicates('device_id')
@@ -613,15 +668,40 @@ def run_all48_oof(n_folds=5, seed=0, origins=range(48),
         eol = c['eol'].reindex(rows.index)
         tr_fold = tr['device_id'].map(dev_bldg).map(bldg_fold)
         row_fold = rows.index.to_series().map(dev_bldg).map(bldg_fold)
-        for arm in ('self', 'oof'):
+
+        def _pred(k, arm):
+            if k == 'tau':
+                # fit-free pseudo-model: the physics timer as a rank source
+                # (only meaningful inside a value-preserving blend)
+                return pd.DataFrame(
+                    {'p42': -rows['tau_projected'].astype(float).fillna(np.inf)},
+                    index=rows.index)
+            hazard = k in ('aft', 'dthaz', 'dthaz_aux', 'logbag_open',
+                           'blend_dh', 'blend_dhaux')
+            trk = mc.censored_pairs(t, T) if hazard else tr
+            fold_k = (trk['device_id'].map(dev_bldg).map(bldg_fold)
+                      if hazard else tr_fold)
+            make = ((lambda: mc.MODELS[k]()) if k in mc.MODELS
+                    else (lambda: mc.SklearnRUL(k)))
             if arm == 'self':
-                pred = mc.SklearnRUL(CHAMPION['model']).fit(tr).predict(rows)
-            else:
-                parts = []
-                for f in sorted(row_fold.dropna().unique()):
-                    m = mc.SklearnRUL(CHAMPION['model']).fit(tr[tr_fold != f])
-                    parts.append(m.predict(rows[(row_fold == f).values]))
-                pred = pd.concat(parts).reindex(rows.index)
+                return make().fit(trk).predict(rows)
+            parts = []
+            for f in sorted(row_fold.dropna().unique()):
+                m = make().fit(trk[fold_k != f])
+                parts.append(m.predict(rows[(row_fold == f).values]))
+            return pd.concat(parts).reindex(rows.index)
+
+        for arm in arms:
+            pred = _pred(kind, arm)
+            if blend is not None:
+                k2, w2 = blend
+                r = ((1 - w2) * pred['p42'].rank(pct=True)
+                     + w2 * _pred(k2, arm)['p42'].rank(pct=True))
+                vals = np.sort(pred['p42'].values)[::-1]
+                pred = pred.copy()
+                pred['p42'] = pd.Series(
+                    vals, index=r.sort_values(ascending=False).index
+                ).reindex(pred.index)
             plan = mc.plan_tsroute(pred, None, locs, T, CHAMPION['day_q'],
                                    travel_costs=c['travel_costs'],
                                    settings=c['settings'])
@@ -630,16 +710,90 @@ def run_all48_oof(n_folds=5, seed=0, origins=range(48),
                                           start_time=T, verbose=0)
             rs = mc.rank_stats(pred, plan, eol, T)
             recs.append(dict(overall, **rs, origin=idx, arm=arm, seed=seed))
-        a, b = recs[-2], recs[-1]
-        print(f"s_{idx}: self {a['total_cost']:7.1f}  oof {b['total_cost']:7.1f} "
-              f"delta {b['total_cost'] - a['total_cost']:+7.1f}",
-              flush=True, file=sys.stderr)
+        if len(arms) == 2:
+            a, b = recs[-2], recs[-1]
+            print(f"s_{idx}: self {a['total_cost']:7.1f}  oof {b['total_cost']:7.1f} "
+                  f"delta {b['total_cost'] - a['total_cost']:+7.1f}",
+                  flush=True, file=sys.stderr)
+        else:
+            print(f"s_{idx}: {arms[0]} {recs[-1]['total_cost']:7.1f}",
+                  flush=True, file=sys.stderr)
     return pd.DataFrame(recs)
 
 
+# The re-test grid for the phase 52 addendum sweep: every prediction-side
+# candidate the campaign rejected on a self-fit or scenario-grouped surface,
+# re-asked on the building-OOF cost harness. Planner/appetite candidates are
+# deliberately absent - that axis is closed and this harness would not
+# change its answer.
+OOF_CANDIDATES = {
+    'logistic': dict(kind='logistic'),
+    'rf': dict(kind='rf'),
+    'gbm': dict(kind='gbm'),
+    'zone': dict(variant='zone'),
+    'vcorr': dict(variant='vcorr'),
+    'vrel': dict(variant='vrel'),
+    'vrel_only': dict(variant='vrel_only'),
+    'vmin7_only': dict(variant='vmin7_only'),
+    'v_last_only': dict(variant='v_last_only'),
+    'blind4': dict(variant='blind4'),
+    'blind3': dict(variant='blind3'),
+    'rankavg25rf': dict(blend=('rf', 0.25)),
+    'rankavg25gbm': dict(blend=('gbm', 0.25)),
+    # wave 2 - the A-list from the phase-52-addendum re-adjudication:
+    # candidates whose rejection came from a self-fit surface while a
+    # building-split instrument disagreed or was never consulted
+    'logbagC03': dict(kind='logbag_c03'),
+    'logbagC01': dict(kind='logbag_c01'),
+    'logbag_rank': dict(kind='logbag_rank'),
+    'ranks': dict(variant='ranks'),
+    'vrel_bldg': dict(variant='vrel_bldg'),
+    'logbag_aux': dict(kind='logbag_aux'),
+    'aux_ucb05': dict(kind='aux_ucb05'),
+    'aux_ucb10': dict(kind='aux_ucb10'),
+    'dthaz': dict(kind='dthaz'),
+    'rankavg25tau': dict(blend=('tau', 0.25)),
+    'rankavg25fht': dict(blend=('wiener', 0.25)),
+}
+
+
+def oof_candidate_report(base_csv='/tmp/cv_all48oof.csv',
+                         pattern='/tmp/cv_oofcand_*.csv'):
+    import glob
+    base = pd.read_csv(base_csv)
+    b = base[base['arm'] == 'oof'].set_index('origin')
+    print('paired vs champion oof arm (logbag/base, floor off, fold seed 0).')
+    print('sched/rec12 shown because a cost delta can be appetite in '
+          'disguise (phase 39/51): trust a win only if sched is ~flat.\n')
+    hdr = (f"{'candidate':14s} {'ALL48':>7s} {'delta':>7s} {'W/L':>7s} "
+           f"{'panel':>7s} {'pdelta':>7s} {'pW/L':>7s} {'sched':>5s} {'rec12':>6s}")
+    print(hdr)
+    print(f"{'base(oof)':14s} {b['total_cost'].mean():7.1f} {'-':>7s} {'-':>7s} "
+          f"{b.loc[(b.index >= 8) & (b.index <= 41), 'total_cost'].mean():7.1f} "
+          f"{'-':>7s} {'-':>7s} {b['n_sched'].mean():5.1f} {b['rec12'].mean():6.3f}")
+    out = []
+    for fn in sorted(glob.glob(pattern)):
+        d = pd.read_csv(fn).set_index('origin')
+        label = d['cand'].iloc[0]
+        delta = (d['total_cost'] - b['total_cost']).dropna()
+        pan = delta[(delta.index >= 8) & (delta.index <= 41)]
+        out.append((delta.mean(), label, d, delta, pan))
+    for _, label, d, delta, pan in sorted(out):
+        print(f"{label:14s} {d['total_cost'].mean():7.1f} {delta.mean():+7.1f} "
+              f"{int((delta < 0).sum()):3d}W/{int((delta > 0).sum()):2d}L "
+              f"{d.loc[(d.index >= 8) & (d.index <= 41), 'total_cost'].mean():7.1f} "
+              f"{pan.mean():+7.1f} {int((pan < 0).sum()):3d}W/{int((pan > 0).sum()):2d}L "
+              f"{d['n_sched'].mean():5.1f} {d['rec12'].mean():6.3f}")
+
+
 def oof_report(df):
-    for arm in ('self', 'oof'):
-        scorecard(df[df['arm'] == arm], f'all48oof / arm={arm} / floor off')
+    arms = [a for a in ('oof', 'self') if (df['arm'] == a).any()]
+    for arm in arms:
+        label = 'THE SCORE' if arm == 'oof' else 'validity-check arm'
+        scorecard(df[df['arm'] == arm], f'all48oof / arm={arm} / {label} / floor off')
+    if 'self' not in arms:
+        return
+    # the paired flattery term - only shown when the self arm was requested
     s = df[df['arm'] == 'self'].set_index('origin')['total_cost']
     o = df[df['arm'] == 'oof'].set_index('origin')['total_cost']
     d = (o - s).dropna()
@@ -681,9 +835,21 @@ def main():
     elif proto == 'screen':
         screen()
     elif proto == 'all48oof':
-        df = run_all48_oof()
+        # THE SCORE (user directive 2026-08-22): oof arm only by default.
+        # `cv all48oof self` additionally runs the self arm - the validity
+        # check (must reproduce panel 1679.2) and the flattery-term readout.
+        arms = ('self', 'oof') if variant == 'self' else ('oof',)
+        df = run_all48_oof(arms=arms)
         df.to_csv('/tmp/cv_all48oof.csv', index=False)
         oof_report(df)
+    elif proto == 'oofcand':
+        label = sys.argv[2]
+        df = run_all48_oof(arms=('oof',), **OOF_CANDIDATES[label])
+        df['cand'] = label
+        df.to_csv(f'/tmp/cv_oofcand_{label}.csv', index=False)
+        scorecard(df, f'all48oof / cand={label} / oof arm / floor off')
+    elif proto == 'oofreport':
+        oof_candidate_report()
     elif proto == 'ablate':
         which = sys.argv[2] if len(sys.argv) > 2 else 'all48'
         df, ns = ablate(protocol=which)