FinOps Framework Implementation

import logging
import uuid
from decimal import Decimal, ROUND_HALF_UP
from typing import Any, Dict, Iterable, List

import boto3
from botocore.exceptions import ClientError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

logger = logging.getLogger("finops.source.aws")
CENTS = Decimal("0.000001")


class CostExplorerSource:
    """Acquisition plugin: paginated, idempotent AWS Cost Explorer reader."""

    provider = "aws"

    def __init__(self, region: str = "us-east-1") -> None:
        self.client = boto3.client("ce", region_name=region)

    @retry(
        stop=stop_after_attempt(5),
        wait=wait_exponential(multiplier=2, min=4, max=60),
        retry=retry_if_exception_type(ClientError),
        reraise=True,
    )
    def _fetch_page(self, params: Dict[str, Any]) -> Dict[str, Any]:
        try:
            return self.client.get_cost_and_usage(**params)
        except ClientError as exc:
            if exc.response["Error"]["Code"] == "ThrottlingException":
                logger.warning("ce throttled, backing off and retrying")
            raise

    def fetch(self, run_id: str, start: str, end: str) -> Iterable[Dict[str, Any]]:
        """Yield one normalized-ready dict per service/usage_type group."""
        params: Dict[str, Any] = {
            "TimePeriod": {"Start": start, "End": end},
            "Granularity": "DAILY",
            "Metrics": ["UnblendedCost", "AmortizedCost"],
            "GroupBy": [
                {"Type": "DIMENSION", "Key": "SERVICE"},
                {"Type": "DIMENSION", "Key": "USAGE_TYPE"},
            ],
        }
        emitted: List[str] = []
        next_token = None
        while True:
            if next_token:
                params["NextPageToken"] = next_token
            response = self._fetch_page(params)
            for result in response.get("ResultsByTime", []):
                usage_date = result["TimePeriod"]["Start"]
                for group in result.get("Groups", []):
                    metrics = group["Metrics"]
                    yield {
                        "run_id": run_id,
                        "provider": self.provider,
                        "usage_date": usage_date,
                        "service": group["Keys"][0],
                        "usage_type": group["Keys"][1],
                        "unblended_cost": Decimal(metrics["UnblendedCost"]["Amount"]).quantize(
                            CENTS, rounding=ROUND_HALF_UP
                        ),
                        "amortized_cost": Decimal(metrics["AmortizedCost"]["Amount"]).quantize(
                            CENTS, rounding=ROUND_HALF_UP
                        ),
                        "currency": metrics["UnblendedCost"]["Unit"],
                    }
                    emitted.append(f"{usage_date}:{group['Keys'][0]}")
            next_token = response.get("NextPageToken")
            if not next_token:
                break
        logger.info("aws source emitted %d groups for %s..%s", len(emitted), start, end)

from __future__ import annotations

import json
import logging
import uuid
from dataclasses import dataclass, field, replace
from datetime import date, datetime, timezone
from decimal import Decimal, ROUND_HALF_UP
from enum import Enum
from typing import Callable, Dict, Iterable, List, Protocol, Sequence, Set

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

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

CENTS = Decimal("0.000001")


class Stage(str, Enum):
    INGEST = "ingest"
    NORMALIZE = "normalize"
    ALLOCATE = "allocate"
    STORE = "store"
    ACTUATE = "actuate"


class TransientError(Exception):
    """Retryable source/sink failure: throttling, 5xx, network partition."""


@dataclass(frozen=True)
class CostRecord:
    run_id: str
    provider: str
    usage_date: date
    service: str
    usage_type: str
    unblended_cost: Decimal
    amortized_cost: Decimal
    currency: str
    business_unit: str = "unallocated"
    tags: Dict[str, str] = field(default_factory=dict)

    def idempotency_key(self) -> str:
        # Same logical record across re-runs -> identical key -> de-duplicated.
        return f"{self.provider}:{self.usage_date.isoformat()}:{self.service}:{self.usage_type}"


class Source(Protocol):
    provider: str

    def fetch(self, run_id: str, start: date, end: date) -> Iterable[CostRecord]:
        ...


class Sink(Protocol):
    def write(self, records: Sequence[CostRecord]) -> str:
        ...


def _quantize(amount: Decimal) -> Decimal:
    return amount.quantize(CENTS, rounding=ROUND_HALF_UP)


@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=2, min=4, max=60),
    retry=retry_if_exception_type(TransientError),
    reraise=True,
)
def _retryable(fn: Callable[[], List[CostRecord]]) -> List[CostRecord]:
    """Single transient boundary shared by ingest and store."""
    return fn()


class FinOpsPipeline:
    """Deterministic five-stage FinOps orchestrator."""

    def __init__(
        self,
        source: Source,
        sink: Sink,
        allocation_rules: Dict[str, str],
        fx_rates: Dict[str, Decimal],
        base_currency: str = "USD",
        budget_threshold: Decimal = Decimal("100000"),
    ) -> None:
        self.source = source
        self.sink = sink
        self.allocation_rules = allocation_rules
        self.fx_rates = fx_rates
        self.base_currency = base_currency
        self.budget_threshold = budget_threshold
        self.run_id = str(uuid.uuid4())
        self._seen: Set[str] = set()

    # Stage 1 -----------------------------------------------------------------
    def ingest(self, start: date, end: date) -> List[CostRecord]:
        def _pull() -> List[CostRecord]:
            return list(self.source.fetch(self.run_id, start, end))

        raw = _retryable(_pull)
        deduped: List[CostRecord] = []
        for record in raw:
            key = record.idempotency_key()
            if key in self._seen:
                continue  # exactly-once across re-published windows
            self._seen.add(key)
            deduped.append(record)
        logger.info("INGEST run=%s pulled=%d deduped=%d", self.run_id, len(raw), len(deduped))
        return deduped

    # Stage 2 -----------------------------------------------------------------
    def normalize(self, records: Sequence[CostRecord]) -> List[CostRecord]:
        out: List[CostRecord] = []
        for record in records:
            if record.currency == self.base_currency:
                rate = Decimal("1")
            else:
                rate = self.fx_rates.get(record.currency)
                if rate is None:
                    raise ValueError(f"missing FX rate for {record.currency}")
            out.append(
                replace(
                    record,
                    unblended_cost=_quantize(record.unblended_cost * rate),
                    amortized_cost=_quantize(record.amortized_cost * rate),
                    currency=self.base_currency,
                )
            )
        logger.info("NORMALIZE run=%s records=%d base=%s", self.run_id, len(out), self.base_currency)
        return out

    # Stage 3 -----------------------------------------------------------------
    def allocate(self, records: Sequence[CostRecord]) -> List[CostRecord]:
        out = [replace(r, business_unit=self._resolve_owner(r)) for r in records]
        unallocated = sum(1 for r in out if r.business_unit == "unallocated")
        logger.info("ALLOCATE run=%s records=%d unallocated=%d", self.run_id, len(out), unallocated)
        return out

    def _resolve_owner(self, record: CostRecord) -> str:
        for tag_key in ("cost_center", "project", "owner"):  # tier 1: direct tag
            if tag_key in record.tags:
                return record.tags[tag_key]
        owner = self.allocation_rules.get(record.service)  # tier 2: hierarchy rule
        if owner:
            return owner
        return "unallocated"  # tier 3: defer to proportional distribution

    # Stage 4 -----------------------------------------------------------------
    def store(self, records: Sequence[CostRecord]) -> str:
        def _write() -> List[CostRecord]:
            self.sink.write(records)
            return list(records)

        _retryable(_write)
        checkpoint = f"{self.run_id}:{len(records)}"
        logger.info("STORE run=%s checkpoint=%s", self.run_id, checkpoint)
        return checkpoint

    # Stage 5 -----------------------------------------------------------------
    def actuate(self, records: Sequence[CostRecord]) -> Decimal:
        total = sum((r.amortized_cost for r in records), Decimal("0"))
        if total > self.budget_threshold:
            event = {
                "type": "budget.exceeded",
                "run_id": self.run_id,
                "emitted_at": datetime.now(timezone.utc).isoformat(),
                "projected": str(total),
                "threshold": str(self.budget_threshold),
            }
            logger.warning("ACTUATE budget breach: %s", json.dumps(event))
        else:
            logger.info("ACTUATE run=%s total=%s within budget", self.run_id, total)
        return total

    # Orchestration -----------------------------------------------------------
    def run(self, start: date, end: date) -> List[CostRecord]:
        ingested = self.ingest(start, end)
        normalized = self.normalize(ingested)
        allocated = self.allocate(normalized)
        self.store(allocated)
        self.actuate(allocated)
        logger.info("RUN COMPLETE run=%s records=%d", self.run_id, len(allocated))
        return allocated


class _InMemorySource:
    provider = "demo"

    def fetch(self, run_id: str, start: date, end: date) -> Iterable[CostRecord]:
        yield CostRecord(
            run_id=run_id, provider="aws", usage_date=start, service="AmazonEC2",
            usage_type="BoxUsage:m5.large", unblended_cost=Decimal("412.50"),
            amortized_cost=Decimal("370.10"), currency="USD",
            tags={"cost_center": "platform"},
        )
        yield CostRecord(
            run_id=run_id, provider="aws", usage_date=start, service="AmazonRDS",
            usage_type="Multi-AZ", unblended_cost=Decimal("88.20"),
            amortized_cost=Decimal("88.20"), currency="EUR",
        )


class _StdoutSink:
    def write(self, records: Sequence[CostRecord]) -> str:
        return f"wrote {len(records)} records"


if __name__ == "__main__":
    pipeline = FinOpsPipeline(
        source=_InMemorySource(),
        sink=_StdoutSink(),
        allocation_rules={"AmazonRDS": "data-eng"},
        fx_rates={"EUR": Decimal("1.08")},
        budget_threshold=Decimal("100"),
    )
    results = pipeline.run(start=date(2026, 6, 1), end=date(2026, 6, 2))
    for row in results:
        logger.info("row %s -> %s (%s)", row.service, row.business_unit, row.amortized_cost)