Add aggregator.py

aggregator.py

@@ -0,0 +1,82 @@
+"""Aggregation helpers: month windows, per-tenant / per-metric / daily rollups."""
+from datetime import datetime, timedelta
+from zoneinfo import ZoneInfo
+
+
+def month_range(year: int | None, month: int | None, tz: str):
+    """Return (begin, end) datetimes for a billing month in the cluster tz.
+
+    CloudKitty aligns its rating periods to month boundaries in local time,
+    e.g. 2026-08-01T00:00:00+05:30 -> 2026-09-01T00:00:00+05:30.
+    """
+    zone = ZoneInfo(tz)
+    now = datetime.now(zone)
+    year = year or now.year
+    month = month or now.month
+    begin = datetime(year, month, 1, tzinfo=zone)
+    end = (
+        datetime(year + 1, 1, 1, tzinfo=zone)
+        if month == 12
+        else datetime(year, month + 1, 1, tzinfo=zone)
+    )
+    return begin, end
+
+
+def days_of_month(begin: datetime, end: datetime):
+    """Yield (day_begin, day_end) pairs from month begin until min(today, end)."""
+    now = datetime.now(begin.tzinfo)
+    cursor = begin
+    while cursor < end and cursor < now:
+        nxt = min(cursor + timedelta(days=1), end)
+        yield cursor, nxt
+        cursor = nxt
+
+
+def tenant_costs(rows: list[dict]) -> list[dict]:
+    """Normalize CloudKitty summary rows into a sorted tenant cost list."""
+    out = [
+        {
+            "tenant": r.get("tenant") or "(unknown)",
+            "qty": r.get("qty", 0),
+            "rate": round(r.get("rate", 0) or 0, 6),
+        }
+        for r in rows
+    ]
+    out.sort(key=lambda x: x["rate"], reverse=True)
+    return out
+
+
+def per_metric(rows: list[dict], tenant: str) -> list[dict]:
+    """From groupby=tenant,type rows, extract the metric breakdown of one tenant.
+
+    'type' is the CloudKitty dataframe field holding the metric name.
+    """
+    out = []
+    for r in rows:
+        if (r.get("tenant") or "") != tenant:
+            continue
+        out.append(
+            {
+                "metric": r.get("type", "?"),
+                "qty": r.get("qty", 0),
+                "rate": round(r.get("rate", 0) or 0, 6),
+            }
+        )
+    out.sort(key=lambda x: x["rate"], reverse=True)
+    return out
+
+
+def projection(rate_so_far: float, begin: datetime, end: datetime, tz: str) -> dict:
+    """Linear month-end projection from the rate accrued so far."""
+    now = datetime.now(ZoneInfo(tz))
+    total_hours = (end - begin).total_seconds() / 3600
+    elapsed_hours = max((now - begin).total_seconds() / 3600, 0.0)
+    if elapsed_hours < 1:
+        return {"projected_month_end": None, "note": "less than 1h of data"}
+    projected = rate_so_far * (total_hours / elapsed_hours)
+    return {
+        "projected_month_end": round(projected, 2),
+        "elapsed_hours": round(elapsed_hours, 1),
+        "total_hours": round(total_hours, 1),
+        "pct_elapsed": round(100 * elapsed_hours / total_hours, 1),
+    }