"""Evaluates a TariffPlan against a stream of interval usage readings. No battery/storage modeling is needed: the household has solar-only generation, so metered import/export values are ground truth and are used directly rather than re-derived from generation and load. """ from __future__ import annotations from collections.abc import Iterable from dataclasses import dataclass, field from datetime import datetime from datetime import time as time_ from zoneinfo import ZoneInfo from app.costing.tariff_models import DAYS, FeedInComponent, TariffPlan, TimeContext, UsageComponent # Queensland does not observe daylight saving, but readings are still stored # in UTC and converted here rather than assuming a fixed UTC+10 offset, so # the engine isn't silently wrong if it's ever reused for a DST-observing state. BRISBANE_TZ = ZoneInfo("Australia/Brisbane") @dataclass class IntervalReading: ts: datetime # timezone-aware (UTC) import_kwh: float export_kwh: float @dataclass class BatteryReading: ts: datetime # timezone-aware (UTC) charge_kwh: float discharge_kwh: float @dataclass class BatteryValueResult: charged_kwh: float discharged_kwh: float # What the discharged energy would have cost to import instead, valued # at the current plan's usage rate for whatever time of day it happened. avoided_import_dollars: float # What the charged energy would have earned as a feed-in credit had it # been exported instead of stored -- the "cost" of charging the battery # rather than exporting straight away. foregone_export_dollars: float net_value_dollars: float days_with_data: int @dataclass class ComponentBreakdown: label: str kwh: float cost_dollars: float @dataclass class CostBreakdown: plan_id: str retailer: str plan_name: str usage_breakdown: list[ComponentBreakdown] feed_in_breakdown: list[ComponentBreakdown] supply_charge_dollars: float unconditional_discount_dollars: float potential_conditional_savings_dollars: float total_dollars: float days_with_data: int modelable: bool modeling_notes: list[str] = field(default_factory=list) @property def annualized_total_dollars(self) -> float | None: if self.days_with_data == 0: return None return self.total_dollars * 365 / self.days_with_data @property def is_estimate(self) -> bool: return self.days_with_data < 365 def _first_match( components: list[UsageComponent] | list[FeedInComponent], ctx: TimeContext ) -> UsageComponent | FeedInComponent | None: for c in components: if c.matches(ctx): return c return None @dataclass class TimeBucket: """A (month, day-of-week, time-of-day) point with import/export summed across every raw reading that shares it. Comparing many plans against the same usage history is much cheaper when every plan re-matches this small bucketed set instead of every raw reading -- see `bucket_readings`. """ ctx: TimeContext import_kwh: float export_kwh: float def bucket_readings(readings: Iterable[IntervalReading]) -> tuple[list[TimeBucket], int, dict[str, float]]: """Groups readings by (month, weekday, hour, minute) and sums their kWh, since a time-of-use tariff's cost only ever depends on that triple, never on the specific calendar date. Also returns the count of distinct calendar days with data (for the supply charge) and each calendar day's total export (for tiered daily feed-in tariffs, which depend on same-day cumulative export rather than time of day) -- computed once up front so neither needs to be re-derived per plan. Bucketing turns "N plans x M raw readings" into "N plans x <= a few thousand buckets", which matters once there are hundreds of cached AER plans to evaluate against months of 5-minute interval history. """ sums: dict[tuple[int, str, int, int], list[float]] = {} daily_export_kwh: dict[str, float] = {} for r in readings: local_dt = r.ts.astimezone(BRISBANE_TZ) day = local_dt.date().isoformat() daily_export_kwh[day] = daily_export_kwh.get(day, 0.0) + r.export_kwh key = (local_dt.month, local_dt.weekday(), local_dt.hour, local_dt.minute) bucket = sums.get(key) if bucket is None: sums[key] = [r.import_kwh, r.export_kwh] else: bucket[0] += r.import_kwh bucket[1] += r.export_kwh buckets = [ TimeBucket( ctx=TimeContext(month=month, day_code=DAYS[weekday], time_of_day=time_(hour, minute)), import_kwh=vals[0], export_kwh=vals[1], ) for (month, weekday, hour, minute), vals in sums.items() ] return buckets, len(daily_export_kwh), daily_export_kwh def _evaluate_core( plan: TariffPlan, buckets: list[TimeBucket], days_with_data: int, daily_export_kwh: dict[str, float] ) -> CostBreakdown: notes = list(plan.modeling_notes) usage_kwh: dict[str, float] = {c.label: 0.0 for c in plan.usage_components} feed_in_kwh: dict[str, float] = {c.label: 0.0 for c in plan.feed_in_tariffs} unmatched_usage_kwh = 0.0 unmatched_export_kwh = 0.0 for b in buckets: usage_component = _first_match(plan.usage_components, b.ctx) if usage_component is not None: usage_kwh[usage_component.label] += b.import_kwh else: unmatched_usage_kwh += b.import_kwh if plan.tiered_feed_in or not b.export_kwh: continue # tiered feed-in is computed below from daily totals instead if plan.feed_in_tariffs: fit_component = _first_match(plan.feed_in_tariffs, b.ctx) if fit_component is not None: feed_in_kwh[fit_component.label] += b.export_kwh else: unmatched_export_kwh += b.export_kwh else: unmatched_export_kwh += b.export_kwh if unmatched_usage_kwh > 1e-9: notes.append( f"{unmatched_usage_kwh:.2f} kWh of usage did not match any tariff window and was not " "charged -- this plan's time windows may have gaps." ) if unmatched_export_kwh > 1e-9: notes.append( f"{unmatched_export_kwh:.2f} kWh of solar export received no feed-in credit " "(no matching feed-in window, or this plan has no feed-in tariff)." ) # A tariff period can need more than one disjoint time window under the # same label (e.g. "Shoulder" = 7am-4pm AND 9pm-10pm) -- usage_kwh is # already correctly summed per label above, so build the breakdown from # unique labels only, or a component appearing twice would double-count # its cost here. rate_by_label = {c.label: c.rate_c_per_kwh for c in plan.usage_components} usage_breakdown = [ ComponentBreakdown(label=label, kwh=kwh, cost_dollars=kwh * rate_by_label[label] / 100) for label, kwh in usage_kwh.items() ] if plan.tiered_feed_in: tier = plan.tiered_feed_in below_kwh = sum(min(v, tier.daily_threshold_kwh) for v in daily_export_kwh.values()) above_kwh = sum(max(v - tier.daily_threshold_kwh, 0.0) for v in daily_export_kwh.values()) feed_in_breakdown = [ ComponentBreakdown( label=f"First {tier.daily_threshold_kwh:g} kWh/day", kwh=below_kwh, cost_dollars=-below_kwh * tier.rate_below_c_per_kwh / 100, ), ComponentBreakdown( label="Remaining balance", kwh=above_kwh, cost_dollars=-above_kwh * tier.rate_above_c_per_kwh / 100, ), ] else: fit_rate_by_label = {c.label: c.rate_c_per_kwh for c in plan.feed_in_tariffs} feed_in_breakdown = [ ComponentBreakdown(label=label, kwh=kwh, cost_dollars=-kwh * fit_rate_by_label[label] / 100) for label, kwh in feed_in_kwh.items() ] usage_cost = sum(c.cost_dollars for c in usage_breakdown) feed_in_credit = sum(c.cost_dollars for c in feed_in_breakdown) # already negative # Supply charge is billed per day the meter has data, not per calendar # day in the requested range, so a gap in polling doesn't overstate cost. supply_charge = plan.supply_charge_c_per_day * days_with_data / 100 subtotal = usage_cost + feed_in_credit + supply_charge unconditional_discount = 0.0 potential_conditional = 0.0 for d in plan.discounts: base = usage_cost if d.applies_to == "usage" else subtotal amount = (base * d.percent / 100) if d.percent is not None else (d.amount_dollars or 0.0) if d.auto_apply: unconditional_discount += amount else: potential_conditional += amount notes.append(f"Conditional discount not applied to headline total: {d.description} (~${amount:.2f})") total = subtotal - unconditional_discount return CostBreakdown( plan_id=plan.plan_id, retailer=plan.retailer, plan_name=plan.plan_name, usage_breakdown=usage_breakdown, feed_in_breakdown=feed_in_breakdown, supply_charge_dollars=supply_charge, unconditional_discount_dollars=unconditional_discount, potential_conditional_savings_dollars=potential_conditional, total_dollars=total, days_with_data=days_with_data, modelable=plan.modelable, modeling_notes=notes, ) def evaluate(plan: TariffPlan, readings: Iterable[IntervalReading]) -> CostBreakdown: buckets, days_with_data, daily_export_kwh = bucket_readings(readings) return _evaluate_core(plan, buckets, days_with_data, daily_export_kwh) def evaluate_bucketed( plan: TariffPlan, buckets: list[TimeBucket], days_with_data: int, daily_export_kwh: dict[str, float] ) -> CostBreakdown: """Like `evaluate`, but takes an already-bucketed usage history (from `bucket_readings`) so comparing many plans against the same usage data only pays the bucketing cost once.""" return _evaluate_core(plan, buckets, days_with_data, daily_export_kwh) def _feed_in_rate_c(plan: TariffPlan, ctx: TimeContext) -> float | None: if plan.tiered_feed_in: # A tiered feed-in tariff's rate depends on that calendar day's # *cumulative* export total, not the time of day, so it can't be # matched via TimeContext the way flat/TOU components can. # Approximating with the above-threshold (marginal) rate is the # reasonable default for a solar household already exporting past # the daily cap most days -- charging the battery instead of # exporting a bit more almost always displaces balance-tier kWh, # not capped-tier kWh. Flagged to the caller via # `BatteryValueResult` so the UI can note it's an approximation. return plan.tiered_feed_in.rate_above_c_per_kwh component = _first_match(plan.feed_in_tariffs, ctx) return component.rate_c_per_kwh if component else None def evaluate_battery_value(plan: TariffPlan, readings: Iterable[BatteryReading]) -> BatteryValueResult: """Estimates what a battery is actually worth: energy it discharged is valued at the import rate it let you avoid, energy it charged is "costed" at the feed-in credit you gave up by not exporting it instead. Both sides are matched against the current plan's own time-of-use windows for the moment each charge/discharge reading happened, the same way `evaluate` matches usage/export readings. """ charged = discharged = avoided_import = foregone_export = 0.0 days_with_data: set[str] = set() for r in readings: local_dt = r.ts.astimezone(BRISBANE_TZ) days_with_data.add(local_dt.date().isoformat()) ctx = TimeContext.from_datetime(local_dt) if r.discharge_kwh > 1e-9: usage_component = _first_match(plan.usage_components, ctx) if usage_component is not None: avoided_import += r.discharge_kwh * usage_component.rate_c_per_kwh / 100 discharged += r.discharge_kwh if r.charge_kwh > 1e-9: feed_in_rate_c = _feed_in_rate_c(plan, ctx) if feed_in_rate_c is not None: foregone_export += r.charge_kwh * feed_in_rate_c / 100 charged += r.charge_kwh return BatteryValueResult( charged_kwh=charged, discharged_kwh=discharged, avoided_import_dollars=avoided_import, foregone_export_dollars=foregone_export, net_value_dollars=avoided_import - foregone_export, days_with_data=len(days_with_data), )