Minimise documentation and remove unused build scripts

README.md

@@ -10,60 +10,28 @@ tags:
 
 # BatterySwapAI 2026 — MnesisLab
 
-Causal battery-swap planning. Current submission **011**: hierarchical Wiener
-first-passage reranking plus operational capacity policy, over the byte-preserved
-public V05 base.
+Causal battery-swap planning: hierarchical Wiener first-passage reranking over a
+degradation model, with a cost-aware capacity and routing policy.
 
-## Deployed artifact
+## Artifact
 
-`submission_artifacts/workload_planner.joblib` — SHA-256
-`336713ecfec29f070e2209b71906d1016755363d9fd570a80e40040e7ec2db13`
+`submission_artifacts/weekly99_planner.joblib`, loaded by `script.py`.
 
 | setting | value |
 |---|---|
 | FPT rank residual weight | 0.15 |
 | capacity lookback | 42 days |
 | emergency operational scale | 0.5 |
-| scheduled fraction | 0.046 |
-| capacity weekly limit fraction | 0.95 (public V05 default) |
-| embedded public V05 base | `63d71091d7f46e8fb11798b7c2be936cc7e0022954824a4eb98c07dc1bf9b0b3` |
+| capacity weekly limit fraction | 0.99 |
 
-## Submission history
+## Contract
 
-| # | public score | note |
-|---|---:|---|
-| V05 | 1541.31 | causal anchor, revision `dfa8acbc` |
-| 007 | 1563.95 | identity + temperature rerank |
-| **008** | **1517.22** | FPT residual + emergency repricing |
-| 009 | 1528.06 | FPT residual 0.25, emergency buffer 3 |
-| 010 | pending | weekly margin 0.95 -> 0.98 on the 008 stack |
-| 011 | pending | workload variant, scheduled_fraction 0.046 |
+- Each scenario uses only readings with `end_time <= scenario.start_time`.
+- EOL is reconstructed as the evaluator defines it: strict `10 < T < 30`, daily median,
+  days with fewer than five readings masked, seven-calendar-day rolling median with
+  `min_periods=3`, first smoothed voltage `<= 2.40 V`. This matches all 82 observed train
+  EOL devices; censored devices remain censored.
+- Runtime: `batteryswap_public==0.3.4`, CPU only, no network, every live battery emitted
+  once with a valid plan date, 19,890 rows on the train split.
 
-## Causal and official contract
-
-- Every scenario uses only readings with `end_time <= scenario.start_time`.
-- EOL is reconstructed exactly as the evaluator defines it: strict `10 < T < 30`,
-  daily median, days with fewer than five readings masked, seven-calendar-day
-  rolling median with `min_periods=3`, first smoothed voltage `<= 2.40 V`.
-  The reconstruction matches all 82 observed train EOL devices; censored devices
-  remain censored.
-- Runtime uses `batteryswap_public==0.3.4`, CPU only, no network, and emits every
-  live battery exactly once with a valid plan date.
-
-## Evidence discipline
-
-Local totals are not comparable with the public leaderboard, and neither 010 nor 011
-carries a validated local gain: measured against the shipped 008 configuration, 010 is
--5.25 (paired t = -1.49, worse in the first chronological half) and 011 is a wash locally,
-motivated only by the public/local difference in early-swap pricing. 008 (1517.22) remains
-the anchor. Every artifact passes 57/57 tests with byte-identical replays and 19,890 rows
-well inside the 30-minute budget.
-
-Local totals are not directly comparable with the public leaderboard; candidates
-are supported by paired deltas against the same clean reference, never by claiming
-a public score in advance.
-
-## License
-
-Participant code is MIT (`LICENSE`). Third-party notices are recorded in
-`THIRD_PARTY_LICENSES.md`; reproduction steps are in `REPRODUCIBILITY.md`.
+MIT licensed (`LICENSE`). Third-party notices in `THIRD_PARTY_LICENSES.md`.

REPRODUCIBILITY.md

@@ -1,114 +1,34 @@
-# Reproduce the temperature-gated candidate
+# Reproducing
 
-Dataset revision: `7f423ac4cb6ab146f7ea7a37872eb4dfc3c9705c`.
+Dataset revision `7f423ac4cb6ab146f7ea7a37872eb4dfc3c9705c`; place the train split in
+`data/raw/train/`.
 
-Place the official train split in `data/raw/train/`. All fitting commands below rebuild features
-from raw official data; persisted experiment rows are optional equality checks, never training
-inputs. Do not replace the pinned local V06 full-lookback base artifact. The public V05 artifact is
-preserved separately on the Hub and on the `public/v05` Git branch.
-
-## Build both fallbacks and the primary
-
-Build the pinned evaluator image:
-
-```bash
-docker build -t batteryswap-temperature .
-```
-
-Use the evaluator interpreter for fitting:
-
-```bash
-docker run --rm -v "$(pwd)":/work -w /work -e PYTHONPATH=/work/src \
-  batteryswap-temperature /app/env/bin/python3 scripts/build_identity_submission.py \
-  --output submission_artifacts/identity_planner.joblib \
-  --manifest submission_artifacts/identity_planner.json
-
-docker run --rm -v "$(pwd)":/work -w /work -e PYTHONPATH=/work/src \
-  batteryswap-temperature /app/env/bin/python3 scripts/build_identity_submission.py \
-  --with-seasonal \
-  --output submission_artifacts/temperature_planner.joblib \
-  --manifest submission_artifacts/temperature_planner.json
-```
-
-Each build records the exact resulting artifact hash in its manifest. Re-pickling the wrapped V06
-estimator is not guaranteed to reproduce identical artifact bytes; release determinism is therefore
-checked on the two emitted submission CSVs below. The builder asserts the raw row count, all 82
-exact EOL crossings, AFT reference equality when available, expected donor-library dimensions
-(82 devices, 24 buildings, 7,257 endpoints), rounded temperature coefficient, and byte preservation
-of the local V06 base artifact.
-
-## Leakage-clean validation
-
-Generate outer predictions with all auxiliary state restricted to the outer-train batteries:
-
-```bash
-docker run --rm -v "$(pwd)":/work -w /work -e PYTHONPATH=/work/src \
-  batteryswap-temperature /app/env/bin/python3 scripts/validate_submission.py \
-  --outer-group building --output artifacts/clean_v05_outer_oof.json
-```
-
-Replay production features and the exact evaluator:
+The artifact wraps a byte-preserved base model
+(`63d71091d7f46e8fb11798b7c2be936cc7e0022954824a4eb98c07dc1bf9b0b3`) and differs from it
+only in planner policy constants. `scripts/build_hierarchical_fpt_submission.py` rebuilds
+it and asserts the embedded base hash.
 
 ```bash
-docker run --rm -v "$(pwd)":/work -w /work -e PYTHONPATH=/work/src \
-  batteryswap-temperature /app/env/bin/python3 scripts/verify_identity_feature_parity.py
-
-docker run --rm -v "$(pwd)":/work -w /work -e PYTHONPATH=/work/src \
-  batteryswap-temperature /app/env/bin/python3 scripts/evaluate_clean_identity_candidate.py
-```
-
-Expected primary evidence:
-
-```text
-rows: 19,890
-clean operational reference mean total: 1613.256409722222
-temperature <=2.50 V mean total: 1534.843055555556
-mean paired delta: -78.413354166667
-chronological halves: -110.968895833333 / -45.857812500000
-paired t: -2.303438305449
-strict promotion gate: PASS
-```
+docker build -t batteryswap .
 
-The clean prediction CSV fingerprint is
-`c810f68e73f62f15cf1d4d19574d5a1ec4ab0916edde3096bca9ad6b00332cbb`.
-
-## Tests and evaluator-runtime release gate
-
-```bash
 docker run --rm -v "$(pwd)":/work -w /work -e PYTHONPATH=/work/src \
-  batteryswap-temperature /app/env/bin/python3 -m pytest -q
+  batteryswap /app/env/bin/python3 scripts/build_hierarchical_fpt_submission.py \
+  --output submission_artifacts/weekly99_planner.joblib \
+  --manifest submission_artifacts/weekly99_planner.json
 ```
 
-Rebuild the image after the final artifacts are frozen, then run the submission twice with network
-disabled. Use separate output filenames and compare their SHA-256 hashes:
+Run and test:
 
 ```bash
-docker build -t batteryswap-temperature .
-
-docker run --rm --network none \
-  -e BATTERYSWAP_SPLITS=train \
-  -e BATTERYSWAP_SUBMISSION_PATH=/out/submission-1.csv \
+docker run --rm --network none -e BATTERYSWAP_SPLITS=train \
+  -e BATTERYSWAP_SUBMISSION_PATH=/out/submission.csv \
   -v "$(pwd)/data/raw":/tmp/data:ro -v "$(pwd)/artifacts":/out \
-  batteryswap-temperature \
-  bash -lc 'time /app/env/bin/python3 script.py'
+  batteryswap /app/env/bin/python3 script.py
 
-docker run --rm --network none \
-  -e BATTERYSWAP_SPLITS=train \
-  -e BATTERYSWAP_SUBMISSION_PATH=/out/submission-2.csv \
-  -v "$(pwd)/data/raw":/tmp/data:ro -v "$(pwd)/artifacts":/out \
-  batteryswap-temperature \
-  bash -lc 'time /app/env/bin/python3 script.py'
-
-sha256sum artifacts/submission-1.csv artifacts/submission-2.csv
+docker run --rm -v "$(pwd)":/work -w /work -e PYTHONPATH=/work/src \
+  batteryswap /app/env/bin/python3 -m pytest -q
 ```
 
-The hashes must match, every plan must pass the output invariants in `script.py`, elapsed time must
-remain under 30 minutes per evaluator split, and peak container memory from `docker stats` must
-remain below 32 GB. Record the measured final values in `artifacts/release_gate.json`; artifact
-hashes are authoritative in the two candidate manifests.
-
-## Deliberate staging only
-
-No command in this document pushes, submits, or chooses a final. After reviewing the ranked
-handoff, stage a chosen artifact explicitly with `BATTERYSWAP_PLANNER_PATH`; keep V05 available for
-immediate rollback.
+Expected: 19,890 rows, byte-identical across repeated runs, inside the 30-minute and
+32 GB limits. Re-pickling is not guaranteed byte-reproducible; determinism is checked on
+the emitted CSV.

scripts/build_hierarchical_fpt_submission.py

@@ -0,0 +1,157 @@
+"""Build the frozen hierarchical-FPT wrapper around the public V05 artifact.
+
+The wrapper carries two operational corrections on top of V05's policy, each
+validated independently over 48 dates x 27 bases:
+
+* ``capacity_lookback_days`` 14 -> 42 (V06): the repair pass may relocate work
+  across the whole horizon instead of a fortnight.  -4.68, t=-4.38.
+* ``emergency_operational_scale`` 0 -> 0.5 (V07): a missed battery becomes its own
+  working day with a dedicated round trip, measured at 65.56 beyond the late
+  penalty against the flat 2.0 the selection assumed.  -14.62, t=-4.11.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import importlib.metadata
+import io
+import json
+import subprocess
+from pathlib import Path
+
+import joblib
+from dataclasses import replace
+
+from batteryswapai.hierarchical_fpt import (
+    EOL_VOLTAGE,
+    HORIZON_DAYS,
+    MAX_STATE_STALENESS_DAYS,
+    MIN_WEEKLY_INCREMENTS,
+    RANK_RESIDUAL_WEIGHT,
+    SCHEMA_VERSION,
+    TERMINAL_WEEKS,
+    HierarchicalFPTPlanner,
+)
+
+
+PUBLIC_V05_GIT_SPEC = "public/v05:submission_artifacts/planner.joblib"
+PUBLIC_V05_SHA256 = "63d71091d7f46e8fb11798b7c2be936cc7e0022954824a4eb98c07dc1bf9b0b3"
+
+CAPACITY_LOOKBACK_DAYS = 42
+EMERGENCY_OPERATIONAL_SCALE = 0.5
+CAPACITY_WEEKLY_LIMIT_FRACTION = 0.99
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--output",
+        type=Path,
+        default=Path("submission_artifacts/hierarchical_fpt_planner.joblib"),
+    )
+    parser.add_argument(
+        "--manifest",
+        type=Path,
+        default=Path("submission_artifacts/hierarchical_fpt_planner.json"),
+    )
+    parser.add_argument("--base-git-spec", default=PUBLIC_V05_GIT_SPEC)
+    return parser.parse_args()
+
+
+def _sha_bytes(payload: bytes) -> str:
+    return hashlib.sha256(payload).hexdigest()
+
+
+def _sha(path: Path) -> str:
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def _git_bytes(spec: str) -> bytes:
+    completed = subprocess.run(
+        ["git", "show", spec], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+    )
+    return completed.stdout
+
+
+def main() -> None:
+    args = parse_args()
+    repo = Path(__file__).resolve().parents[1]
+    payload = _git_bytes(args.base_git_spec)
+    base_sha = _sha_bytes(payload)
+    if base_sha != PUBLIC_V05_SHA256:
+        raise AssertionError(
+            f"public V05 SHA mismatch: expected {PUBLIC_V05_SHA256}, found {base_sha}"
+        )
+    base_planner = joblib.load(io.BytesIO(payload))
+    if int(base_planner.policy.capacity_lookback_days) != 14:
+        raise AssertionError("embedded planner is not public V05 capacity policy")
+    base_planner.policy = replace(
+        base_planner.policy,
+        capacity_lookback_days=CAPACITY_LOOKBACK_DAYS,
+        emergency_operational_scale=EMERGENCY_OPERATIONAL_SCALE,
+        capacity_weekly_limit_fraction=CAPACITY_WEEKLY_LIMIT_FRACTION,
+    )
+    wrapper = HierarchicalFPTPlanner(
+        base_planner=base_planner,
+        base_artifact_sha256=base_sha,
+    )
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    joblib.dump(wrapper, args.output, compress=3)
+    loaded = joblib.load(args.output)
+    if loaded.base_artifact_sha256 != PUBLIC_V05_SHA256:
+        raise AssertionError("serialized wrapper lost its public V05 provenance")
+    manifest = {
+        "artifact": args.output.as_posix(),
+        "artifact_sha256": _sha(args.output),
+        "schema_version": SCHEMA_VERSION,
+        "base_artifact": args.base_git_spec,
+        "base_artifact_sha256": base_sha,
+        "configuration": {
+            "eol_voltage": EOL_VOLTAGE,
+            "horizon_days": HORIZON_DAYS,
+            "terminal_nonoverlap_weeks": TERMINAL_WEEKS,
+            "minimum_weekly_increments": MIN_WEEKLY_INCREMENTS,
+            "maximum_state_staleness_days": MAX_STATE_STALENESS_DAYS,
+            "rank_residual_weight": RANK_RESIDUAL_WEIGHT,
+            "capacity_lookback_days": CAPACITY_LOOKBACK_DAYS,
+            "emergency_operational_scale": EMERGENCY_OPERATIONAL_SCALE,
+            "capacity_weekly_limit_fraction": CAPACITY_WEEKLY_LIMIT_FRACTION,
+            "outer_pool": "leave current target building out",
+            "smoothing": (
+                "strict 10<T<30; daily median count>=5; rolling seven-calendar-day "
+                "median min_periods=3; terminal calendar day strictly before cutoff"
+            ),
+        },
+        "promotion_evidence": {
+            "official_48": "artifacts/hierarchical_fpt_official.json",
+            "robust_48x27": "artifacts/hierarchical_fpt_grid.cases.json",
+        },
+        "source_sha256": {
+            name: _sha(repo / name)
+            for name in (
+                "src/batteryswapai/hierarchical_fpt.py",
+                "src/batteryswapai/identity_ensemble.py",
+                "src/batteryswapai/competition_planner.py",
+                "scripts/experiment_hierarchical_fpt.py",
+                "scripts/evaluate_hierarchical_fpt_grid.py",
+                "scripts/build_hierarchical_fpt_submission.py",
+                "script.py",
+                "Dockerfile",
+                "requirements.txt",
+            )
+        },
+        "runtime_versions": {
+            package: importlib.metadata.version(package)
+            for package in (
+                "batteryswap_public", "joblib", "numpy", "pandas", "scipy",
+                "scikit-learn",
+            )
+        },
+    }
+    args.manifest.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
+    print(json.dumps(manifest, indent=2))
+
+
+if __name__ == "__main__":
+    main()

scripts/build_identity_submission.py

@@ -1,490 +0,0 @@
-"""Fit and serialize the deployable LT/sim/seasonal identity wrapper.
-
-This builder never overwrites the frozen base planner.  OOF experiment rows are
-used only for their strictly causal first-passage training covariates; the AFT
-is refit on every train building and the similarity library is rebuilt directly
-from the train split.
-"""
-
-from __future__ import annotations
-
-import argparse
-import dataclasses
-import hashlib
-import importlib.metadata
-import json
-from pathlib import Path
-
-import joblib
-import numpy as np
-import pandas as pd
-from batteryswap_public.utils import iterate_scenarios, load_dataset
-
-from batteryswapai.identity_ensemble import (
-    FROZEN_TEMPERATURE_BETA_V_PER_C,
-    TEMPERATURE_MAX_PREDICTED_MIN_VOLTAGE,
-    CausalHistoryCache,
-    IdentityEnsembleModel,
-    IdentityEnsemblePlanner,
-    LongTermAFTResidual,
-    OriginalSimilarityResidual,
-    SCHEMA_VERSION,
-    SeasonalTemperatureResidual,
-    WeightedIdentityResidual,
-    build_similarity_donor_library,
-    exact_smoothed_voltage,
-    fit_aft,
-    fit_temperature_beta,
-    _first_passage_lifetime,
-)
-
-
-DATASET_REVISION = "7f423ac4cb6ab146f7ea7a37872eb4dfc3c9705c"
-EXPECTED_BASE_SHA256 = "3cfca2e7dd2c05ddd84f2eca42a484168a1ab4ad3cd454806ef4e687fe8f1569"
-EXPECTED_ROWS = 19_890
-EXPECTED_EOL_DEVICES = 82
-EXPECTED_DONOR_DEVICES = 82
-EXPECTED_DONOR_BUILDINGS = 24
-EXPECTED_DONOR_ENDPOINTS = 7_257
-CONTAINER_BASE = (
-    "huggingface/competitions@sha256:"
-    "6cea4ff69a6832761484f48c07ccfbf49f701f285ffcb9fc72a4ecfb81b6b4e5"
-)
-
-
-def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--dataset-path", type=Path, default=Path("data/raw/train"))
-    parser.add_argument(
-        "--base-artifact",
-        type=Path,
-        default=Path("submission_artifacts/planner.joblib"),
-    )
-    parser.add_argument(
-        "--lt-reference-rows",
-        type=Path,
-        default=Path("artifacts/lt_fp_aft_official.rows.csv"),
-        help=(
-            "Optional validation-only reference. Training covariates are always "
-            "rebuilt from raw train prefixes; this file is never required."
-        ),
-    )
-    parser.add_argument(
-        "--output",
-        type=Path,
-        default=Path("submission_artifacts/identity_planner.joblib"),
-    )
-    parser.add_argument(
-        "--manifest",
-        type=Path,
-        default=Path("submission_artifacts/identity_planner.json"),
-    )
-    parser.add_argument(
-        "--expected-base-sha256", default=EXPECTED_BASE_SHA256
-    )
-    parser.add_argument(
-        "--with-seasonal",
-        action="store_true",
-        help="Add the promoted <=2.50 V seasonal-temperature post-rerank.",
-    )
-    return parser.parse_args()
-
-
-def _sha256(path: Path) -> str:
-    digest = hashlib.sha256()
-    with path.open("rb") as handle:
-        while chunk := handle.read(1024 * 1024):
-            digest.update(chunk)
-    return digest.hexdigest()
-
-
-def _array_sha256(values: np.ndarray) -> str:
-    contiguous = np.ascontiguousarray(np.asarray(values))
-    if contiguous.dtype.hasobject:
-        payload = "\n".join(contiguous.astype(str).ravel()).encode("utf-8")
-    else:
-        payload = contiguous.view(np.uint8)
-    return hashlib.sha256(payload).hexdigest()
-
-
-def _dataset_manifest(dataset_path: Path) -> dict[str, str]:
-    return {
-        path.relative_to(dataset_path).as_posix(): _sha256(path)
-        for path in sorted(dataset_path.rglob("*"))
-        if path.is_file()
-    }
-
-
-def _source_hashes(repo_root: Path) -> dict[str, str]:
-    paths = (
-        repo_root / "src/batteryswapai/identity_ensemble.py",
-        repo_root / "src/batteryswapai/competition_planner.py",
-        repo_root / "scripts/build_identity_submission.py",
-        repo_root / "scripts/experiment_lt_fp_aft.py",
-        repo_root / "scripts/experiment_similarity_eol.py",
-        repo_root / "scripts/experiment_identity_ensemble.py",
-        repo_root / "scripts/experiment_temperature_physics_ensemble.py",
-        repo_root / "scripts/evaluate_clean_identity_candidate.py",
-        repo_root / "scripts/verify_identity_feature_parity.py",
-        repo_root / "script.py",
-        repo_root / "Dockerfile",
-        repo_root / "requirements.txt",
-        repo_root / "pyproject.toml",
-        repo_root / "LICENSE",
-        repo_root / "THIRD_PARTY_LICENSES.md",
-        repo_root / "README.md",
-        repo_root / "REPRODUCIBILITY.md",
-    )
-    return {
-        path.relative_to(repo_root).as_posix(): _sha256(path)
-        for path in paths
-        if path.exists()
-    }
-
-
-def _evidence_hashes(repo_root: Path) -> dict[str, str]:
-    paths = (
-        "artifacts/lt_fp_aft_official.json",
-        "artifacts/lt_fp_aft_official.rows.csv",
-        "artifacts/similarity_eol_official.json",
-        "artifacts/similarity_eol_official.rows.csv",
-        "artifacts/identity_ensemble_official.json",
-        "artifacts/identity_ensemble_official.rows.csv",
-        "artifacts/temperature_physics_ensemble_official.json",
-        "artifacts/temperature_physics_ensemble_official.rows.csv",
-        "artifacts/temperature_physics_globalbeta_diagnostic.json",
-    )
-    return {
-        name: _sha256(repo_root / name)
-        for name in paths
-        if (repo_root / name).exists()
-    }
-
-
-def _assert_eol_reconstruction(
-    smoothed: pd.DataFrame, eol_times: pd.Series
-) -> dict[str, int]:
-    crossings = (
-        smoothed[smoothed["smooth_voltage"].le(2.40)]
-        .groupby("device_id", observed=True)["end_time"]
-        .min()
-    )
-    crossings.index = crossings.index.astype(str)
-    observed = pd.to_datetime(eol_times.dropna(), errors="raise")
-    observed.index = observed.index.astype(str)
-    comparison = pd.DataFrame(
-        {
-            "official": observed.dt.normalize(),
-            "reconstructed": pd.to_datetime(crossings).dt.normalize(),
-        }
-    )
-    exact = int(
-        comparison.dropna()["official"].eq(comparison.dropna()["reconstructed"]).sum()
-    )
-    if len(observed) != EXPECTED_EOL_DEVICES or exact != EXPECTED_EOL_DEVICES:
-        raise AssertionError(
-            f"exact EOL reconstruction failed: observed={len(observed)}, exact={exact}"
-        )
-    return {"observed_eol_devices": len(observed), "exact_eol_matches": exact}
-
-
-def _build_aft_training_rows(
-    locations: pd.DataFrame,
-    timeseries: pd.DataFrame,
-    eol_times: pd.Series,
-    scenarios: list[dict],
-) -> pd.DataFrame:
-    """Rebuild the full-train AFT landmarks from official visible prefixes.
-
-    The serialized winner must be reproducible from the allowed train split alone.
-    Persisted OOF rows are useful parity evidence, but are deliberately not an input
-    to this builder.
-    """
-
-    cache = CausalHistoryCache(
-        beta_v_per_c=FROZEN_TEMPERATURE_BETA_V_PER_C,
-        split_id="train",
-    )
-    records: list[dict[str, object]] = []
-    for scenario, locs, visible, _ in iterate_scenarios(
-        locations, timeseries, eol_times, scenarios
-    ):
-        start = pd.Timestamp(scenario["start_time"])
-        history = cache.update(visible, start)
-        for row in locs.itertuples(index=False):
-            battery = str(row.battery)
-            installation = pd.Timestamp(row.start_time)
-            lifetime, reliable = _first_passage_lifetime(
-                battery,
-                start,
-                installation,
-                history.smooth_lookup,
-            )
-            event_time = pd.to_datetime(eol_times.get(battery), errors="coerce")
-            event_observed = bool(pd.notna(event_time))
-            outcome_time = (
-                pd.Timestamp(event_time)
-                if event_observed
-                else pd.Timestamp(row.end_time)
-            )
-            records.append(
-                {
-                    "scenario": str(scenario["name"]),
-                    "battery": battery,
-                    "fp_reliable": reliable,
-                    "fp_lifetime_days": lifetime,
-                    "landmark_age_days": float(
-                        (start - installation) / pd.Timedelta(days=1)
-                    ),
-                    "outcome_lifetime_days": float(
-                        (outcome_time - installation) / pd.Timedelta(days=1)
-                    ),
-                    "event_observed": event_observed,
-                }
-            )
-    result = pd.DataFrame(records)
-    if len(result) != EXPECTED_ROWS or result.duplicated(
-        ["scenario", "battery"]
-    ).any():
-        raise AssertionError(
-            "raw AFT landmark rebuild did not produce 19,890 unique rows"
-        )
-    return result
-
-
-def _verify_optional_aft_reference(
-    rebuilt: pd.DataFrame, reference_path: Path
-) -> dict[str, object] | None:
-    if not reference_path.exists():
-        return None
-    columns = [
-        "scenario",
-        "battery",
-        "fp_reliable",
-        "fp_lifetime_days",
-        "landmark_age_days",
-        "outcome_lifetime_days",
-        "event_observed",
-    ]
-    reference = pd.read_csv(reference_path, usecols=columns)
-    if len(reference) != len(rebuilt):
-        raise AssertionError("optional LT reference row count differs from raw rebuild")
-    if not rebuilt[["scenario", "battery"]].astype(str).equals(
-        reference[["scenario", "battery"]].astype(str)
-    ):
-        raise AssertionError("optional LT reference key order differs from raw rebuild")
-    for column in ("fp_reliable", "event_observed"):
-        if not np.array_equal(
-            rebuilt[column].to_numpy(bool),
-            reference[column].to_numpy(bool),
-        ):
-            raise AssertionError(
-                f"raw AFT {column} flags differ from LT reference"
-            )
-    for column in (
-        "fp_lifetime_days",
-        "landmark_age_days",
-        "outcome_lifetime_days",
-    ):
-        np.testing.assert_allclose(
-            rebuilt[column].to_numpy(float),
-            reference[column].to_numpy(float),
-            rtol=0.0,
-            atol=1e-12,
-            equal_nan=True,
-        )
-    return {
-        "path": reference_path.as_posix(),
-        "sha256": _sha256(reference_path),
-        "rows": len(reference),
-        "raw_rebuild_exact": True,
-    }
-
-
-def main() -> None:
-    args = parse_args()
-    repo_root = Path(__file__).resolve().parents[1]
-    base_sha_before = _sha256(args.base_artifact)
-    if base_sha_before != args.expected_base_sha256:
-        raise AssertionError(
-            "frozen base artifact changed: "
-            f"expected={args.expected_base_sha256}, actual={base_sha_before}"
-        )
-
-    locations, timeseries, eol_times, scenarios = load_dataset(args.dataset_path)
-    if len(scenarios) != 48:
-        raise AssertionError(f"expected 48 train scenarios, found {len(scenarios)}")
-    eol_times = eol_times.copy()
-    eol_times.index = eol_times.index.astype(str)
-
-    aft_rows = _build_aft_training_rows(
-        locations, timeseries, eol_times, scenarios
-    )
-    aft_reference = _verify_optional_aft_reference(
-        aft_rows, args.lt_reference_rows
-    )
-    aft_parameters, aft_diagnostics = fit_aft(aft_rows)
-
-    full_smoothed = exact_smoothed_voltage(timeseries)
-    smoothing_audit = _assert_eol_reconstruction(full_smoothed, eol_times)
-    mapping_rows = locations[["battery", "building"]].drop_duplicates()
-    if mapping_rows["battery"].duplicated().any():
-        raise AssertionError("a train battery maps to multiple buildings")
-    battery_building = dict(
-        zip(
-            mapping_rows["battery"].astype(str),
-            mapping_rows["building"].astype(str),
-            strict=True,
-        )
-    )
-    donor_library = build_similarity_donor_library(
-        full_smoothed, eol_times, battery_building
-    )
-    donor_counts = (
-        donor_library.donor_count,
-        donor_library.building_count,
-        donor_library.endpoint_count,
-    )
-    expected_counts = (
-        EXPECTED_DONOR_DEVICES,
-        EXPECTED_DONOR_BUILDINGS,
-        EXPECTED_DONOR_ENDPOINTS,
-    )
-    if donor_counts != expected_counts:
-        raise AssertionError(
-            f"full donor library changed: expected={expected_counts}, actual={donor_counts}"
-        )
-
-    fitted_beta, beta_diagnostics = fit_temperature_beta(timeseries)
-    if round(fitted_beta, 5) != FROZEN_TEMPERATURE_BETA_V_PER_C:
-        raise AssertionError(
-            "full-train temperature beta no longer reproduces the frozen value: "
-            f"fit={fitted_beta}, frozen={FROZEN_TEMPERATURE_BETA_V_PER_C}"
-        )
-    seasonal = SeasonalTemperatureResidual(
-        beta_v_per_c=FROZEN_TEMPERATURE_BETA_V_PER_C,
-        fitted_raw_beta_v_per_c=fitted_beta,
-        training_readings=int(beta_diagnostics["training_readings"]),
-        maximum_predicted_min_voltage=TEMPERATURE_MAX_PREDICTED_MIN_VOLTAGE,
-    )
-    ensemble = IdentityEnsembleModel(
-        identity_residuals=(
-            WeightedIdentityResidual(LongTermAFTResidual(aft_parameters), 0.50),
-            WeightedIdentityResidual(OriginalSimilarityResidual(donor_library), 0.50),
-        ),
-        post_residuals=(seasonal,) if args.with_seasonal else (),
-        history_temperature_beta_v_per_c=FROZEN_TEMPERATURE_BETA_V_PER_C,
-    )
-    base_planner = joblib.load(args.base_artifact)
-    wrapper = IdentityEnsemblePlanner(
-        base_planner=base_planner,
-        ensemble=ensemble,
-        base_artifact_sha256=base_sha_before,
-    )
-
-    args.output.parent.mkdir(parents=True, exist_ok=True)
-    joblib.dump(wrapper, args.output, compress=3)
-    base_sha_after = _sha256(args.base_artifact)
-    if base_sha_after != base_sha_before:
-        raise AssertionError("builder modified the frozen base artifact")
-
-    library_arrays = {
-        "values": donor_library.values,
-        "masks": donor_library.masks,
-        "donor_ids": donor_library.donor_ids,
-        "donor_buildings": donor_library.donor_buildings,
-        "donor_codes": donor_library.donor_codes,
-        "endpoint_days": donor_library.endpoint_days,
-        "residual_days": donor_library.residual_days,
-        "device_ids_by_code": donor_library.device_ids_by_code,
-        "device_buildings_by_code": donor_library.device_buildings_by_code,
-    }
-    manifest = {
-        "schema_version": SCHEMA_VERSION,
-        "dataset_revision": DATASET_REVISION,
-        "container_base": CONTAINER_BASE,
-        "artifact": args.output.as_posix(),
-        "artifact_sha256": _sha256(args.output),
-        "artifact_size_bytes": args.output.stat().st_size,
-        "base_artifact": args.base_artifact.as_posix(),
-        "base_artifact_sha256_before": base_sha_before,
-        "base_artifact_sha256_after": base_sha_after,
-        "base_artifact_byte_preserved": base_sha_before == base_sha_after,
-        "dataset_files": _dataset_manifest(args.dataset_path),
-        "training_source": {
-            "aft_landmarks": "rebuilt from official raw train visible prefixes",
-            "optional_lt_reference": aft_reference,
-        },
-        "source_sha256": _source_hashes(repo_root),
-        "validation_evidence_sha256": _evidence_hashes(repo_root),
-        "aft": {
-            "parameters": dataclasses.asdict(aft_parameters),
-            "diagnostics": aft_diagnostics,
-        },
-        "similarity_library": {
-            "donor_devices": donor_library.donor_count,
-            "donor_buildings": donor_library.building_count,
-            "donor_endpoints": donor_library.endpoint_count,
-            "prefix_lags": donor_library.values.shape[1],
-            "arrays": {
-                name: {
-                    "shape": list(np.asarray(values).shape),
-                    "dtype": str(np.asarray(values).dtype),
-                    "sha256": _array_sha256(values),
-                }
-                for name, values in library_arrays.items()
-            },
-        },
-        "seasonal_temperature": {
-            "enabled": args.with_seasonal,
-            "frozen_beta_v_per_c": FROZEN_TEMPERATURE_BETA_V_PER_C,
-            "maximum_predicted_min_voltage": (
-                TEMPERATURE_MAX_PREDICTED_MIN_VOLTAGE
-                if args.with_seasonal
-                else None
-            ),
-            "full_train_fit": beta_diagnostics,
-        },
-        "ensemble": {
-            "identity_components": [
-                {"name": item.residual.name, "weight": item.weight}
-                for item in ensemble.identity_residuals
-            ],
-            "post_components": [item.name for item in ensemble.post_residuals],
-            "freshness_stratified": True,
-            "raw_risk_multiset_preserved": True,
-            "planner_effective_risk_multiset_preserved": True,
-            "v07_emergency_scale": 0.75,
-            "v07_mean_trip_gate_hours": 8.0,
-        },
-        "smoothing_audit": smoothing_audit,
-        "runtime_versions": {
-            package: importlib.metadata.version(package)
-            for package in (
-                "batteryswap_public",
-                "fastparquet",
-                "joblib",
-                "numpy",
-                "pandas",
-                "pydantic-settings",
-                "scikit-learn",
-                "scipy",
-                "structlog",
-            )
-        },
-        "random_seeds": {
-            "base_model": 2026,
-            "identity_residuals": None,
-        },
-        "official_evaluator_run": False,
-    }
-    args.manifest.parent.mkdir(parents=True, exist_ok=True)
-    args.manifest.write_text(
-        json.dumps(manifest, indent=2, sort_keys=True, default=float),
-        encoding="utf-8",
-    )
-    print(json.dumps(manifest, indent=2, sort_keys=True, default=float))
-
-
-if __name__ == "__main__":
-    main()

scripts/evaluate_clean_identity_candidate.py

@@ -1,450 +0,0 @@
-"""Evaluate the frozen identity+temperature rerank on leakage-clean base OOF risks.
-
-The base predictions must come from outer-building fits whose trajectory matrix and
-EOL targets contain outer-train batteries only. Identity signals are the already
-frozen building-held-out AFT/similarity/temperature signals. No weights or policies
-are searched here.
-"""
-
-from __future__ import annotations
-
-import argparse
-import hashlib
-import json
-import math
-from dataclasses import replace
-from pathlib import Path
-
-import joblib
-import numpy as np
-import pandas as pd
-from batteryswap_public.evaluate import evaluate_plan
-from batteryswap_public.utils import iterate_scenarios, load_dataset
-
-from batteryswapai.competition_planner import CompetitionPlanner
-from batteryswapai.identity_ensemble import (
-    IdentityEnsembleModel,
-    ResidualSignal,
-    SeasonalTemperatureResidual,
-    planner_freshness_factors,
-    v07_emergency_scale,
-)
-
-
-KEYS = ["scenario", "battery"]
-COST_COLUMNS = (
-    "battery_swap",
-    "building_change",
-    "room_change",
-    "travel",
-    "overtime",
-    "daily_limit",
-    "weekly_limit",
-    "late_swap",
-    "early_swap",
-    "total_cost",
-)
-
-
-def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--dataset-path", type=Path, default=Path("data/raw/train"))
-    parser.add_argument(
-        "--clean-predictions",
-        type=Path,
-        default=Path("artifacts/clean_v05_outer_oof.csv"),
-    )
-    parser.add_argument(
-        "--artifact",
-        type=Path,
-        default=Path("submission_artifacts/identity_planner.joblib"),
-    )
-    parser.add_argument(
-        "--lt-rows",
-        type=Path,
-        default=Path("artifacts/lt_fp_aft_official.rows.csv"),
-    )
-    parser.add_argument(
-        "--similarity-rows",
-        type=Path,
-        default=Path("artifacts/similarity_eol_official.rows.csv"),
-    )
-    parser.add_argument(
-        "--temperature-rows",
-        type=Path,
-        default=Path("artifacts/temperature_physics_ensemble_official.rows.csv"),
-    )
-    parser.add_argument(
-        "--output-prefix",
-        type=Path,
-        default=Path("artifacts/clean_identity_temperature_official"),
-    )
-    return parser.parse_args()
-
-
-def _sha256(path: Path) -> str:
-    return hashlib.sha256(path.read_bytes()).hexdigest()
-
-
-def _bool(values: pd.Series) -> np.ndarray:
-    if pd.api.types.is_bool_dtype(values):
-        return values.to_numpy(bool)
-    return values.astype(str).str.lower().isin({"true", "1"}).to_numpy(bool)
-
-
-def _paired_t(delta: np.ndarray) -> float:
-    delta = np.asarray(delta, dtype=float)
-    standard_error = float(delta.std(ddof=1) / math.sqrt(len(delta)))
-    if standard_error == 0.0:
-        return -1e99 if delta.mean() < 0.0 else 1e99
-    return float(delta.mean() / standard_error)
-
-
-def _score(
-    planner: CompetitionPlanner,
-    scenario: dict,
-    locs: pd.DataFrame,
-    not_dead: pd.Series,
-    snapshot: pd.DataFrame,
-    risk: np.ndarray,
-    rul: np.ndarray,
-    survivor: np.ndarray,
-) -> tuple[pd.DataFrame, dict[str, float]]:
-    plan = planner.plan_snapshot(
-        snapshot,
-        locs,
-        scenario["travel_costs"],
-        scenario["settings"],
-        scenario["start_time"],
-        predicted_risk=risk,
-        predicted_rul=rul,
-        predicted_survivor_rul=survivor,
-    )
-    _, _, score = evaluate_plan(
-        plan,
-        locs,
-        scenario["travel_costs"],
-        scenario["settings"],
-        eol_times=not_dead,
-        start_time=pd.Timestamp(scenario["start_time"]),
-        verbose=0,
-    )
-    return plan, {column: float(score[column]) for column in COST_COLUMNS}
-
-
-def main() -> None:
-    args = parse_args()
-    clean = pd.read_csv(args.clean_predictions)
-    lt = pd.read_csv(
-        args.lt_rows, usecols=KEYS + ["fp_aft_risk_oof", "fp_reliable"]
-    )
-    similarity = pd.read_csv(
-        args.similarity_rows,
-        usecols=KEYS
-        + [
-            "planner_data_gap_days",
-            "planner_freshness_factor",
-            "sim_neighbor_risk_oof",
-            "sim_neighbor_reliable",
-        ],
-    )
-    temperature = pd.read_csv(
-        args.temperature_rows,
-        usecols=KEYS + ["reliable", "predicted_min_smooth_voltage"],
-    )
-    frames = {
-        "clean": clean,
-        "lt": lt,
-        "similarity": similarity,
-        "temperature": temperature,
-    }
-    for name, frame in frames.items():
-        frame["scenario"] = frame["scenario"].astype(str)
-        frame["battery"] = frame["battery"].astype(str)
-        if len(frame) != 19_890 or frame.duplicated(KEYS).any():
-            raise AssertionError(f"{name} does not contain 19,890 unique keys")
-        if not clean[KEYS].equals(frame[KEYS]):
-            raise AssertionError(f"{name} key order differs from clean base")
-
-    rows = clean[
-        KEYS
-        + [
-            "oof_event_risk",
-            "oof_rul_days",
-            "oof_survivor_rul",
-            "target_rul_days",
-            "event_observed",
-        ]
-    ].copy()
-    for column in (
-        "planner_data_gap_days",
-        "planner_freshness_factor",
-        "sim_neighbor_risk_oof",
-        "sim_neighbor_reliable",
-    ):
-        rows[column] = similarity[column].to_numpy()
-    rows["fp_aft_risk_oof"] = lt["fp_aft_risk_oof"].to_numpy()
-    rows["fp_reliable"] = lt["fp_reliable"].to_numpy()
-    rows["temperature_reliable"] = temperature["reliable"].to_numpy()
-    rows["temperature_predicted_min_voltage"] = temperature[
-        "predicted_min_smooth_voltage"
-    ].to_numpy(float)
-    rows["temperature_urgency"] = -rows[
-        "temperature_predicted_min_voltage"
-    ].to_numpy(float)
-
-    wrapper = joblib.load(args.artifact)
-    identity_only = IdentityEnsembleModel(wrapper.ensemble.identity_residuals)
-    temperature_ensemble = IdentityEnsembleModel(
-        wrapper.ensemble.identity_residuals,
-        (SeasonalTemperatureResidual(),),
-    )
-    row_index = rows.set_index(KEYS)
-    locations, timeseries, eol_times, scenarios = load_dataset(args.dataset_path)
-    case_rows: list[dict[str, object]] = []
-    risk_rows: list[pd.DataFrame] = []
-    scale_counts = {"0.0": 0, "0.75": 0}
-
-    for scenario, locs, _, not_dead in iterate_scenarios(
-        locations,
-        timeseries,
-        eol_times,
-        scenarios,
-    ):
-        name = str(scenario["name"])
-        batteries = locs["battery"].astype(str).to_numpy()
-        keys = pd.MultiIndex.from_arrays(
-            [np.repeat(name, len(batteries)), batteries], names=KEYS
-        )
-        aligned = row_index.reindex(keys)
-        if aligned.isna().all(axis=1).any():
-            raise AssertionError(f"clean row alignment failed for {name}")
-        base = aligned["oof_event_risk"].to_numpy(float)
-        rul = aligned["oof_rul_days"].to_numpy(float)
-        survivor = aligned["oof_survivor_rul"].to_numpy(float)
-        freshness = aligned["planner_freshness_factor"].to_numpy(float)
-        expected_freshness = planner_freshness_factors(
-            aligned["planner_data_gap_days"].to_numpy(float),
-            wrapper.base_planner.policy,
-        )
-        np.testing.assert_array_equal(freshness, expected_freshness)
-        identity_signals = (
-            ResidualSignal(
-                aligned["fp_aft_risk_oof"].to_numpy(float),
-                _bool(aligned["fp_reliable"]),
-            ),
-            ResidualSignal(
-                aligned["sim_neighbor_risk_oof"].to_numpy(float),
-                _bool(aligned["sim_neighbor_reliable"]),
-            ),
-        )
-        temperature_urgency = aligned["temperature_urgency"].to_numpy(float)
-        temperature_reliable = _bool(aligned["temperature_reliable"])
-        predicted_min_voltage = aligned[
-            "temperature_predicted_min_voltage"
-        ].to_numpy(float)
-        temperature_signals = {
-            "temperature_all": ResidualSignal(
-                temperature_urgency, temperature_reliable
-            ),
-            "temperature_below_250": ResidualSignal(
-                temperature_urgency,
-                temperature_reliable & (predicted_min_voltage <= 2.50),
-            ),
-            "temperature_below_240": ResidualSignal(
-                temperature_urgency,
-                temperature_reliable & (predicted_min_voltage <= 2.40),
-            ),
-        }
-        identity_risk = identity_only.rerank_from_signals(
-            base, freshness, batteries, identity_signals
-        )
-        temperature_risks = {
-            name: temperature_ensemble.rerank_from_signals(
-                base,
-                freshness,
-                batteries,
-                identity_signals,
-                (signal,),
-            )
-            for name, signal in temperature_signals.items()
-        }
-        snapshot = locs[["battery", "building", "room"]].copy()
-        snapshot["data_gap_days"] = aligned[
-            "planner_data_gap_days"
-        ].to_numpy(float)
-        scale = v07_emergency_scale(
-            snapshot, scenario["travel_costs"], scenario["settings"]
-        )
-        scale_counts[str(scale)] += 1
-        planner = CompetitionPlanner(
-            None,
-            replace(
-                wrapper.base_planner.policy,
-                emergency_operational_scale=scale,
-            ),
-        )
-        arm_scores: dict[str, dict[str, float]] = {}
-        arm_plans: dict[str, pd.DataFrame] = {}
-        risks_by_arm = {
-            "baseline": base,
-            "identity": identity_risk,
-            **temperature_risks,
-        }
-        for arm, risk in risks_by_arm.items():
-            arm_plans[arm], arm_scores[arm] = _score(
-                planner, scenario, locs, not_dead, snapshot, risk, rul, survivor
-            )
-        start = pd.Timestamp(scenario["start_time"])
-        horizon_end = start + pd.Timedelta(
-            days=float(scenario["settings"].planning_window_days)
-        )
-        selected = {
-            arm: set(plan.loc[plan["day"].le(horizon_end), "battery"].astype(str))
-            for arm, plan in arm_plans.items()
-        }
-        expected_quota = min(max(int(round(0.038 * len(locs))), 8), 24)
-        if any(len(value) != expected_quota for value in selected.values()):
-            raise AssertionError(f"fixed quota drifted in {name}")
-        case: dict[str, object] = {
-            "scenario": name,
-            "start_time": start.isoformat(),
-            "effective_emergency_scale": scale,
-            "quota": expected_quota,
-        }
-        for arm, score in arm_scores.items():
-            case.update({f"{arm}_{key}": value for key, value in score.items()})
-        case_rows.append(case)
-        risk_rows.append(
-            pd.DataFrame(
-                {
-                    "scenario": name,
-                    "battery": batteries,
-                    "planner_freshness_factor": freshness,
-                    "clean_baseline_event_risk": base,
-                    "clean_identity_event_risk": identity_risk,
-                    **{
-                        f"clean_{arm}_event_risk": risk
-                        for arm, risk in temperature_risks.items()
-                    },
-                    "baseline_selected": [battery in selected["baseline"] for battery in batteries],
-                    "identity_selected": [battery in selected["identity"] for battery in batteries],
-                    **{
-                        f"{arm}_selected": [
-                            battery in selected[arm] for battery in batteries
-                        ]
-                        for arm in temperature_risks
-                    },
-                }
-            )
-        )
-
-    cases = pd.DataFrame(case_rows).sort_values("start_time").reset_index(drop=True)
-    risks = pd.concat(risk_rows, ignore_index=True)
-    baseline = cases["baseline_total_cost"].to_numpy(float)
-    identity = cases["identity_total_cost"].to_numpy(float)
-    identity_delta = identity - baseline
-    temperature_deltas_vs_identity = {
-        arm: cases[f"{arm}_total_cost"].to_numpy(float) - identity
-        for arm in (
-            "temperature_all",
-            "temperature_below_250",
-            "temperature_below_240",
-        )
-    }
-    temperature_variants = {}
-    for arm, incremental in temperature_deltas_vs_identity.items():
-        values = cases[f"{arm}_total_cost"].to_numpy(float)
-        delta = values - baseline
-        temperature_variants[arm] = {
-            "total": float(values.mean()),
-            "delta_vs_baseline": float(delta.mean()),
-            "delta_vs_identity": float(incremental.mean()),
-            "paired_t_vs_baseline": _paired_t(delta),
-            "paired_t_vs_identity": _paired_t(incremental),
-            "first_half_delta_vs_baseline": float(delta[:24].mean()),
-            "second_half_delta_vs_baseline": float(delta[24:].mean()),
-            "first_half_delta_vs_identity": float(incremental[:24].mean()),
-            "second_half_delta_vs_identity": float(incremental[24:].mean()),
-            "improved": int((delta < 0).sum()),
-            "unchanged": int((delta == 0).sum()),
-            "worse": int((delta > 0).sum()),
-            "strict_gate_vs_baseline": bool(
-                delta.mean() <= -40.0
-                and delta[:24].mean() <= 0.0
-                and delta[24:].mean() <= 0.0
-                and _paired_t(delta) <= -2.0
-            ),
-        }
-    candidate = cases["temperature_below_250_total_cost"].to_numpy(float)
-    delta = candidate - baseline
-    report = {
-        "experiment": (
-            "outer-clean V05 plus frozen LT/sim identity and promoted "
-            "seasonal temperature <=2.50 V gate"
-        ),
-        "rows": len(risks),
-        "cases": len(cases),
-        "baseline_total": float(baseline.mean()),
-        "identity_total": float(identity.mean()),
-        "candidate_total": float(candidate.mean()),
-        "temperature_variants": temperature_variants,
-        "candidate_vs_baseline": {
-            "mean_delta": float(delta.mean()),
-            "paired_t": _paired_t(delta),
-            "first_half_delta": float(delta[:24].mean()),
-            "second_half_delta": float(delta[24:].mean()),
-            "improved": int((delta < 0).sum()),
-            "unchanged": int((delta == 0).sum()),
-            "worse": int((delta > 0).sum()),
-        },
-        "identity_vs_baseline": {
-            "mean_delta": float(identity_delta.mean()),
-            "paired_t": _paired_t(identity_delta),
-            "first_half_delta": float(identity_delta[:24].mean()),
-            "second_half_delta": float(identity_delta[24:].mean()),
-        },
-        "components": {
-            column: {
-                "baseline_mean": float(cases[f"baseline_{column}"].mean()),
-                "candidate_mean": float(
-                    cases[f"temperature_below_250_{column}"].mean()
-                ),
-                "mean_delta": float(
-                    (
-                        cases[f"temperature_below_250_{column}"]
-                        - cases[f"baseline_{column}"]
-                    ).mean()
-                ),
-            }
-            for column in COST_COLUMNS
-        },
-        "gate": {
-            "definition": "delta <= -40, both halves <= 0, paired_t <= -2",
-            "pass": bool(
-                delta.mean() <= -40.0
-                and delta[:24].mean() <= 0.0
-                and delta[24:].mean() <= 0.0
-                and _paired_t(delta) <= -2.0
-            ),
-        },
-        "v07_scale_counts": scale_counts,
-        "fingerprints": {
-            "clean_predictions": _sha256(args.clean_predictions),
-            "artifact": _sha256(args.artifact),
-            "lt_rows": _sha256(args.lt_rows),
-            "similarity_rows": _sha256(args.similarity_rows),
-            "temperature_rows": _sha256(args.temperature_rows),
-        },
-    }
-    args.output_prefix.parent.mkdir(parents=True, exist_ok=True)
-    cases.to_csv(args.output_prefix.with_suffix(".cases.csv"), index=False)
-    risks.to_csv(args.output_prefix.with_suffix(".rows.csv"), index=False)
-    args.output_prefix.with_suffix(".json").write_text(
-        json.dumps(report, indent=2), encoding="utf-8"
-    )
-    print(json.dumps(report, indent=2))
-
-
-if __name__ == "__main__":
-    main()

scripts/verify_identity_feature_parity.py

@@ -1,261 +0,0 @@
-"""Verify production feature extraction against frozen causal experiment rows.
-
-This is a release test, not a model fit. It replays every official train cut through
-the deployable raw-history cache and compares AFT, similarity-query, and rounded-beta
-seasonal physics features to the persisted experiment evidence.
-"""
-
-from __future__ import annotations
-
-import argparse
-import hashlib
-import json
-from pathlib import Path
-
-import numpy as np
-import pandas as pd
-from batteryswap_public.utils import iterate_scenarios, load_dataset
-
-from batteryswapai.identity_ensemble import (
-    FROZEN_TEMPERATURE_BETA_V_PER_C,
-    TEMPERATURE_MAX_PREDICTED_MIN_VOLTAGE,
-    CausalHistoryCache,
-    _first_passage_lifetime,
-    _query_similarity,
-    _seasonal_physics_signal,
-)
-
-
-KEYS = ["scenario", "battery"]
-
-
-def parse_args() -> argparse.Namespace:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("--dataset-path", type=Path, default=Path("data/raw/train"))
-    parser.add_argument(
-        "--lt-rows",
-        type=Path,
-        default=Path("artifacts/lt_fp_aft_official.rows.csv"),
-    )
-    parser.add_argument(
-        "--similarity-rows",
-        type=Path,
-        default=Path("artifacts/similarity_eol_official.rows.csv"),
-    )
-    parser.add_argument(
-        "--temperature-rows",
-        type=Path,
-        default=Path("artifacts/temperature_physics_globalbeta_diagnostic.rows.csv"),
-    )
-    parser.add_argument(
-        "--output",
-        type=Path,
-        default=Path("artifacts/identity_feature_parity.json"),
-    )
-    return parser.parse_args()
-
-
-def _sha256(path: Path) -> str:
-    return hashlib.sha256(path.read_bytes()).hexdigest()
-
-
-def _bool(values: pd.Series) -> np.ndarray:
-    if pd.api.types.is_bool_dtype(values):
-        return values.to_numpy(bool)
-    return values.astype(str).str.lower().isin({"true", "1"}).to_numpy(bool)
-
-
-def _max_abs(actual: np.ndarray, expected: np.ndarray) -> float:
-    finite = np.isfinite(actual) & np.isfinite(expected)
-    return float(np.max(np.abs(actual[finite] - expected[finite]))) if finite.any() else 0.0
-
-
-def main() -> None:
-    args = parse_args()
-    lag_columns = [f"sim_hi_lag_{lag}d" for lag in range(0, 337, 7)]
-    lt = pd.read_csv(
-        args.lt_rows,
-        usecols=KEYS
-        + ["installation_start", "fp_lifetime_days", "fp_reliable"],
-    ).set_index(KEYS)
-    similarity = pd.read_csv(
-        args.similarity_rows,
-        usecols=KEYS
-        + ["sim_query_ready", "sim_staleness_days", *lag_columns],
-    ).set_index(KEYS)
-    temperature = pd.read_csv(
-        args.temperature_rows,
-        usecols=KEYS + ["reliable", "predicted_min_smooth_voltage"],
-    ).set_index(KEYS)
-    if not (len(lt) == len(similarity) == len(temperature) == 19_890):
-        raise AssertionError("feature references must contain 19,890 rows")
-
-    locations, timeseries, eol_times, scenarios = load_dataset(args.dataset_path)
-    cache = CausalHistoryCache(
-        beta_v_per_c=FROZEN_TEMPERATURE_BETA_V_PER_C,
-        split_id="train",
-    )
-    aft_max_error = 0.0
-    similarity_max_error = 0.0
-    similarity_staleness_max_error = 0.0
-    temperature_max_error = 0.0
-    aft_reliable_matches = 0
-    similarity_ready_matches = 0
-    temperature_reliable_matches = 0
-    temperature_gate_matches = 0
-    rows = 0
-
-    for scenario, locs, visible, _ in iterate_scenarios(
-        locations, timeseries, eol_times, scenarios
-    ):
-        name = str(scenario["name"])
-        start = pd.Timestamp(scenario["start_time"])
-        history = cache.update(visible, start)
-        batteries = locs["battery"].astype(str).to_numpy()
-        keys = pd.MultiIndex.from_arrays(
-            [np.repeat(name, len(batteries)), batteries], names=KEYS
-        )
-        lt_expected = lt.reindex(keys)
-        sim_expected = similarity.reindex(keys)
-        temp_expected = temperature.reindex(keys)
-        if (
-            lt_expected.isna().all(axis=1).any()
-            or sim_expected.isna().all(axis=1).any()
-            or temp_expected.isna().all(axis=1).any()
-        ):
-            raise AssertionError(f"reference alignment failed for {name}")
-
-        actual_lifetime = np.full(len(batteries), np.nan)
-        actual_aft_reliable = np.zeros(len(batteries), dtype=bool)
-        installation = pd.to_datetime(lt_expected["installation_start"])
-        for position, battery in enumerate(batteries):
-            lifetime, reliable = _first_passage_lifetime(
-                battery,
-                start,
-                pd.Timestamp(installation.iloc[position]),
-                history.smooth_lookup,
-            )
-            actual_lifetime[position] = lifetime
-            actual_aft_reliable[position] = reliable
-        expected_lifetime = lt_expected["fp_lifetime_days"].to_numpy(float)
-        expected_aft_reliable = _bool(lt_expected["fp_reliable"])
-        np.testing.assert_allclose(
-            actual_lifetime,
-            expected_lifetime,
-            rtol=0.0,
-            atol=1e-12,
-            equal_nan=True,
-        )
-        np.testing.assert_array_equal(actual_aft_reliable, expected_aft_reliable)
-        aft_max_error = max(aft_max_error, _max_abs(actual_lifetime, expected_lifetime))
-        aft_reliable_matches += int((actual_aft_reliable == expected_aft_reliable).sum())
-
-        query, ready, staleness = _query_similarity(
-            batteries, start, history.smooth_lookup
-        )
-        expected_query = sim_expected[lag_columns].to_numpy(float)
-        expected_ready = _bool(sim_expected["sim_query_ready"])
-        expected_staleness = sim_expected["sim_staleness_days"].to_numpy(float)
-        np.testing.assert_allclose(
-            query, expected_query, rtol=0.0, atol=1e-12, equal_nan=True
-        )
-        np.testing.assert_array_equal(ready, expected_ready)
-        np.testing.assert_allclose(
-            staleness, expected_staleness, rtol=0.0, atol=1e-12, equal_nan=True
-        )
-        similarity_max_error = max(
-            similarity_max_error, _max_abs(query, expected_query)
-        )
-        similarity_staleness_max_error = max(
-            similarity_staleness_max_error,
-            _max_abs(staleness, expected_staleness),
-        )
-        similarity_ready_matches += int((ready == expected_ready).sum())
-
-        temp_signal = _seasonal_physics_signal(
-            history,
-            batteries,
-            start,
-            FROZEN_TEMPERATURE_BETA_V_PER_C,
-        )
-        expected_temp_reliable = _bool(temp_expected["reliable"])
-        expected_urgency = -temp_expected[
-            "predicted_min_smooth_voltage"
-        ].to_numpy(float)
-        np.testing.assert_array_equal(
-            temp_signal.reliable, expected_temp_reliable
-        )
-        np.testing.assert_allclose(
-            temp_signal.values,
-            expected_urgency,
-            rtol=0.0,
-            atol=2e-7,
-            equal_nan=True,
-        )
-        temperature_max_error = max(
-            temperature_max_error,
-            _max_abs(temp_signal.values, expected_urgency),
-        )
-        temperature_reliable_matches += int(
-            (temp_signal.reliable == expected_temp_reliable).sum()
-        )
-        gated_temp_signal = _seasonal_physics_signal(
-            history,
-            batteries,
-            start,
-            FROZEN_TEMPERATURE_BETA_V_PER_C,
-            TEMPERATURE_MAX_PREDICTED_MIN_VOLTAGE,
-        )
-        expected_gated_reliable = expected_temp_reliable & (
-            temp_expected["predicted_min_smooth_voltage"].to_numpy(float)
-            <= TEMPERATURE_MAX_PREDICTED_MIN_VOLTAGE
-        )
-        np.testing.assert_array_equal(
-            gated_temp_signal.reliable, expected_gated_reliable
-        )
-        temperature_gate_matches += int(
-            (gated_temp_signal.reliable == expected_gated_reliable).sum()
-        )
-        rows += len(batteries)
-
-    if rows != 19_890:
-        raise AssertionError(f"feature replay produced {rows} rows")
-    report = {
-        "rows": rows,
-        "all_feature_flags_match": bool(
-            aft_reliable_matches
-            == similarity_ready_matches
-            == temperature_reliable_matches
-            == temperature_gate_matches
-            == rows
-        ),
-        "aft": {
-            "reliable_flag_matches": aft_reliable_matches,
-            "maximum_absolute_lifetime_error_days": aft_max_error,
-        },
-        "similarity_query": {
-            "ready_flag_matches": similarity_ready_matches,
-            "maximum_absolute_prefix_error": similarity_max_error,
-            "maximum_absolute_staleness_error_days": similarity_staleness_max_error,
-        },
-        "seasonal_temperature": {
-            "reliable_flag_matches": temperature_reliable_matches,
-            "below_250_gate_flag_matches": temperature_gate_matches,
-            "maximum_predicted_min_voltage": (
-                TEMPERATURE_MAX_PREDICTED_MIN_VOLTAGE
-            ),
-            "maximum_absolute_urgency_error_v": temperature_max_error,
-        },
-        "evidence_sha256": {
-            "lt_rows": _sha256(args.lt_rows),
-            "similarity_rows": _sha256(args.similarity_rows),
-            "temperature_rows": _sha256(args.temperature_rows),
-        },
-    }
-    args.output.parent.mkdir(parents=True, exist_ok=True)
-    args.output.write_text(json.dumps(report, indent=2), encoding="utf-8")
-    print(json.dumps(report, indent=2))
-
-
-if __name__ == "__main__":
-    main()