pluginbattery/scripts/battery_leaderboard.py

#!/usr/bin/env python3
"""Run every battery in the catalog through both calculators and rank by payback.

Default scenario matches the dynamic-mode store-test:
  4000 kWh/yr net demand, avg retail €0.25, no saldering, 3% inflation.

The simulator works on the raw P1 net-meter trace — whatever PV is on the
roof is already netted in. To model a different PV setup, swap the input
data, not a CLI flag. Output: data/processed/battery_leaderboard.csv.
"""
from __future__ import annotations

import argparse
import csv
from pathlib import Path

from pluginbattery.sim import (
    Battery,
    apply_nl_tariff,
    load_hourly,
    oracle_daily_schedule,
    simulate,
)
from pluginbattery.store_calc import (
    Scenario,
    StoreParams,
    payback_years,
    quote as store_quote,
)


def main() -> None:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("--catalog", default="data/raw/thuisbatterijgids_catalog.csv")
    p.add_argument("--out", default="data/processed/battery_leaderboard.csv")
    p.add_argument("--retail", type=float, default=0.25)
    p.add_argument("--demand", type=float, default=4000.0)
    p.add_argument("--eta", type=float, default=0.88)
    p.add_argument("--saldering", action="store_true")
    p.add_argument("--inflation", type=float, default=0.03)
    p.add_argument("--top", type=int, default=20, help="How many rows to print")
    args = p.parse_args()

    # ─── Load and prep the simulation data once ────────────────────────
    base = load_hourly("data/raw"); base = apply_nl_tariff(base)
    df = base.copy()
    annual_net_demand = base["raw_demand_kw"].sum() * (8766.0 / len(base))
    df["raw_demand_kw"] = base["raw_demand_kw"] * (args.demand / annual_net_demand)
    scale = args.retail / base["eur_per_kwh"].mean()
    df["eur_per_kwh"] = base["eur_per_kwh"] * scale
    df["epex_eur_per_kwh"] = base["epex_eur_per_kwh"] * scale
    df["export_eur_per_kwh"] = df["eur_per_kwh"] if args.saldering else 0.0
    # Existing-PV export visible at the meter (any hour where raw_demand_kw < 0).
    has_existing_pv = bool((df["raw_demand_kw"] < 0).any())

    avg_epex = float(df["epex_eur_per_kwh"].mean())
    store_params = StoreParams(avg_epex_eur_per_kwh=avg_epex)

    # ─── Load catalog ─────────────────────────────────────────────────
    rows_in = []
    with open(args.catalog) as f:
        for r in csv.DictReader(f):
            try:
                cap = float(r["capacity_kwh"])
                pw = float(r["power_w"]) / 1000.0  # → kW
                price = float(r["price_eur"])
            except (TypeError, ValueError):
                continue
            if cap <= 0 or pw <= 0 or price <= 0:
                continue
            rows_in.append({
                "title": r["title"], "brand": r["brand"],
                "capacity_kwh": cap, "power_kw": pw, "price_eur": price,
                "url": r["url"],
            })

    # Cache LP results by (capacity, power) — many batteries share identical specs.
    lp_cache: dict[tuple[float, float], float] = {}

    print(f"Scenario: net demand {args.demand:.0f} kWh/yr, avg retail €{args.retail}, "
          f"existing-PV exports={'yes' if has_existing_pv else 'no'}, "
          f"saldering={args.saldering}, inflation {args.inflation*100:.1f}%/yr")
    print(f"Running {len(rows_in)} batteries (caching by (capacity, power))...")
    print()

    rows_out = []
    for i, r in enumerate(rows_in, 1):
        key = (round(r["capacity_kwh"], 3), round(r["power_kw"], 3))
        if key in lp_cache:
            lp_year1 = lp_cache[key]
        else:
            bat = Battery(
                capacity_kwh=r["capacity_kwh"], max_charge_kw=r["power_kw"],
                max_discharge_kw=r["power_kw"], round_trip_eff=args.eta,
                allows_export=False,
            )
            out = simulate(df, bat, oracle_daily_schedule(df, bat))
            lp_year1 = float(out["savings"].sum())
            lp_cache[key] = lp_year1
        lp_payback = payback_years(lp_year1, r["price_eur"], args.inflation)

        scn = Scenario(
            capacity_kwh=r["capacity_kwh"], max_charge_kw=r["power_kw"],
            battery_cost_eur=r["price_eur"], avg_retail_eur_per_kwh=args.retail,
            has_pv=has_existing_pv, has_saldering=args.saldering, dynamic_rate=True,
        )
        sq = store_quote(scn, store_params, inflation=args.inflation)

        rows_out.append({
            "title": r["title"], "brand": r["brand"],
            "capacity_kwh": r["capacity_kwh"], "power_kw": r["power_kw"],
            "price_eur": r["price_eur"],
            "lp_year1_eur": round(lp_year1, 2),
            "lp_payback_yr": round(lp_payback, 2),
            "store_year1_eur": round(sq["year1_total"], 2),
            "store_payback_yr": round(sq["payback_years"], 2),
            "overstatement": round(sq["year1_total"] / lp_year1, 2) if lp_year1 > 0 else None,
            "url": r["url"],
        })

    # Sort by honest payback (ascending = best first).
    rows_out.sort(key=lambda r: r["lp_payback_yr"])

    out_path = Path(args.out)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    with out_path.open("w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=rows_out[0].keys())
        writer.writeheader()
        writer.writerows(rows_out)

    # Pretty-print top N.
    print(f"{'rank':>4s}  {'battery':40s} {'cap':>5s} {'kW':>4s} "
          f"{'price':>7s}  {'LP €/yr':>9s} {'LP yr':>7s}  {'Store €/yr':>11s} {'Store yr':>9s}  {'×over':>6s}")
    print("-" * 120)
    for i, r in enumerate(rows_out[:args.top], 1):
        title = (r["title"][:38] + "…") if len(r["title"]) > 39 else r["title"]
        over = f"{r['overstatement']:.2f}" if r["overstatement"] else "—"
        print(f"{i:>4d}  {title:40s} {r['capacity_kwh']:>5.2f} {r['power_kw']:>4.1f} "
              f"€{r['price_eur']:>5.0f}  €{r['lp_year1_eur']:>7.2f} {r['lp_payback_yr']:>5.2f}  "
              f"€{r['store_year1_eur']:>9.2f} {r['store_payback_yr']:>7.2f}  {over:>5s}×")
    print(f"\nFull leaderboard ({len(rows_out)} batteries) → {out_path}")


if __name__ == "__main__":
    main()