Time-Series Aggregation for Daily Cloud Cost Tracking

The bottleneck in daily cloud cost tracking is almost never missing raw data — it is the rollup. Billing APIs hand back event-grained line items, not finalized daily totals, and the moment you group them by calendar day three forces collide: late-arriving adjustments that land in past days, pagination retries that re-deliver rows you already counted, and timezone drift that shifts spend across the midnight boundary. Get any one of them wrong and the series grows phantom spikes, double charges, and reconciliation drift that no downstream dashboard can quietly absorb. This page ships the aggregation layer that sits at the tail of the Cloud Billing Data Ingestion & Parsing pipeline and converts heterogeneous line items into a deterministic, idempotent daily cost series. The fix is not bigger machines — it is UTC normalization, a cryptographic dedup key applied before rows reach storage, and an append-as-ledger partition model that absorbs backfills without rewriting history.

# Root Cause & Failure Modes

Naive aggregation assumes billing data is append-only and final. It is neither. Every major provider finalizes asynchronously and revises retroactively, so a GROUP BY billing_date over a freshly fetched window is a guess, not a total.

• Asynchronous finalization. AWS Cost and Usage Reports deliver with up to a 24-hour delay and then restate rows as reservation and savings-plan credits settle. GCP billing exports inject adjustment, credit, and tax rows into already-exported days for up to 48 hours. Azure Cost Management finalizes 24–72 hours after usage. A series built only from the latest fetch silently diverges from the invoice.
• Retry-induced duplication. The retry layer that wraps every metered call — see Handling Billing API Rate Limits & Retries — re-issues requests after a 429 or 5xx. If a page is re-fetched after a partial write, an append-only rollup counts those line items twice and the day inflates.
• Timezone-boundary drift. Providers stamp line_item_start_time in UTC, but ingestion code that truncates in local time pushes a slice of one UTC day into the adjacent bucket. Cross-region spend then straddles two days, tripping budget thresholds on a day that never actually overspent.
• Pagination-window overlap. Cursor windows that overlap at the edges re-deliver boundary rows, the same class of double-count solved for one provider in the Azure Cost API pagination and deduplication guide.

Quantitatively, a single duplicated 10,000-row page on a high-volume account can skew a day’s total by thousands of dollars, and a one-hour timezone shift misattributes every record in that hour. The instinct is to fix all of this downstream with UPSERT or a nightly dedup job, but that pushes cost onto every query and leaves a window where the dashboard is wrong. Determinism belongs at aggregation time, keyed on immutable dimensions.

# Production Pipeline Architecture

The execution model is four phases, each with a single responsibility so a fault in one never corrupts the others. It mirrors the four-stage acquisition → normalization → allocation → persistence contract of the parent pipeline, scoped down to the daily-rollup task.

1. Fetch — pull paginated billing records under exponential backoff so a transient throttle never propagates into pipeline state. This phase is read-only and owns no aggregation logic; it simply yields raw pages, deferring all rate-limit semantics to the retry layer.
2. Normalize — parse every line_item_start_time to UTC, truncate to a calendar billing_date, coerce the canonical schema, and explicitly flag adjustment rows. Forcing a single reference timezone is the one decision that eliminates boundary drift outright.
3. Deduplicate — reduce each row to a SHA-256 digest over its immutable dimensions (account_id, service_name, billing_date, cost_amount) and drop repeats in O(1) before they leave the process. This is the same composite-key contract used across the ingestion pipeline; it neutralizes retry overlap and pagination double-delivery in one pass.
4. Idempotent rollup — group to daily service-level totals and upsert into per-day Parquet partitions treated as an appendable ledger. A backfill re-reads only the affected partition, unions the delta, and rewrites it — the same watermark-and-merge discipline detailed for incremental sync strategies on GCP billing exports, applied here to a columnar file store.

The load-bearing decision is the composite hash. For single-node aggregation a native Python set of hex digests holds millions of keys in modest memory; for distributed workers, offload the registry to Redis and expire hashes past the 30–45 day reconciliation window. Daily partitions aligned to billing_date=YYYY-MM-DD let downstream warehouses prune by date and keep backfills surgical.

# Step-by-Step Python Implementation

The module below uses polars for vectorized time-series operations, tenacity for resilient pagination, and an idempotent partitioned-Parquet upsert. It is import-complete and runs end-to-end against the mock payload in the __main__ guard.

import hashlib
import logging
from pathlib import Path
from typing import Any, Dict

import polars as pl
import requests
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

# Canonical schema the daily series is contracted to emit.
AGGREGATION_SCHEMA = pl.Schema(
    {
        "billing_date": pl.Date,
        "account_id": pl.Utf8,
        "service_name": pl.Utf8,
        "cost_amount": pl.Float64,
        "currency": pl.Utf8,
        "is_adjustment": pl.Boolean,
        "record_hash": pl.Utf8,
    }
)

HASH_DIMENSIONS = ["account_id", "service_name", "billing_date", "cost_amount"]


def normalize_billing_timestamps(df: pl.DataFrame) -> pl.DataFrame:
    """Parse ISO timestamps to UTC, truncate to a calendar day, map to canonical names."""
    return df.with_columns(
        pl.col("line_item_start_time")
        .str.to_datetime(time_zone="UTC", strict=False)
        .dt.truncate("1d")
        .cast(pl.Date)
        .alias("billing_date")
    ).rename({"unblended_cost": "cost_amount"})


def generate_record_hash(df: pl.DataFrame) -> pl.DataFrame:
    """Attach a deterministic SHA-256 digest over immutable dimensions for dedup."""
    return df.with_columns(
        pl.concat_str(
            [pl.col(c).cast(pl.Utf8) for c in HASH_DIMENSIONS], separator="|"
        )
        .map_elements(
            lambda s: hashlib.sha256(s.encode("utf-8")).hexdigest(),
            return_dtype=pl.Utf8,
        )
        .alias("record_hash")
    )


@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=2, min=4, max=30),
    retry=retry_if_exception_type(requests.exceptions.RequestException),
)
def fetch_billing_page(api_url: str, params: Dict[str, Any]) -> pl.DataFrame:
    """Fetch one paginated billing page; tenacity owns all transient-error backoff."""
    response = requests.get(api_url, params=params, timeout=30)
    response.raise_for_status()
    payload = response.json()
    return pl.DataFrame(payload.get("items", []))


def aggregate_daily_costs(raw_df: pl.DataFrame) -> pl.DataFrame:
    """Normalize, deduplicate, and roll up to daily service-level aggregates."""
    if raw_df.is_empty():
        return pl.DataFrame(schema=AGGREGATION_SCHEMA)

    normalized = normalize_billing_timestamps(raw_df)
    hashed = generate_record_hash(normalized)

    # Drop retry/pagination duplicates before any sum is taken.
    deduped = hashed.unique(subset=["record_hash"], keep="first")

    aggregated = deduped.group_by(["billing_date", "account_id", "service_name"]).agg(
        [
            pl.col("cost_amount").sum().alias("cost_amount"),
            pl.col("currency").first().alias("currency"),
            pl.col("is_adjustment").any().alias("is_adjustment"),
            pl.col("record_hash").first().alias("record_hash"),
        ]
    )

    return aggregated.select(list(AGGREGATION_SCHEMA.names())).sort(
        ["billing_date", "account_id", "service_name"]
    )


def upsert_to_parquet(df: pl.DataFrame, base_path: Path) -> int:
    """Idempotent partitioned write: read existing day, union delta, dedup, overwrite."""
    partitions_written = 0
    for (billing_date,), partition_df in df.group_by(["billing_date"]):
        partition_path = base_path / f"billing_date={billing_date}"
        partition_path.mkdir(parents=True, exist_ok=True)
        target = partition_path / "data.parquet"

        if target.exists():
            merged = pl.concat([pl.read_parquet(target), partition_df], how="vertical")
        else:
            merged = partition_df

        # Late-arriving adjustments overwrite stale rows keyed on the same hash.
        final_df = merged.unique(subset=["record_hash"], keep="last")
        final_df.write_parquet(target, compression="zstd")
        partitions_written += 1
        logger.info("Wrote %d rows to %s", final_df.height, target)
    return partitions_written


if __name__ == "__main__":
    mock_items = [
        {"account_id": "123456", "service_name": "AmazonEC2", "line_item_start_time": "2024-05-10T14:30:00Z", "unblended_cost": 12.50, "currency": "USD", "is_adjustment": False},
        {"account_id": "123456", "service_name": "AmazonEC2", "line_item_start_time": "2024-05-10T18:00:00Z", "unblended_cost": 8.75, "currency": "USD", "is_adjustment": False},
        {"account_id": "789012", "service_name": "AmazonS3", "line_item_start_time": "2024-05-11T02:15:00Z", "unblended_cost": 3.20, "currency": "USD", "is_adjustment": True},
    ]
    raw = pl.DataFrame(mock_items)
    daily = aggregate_daily_costs(raw)
    written = upsert_to_parquet(daily, Path("./data/finops/daily_costs"))
    logger.info("Aggregated %d daily rows across %d partitions", daily.height, written)

# Verification & Testing

Correctness here means the same input always yields the same series, and re-running never inflates a total. Assert both directly rather than eyeballing dashboards.

• Idempotency replay. Run aggregate_daily_costs followed by upsert_to_parquet twice on the identical input. The second run must leave every partition’s row count and cost_amount sum unchanged — the dedup hash guarantees it.

def test_aggregation_is_idempotent() -> None:
    items = [
        {"account_id": "1", "service_name": "EC2", "line_item_start_time": "2024-05-10T23:30:00Z", "unblended_cost": 5.0, "currency": "USD", "is_adjustment": False},
    ]
    raw = pl.DataFrame(items)
    first = aggregate_daily_costs(raw)
    # Re-feeding the same rows (a retry replay) must not double the total.
    replay = aggregate_daily_costs(pl.concat([raw, raw]))
    assert first["cost_amount"].sum() == replay["cost_amount"].sum()
    assert first.equals(replay)


if __name__ == "__main__":
    test_aggregation_is_idempotent()
    print("idempotency invariant holds")

• Boundary check. Feed a 23:30Z and a 00:30Z record for the same service; assert they land in different billing_date partitions and neither leaks across the UTC midnight.
• Reconciliation gate. After each run, compare the summed daily series against the provider invoice summary. Any variance beyond a configurable threshold (for example ±0.5%) should raise an alert and trigger a targeted re-aggregation of the affected window rather than a full rescan.
• Dry-run drop count. Run aggregation without the write and log the number of rows dropped by unique. A non-zero count confirms retry or pagination overlap is real and being caught upstream of storage.

# Common Pitfalls Checklist

• Truncating in local time. Local-time truncation shifts spend across midnight. Fix: parse to UTC and truncate with dt.truncate("1d") before casting to Date.
• Hashing mutable cost on revised rows. Including a value that the provider later restates makes a corrected row read as new. Fix: hash only immutable dimensions and let the upsert keep="last" apply the correction.
• Append-only writes. Treating each partition as a static snapshot drops backfilled adjustments on the floor. Fix: read-union-dedup-overwrite the affected day as an appendable ledger.
• Unbounded hash registry. A seen-hashes set that never expires OOMs the worker on multi-terabyte history. Fix: bound it to the 30–45 day reconciliation window or offload to Redis with TTL.
• Silent schema coercion. A new provider tag or currency field can be coerced rather than rejected. Fix: enforce AGGREGATION_SCHEMA so the pipeline fails fast instead of emitting corrupted metrics.

# Frequently Asked Questions

# Related

• Handling Billing API Rate Limits & Retries — the parent layer that wraps every fetch in backoff and feeds clean pages into this rollup.
• Cloud Billing Data Ingestion & Parsing — the end-to-end pipeline whose idempotent-ingestion guarantee this daily series upholds.
• Azure Cost API Pagination and Deduplication — the sibling that solves the same composite-key double-count at the pagination boundary.
• Incremental Sync Strategies for GCP Billing Exports — the watermark-and-merge backfill discipline this page applies to a Parquet store.
• Async Billing Data Processing Patterns — the queue topology and cursor-state contract that scale this aggregation across workers.

Up: Cloud Cost Optimization & FinOps Automation · Parent reference: Handling Billing API Rate Limits & Retries