pluginbattery/scripts/compare_batteries.py

#!/usr/bin/env python3
"""Run the price-arbitrage oracle for several battery configs and print a comparison.

Saves per-hour CSVs for each config plus a summary table. Tariff: NL consumer.
The simulator works on the raw P1 net-meter trace — whatever PV is on the
roof is already netted in. Existing-PV exports show up as negative
``raw_demand_kw`` and are absorbed by the battery before crossing the meter.
"""
from __future__ import annotations

import argparse
import csv
import re
from pathlib import Path

from pluginbattery.sim import (
    Battery,
    apply_nl_tariff,
    load_hourly,
    oracle_daily_schedule,
    simulate,
)

CONFIGS = [
    (
        "EcoFlow Stream AC (1.92 kWh, 0.8 kW, plug-in)",
        Battery(
            capacity_kwh=1.92,
            max_charge_kw=0.8,
            max_discharge_kw=0.8,
            round_trip_eff=0.90,
            allows_export=False,
        ),
        700.0,
    ),
    (
        "Marstek 5.12 kWh plug-in (800 W)",
        Battery(
            capacity_kwh=5.12,
            max_charge_kw=0.8,
            max_discharge_kw=0.8,
            round_trip_eff=0.90,
            allows_export=False,
        ),
        1339.0,
    ),
    (
        "Marstek 5.12 kWh hardwired (2.5 kW)",
        Battery(
            capacity_kwh=5.12,
            max_charge_kw=2.5,
            max_discharge_kw=2.5,
            round_trip_eff=0.90,
            allows_export=False,
        ),
        1339.0,
    ),
]


def slugify(s: str) -> str:
    s = re.sub(r"[^a-zA-Z0-9]+", "_", s).strip("_").lower()
    return s


def payback_years(year1_savings: float, cost: float, inflation: float = 0.0) -> float:
    """Solve Σ_{k=0}^{N-1} year1 × (1 + i)^k = cost for N.

    With inflation = 0, reduces to cost / year1. With inflation > 0, the
    closed form is N = log(1 + cost × i / year1) / log(1 + i).
    """
    if year1_savings <= 0:
        return float("inf")
    if inflation == 0.0:
        return cost / year1_savings
    import math
    return math.log(1 + cost * inflation / year1_savings) / math.log(1 + inflation)


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--saldering", choices=["none", "full"], default="none",
        help="'full' = export at consumer price (NL pre-2027). "
             "'none' = export at raw EPEX (post-2027 default).",
    )
    parser.add_argument(
        "--export-rate", type=float, default=None,
        help="Fixed export EUR/kWh, e.g. 0 for 'no compensation' (sales-calculator style). "
             "Overrides --saldering when set.",
    )
    parser.add_argument(
        "--price-mult", type=float, default=1.0,
        help="Scale consumer price by this factor (default 1.0). Use ~1.5 to project "
             "from our 2023-24 backtest prices to current 2025 retail levels.",
    )
    parser.add_argument(
        "--flat-retail", type=float, default=None,
        help="Replace time-varying consumer price with a flat EUR/kWh value (kills arbitrage). "
             "Mirrors the 'fixed rate' switch on online sales calculators.",
    )
    parser.add_argument(
        "--price-inflation", type=float, default=0.0,
        help="Annual energy-price inflation rate (e.g. 0.03 for 3%%). Affects payback only; "
             "year-1 savings displayed are nominal.",
    )
    args = parser.parse_args()

    df = load_hourly("data/raw")
    df = apply_nl_tariff(df)
    if args.flat_retail is not None:
        df["eur_per_kwh"] = float(args.flat_retail)
    elif args.price_mult != 1.0:
        df["eur_per_kwh"] = df["eur_per_kwh"] * args.price_mult

    # Existing PV (if any) is already in raw_demand_kw via the meter signal.
    annual_export = float(-df["raw_demand_kw"].clip(upper=0).sum() * 8766.0 / len(df))
    if annual_export > 1.0:
        pv_note = f"{annual_export:.0f} kWh/yr exported at meter (existing PV)"
    else:
        pv_note = "no measurable existing PV in the data"

    if args.export_rate is not None:
        df["export_eur_per_kwh"] = float(args.export_rate)
        sald_note = f"export = {args.export_rate:.3f} EUR/kWh (fixed)"
    elif args.saldering == "none":
        df["export_eur_per_kwh"] = df["epex_eur_per_kwh"]
        sald_note = "no saldering (export = raw EPEX)"
    else:
        sald_note = "full saldering (export = consumer price)"

    if args.flat_retail is not None:
        price_note = f"FLAT €{args.flat_retail}/kWh (fixed-rate mode)"
    elif args.price_mult != 1.0:
        price_note = f"EPEX × 1.21 + 0.136 × {args.price_mult:.2f}"
    else:
        price_note = "EPEX × 1.21 + 0.136 (dynamic, as in raw data)"
    print(f"Window: {df.index[0]} → {df.index[-1]}, {len(df)} hours")
    print(f"Tariff: import = {price_note}; {sald_note}")
    if args.price_inflation > 0:
        print(f"Payback uses {args.price_inflation*100:.1f}%/yr energy-price inflation")
    print(f"Meter : {pv_note}\n")

    # Headline figures with no battery, just for context.
    g_no = df["raw_demand_kw"]
    imp_p = df["eur_per_kwh"]
    exp_p = df.get("export_eur_per_kwh", df["eur_per_kwh"])
    cost_no_battery_total = float((g_no.where(g_no > 0, 0) * imp_p +
                                    g_no.where(g_no < 0, 0) * exp_p).sum())
    import_kwh = float(g_no.where(g_no > 0, 0).sum())
    export_kwh = float(-g_no.where(g_no < 0, 0).sum())
    print(f"  Without battery:  {import_kwh:6.0f} kWh imported, {export_kwh:6.0f} kWh exported, "
          f"net bill €{cost_no_battery_total:.2f}\n")

    out_dir = Path("data/processed")
    out_dir.mkdir(parents=True, exist_ok=True)

    summary_rows = []
    header = (
        f"{'Battery':47s}{'€/yr saved':>12s}{'cycles':>9s}"
        f"{'kWh shifted':>13s}{'payback':>10s}"
    )
    print(header)
    print("-" * len(header))

    for name, battery, price in CONFIGS:
        schedule = oracle_daily_schedule(df, battery)
        sim = simulate(df, battery, schedule)

        savings = float(sim["savings"].sum())
        kwh_shifted = float(sim["discharge_kwh"].sum())
        cycles = kwh_shifted / battery.capacity_kwh
        payback = payback_years(savings, price, args.price_inflation)

        sim.to_csv(out_dir / f"hourly_{slugify(name)}.csv")
        summary_rows.append(
            {
                "battery": name,
                "capacity_kwh": battery.capacity_kwh,
                "max_charge_kw": battery.max_charge_kw,
                "max_discharge_kw": battery.max_discharge_kw,
                "price_eur": price,
                "saldering": args.saldering,
                "annual_savings_eur": round(savings, 2),
                "kwh_shifted": round(kwh_shifted, 1),
                "equivalent_cycles": round(cycles, 1),
                "payback_years": round(payback, 2),
                "price_inflation": args.price_inflation,
            }
        )

        print(
            f"{name:47s}€{savings:9.2f}{cycles:9.1f}{kwh_shifted:11.0f} kWh"
            f"{payback:7.2f} yr"
        )

    summary_path = out_dir / "comparison.csv"
    with summary_path.open("w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=summary_rows[0].keys())
        writer.writeheader()
        writer.writerows(summary_rows)
    print(f"\nSummary -> {summary_path}")
    print(f"Per-hour outputs -> {out_dir}/hourly_*.csv")


if __name__ == "__main__":
    main()