goodwe-addon/goodwe_controller/app/priceruns.py

"""OPT-11 P3: price runs. Which consecutive firm price slots are cheap, which
are expensive, and which one a battery may sell in.

Pure: prices, the time and a few energy figures in, a `RunPlan` out. No I/O,
no clock. `modetrade.py` turns the plan into each battery's range and the
site's trade; main.py rebuilds the plan at most once per 15 minutes and keeps
a run that has started until its end (`keep_started()`).

The owner's SAFETY-46 answer (2026-09-25): a trade is a run of consecutive
slots, not one 15-minute interval. OPT-11-design section 2 ("Prices") and
DECISIONS/OPT-11-battery-modes.md have the record.

* The CHEAP run is the cheapest consecutive firm window long enough to put
  the planned energy in at the chargers' combined rating.
* A slot is EXPENSIVE when its price is above the cheap run's mean by at
  least the required spread (`cycle_cost_threshold_eur`, widened as
  `TariffPlan.arbitrage()` widens it). Give runs are the maximal consecutive
  expensive slots. Holding batteries cover the house in them. The mean is
  the lower of the next cheap run's and `ref_eur_kwh`, the cheap run the
  batteries last filled in (main.py keeps it for 12 h): energy bought at
  03:00 is still worth that price at 08:00, when the next cheap run in sight
  is today's dearer midday (measured: without it a Cheap hours battery spent
  its night charge on flat morning prices, then had nothing at 19:00).
* The SELL run is the dearest consecutive window inside one give run long
  enough to sell the sellers' energy at their rating, with no slot below 0
  (section 17 of the spec: never export at a negative price).
* A provisional (carried) price is never part of any run (section 8.3).
"""

import math
from dataclasses import dataclass
from datetime import datetime, timedelta

from .price import _cover_table

# How far ahead runs are looked for. A day: tonight's cheap hours and the next
# evening's expensive ones, without holding a battery for a peak two days out.
HORIZON = timedelta(hours=24)

# No plan, and why (the page words are in homeview / ui_text).
NO_PRICES = "no_prices"           # fewer than two firm slots ahead
NOT_WORTH = "not_worth"           # no slot clears the required spread


@dataclass(frozen=True)
class Run:
    start: datetime
    end: datetime
    mean_eur_kwh: float

    def holds(self, when: datetime) -> bool:
        return self.start <= when < self.end


@dataclass(frozen=True)
class RunPlan:
    """What the prices say for the next day. `reason` is None when there is
    a plan (a cheap run and at least one give run), else why there is none."""
    cheap: Run | None
    give: tuple
    sell: Run | None
    level_eur_kwh: float | None
    reason: str | None
    # A slot after the cheap run clears the spread over the CHEAP RUN'S OWN
    # price (not the reference): only then is buying in it worth a cycle.
    charge_worth: bool = False

    @property
    def on(self) -> bool:
        return self.reason is None

    def give_at(self, when):
        return next((g for g in self.give if g.holds(when)), None)

    def next_give(self, when):
        """The give run under way at `when`, else the next one, else None."""
        return next((g for g in self.give if g.end > when), None)

    def charge_useful(self) -> bool:
        """Charging in the cheap run pays only if, after it, a slot clears
        the spread over the cheap run's own price."""
        return self.cheap is not None and self.charge_worth


def firm_slots(points, now: datetime, horizon: timedelta = HORIZON) -> list:
    """`[(start, end, eur_kwh)]`, ascending: the firm slots that have not
    ended and start within the horizon. A slot's span is TARIFF-14's rule
    (`price._cover_table()`), so 15-minute and hourly series both work."""
    out = []
    for start, end, p in _cover_table(points):
        if p.provisional or end <= now or start >= now + horizon:
            continue
        if end > start:
            out.append((start, end, float(p.eur_per_kwh)))
    return out


def _segments(slots) -> list:
    """Split slots into runs of consecutive ones (each starts where the last
    ended). A gap or a provisional slot breaks a run."""
    segs, cur = [], []
    for s in slots:
        if cur and s[0] != cur[-1][1]:
            segs.append(cur)
            cur = []
        cur.append(s)
    if cur:
        segs.append(cur)
    return segs


def _hours(s) -> float:
    return (s[1] - s[0]).total_seconds() / 3600.0


def _mean(win) -> float:
    h = sum(_hours(s) for s in win)
    return sum(_hours(s) * s[2] for s in win) / h


def _windows(segs, need_h):
    """Every consecutive window that lasts at least `need_h` hours, shortest
    such from each start. If no segment is that long, each whole segment."""
    out = []
    for seg in segs:
        for i in range(len(seg)):
            h, j = 0.0, i
            while j < len(seg) and h < need_h - 1e-9:
                h += _hours(seg[j])
                j += 1
            if h >= need_h - 1e-9:
                out.append(seg[i:j])
    return out or [seg for seg in segs]


def need_hours(kwh, kw, slots) -> float:
    """How long a run must last: `kwh` at `kw`, and at least one slot."""
    one = min((_hours(s) for s in slots), default=1.0)
    if not (kwh and kw) or kwh <= 0 or kw <= 0:
        return one
    return max(one, float(kwh) / float(kw))


def plan(points, now: datetime, *, required_eur_kwh: float, charge_kwh,
         charge_kw, sell_kwh, sell_kw, horizon: timedelta = HORIZON,
         ref_eur_kwh=None) -> RunPlan:
    """The run plan for the next `horizon`. Ties go to the earliest window."""
    if not (isinstance(required_eur_kwh, (int, float))
            and math.isfinite(required_eur_kwh) and required_eur_kwh >= 0):
        raise ValueError("required_eur_kwh must be a finite number >= 0")
    slots = firm_slots(points, now, horizon)
    if len(slots) < 2:
        return RunPlan(None, (), None, None, NO_PRICES)
    segs = _segments(slots)
    wins = _windows(segs, need_hours(charge_kwh, charge_kw, slots))
    best = min(wins, key=lambda w: (round(_mean(w), 9), w[0][0]))
    cheap = Run(best[0][0], best[-1][1], _mean(best))
    base = cheap.mean_eur_kwh if ref_eur_kwh is None else min(
        cheap.mean_eur_kwh, float(ref_eur_kwh))
    level = base + float(required_eur_kwh)
    # Expensive: clears the spread, strictly dearer than the cheap mean, and
    # not part of the cheap run itself (a threshold of 0 must not overlap).
    dear = [s for s in slots
            if s[2] >= level and s[2] > cheap.mean_eur_kwh
            and not (cheap.start <= s[0] < cheap.end)]
    give_segs = _segments(dear)
    give = tuple(Run(g[0][0], g[-1][1], _mean(g)) for g in give_segs)
    worth = any(s[0] >= cheap.end and s[2] >= cheap.mean_eur_kwh
                + float(required_eur_kwh) for s in dear)
    if not give:
        return RunPlan(cheap, (), None, level, NOT_WORTH)
    # Sell: inside one give run, never over a slot priced below 0.
    sellable = [part for g in give_segs
                for part in _segments([s for s in g if s[2] >= 0.0])]
    sell = None
    if sellable:
        sw = _windows(sellable, need_hours(sell_kwh, sell_kw, slots))
        top = max(sw, key=lambda w: (round(_mean(w), 9),
                                     -w[0][0].timestamp()))
        sell = Run(top[0][0], top[-1][1], _mean(top))
    return RunPlan(cheap, give, sell, level, None, worth)


def keep_started(old: RunPlan | None, new: RunPlan, now: datetime) -> RunPlan:
    """A run under way keeps going to its end (no flapping on a new plan): the
    cheap run, the give run and the sell run each come from `old` while `old`
    holds `now` in it. The rest is `new`. A safety or peak rule can still stop
    a run at any tick; that is not decided here."""
    if old is None or not old.on:
        return new
    cheap = old.cheap if old.cheap and old.cheap.holds(now) else new.cheap
    sell = old.sell if old.sell and old.sell.holds(now) else new.sell
    g_old = old.give_at(now)
    give = new.give
    if g_old is not None:
        give = (g_old,) + tuple(g for g in new.give
                                if g.start >= g_old.end)
    if cheap is new.cheap and sell is new.sell and g_old is None:
        return new
    worth = old.charge_worth if cheap is old.cheap else new.charge_worth
    return RunPlan(cheap, give, sell, new.level_eur_kwh if new.on
                   else old.level_eur_kwh, None if give else new.reason,
                   worth)