"""THE CV (user directive 2026-08-22 v5): building-GroupKFold, deployment-faithful.

ONE protocol lives here, deliberately. Buildings are assigned to K folds,
stratified on battery-failure count and location (travel centrality). For
each fold: ONE fit on the TRAIN buildings' full closed-pair history (that is
exactly what the deployed held-out path does - a foreign fleet's labels were
all known when the submission was built, so "before and after" is
deployment-legal, not a leak), then for each of the 48 scenario times the
TEST buildings are planned as a STANDALONE fleet with the DEPLOYMENT-EXACT
planner state (submit.py's static-path knobs, including the cold-start
floor gate, which is structurally inert here because every fold trains on
60+ unique deaths vs the gate's 20) and costed by the real evaluator.
THE SCORE: per scenario, the fold costs are SUMMED - every building sits
in exactly one test fold, so the sum is the out-of-fold cost of ALL 24
buildings for that scenario - and THE SCORE is the mean of those 48
all-building totals. Same information as the fold-mean average (x5), but in
the familiar full-fleet-per-scenario magnitude.

Fidelity pass (v5): this harness is the closest possible local twin of the
held-out deployment path - both public and private splits are ~2x unseen
fleets served by that path (Docker dress rehearsal, phase 27). Coherence
with the public board is CHECKED against recorded public deltas, never
fitted (a calibrated pricing layer was explicitly rejected). The in-fold
purge (v4) was dropped: it made the CV stricter than the deployment it
represents and bought no public coherence (identical sign flips).

Why this protocol (2026-08-22, the day the temporal purge fell):

* It simulates the HELD-OUT deployment path: the competition's private split
  contains fleets not in the packaged train table (submit.py's static
  foreign-fit branch exists for exactly this). Train-on-some-buildings /
  plan-the-others-as-a-fleet is that path, measured.

* The test group is planned as a COMPLETE fleet, so there is no per-building
  cost-attribution problem (the old LOBO objection) and no partial-fleet
  count bias (phase 26): the count rule operates on the test fleet's own
  sum(p42), exactly as deployment does on a held-out fleet.

* Instrument history: the pure temporal purge (fleet-wide) fell to three
  public refutations in one day (phase 56: count-coupling, label-process
  ghosts, era-calibration leak - its fits saw the future of the same fleet).
  The pure building split (v3) fell the same evening: it still sign-flipped
  all three public killers (phase 57 alignment check) because its fits saw
  the scenario's own era through other buildings. v4 removes both channels
  structurally. What NO local split can price (public-measured, 4x): the
  public fleet's uncatchable tail and its train-absent ghost devices -
  count-coupled and label-process candidates stay report-only regardless of
  what this harness says.

* Diagnostics are PLAN-DERIVED only (cost blocks + in-window swap count), so
  ANY method that yields a plan can be scored. Candidates must win at ~flat
  swaps; count-coupled wins are untrustworthy (public evidence, 4x).

Superseded protocols live in git: the temporal-purge harness (sole validator
2026-08-22, retired same day) at f5f3fda; the earlier walk-forward/static/
screen/LOBO family at 2000281.

Usage
    venv/bin/python -m batteryswap_example.cv bgkf          # THE SCORE
    venv/bin/python -m batteryswap_example.cv cand <label>
    venv/bin/python -m batteryswap_example.cv report
"""
import sys
import time
from pathlib import Path

import numpy as np
import pandas as pd

from batteryswap_public.evaluate import evaluate_plan

from batteryswap_example import models_cv as mc

WINDOW_DAYS = int(mc.WINDOW_DAYS)
# pinned at import: variants build off this, never off the possibly
# already-mutated mc.FEATURES (a second seed used to append twice)
BASE_FEATURES = list(mc.FEATURES)

# The deployed champion (submit.py default after phase 56 reverted the
# frontier): plain logbag / tsroute alpha 0.8 + break-even top-up, censor-
# aware pricing. Frontier dials exist in submit.py for explicit A/B only.
CHAMPION = dict(model='logbag', planner='tsroute', p_star=0.8, day_q=0.05,
                ec_margin=0.8, ec_pmax=25.0, split_days=7,
                floor_scale=2.5, floor_gate=20)   # submit.py FLOOR_* defaults

N_FOLDS = 5

BLOCKS = (('warmup  s_0-7', lambda o: o <= 7),
          ('panel   s_8-41', lambda o: (o >= 8) & (o <= 41)),
          ('nearend s_42-47', lambda o: o >= 42),
          ('ALL', lambda o: o >= 0))


def load_table(dataset_path=Path('dataset')):
    """The one table build. Every run starts here so a variant can never
    silently differ from the champion by a feature-prep line."""
    t = mc.load_features(dataset_path)
    t['v_proj'] = t['v_last'] + t['slope_lastk'] * t['staleness']
    # 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)
    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


def build_folds(t, ctx, dataset_path=Path('dataset'), k=N_FOLDS):
    """Deterministic stratified building->fold assignment.

    Stratified on (1) battery-failure count - unique dying devices per
    building - and (2) location, proxied by travel centrality (mean hours to
    every other building). Buildings are sorted by (deaths desc, centrality,
    id) and snake-assigned to k folds, which balances the lumpy death counts
    (a test fold with ~0 deaths measures nothing) while spreading central
    and remote buildings across folds. No randomness: reruns are identical.
    """
    dev_b = (pd.read_csv(dataset_path / 'train' / 'devices.csv')
             .drop_duplicates('device_id').set_index('device_id')['building_id'])
    b = t['device_id'].map(dev_b)
    deaths = (t.loc[t['dies42'] == 1].groupby(b[t['dies42'] == 1])['device_id']
              .nunique())
    tc = ctx[0]['travel_costs']
    central = tc[tc['from'] != tc['to']].groupby('from')['hours'].mean()
    rows = pd.DataFrame({'deaths': deaths, 'central': central}).fillna(0)
    rows = rows.sort_values(['deaths', 'central'],
                            ascending=[False, True]).reset_index()
    order = list(range(k)) + list(range(k - 1, -1, -1))     # snake
    rows['fold'] = [order[i % (2 * k)] for i in range(len(rows))]
    return rows.set_index('index')['fold'], dev_b


# --------------------------------------------------------------------------
# feature variants (candidate re-tests run them through the one harness)
# --------------------------------------------------------------------------

def _base(t):
    return BASE_FEATURES


def _ranks_add(t):
    # phase-56 rank features. NOTE: under this harness the prediction-time
    # rank is over the TEST fleet only (a few buildings), which is also what
    # a held-out deployment would compute.
    return BASE_FEATURES + ['r_vlast', 'r_tau']


def _vrel(t):
    return BASE_FEATURES + [c for c in ('vrel', 'vmin7_rank') if c in t.columns]


def _vm7(t):
    # the ADOPTED feature set (campaign 2, phase 59): champion features +
    # fleet pct-rank of the 7d-min voltage
    return BASE_FEATURES + ['vmin7_rank']


VARIANTS = {'base': _base, 'ranks_add': _ranks_add, 'vrel': _vrel,
            'vm7': _vm7}


# --------------------------------------------------------------------------
# THE protocol: building-GroupKFold, test group planned as a standalone fleet
# --------------------------------------------------------------------------

def run_bgkf(dataset_path=Path('dataset'), kind=None, variant='base',
             blend=None, n_folds=N_FOLDS, origins=range(48), p_star=None,
             split_days=None, timing_weight=None):
    """THE SCORE. Per fold: one fit on the TRAIN buildings' full closed-pair
    history (deployment-exact for a held-out fleet), then per scenario the
    TEST buildings are planned as a standalone fleet under submit.py's
    static-path planner state (incl. the cold-start floor gate) and costed.
    Candidate hooks: `kind` swaps the model family (mc.MODELS key or
    SklearnRUL spec), `variant` the feature set, `blend`=(kind2, w2) a
    value-preserving rank blend (order changes, the primary's p42 values
    and hence the count machinery do not).
    """
    kind = kind or CHAMPION['model']
    t = load_table(dataset_path)
    mc.FEATURES = list(VARIANTS[variant](t))
    ctx = mc.load_context(dataset_path)
    W = pd.Timedelta(days=WINDOW_DAYS)
    closed = t[t['origin'] <= t['origin'].max() - W]
    folds, dev_b = build_folds(t, ctx, dataset_path, k=n_folds)
    t_bldg = t['device_id'].map(dev_b)

    # deployment-exact planner state: exactly the attributes submit.py's
    # static path sets, everything else at models_cv module defaults
    # (p_be=0, floor_day=0, p_conf=0.5 - deployment does not touch them)
    p = mc.plan_tsroute
    p.alpha = CHAMPION['p_star'] if p_star is None else p_star
    p.ec_margin, p.ec_pmax = CHAMPION['ec_margin'], CHAMPION['ec_pmax']
    p.split_days = CHAMPION['split_days'] if split_days is None else split_days
    p.timing_weight = 1.0 if timing_weight is None else timing_weight
    p.floor_scale, p.recent_deaths = CHAMPION['floor_scale'], 0.0

    def _fit(k_, train):
        hazard = k_ in ('aft', 'dthaz', 'dthaz_aux', 'logbag_open',
                        'blend_dh', 'blend_dhaux')
        if hazard:
            pool = mc.censored_pairs(t, t['origin'].max())
            keep = set(train['device_id'])
            train = pool[pool['device_id'].isin(keep)]
        make = ((lambda: mc.MODELS[k_]()) if k_ in mc.MODELS
                else (lambda: mc.SklearnRUL(k_)))
        return make().fit(train)

    recs = []
    for f in range(n_folds):
        test_b = set(folds.index[folds == f])
        train = closed[~t_bldg.loc[closed.index].isin(test_b)]
        model = _fit(kind, train)
        partner = _fit(blend[0], train) if blend is not None else None
        train_uniq = int(train.loc[train['dies42'] == 1, 'device_id'].nunique())
        for idx in origins:
            c = ctx[idx]
            T = c['start_time']
            fleet = c['locs'][c['locs']['building'].isin(test_b)]
            rows = (t[t['origin_idx'] == idx].set_index('device_id')
                    .reindex(fleet['battery']).dropna(subset=['v_last']))
            if rows.empty:
                continue
            locs = fleet[fleet['battery'].isin(rows.index)]
            eol = c['eol'].reindex(rows.index)
            pred = model.predict(rows)
            if partner is not None:
                w2 = blend[1]
                r = ((1 - w2) * pred['p42'].rank(pct=True)
                     + w2 * partner.predict(rows)['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)
            # cold-start floor, gated exactly as submit.py: observable
            # recent deaths / voltage-zone census, armed only when the fit
            # saw < floor_gate unique deaths (inert at fold sizes >= 60)
            p.recent_deaths = 0.0
            if CHAMPION['floor_scale'] > 0 and train_uniq < CHAMPION['floor_gate']:
                recent = int(((eol > T - W) & (eol <= T)).sum())
                zone = int(((rows['days_below_245'] >= 0)
                            & (rows['days_below_245'] <= WINDOW_DAYS)
                            & (rows['staleness'] <= 7)).sum())
                p.recent_deaths = float(max(recent, zone))
            plan = mc.plan_tsroute(pred, None, locs, T, CHAMPION['day_q'],
                                   travel_costs=c['travel_costs'],
                                   settings=c['settings'])
            _, _, overall = evaluate_plan(plan, locs, c['travel_costs'],
                                          c['settings'], eol_times=eol,
                                          start_time=T, verbose=0)
            in_win = plan.loc[plan['day'] <= T + W, 'battery']
            recs.append(dict(overall, n_swaps=int(in_win.nunique()),
                             origin=idx, fold=f, arm='bgkf'))
            print(f"f{f} s_{idx}: {recs[-1]['total_cost']:7.1f}",
                  flush=True, file=sys.stderr)
    return pd.DataFrame(recs)


def report(df):
    """Fold table + all-building scenario scorecard + THE SCORE."""
    print('\n=== building-GroupKFold / test groups planned as fleets / '
          'folds summed per scenario ===')
    print(f"{'fold':>4s} {'n':>3s} {'total':>8s} {'late':>7s} {'early':>6s} "
          f"{'swaps':>5s}")
    for f, d in df.groupby('fold'):
        print(f"{f:4d} {len(d):3d} {d['total_cost'].mean():8.1f} "
              f"{d['late_swap'].mean():7.1f} {d['early_swap'].mean():6.1f} "
              f"{d['n_swaps'].mean():5.1f}")
    # per-scenario all-building totals: sum across folds (each building is
    # out-of-fold exactly once per scenario)
    g = df.groupby('origin')[['total_cost', 'late_swap', 'early_swap',
                              'n_swaps']].sum()
    print(f"\n{'block':16s} {'n':>3s} {'total':>8s} {'late':>7s} "
          f"{'early':>6s} {'swaps':>5s}   (all buildings, per scenario)")
    for name, sel in BLOCKS:
        d = g[sel(g.index)]
        if d.empty:
            continue
        print(f"{name:16s} {len(d):3d} {d['total_cost'].mean():8.1f} "
              f"{d['late_swap'].mean():7.1f} {d['early_swap'].mean():6.1f} "
              f"{d['n_swaps'].mean():5.1f}")
    tc = g['total_cost']
    se = tc.std(ddof=1) / np.sqrt(len(tc))
    print(f"\nSCORE (all-building out-of-fold cost, mean over the {len(tc)} "
          f"scenarios): {tc.mean():.1f} ± {se:.1f} (SE; 95% CI "
          f"{tc.mean() - 1.96 * se:.1f}..{tc.mean() + 1.96 * se:.1f})")
    print("      NOTE: this SE is scenario-difficulty scatter, shared by "
          "every candidate.\n      For comparisons use the PAIRED deltas "
          "(`cv report`), whose SE is far smaller.")


# --------------------------------------------------------------------------
# candidate sweep on the one harness
# --------------------------------------------------------------------------

CANDIDATES = {
    'logistic': dict(kind='logistic'),
    'rf': dict(kind='rf'),
    'gbm': dict(kind='gbm'),
    'dthaz': dict(kind='dthaz'),
    'logbag_rank': dict(kind='logbag_rank'),
    'ranks_add': dict(variant='ranks_add'),
    'vrel': dict(variant='vrel'),
    'rankavg25rf': dict(blend=('rf', 0.25)),
    'rankavg25gbm': dict(blend=('gbm', 0.25)),
    'frontier': dict(kind='logbag_rank', blend=('gbm', 0.30)),
    # ft2 = ft_vmin7r_wnr20, ADOPTED at 3f9bf12 (submit.py Ft2RUL default);
    # this cv-side recipe is bit-identical to running the deployed class
    # (parity check, phase 59): SCORE 1688.4, -71.8 +/- 23.5 vs baseline
    'ft2': dict(variant='vm7', blend=('wiener', 0.20)),
    # the alpha-1.0 A/B (head 401f171): ft2 with planner appetite 1.0.
    # Count-coupled vs ft2 (17W/24L on the increment) - public arbitrates.
    # ft2a10 / head = THE CURRENT MODEL: runs the literal deployed class
    # (submit.Ft2RUL - lazily registered in main()), so these labels always
    # score exactly what a public trigger would run. Older forms kept below.
    # 2026-08-23 night: head alpha -> 1.4 (62e9d91, public 1480.85 vs 1575.31
    # at alpha 1.0; bgkf 1498.3 vs 1541.3). Old head kept as ft2a10_a10.
    'ft2a10': dict(kind='@deployed', variant='base', p_star=1.4,
                   split_days=3, timing_weight=2.0),
    'ft2a10_a10': dict(kind='@deployed', variant='base', p_star=1.0,
                       split_days=3, timing_weight=2.0),
    'ft2a10_w20': dict(variant='vm7', blend=('wiener', 0.20), p_star=1.0),
    'ft2a10_w25': dict(variant='vm7', blend=('wiener', 0.25), p_star=1.0),
    'head': dict(kind='@deployed', variant='base', p_star=1.4,
                 split_days=3, timing_weight=2.0),
    # pre-T2 plan path (split 7, weight 1) kept for paired A/B:
    'ft2a10_sp7': dict(kind='@deployed', variant='base', p_star=1.0),
}


def candidate_report(base_csv='/tmp/cv_bgkf.csv',
                     pattern='/tmp/cv_bgkfcand_*.csv'):
    import glob
    base = pd.read_csv(base_csv)
    b = base.groupby('origin')[['total_cost', 'n_swaps']].sum()
    print('paired vs champion (logbag/base, building-GroupKFold; folds '
          'summed per scenario -> 48 paired all-building totals).')
    print('swaps (plan-derived) shown because a cost delta can be appetite '
          'in disguise: trust a win only if swaps is ~flat, and check the '
          'PER-SCENARIO count swing, not just the mean (phase 56 lesson).\n')
    print(f"{'candidate':14s} {'SCORE':>7s} {'delta':>7s} {'SE':>6s} "
          f"{'W/L':>7s} {'swaps':>5s}")
    print(f"{'base':14s} {b['total_cost'].mean():7.1f} {'-':>7s} {'-':>6s} "
          f"{'-':>7s} {b['n_swaps'].mean():5.1f}")
    out = []
    for fn in sorted(glob.glob(pattern)):
        d = (pd.read_csv(fn).groupby('origin')
             [['total_cost', 'n_swaps']].sum())
        label = pd.read_csv(fn)['cand'].iloc[0]
        delta = (d['total_cost'] - b['total_cost']).dropna()
        out.append((delta.mean(), label, d, delta))
    for _, label, d, delta in sorted(out):
        se = delta.std() / np.sqrt(len(delta))
        print(f"{label:14s} {d['total_cost'].mean():7.1f} "
              f"{delta.mean():+7.1f} {se:6.1f} "
              f"{int((delta < 0).sum()):3d}W/{int((delta > 0).sum()):2d}L "
              f"{d['n_swaps'].mean():5.1f}")


def main():
    proto = sys.argv[1] if len(sys.argv) > 1 else 'bgkf'
    t0 = time.time()
    if proto == 'bgkf':
        df = run_bgkf()
        df.to_csv('/tmp/cv_bgkf.csv', index=False)
        report(df)
    elif proto == 'cand':
        label = sys.argv[2]
        spec = dict(CANDIDATES[label])
        if spec.get('kind') == '@deployed':
            from batteryswap_example import submit as _sub
            mc.MODELS['@deployed'] = _sub.Ft2RUL
        df = run_bgkf(**spec)
        df['cand'] = label
        df.to_csv(f'/tmp/cv_bgkfcand_{label}.csv', index=False)
        report(df)
    elif proto == 'report':
        candidate_report()
    else:
        raise SystemExit(
            f'unknown protocol {proto!r} - the building-GroupKFold x purge '
            f'harness is the only CV (user directive 2026-08-22 v4). The '
            f'fleet-wide purge lives in git at f5f3fda; the pure building '
            f'split at 4be2283; the walk-forward family at 2000281.')
    print(f'\n[{time.time() - t0:.0f}s]')


if __name__ == '__main__':
    main()