Submission 010: weekly capacity margin on the 008 stack

script.py

@@ -45,7 +45,7 @@ def main() -> None:
     artifact_path = Path(
         os.environ.get(
             "BATTERYSWAP_PLANNER_PATH",
-            "submission_artifacts/final_planner.joblib",
+            "submission_artifacts/weekly_planner.joblib",
         )
     )
     splits = [

src/batteryswapai/audit.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+from .schema import InferredSchema, leakage_columns
+
+
+def _safe_quantiles(series: pd.Series) -> dict[str, float | None]:
+    values = pd.to_numeric(series, errors="coerce").dropna()
+    if values.empty:
+        return {"p01": None, "p10": None, "p50": None, "p90": None, "p99": None}
+    quantiles = values.quantile([0.01, 0.10, 0.50, 0.90, 0.99])
+    return {
+        "p01": float(quantiles.loc[0.01]),
+        "p10": float(quantiles.loc[0.10]),
+        "p50": float(quantiles.loc[0.50]),
+        "p90": float(quantiles.loc[0.90]),
+        "p99": float(quantiles.loc[0.99]),
+    }
+
+
+def audit_longitudinal_table(df: pd.DataFrame, schema: InferredSchema) -> dict:
+    report: dict = {
+        "rows": int(len(df)),
+        "columns": int(len(df.columns)),
+        "possible_leakage_columns": leakage_columns(df, schema.target),
+    }
+
+    if schema.entity_id:
+        counts = df.groupby(schema.entity_id, dropna=False).size()
+        report["entities"] = {
+            "count": int(counts.size),
+            "rows_per_entity": _safe_quantiles(counts),
+        }
+
+    if schema.timestamp:
+        timestamps = pd.to_datetime(df[schema.timestamp], errors="coerce", utc=True)
+        valid = timestamps.dropna()
+        if not valid.empty:
+            report["time"] = {
+                "start": valid.min().isoformat(),
+                "end": valid.max().isoformat(),
+                "span_days": float((valid.max() - valid.min()).total_seconds() / 86400.0),
+            }
+
+        if schema.entity_id:
+            ordered = pd.DataFrame({"entity": df[schema.entity_id], "time": timestamps}).dropna()
+            ordered = ordered.sort_values(["entity", "time"])
+            cadence = ordered.groupby("entity", sort=False)["time"].diff().dt.total_seconds() / 3600.0
+            report.setdefault("time", {})["cadence_hours"] = _safe_quantiles(cadence)
+
+    if schema.target:
+        target = pd.to_numeric(df[schema.target], errors="coerce")
+        report["target"] = {
+            "non_null": int(target.notna().sum()),
+            "quantiles": _safe_quantiles(target),
+            "negative_count": int((target < 0).sum()),
+            "zero_count": int((target == 0).sum()),
+        }
+        if schema.entity_id and schema.timestamp:
+            temp = pd.DataFrame(
+                {
+                    "entity": df[schema.entity_id],
+                    "time": pd.to_datetime(df[schema.timestamp], errors="coerce", utc=True),
+                    "target": target,
+                }
+            ).dropna().sort_values(["entity", "time"])
+            delta = temp.groupby("entity", sort=False)["target"].diff()
+            report["target"]["delta_quantiles"] = _safe_quantiles(delta)
+            report["target"]["fraction_increasing"] = float((delta > 1e-9).mean()) if delta.notna().any() else None
+
+    if schema.voltage:
+        voltage = pd.to_numeric(df[schema.voltage], errors="coerce")
+        report["voltage"] = {"quantiles": _safe_quantiles(voltage)}
+        if schema.entity_id and schema.timestamp:
+            temp = pd.DataFrame(
+                {
+                    "entity": df[schema.entity_id],
+                    "time": pd.to_datetime(df[schema.timestamp], errors="coerce", utc=True),
+                    "voltage": voltage,
+                }
+            ).dropna().sort_values(["entity", "time"])
+            jumps = temp.groupby("entity", sort=False)["voltage"].diff()
+            report["voltage"]["jump_quantiles"] = _safe_quantiles(jumps)
+            report["voltage"]["jumps_over_0_25v"] = int((jumps >= 0.25).sum())
+            report["voltage"]["jumps_over_0_35v"] = int((jumps >= 0.35).sum())
+
+    if schema.site_id:
+        report["sites"] = int(df[schema.site_id].nunique(dropna=True))
+
+    return report

src/batteryswapai/cli.py

@@ -0,0 +1,77 @@
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from .audit import audit_longitudinal_table
+from .config import load_config
+from .io import download_dataset, load_named_table
+from .pipeline import inspect_dataset, run_cv
+from .schema import resolve_schema
+from .synthetic import make_synthetic_dataset
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="bsa")
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    download = sub.add_parser("download", help="Download the public Hugging Face dataset")
+    download.add_argument("--config", default="configs/default.yaml")
+
+    inspect = sub.add_parser("inspect", help="Inspect local tables and infer semantic columns")
+    inspect.add_argument("--config", default="configs/default.yaml")
+    inspect.add_argument("--table", default=None)
+
+    cv = sub.add_parser("cv", help="Run leakage-safe grouped cross validation")
+    cv.add_argument("--config", default="configs/default.yaml")
+    cv.add_argument("--table", default=None)
+
+    audit = sub.add_parser("audit", help="Audit longitudinal structure, cadence, target and voltage resets")
+    audit.add_argument("--config", default="configs/default.yaml")
+    audit.add_argument("--table", default=None)
+
+    synthetic = sub.add_parser("synthetic", help="Generate a local smoke-test dataset")
+    synthetic.add_argument("--output", default="data/raw/synthetic.csv")
+    synthetic.add_argument("--devices", type=int, default=24)
+    synthetic.add_argument("--days", type=int, default=180)
+    return parser
+
+
+def main() -> None:
+    args = _parser().parse_args()
+    if args.command == "download":
+        config = load_config(args.config)
+        path = download_dataset(
+            config["dataset"]["repo_id"],
+            config["dataset"]["data_dir"],
+            config["dataset"].get("revision", "main"),
+        )
+        print(path)
+        return
+
+    if args.command == "inspect":
+        print(json.dumps(inspect_dataset(args.config, args.table), indent=2, default=str))
+        return
+
+    if args.command == "cv":
+        print(json.dumps(run_cv(args.config, args.table), indent=2))
+        return
+
+    if args.command == "audit":
+        config = load_config(args.config)
+        _, frame = load_named_table(config["dataset"]["data_dir"], args.table)
+        schema = resolve_schema(frame, config["schema"])
+        print(json.dumps(audit_longitudinal_table(frame, schema), indent=2, default=str))
+        return
+
+    if args.command == "synthetic":
+        output = Path(args.output)
+        output.parent.mkdir(parents=True, exist_ok=True)
+        make_synthetic_dataset(args.devices, args.days).to_csv(output, index=False)
+        print(output)
+        return
+
+
+if __name__ == "__main__":
+    main()

src/batteryswapai/config.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+
+def load_config(path: str | Path) -> dict[str, Any]:
+    with Path(path).open("r", encoding="utf-8") as handle:
+        return yaml.safe_load(handle)

src/batteryswapai/features.py

@@ -0,0 +1,225 @@
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+
+import numpy as np
+import pandas as pd
+
+from .schema import leakage_columns
+
+
+SIGNAL_NAME_WEIGHTS = (
+    "battery",
+    "voltage",
+    "vbat",
+    "temperature",
+    "temp",
+    "rssi",
+    "signal",
+    "current",
+    "power",
+    "humidity",
+)
+
+
+@dataclass(frozen=True)
+class FeatureResult:
+    frame: pd.DataFrame
+    feature_columns: list[str]
+
+
+def _signal_score(name: str) -> int:
+    lowered = name.lower()
+    return sum(max(1, 20 - index) for index, token in enumerate(SIGNAL_NAME_WEIGHTS) if token in lowered)
+
+
+def choose_numeric_signals(
+    df: pd.DataFrame,
+    exclude: set[str],
+    max_signals: int,
+) -> list[str]:
+    candidates = []
+    for column in df.columns:
+        if column in exclude or not pd.api.types.is_numeric_dtype(df[column]):
+            continue
+        variance = float(pd.to_numeric(df[column], errors="coerce").var(skipna=True) or 0.0)
+        if not np.isfinite(variance) or variance <= 0:
+            continue
+        candidates.append((column, _signal_score(column), variance))
+    candidates.sort(key=lambda item: (item[1], math.log1p(item[2])), reverse=True)
+    return [column for column, _, _ in candidates[:max_signals]]
+
+
+def daily_panel(
+    df: pd.DataFrame,
+    entity_col: str,
+    timestamp_col: str,
+    numeric_signals: list[str],
+    passthrough_cols: list[str] | None = None,
+) -> pd.DataFrame:
+    passthrough_cols = passthrough_cols or []
+    frame = df.copy()
+    frame[timestamp_col] = pd.to_datetime(frame[timestamp_col], errors="coerce", utc=True)
+    frame = frame.dropna(subset=[entity_col, timestamp_col]).sort_values([entity_col, timestamp_col])
+    frame["_day"] = frame[timestamp_col].dt.floor("D")
+
+    aggregation: dict[str, str] = {column: "mean" for column in numeric_signals}
+    for column in passthrough_cols:
+        if column in frame.columns and column not in aggregation:
+            aggregation[column] = "last"
+    panel = frame.groupby([entity_col, "_day"], as_index=False, sort=False).agg(aggregation)
+    counts = frame.groupby([entity_col, "_day"], sort=False).size().rename("observations_day").reset_index()
+    panel = panel.merge(counts, on=[entity_col, "_day"], how="left")
+    return panel.rename(columns={"_day": timestamp_col})
+
+
+def _add_calendar_lags(
+    panel: pd.DataFrame,
+    entity_col: str,
+    timestamp_col: str,
+    signals: list[str],
+    lags_days: tuple[int, ...],
+) -> tuple[pd.DataFrame, list[str]]:
+    result = panel.copy()
+    feature_columns: list[str] = []
+    right = result[[entity_col, timestamp_col, *signals]].copy()
+    right = right.rename(columns={timestamp_col: "_source_time"})
+    right = right.sort_values(["_source_time", entity_col])
+
+    for lag in lags_days:
+        left = result[[entity_col, timestamp_col]].copy()
+        left["_row_id"] = np.arange(len(left))
+        left["_lookup_time"] = left[timestamp_col] - pd.Timedelta(days=lag)
+        left = left.sort_values(["_lookup_time", entity_col])
+        tolerance_days = max(1, min(3, lag // 7 if lag >= 7 else 1))
+
+        matched = pd.merge_asof(
+            left,
+            right,
+            left_on="_lookup_time",
+            right_on="_source_time",
+            by=entity_col,
+            direction="backward",
+            tolerance=pd.Timedelta(days=tolerance_days),
+        ).sort_values("_row_id")
+
+        actual_age_col = f"lag_age_{lag}d"
+        source_time = pd.to_datetime(matched["_source_time"], errors="coerce", utc=True).reset_index(drop=True)
+        current_time = result[timestamp_col].reset_index(drop=True)
+        result[actual_age_col] = (current_time - source_time).dt.total_seconds() / 86400.0
+        feature_columns.append(actual_age_col)
+
+        for signal in signals:
+            lag_col = f"{signal}__lag_{lag}d"
+            delta_col = f"{signal}__delta_{lag}d"
+            rate_col = f"{signal}__rate_{lag}d"
+            result[lag_col] = matched[signal].to_numpy()
+            result[delta_col] = result[signal] - result[lag_col]
+            denominator = result[actual_age_col].replace(0.0, np.nan)
+            result[rate_col] = result[delta_col] / denominator
+            feature_columns.extend([lag_col, delta_col, rate_col])
+
+    return result, feature_columns
+
+
+def _add_time_rolling_features(
+    panel: pd.DataFrame,
+    entity_col: str,
+    timestamp_col: str,
+    signals: list[str],
+    windows_days: tuple[int, ...],
+) -> tuple[pd.DataFrame, list[str]]:
+    result = panel.copy()
+    feature_columns: list[str] = []
+    index = pd.MultiIndex.from_arrays(
+        [result[entity_col], result[timestamp_col]],
+        names=[entity_col, timestamp_col],
+    )
+
+    for window in windows_days:
+        min_periods = max(2, window // 3)
+        stats = (
+            result.groupby(entity_col, sort=False)
+            .rolling(f"{window}D", on=timestamp_col, min_periods=min_periods)[signals]
+            .agg(["mean", "std", "min", "max"])
+        )
+        for signal in signals:
+            for stat in ("mean", "std", "min", "max"):
+                column = f"{signal}__{stat}_{window}d"
+                result[column] = stats[(signal, stat)].reindex(index).to_numpy()
+                feature_columns.append(column)
+
+    return result, feature_columns
+
+
+def build_features(
+    df: pd.DataFrame,
+    entity_col: str,
+    timestamp_col: str,
+    target_col: str | None = None,
+    site_col: str | None = None,
+    max_signals: int = 12,
+    lags_days: tuple[int, ...] = (1, 3, 7, 14, 30, 60, 90),
+    windows_days: tuple[int, ...] = (3, 7, 14, 30, 60, 90),
+    min_history_days: int = 7,
+    include_entity_id: bool = False,
+    include_site_id: bool = True,
+) -> FeatureResult:
+    blocked = set(leakage_columns(df, target_col)) | {entity_col, timestamp_col}
+    if site_col:
+        blocked.add(site_col)
+    signals = choose_numeric_signals(df, blocked, max_signals=max_signals)
+    if not signals:
+        raise ValueError("No usable numeric sensor signals were found")
+
+    passthrough = [column for column in (target_col, site_col) if column]
+    panel = daily_panel(df, entity_col, timestamp_col, signals, passthrough)
+    panel = panel.sort_values([entity_col, timestamp_col]).reset_index(drop=True)
+
+    grouped = panel.groupby(entity_col, sort=False)
+    first_seen = grouped[timestamp_col].transform("min")
+    panel["age_days"] = (panel[timestamp_col] - first_seen).dt.total_seconds() / 86400.0
+    panel["days_since_prev_observation"] = (
+        grouped[timestamp_col].diff().dt.total_seconds() / 86400.0
+    )
+    panel["dow"] = panel[timestamp_col].dt.dayofweek.astype("int8")
+    panel["month_sin"] = np.sin(2 * np.pi * panel[timestamp_col].dt.month / 12.0)
+    panel["month_cos"] = np.cos(2 * np.pi * panel[timestamp_col].dt.month / 12.0)
+
+    feature_columns = [
+        "age_days",
+        "days_since_prev_observation",
+        "dow",
+        "month_sin",
+        "month_cos",
+        "observations_day",
+        *signals,
+    ]
+
+    panel, lag_features = _add_calendar_lags(
+        panel,
+        entity_col=entity_col,
+        timestamp_col=timestamp_col,
+        signals=signals,
+        lags_days=lags_days,
+    )
+    feature_columns.extend(lag_features)
+
+    panel, rolling_features = _add_time_rolling_features(
+        panel,
+        entity_col=entity_col,
+        timestamp_col=timestamp_col,
+        signals=signals,
+        windows_days=windows_days,
+    )
+    feature_columns.extend(rolling_features)
+
+    if include_entity_id:
+        feature_columns.append(entity_col)
+    if include_site_id and site_col and site_col in panel.columns:
+        feature_columns.append(site_col)
+
+    panel = panel[panel["age_days"] >= min_history_days].copy()
+    feature_columns = [column for column in feature_columns if column in panel.columns]
+    return FeatureResult(frame=panel, feature_columns=feature_columns)

src/batteryswapai/hierarchical_fpt.py

@@ -30,10 +30,7 @@ EOL_VOLTAGE = 2.40
 TERMINAL_WEEKS = 12
 MIN_WEEKLY_INCREMENTS = 4
 MAX_STATE_STALENESS_DAYS = 7.0
-RANK_RESIDUAL_WEIGHT = 0.25
-# Codex froze this at 0.15 with no_sweep=true, so the one dial on the mechanism
-# that actually works had never been tested.  Over 48 dates x 27 bases, 0.25 is
-# -35.16 against 0.15, both calendar halves negative and all 27 bases improving.
+RANK_RESIDUAL_WEIGHT = 0.15
 
 
 @dataclass(frozen=True)

src/batteryswapai/io.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pandas as pd
+from huggingface_hub import snapshot_download
+
+SUPPORTED_SUFFIXES = {".csv", ".parquet", ".json", ".jsonl", ".feather"}
+
+
+def download_dataset(repo_id: str, data_dir: str | Path, revision: str = "main") -> Path:
+    destination = Path(data_dir)
+    destination.mkdir(parents=True, exist_ok=True)
+    snapshot_download(
+        repo_id=repo_id,
+        repo_type="dataset",
+        revision=revision,
+        local_dir=destination,
+    )
+    return destination
+
+
+def discover_tables(data_dir: str | Path) -> list[Path]:
+    root = Path(data_dir)
+    if not root.exists():
+        return []
+    return sorted(path for path in root.rglob("*") if path.is_file() and path.suffix.lower() in SUPPORTED_SUFFIXES)
+
+
+def read_table(path: str | Path) -> pd.DataFrame:
+    path = Path(path)
+    suffix = path.suffix.lower()
+    if suffix == ".csv":
+        return pd.read_csv(path)
+    if suffix == ".parquet":
+        return pd.read_parquet(path)
+    if suffix == ".json":
+        try:
+            return pd.read_json(path)
+        except ValueError:
+            return pd.read_json(path, lines=True)
+    if suffix == ".jsonl":
+        return pd.read_json(path, lines=True)
+    if suffix == ".feather":
+        return pd.read_feather(path)
+    raise ValueError(f"Unsupported data file: {path}")
+
+
+def load_named_table(data_dir: str | Path, name_hint: str | None = None) -> tuple[Path, pd.DataFrame]:
+    tables = discover_tables(data_dir)
+    if not tables:
+        raise FileNotFoundError(f"No supported tables found under {data_dir}")
+
+    if name_hint:
+        matches = [path for path in tables if name_hint.lower() in path.name.lower()]
+        if len(matches) == 1:
+            return matches[0], read_table(matches[0])
+        if len(matches) > 1:
+            matches.sort(key=lambda path: path.stat().st_size, reverse=True)
+            return matches[0], read_table(matches[0])
+
+    tables.sort(key=lambda path: path.stat().st_size, reverse=True)
+    return tables[0], read_table(tables[0])

src/batteryswapai/lifecycles.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+
+def assign_lifecycles(
+    df: pd.DataFrame,
+    entity_col: str,
+    timestamp_col: str,
+    voltage_col: str,
+    reset_jump_volts: float = 0.35,
+    min_points: int = 14,
+    min_duration_days: int = 7,
+) -> pd.DataFrame:
+    frame = df.copy()
+    frame[timestamp_col] = pd.to_datetime(frame[timestamp_col], errors="coerce", utc=True)
+    frame = frame.dropna(subset=[entity_col, timestamp_col]).sort_values([entity_col, timestamp_col])
+
+    voltage = pd.to_numeric(frame[voltage_col], errors="coerce")
+    jump = voltage.groupby(frame[entity_col], sort=False).diff()
+    replacement = jump >= reset_jump_volts
+    episode = replacement.groupby(frame[entity_col], sort=False).cumsum().astype("int64")
+    frame["lifecycle_id"] = frame[entity_col].astype(str) + "::" + episode.astype(str)
+
+    stats = frame.groupby("lifecycle_id", sort=False).agg(
+        lifecycle_points=(timestamp_col, "size"),
+        lifecycle_start=(timestamp_col, "min"),
+        lifecycle_end=(timestamp_col, "max"),
+    )
+    stats["lifecycle_duration_days"] = (
+        (stats["lifecycle_end"] - stats["lifecycle_start"]).dt.total_seconds() / 86400.0
+    )
+    valid_ids = stats.index[
+        (stats["lifecycle_points"] >= min_points)
+        & (stats["lifecycle_duration_days"] >= min_duration_days)
+    ]
+    return frame[frame["lifecycle_id"].isin(valid_ids)].copy()
+
+
+def add_rul_labels(
+    df: pd.DataFrame,
+    entity_col: str,
+    timestamp_col: str,
+    lifecycle_col: str = "lifecycle_id",
+    censor_last_episode: bool = True,
+) -> pd.DataFrame:
+    frame = df.copy()
+    frame[timestamp_col] = pd.to_datetime(frame[timestamp_col], errors="coerce", utc=True)
+    ends = frame.groupby(lifecycle_col, sort=False)[timestamp_col].transform("max")
+    frame["rul_days"] = (ends - frame[timestamp_col]).dt.total_seconds() / 86400.0
+    frame["event_observed"] = 1
+
+    if censor_last_episode:
+        last_lifecycle = (
+            frame.groupby(entity_col, sort=False)[lifecycle_col]
+            .last()
+            .astype(str)
+            .to_dict()
+        )
+        is_last = frame.apply(lambda row: str(row[lifecycle_col]) == last_lifecycle.get(row[entity_col]), axis=1)
+        frame.loc[is_last, "event_observed"] = 0
+        frame.loc[is_last, "rul_days"] = np.nan
+
+    return frame

src/batteryswapai/metrics.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+import numpy as np
+from sklearn.metrics import mean_absolute_error, mean_squared_error
+
+
+def regression_metrics(y_true, y_pred) -> dict[str, float]:
+    y_true = np.asarray(y_true, dtype=float)
+    y_pred = np.asarray(y_pred, dtype=float)
+    return {
+        "mae": float(mean_absolute_error(y_true, y_pred)),
+        "rmse": float(np.sqrt(mean_squared_error(y_true, y_pred))),
+        "bias": float(np.mean(y_pred - y_true)),
+        "p90_abs_error": float(np.quantile(np.abs(y_pred - y_true), 0.90)),
+    }

src/batteryswapai/models.py

@@ -0,0 +1,151 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+import numpy as np
+import pandas as pd
+from sklearn.ensemble import ExtraTreesRegressor, HistGradientBoostingRegressor
+
+from .metrics import regression_metrics
+from .validation import group_folds
+
+
+@dataclass
+class TrainedEnsemble:
+    histogram: HistGradientBoostingRegressor
+    extra_trees: ExtraTreesRegressor
+    feature_columns: list[str]
+    categorical_columns: list[str]
+    encoded_columns: list[str]
+    fill_values: pd.Series
+    weights: tuple[float, float] = (0.5, 0.5)
+
+    def predict(self, frame: pd.DataFrame) -> np.ndarray:
+        features, _, _ = encode_features(
+            frame[self.feature_columns],
+            self.categorical_columns,
+            encoded_columns=self.encoded_columns,
+            fill_values=self.fill_values,
+        )
+        histogram_prediction = self.histogram.predict(features)
+        tree_prediction = self.extra_trees.predict(features)
+        prediction = self.weights[0] * histogram_prediction + self.weights[1] * tree_prediction
+        return np.clip(prediction, 0.0, None)
+
+
+def _categorical_columns(frame: pd.DataFrame, feature_columns: list[str]) -> list[str]:
+    categorical = []
+    for column in feature_columns:
+        dtype = frame[column].dtype
+        if (
+            pd.api.types.is_object_dtype(dtype)
+            or pd.api.types.is_string_dtype(dtype)
+            or isinstance(dtype, pd.CategoricalDtype)
+        ):
+            categorical.append(column)
+    return categorical
+
+
+def encode_features(
+    frame: pd.DataFrame,
+    categorical_columns: list[str],
+    *,
+    encoded_columns: list[str] | None = None,
+    fill_values: pd.Series | None = None,
+) -> tuple[pd.DataFrame, list[str], pd.Series]:
+    result = frame.copy()
+    for column in categorical_columns:
+        if column in result:
+            result[column] = result[column].astype("string").fillna("__missing__")
+    result = pd.get_dummies(result, columns=categorical_columns, dummy_na=False, dtype=float)
+    result = result.apply(pd.to_numeric, errors="coerce").replace([np.inf, -np.inf], np.nan)
+    if encoded_columns is None:
+        encoded_columns = result.columns.tolist()
+    result = result.reindex(columns=encoded_columns)
+    if fill_values is None:
+        fill_values = result.median(numeric_only=True).fillna(0.0)
+    result = result.fillna(fill_values).fillna(0.0).astype(float)
+    return result, encoded_columns, fill_values
+
+
+def _estimators(model_config: dict[str, Any], seed_offset: int = 0):
+    histogram_config = {
+        "loss": "absolute_error",
+        "learning_rate": 0.055,
+        "max_iter": 320,
+        "max_leaf_nodes": 31,
+        "min_samples_leaf": 25,
+        "l2_regularization": 2.0,
+        "random_state": 2026 + seed_offset,
+        **model_config.get("hist_gradient_boosting", {}),
+    }
+    extra_config = {
+        "n_estimators": 280,
+        "min_samples_leaf": 3,
+        "max_features": 0.8,
+        "n_jobs": -1,
+        "random_state": 2026 + seed_offset,
+        **model_config.get("extra_trees", {}),
+    }
+    return HistGradientBoostingRegressor(**histogram_config), ExtraTreesRegressor(**extra_config)
+
+
+def cross_validate(
+    frame: pd.DataFrame,
+    feature_columns: list[str],
+    target_col: str,
+    group_col: str,
+    model_config: dict[str, Any],
+    folds: int = 5,
+) -> tuple[pd.DataFrame, dict[str, float]]:
+    work = frame.dropna(subset=[target_col]).reset_index(drop=True).copy()
+    categorical = _categorical_columns(work, feature_columns)
+    target = pd.to_numeric(work[target_col], errors="coerce").astype(float)
+
+    oof = np.full(len(work), np.nan, dtype=float)
+    for fold_number, fold in enumerate(group_folds(work, group_col=group_col, folds=folds)):
+        histogram, extra_trees = _estimators(model_config, seed_offset=fold_number)
+        train_frame = work.iloc[fold.train_idx][feature_columns]
+        valid_frame = work.iloc[fold.valid_idx][feature_columns]
+        x_train, encoded_columns, fill_values = encode_features(train_frame, categorical)
+        x_valid, _, _ = encode_features(
+            valid_frame,
+            categorical,
+            encoded_columns=encoded_columns,
+            fill_values=fill_values,
+        )
+        y_train = target.iloc[fold.train_idx]
+        histogram.fit(x_train, y_train)
+        extra_trees.fit(x_train, y_train)
+        prediction = 0.5 * histogram.predict(x_valid) + 0.5 * extra_trees.predict(x_valid)
+        oof[fold.valid_idx] = np.clip(prediction, 0.0, None)
+
+    metrics = regression_metrics(target, oof)
+    output = work[[target_col, group_col]].copy()
+    output["prediction"] = oof
+    output["abs_error"] = np.abs(output["prediction"] - output[target_col])
+    return output, metrics
+
+
+def train_ensemble(
+    frame: pd.DataFrame,
+    feature_columns: list[str],
+    target_col: str,
+    model_config: dict[str, Any],
+) -> TrainedEnsemble:
+    work = frame.dropna(subset=[target_col]).copy()
+    categorical = _categorical_columns(work, feature_columns)
+    features, encoded_columns, fill_values = encode_features(work[feature_columns], categorical)
+    target = pd.to_numeric(work[target_col], errors="coerce").astype(float)
+    histogram, extra_trees = _estimators(model_config)
+    histogram.fit(features, target)
+    extra_trees.fit(features, target)
+    return TrainedEnsemble(
+        histogram,
+        extra_trees,
+        feature_columns,
+        categorical,
+        encoded_columns,
+        fill_values,
+    )

src/batteryswapai/pipeline.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pandas as pd
+
+from .config import load_config
+from .features import build_features
+from .io import load_named_table
+from .models import cross_validate, train_ensemble
+from .schema import describe_table, leakage_columns, resolve_schema
+
+
+def inspect_dataset(config_path: str | Path, table_hint: str | None = None) -> dict:
+    config = load_config(config_path)
+    path, frame = load_named_table(config["dataset"]["data_dir"], table_hint)
+    schema = resolve_schema(frame, config["schema"])
+    report = {
+        "file": str(path),
+        "table": describe_table(frame),
+        "schema": schema.as_dict(),
+        "possible_leakage_columns": leakage_columns(frame, schema.target),
+    }
+    return report
+
+
+def prepare_training_frame(config_path: str | Path, table_hint: str | None = None):
+    config = load_config(config_path)
+    path, frame = load_named_table(config["dataset"]["data_dir"], table_hint)
+    schema = resolve_schema(frame, config["schema"])
+    if not schema.timestamp or not schema.entity_id:
+        raise ValueError("Could not infer timestamp and entity ID. Set them in configs/default.yaml")
+    if not schema.target:
+        raise ValueError("Could not infer an RUL target. Set schema.target or construct lifecycle labels first")
+
+    feature_cfg = config["features"]
+    result = build_features(
+        frame,
+        entity_col=schema.entity_id,
+        timestamp_col=schema.timestamp,
+        target_col=schema.target,
+        site_col=schema.site_id,
+        max_signals=int(feature_cfg["max_signals"]),
+        lags_days=tuple(feature_cfg["lags_days"]),
+        windows_days=tuple(feature_cfg["windows_days"]),
+        min_history_days=int(feature_cfg["min_history_days"]),
+        include_entity_id=bool(feature_cfg["include_entity_id"]),
+        include_site_id=bool(feature_cfg["include_site_id"]),
+    )
+    return config, path, schema, result
+
+
+def run_cv(config_path: str | Path, table_hint: str | None = None) -> dict:
+    config, _, schema, result = prepare_training_frame(config_path, table_hint)
+    oof, metrics = cross_validate(
+        result.frame,
+        result.feature_columns,
+        target_col=schema.target,
+        group_col=schema.entity_id,
+        model_config=config["model"],
+        folds=int(config["validation"]["folds"]),
+    )
+    artifacts = Path("artifacts")
+    artifacts.mkdir(exist_ok=True)
+    oof.to_csv(artifacts / "oof_predictions.csv", index=False)
+    (artifacts / "cv_metrics.json").write_text(json.dumps(metrics, indent=2), encoding="utf-8")
+    return metrics
+
+
+def fit_final(config_path: str | Path, table_hint: str | None = None):
+    config, _, schema, result = prepare_training_frame(config_path, table_hint)
+    model = train_ensemble(result.frame, result.feature_columns, schema.target, config["model"])
+    return model, schema, result

src/batteryswapai/scheduling.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import numpy as np
+import pandas as pd
+
+
+@dataclass(frozen=True)
+class ScheduleConfig:
+    failure_cost: float = 100.0
+    early_replacement_cost_per_day: float = 0.08
+    visit_cost: float = 15.0
+    max_replacements_per_day: int = 25
+    horizon_days: int = 120
+
+
+def expected_replacement_cost(
+    replacement_day: int,
+    failure_samples: np.ndarray,
+    config: ScheduleConfig,
+) -> float:
+    failures_before_visit = failure_samples < replacement_day
+    failure_cost = config.failure_cost * failures_before_visit.mean()
+    unused_life = np.maximum(failure_samples - replacement_day, 0.0)
+    early_cost = config.early_replacement_cost_per_day * unused_life.mean()
+    return float(failure_cost + early_cost)
+
+
+def choose_replacement_days(
+    predictions: pd.DataFrame,
+    id_col: str,
+    median_col: str = "rul_days",
+    uncertainty_fraction: float = 0.25,
+    config: ScheduleConfig | None = None,
+    seed: int = 2026,
+) -> pd.DataFrame:
+    config = config or ScheduleConfig()
+    rng = np.random.default_rng(seed)
+    rows = []
+
+    for _, row in predictions.iterrows():
+        median = max(float(row[median_col]), 0.1)
+        sigma = max(median * uncertainty_fraction, 1.0)
+        samples = np.clip(rng.normal(median, sigma, size=2000), 0.0, None)
+        costs = [expected_replacement_cost(day, samples, config) for day in range(config.horizon_days + 1)]
+        best_day = int(np.argmin(costs))
+        rows.append({id_col: row[id_col], "replacement_day": best_day, "expected_risk_cost": costs[best_day]})
+
+    schedule = pd.DataFrame(rows).sort_values(["replacement_day", "expected_risk_cost"], ascending=[True, False])
+    scheduled = []
+    day_load: dict[int, int] = {}
+    for _, row in schedule.iterrows():
+        day = int(row["replacement_day"])
+        while day_load.get(day, 0) >= config.max_replacements_per_day and day < config.horizon_days:
+            day += 1
+        day_load[day] = day_load.get(day, 0) + 1
+        record = row.to_dict()
+        record["replacement_day"] = day
+        scheduled.append(record)
+    return pd.DataFrame(scheduled).sort_values("replacement_day").reset_index(drop=True)
+
+
+def group_site_visits(
+    schedule: pd.DataFrame,
+    site_col: str,
+    tolerance_days: int = 3,
+) -> pd.DataFrame:
+    if site_col not in schedule.columns:
+        return schedule
+    result = schedule.copy().sort_values([site_col, "replacement_day"])
+    for _, index in result.groupby(site_col, sort=False).groups.items():
+        positions = list(index)
+        anchor = int(result.loc[positions[0], "replacement_day"])
+        for position in positions[1:]:
+            current = int(result.loc[position, "replacement_day"])
+            if current - anchor <= tolerance_days:
+                result.loc[position, "replacement_day"] = anchor
+            else:
+                anchor = current
+    return result.sort_values("replacement_day").reset_index(drop=True)

src/batteryswapai/schema.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, asdict
+from typing import Iterable
+
+import numpy as np
+import pandas as pd
+
+
+ROLE_HINTS = {
+    "timestamp": ("timestamp", "datetime", "recorded_at", "created_at", "time", "date"),
+    "entity_id": ("sensor_id", "device_id", "serial_number", "serial", "sensor", "device", "uuid", "mac"),
+    "target": ("remaining_useful_life", "remaining_life", "rul_days", "rul", "remaining_days", "target"),
+    "voltage": ("battery_voltage", "voltage", "vbat", "battery_v", "volt"),
+    "site_id": ("building_id", "site_id", "location_id", "building", "site", "location"),
+    "event": ("battery_replaced", "replacement", "replaced", "failure", "failed", "event"),
+}
+
+FUTURE_NAME_FRAGMENTS = (
+    "rul",
+    "remaining",
+    "future",
+    "failure_time",
+    "failure_date",
+    "end_time",
+    "end_date",
+    "next_replacement",
+)
+
+
+def _norm(value: str) -> str:
+    return "".join(char.lower() if char.isalnum() else "_" for char in value).strip("_")
+
+
+def _score_name(column: str, hints: Iterable[str]) -> int:
+    normalized = _norm(column)
+    score = 0
+    for rank, hint in enumerate(hints):
+        if normalized == hint:
+            score = max(score, 100 - rank)
+        elif hint in normalized:
+            score = max(score, 60 - rank)
+    return score
+
+
+@dataclass(frozen=True)
+class InferredSchema:
+    timestamp: str | None
+    entity_id: str | None
+    target: str | None
+    voltage: str | None
+    site_id: str | None
+    event: str | None
+
+    def as_dict(self) -> dict[str, str | None]:
+        return asdict(self)
+
+
+def infer_schema(df: pd.DataFrame) -> InferredSchema:
+    candidates: dict[str, str | None] = {}
+    for role, hints in ROLE_HINTS.items():
+        ranked = sorted(
+            ((column, _score_name(column, hints)) for column in df.columns),
+            key=lambda item: item[1],
+            reverse=True,
+        )
+        candidates[role] = ranked[0][0] if ranked and ranked[0][1] > 0 else None
+
+    if candidates["timestamp"] is None:
+        for column in df.columns:
+            if pd.api.types.is_datetime64_any_dtype(df[column]):
+                candidates["timestamp"] = column
+                break
+
+    if candidates["entity_id"] is None:
+        object_columns = [
+            column
+            for column in df.columns
+            if pd.api.types.is_object_dtype(df[column]) or isinstance(df[column].dtype, pd.CategoricalDtype)
+        ]
+        if object_columns:
+            ratios = []
+            n = max(len(df), 1)
+            for column in object_columns:
+                unique_ratio = df[column].nunique(dropna=True) / n
+                score = 1.0 - abs(unique_ratio - 0.05)
+                ratios.append((column, score))
+            candidates["entity_id"] = max(ratios, key=lambda item: item[1])[0]
+
+    return InferredSchema(**candidates)
+
+
+def resolve_schema(df: pd.DataFrame, config: dict) -> InferredSchema:
+    inferred = infer_schema(df)
+    resolved = {}
+    for role in ROLE_HINTS:
+        configured = config.get(role, "auto")
+        value = getattr(inferred, role) if configured in (None, "auto") else configured
+        if value is not None and value not in df.columns:
+            raise ValueError(f"Configured {role} column '{value}' does not exist")
+        resolved[role] = value
+    return InferredSchema(**resolved)
+
+
+def leakage_columns(df: pd.DataFrame, target: str | None = None) -> list[str]:
+    blocked: list[str] = []
+    for column in df.columns:
+        normalized = _norm(column)
+        if target and column == target:
+            blocked.append(column)
+            continue
+        if any(fragment in normalized for fragment in FUTURE_NAME_FRAGMENTS):
+            blocked.append(column)
+    return sorted(set(blocked))
+
+
+def describe_table(df: pd.DataFrame) -> dict:
+    return {
+        "rows": int(len(df)),
+        "columns": int(len(df.columns)),
+        "duplicate_rows": int(df.duplicated().sum()),
+        "missing_fraction": {
+            column: float(value)
+            for column, value in df.isna().mean().sort_values(ascending=False).head(15).items()
+        },
+        "dtypes": {column: str(dtype) for column, dtype in df.dtypes.items()},
+    }

src/batteryswapai/submission.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pandas as pd
+
+
+def write_predictions(
+    ids: pd.Series,
+    predictions,
+    output_path: str | Path,
+    id_name: str,
+    prediction_name: str = "rul_days",
+) -> Path:
+    path = Path(output_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    output = pd.DataFrame({id_name: ids.to_numpy(), prediction_name: predictions})
+    output.to_csv(path, index=False)
+    return path

src/batteryswapai/synthetic.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+
+def make_synthetic_dataset(
+    n_devices: int = 24,
+    days: int = 180,
+    seed: int = 2026,
+) -> pd.DataFrame:
+    rng = np.random.default_rng(seed)
+    start = pd.Timestamp("2025-01-01", tz="UTC")
+    rows = []
+
+    for device in range(n_devices):
+        lifetime = int(rng.integers(days // 2, days + 60))
+        base_voltage = rng.normal(3.62, 0.025)
+        temperature_base = rng.normal(21.0, 3.0)
+        site = f"site_{device % 5}"
+        for day in range(days):
+            timestamp = start + pd.Timedelta(days=day)
+            health = max(0.0, 1.0 - day / lifetime)
+            knee = 1.0 / (1.0 + np.exp(-(day - 0.82 * lifetime) / 7.0))
+            voltage = base_voltage - 0.10 * (1.0 - health) - 0.48 * knee + rng.normal(0, 0.012)
+            temperature = temperature_base + 5.0 * np.sin(2 * np.pi * day / 365.0) + rng.normal(0, 1.0)
+            observations = max(1, int(rng.poisson(24)))
+            rul = max(float(lifetime - day), 0.0)
+            rows.append(
+                {
+                    "device_id": f"device_{device:03d}",
+                    "site_id": site,
+                    "timestamp": timestamp,
+                    "battery_voltage": voltage,
+                    "temperature": temperature,
+                    "message_count": observations,
+                    "rul_days": rul,
+                }
+            )
+    return pd.DataFrame(rows)

src/batteryswapai/uncertainty.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import numpy as np
+import pandas as pd
+from sklearn.ensemble import HistGradientBoostingRegressor
+
+from .models import _categorical_columns, encode_features
+
+
+@dataclass
+class QuantileRULModel:
+    models: dict[float, HistGradientBoostingRegressor]
+    feature_columns: list[str]
+    categorical_columns: list[str]
+    encoded_columns: list[str]
+    fill_values: pd.Series
+
+    def predict(self, frame: pd.DataFrame) -> pd.DataFrame:
+        features, _, _ = encode_features(
+            frame[self.feature_columns],
+            self.categorical_columns,
+            encoded_columns=self.encoded_columns,
+            fill_values=self.fill_values,
+        )
+        output = {}
+        for quantile, model in sorted(self.models.items()):
+            output[f"q{int(round(quantile * 100)):02d}"] = np.clip(model.predict(features), 0.0, None)
+        result = pd.DataFrame(output, index=frame.index)
+        columns = list(result.columns)
+        result[columns] = np.maximum.accumulate(result[columns].to_numpy(), axis=1)
+        return result
+
+
+def fit_quantile_models(
+    frame: pd.DataFrame,
+    feature_columns: list[str],
+    target_col: str,
+    categorical_columns: list[str] | None = None,
+    quantiles: tuple[float, ...] = (0.1, 0.5, 0.9),
+    iterations: int = 700,
+    depth: int = 8,
+    learning_rate: float = 0.04,
+    seed: int = 2026,
+) -> QuantileRULModel:
+    work = frame.dropna(subset=[target_col]).copy()
+    categorical_columns = categorical_columns or _categorical_columns(work, feature_columns)
+    features, encoded_columns, fill_values = encode_features(
+        work[feature_columns], categorical_columns
+    )
+    target = pd.to_numeric(work[target_col], errors="coerce").astype(float)
+    models = {}
+    for quantile in quantiles:
+        model = HistGradientBoostingRegressor(
+            loss="quantile",
+            quantile=quantile,
+            max_iter=iterations,
+            max_leaf_nodes=max(7, 2**depth - 1),
+            learning_rate=learning_rate,
+            l2_regularization=2.0,
+            random_state=seed,
+        )
+        model.fit(features, target)
+        models[quantile] = model
+    return QuantileRULModel(
+        models,
+        feature_columns,
+        categorical_columns,
+        encoded_columns,
+        fill_values,
+    )
+
+
+def conformal_radius(y_true, y_pred, coverage: float = 0.9) -> float:
+    residuals = np.abs(np.asarray(y_true, dtype=float) - np.asarray(y_pred, dtype=float))
+    residuals = residuals[np.isfinite(residuals)]
+    if residuals.size == 0:
+        raise ValueError("No finite residuals available for conformal calibration")
+    return float(np.quantile(residuals, coverage, method="higher"))
+
+
+def add_conformal_interval(predictions, radius: float) -> pd.DataFrame:
+    median = np.asarray(predictions, dtype=float)
+    return pd.DataFrame(
+        {
+            "lower": np.clip(median - radius, 0.0, None),
+            "median": np.clip(median, 0.0, None),
+            "upper": np.clip(median + radius, 0.0, None),
+        }
+    )

src/batteryswapai/validation.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import numpy as np
+import pandas as pd
+from sklearn.model_selection import GroupKFold
+
+
+@dataclass(frozen=True)
+class Fold:
+    train_idx: np.ndarray
+    valid_idx: np.ndarray
+
+
+def group_folds(df: pd.DataFrame, group_col: str, folds: int = 5) -> list[Fold]:
+    groups = df[group_col].astype(str)
+    unique_groups = groups.nunique()
+    if unique_groups < 2:
+        raise ValueError("At least two groups are required for leakage-safe validation")
+    n_splits = min(folds, unique_groups)
+    splitter = GroupKFold(n_splits=n_splits)
+    dummy = np.zeros(len(df))
+    return [Fold(train, valid) for train, valid in splitter.split(dummy, groups=groups)]
+
+
+def temporal_holdout(
+    df: pd.DataFrame,
+    timestamp_col: str,
+    valid_fraction: float = 0.2,
+) -> Fold:
+    timestamps = pd.to_datetime(df[timestamp_col], errors="coerce", utc=True)
+    cutoff = timestamps.quantile(1.0 - valid_fraction)
+    train = np.flatnonzero((timestamps < cutoff).to_numpy())
+    valid = np.flatnonzero((timestamps >= cutoff).to_numpy())
+    if len(train) == 0 or len(valid) == 0:
+        raise ValueError("Temporal holdout produced an empty split")
+    return Fold(train, valid)

submission_artifacts/weekly_planner.joblib

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:228d48ca831b806c44b66c1e082ed4b8ec1e8ca2b43978bc14af15be731a2295
+size 16259582