84c4922bbc
FastAPI app that pulls interval usage from Sungrow iSolarCloud (OAuth2), caches AER Consumer Data Right electricity plan data, and compares actual metered usage against every available plan to estimate real annual cost (TOU, tiered feed-in, and flat tariffs). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
84 lines
3.4 KiB
Python
84 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
|
|
from app.costing.engine import CostBreakdown, bucket_readings, evaluate_bucketed
|
|
from app.db.repositories import get_cached_plans, get_current_plan, get_interval_readings
|
|
from app.db.session import get_session
|
|
|
|
# The AER plan cache can hold thousands of near-duplicate whitelabel/partner
|
|
# variants of the same underlying offer across all configured retailers;
|
|
# dumping every one into the comparison table is neither fast nor useful, so
|
|
# only the cheapest N modelable plans are returned (see `compare_plans`).
|
|
DEFAULT_PLAN_LIMIT = 50
|
|
|
|
|
|
@dataclass
|
|
class PlanComparisonRow:
|
|
breakdown: CostBreakdown
|
|
is_current: bool
|
|
savings_vs_current_dollars: float | None # positive = cheaper than current plan
|
|
|
|
|
|
@dataclass
|
|
class PlanComparisonResult:
|
|
rows: list[PlanComparisonRow]
|
|
total_plans_considered: int
|
|
|
|
|
|
def compare_plans(start: datetime, end: datetime, limit: int = DEFAULT_PLAN_LIMIT) -> PlanComparisonResult:
|
|
with get_session() as session:
|
|
readings = get_interval_readings(session, start, end)
|
|
plans = get_cached_plans(session)
|
|
current_plan = get_current_plan(session)
|
|
|
|
# Retailers commonly publish the same underlying offer multiple times
|
|
# under different partner/channel plan IDs (e.g. a bank-affiliated
|
|
# sign-up variant) with identical terms -- collapse those before
|
|
# evaluating so the table isn't dominated by exact duplicates.
|
|
seen: set[tuple[str, str, float, tuple[float, ...]]] = set()
|
|
deduped_plans = []
|
|
for plan in plans:
|
|
fingerprint = (
|
|
plan.retailer,
|
|
plan.plan_name,
|
|
plan.supply_charge_c_per_day,
|
|
tuple(sorted(c.rate_c_per_kwh for c in plan.usage_components)),
|
|
)
|
|
if fingerprint in seen:
|
|
continue
|
|
seen.add(fingerprint)
|
|
deduped_plans.append(plan)
|
|
plans = deduped_plans
|
|
|
|
# Bucketing once and re-matching each plan against the small bucketed
|
|
# set (instead of every raw reading) is what keeps this fast with
|
|
# hundreds of cached plans and months of 5-minute interval history.
|
|
buckets, days_with_data, daily_export_kwh = bucket_readings(readings)
|
|
|
|
current_result = (
|
|
evaluate_bucketed(current_plan, buckets, days_with_data, daily_export_kwh) if current_plan else None
|
|
)
|
|
current_annual = current_result.annualized_total_dollars if current_result else None
|
|
|
|
rows: list[PlanComparisonRow] = []
|
|
if current_result:
|
|
rows.append(PlanComparisonRow(breakdown=current_result, is_current=True, savings_vs_current_dollars=0.0))
|
|
|
|
plan_rows: list[PlanComparisonRow] = []
|
|
for plan in plans:
|
|
result = evaluate_bucketed(plan, buckets, days_with_data, daily_export_kwh)
|
|
savings = None
|
|
if current_annual is not None and result.annualized_total_dollars is not None:
|
|
savings = current_annual - result.annualized_total_dollars
|
|
plan_rows.append(PlanComparisonRow(breakdown=result, is_current=False, savings_vs_current_dollars=savings))
|
|
|
|
def sort_key(row: PlanComparisonRow):
|
|
if not row.breakdown.modelable or row.breakdown.annualized_total_dollars is None:
|
|
return (1, 0.0)
|
|
return (0, row.breakdown.annualized_total_dollars)
|
|
|
|
plan_rows.sort(key=sort_key)
|
|
rows.extend(plan_rows[:limit])
|
|
return PlanComparisonResult(rows=rows, total_plans_considered=len(plan_rows))
|