Indirect Cost Rate Automation

# requirements.txt — pinned for reproducible indirect-cost calculation
pydantic==2.9.2      # rate-agreement contract validation only
python-dateutil==2.9.0.post0   # NICRA effective-date windows
# calculation core uses the stdlib: decimal, dataclasses, enum, logging

import logging
from dataclasses import dataclass, field
from decimal import Decimal, ROUND_HALF_UP
from enum import Enum
from typing import Dict, List

logger = logging.getLogger("indirect_cost.calculator")

CENTS = Decimal("0.01")
SUBAWARD_CAP = Decimal("50000")


class CostElementType(str, Enum):
    SALARIES_WAGES = "salaries_wages"
    FRINGE_BENEFITS = "fringe_benefits"
    MATERIALS_SUPPLIES = "materials_supplies"
    TRAVEL = "travel"
    CONTRACTUAL_SERVICES = "contractual_services"
    EQUIPMENT = "equipment"                 # excluded: capital expenditure
    RENTAL_COSTS = "rental_costs"           # excluded
    TUITION_REMISSION = "tuition_remission" # excluded
    SCHOLARSHIPS = "scholarships"           # excluded (incl. fellowships)
    PARTICIPANT_SUPPORT = "participant_support"  # excluded
    SUBAWARD = "subaward"                   # first $50,000 included, remainder excluded


# 2 CFR §200.1 — elements wholly removed from the MTDC base.
MTDC_EXCLUSIONS: frozenset = frozenset({
    CostElementType.EQUIPMENT,
    CostElementType.RENTAL_COSTS,
    CostElementType.TUITION_REMISSION,
    CostElementType.SCHOLARSHIPS,
    CostElementType.PARTICIPANT_SUPPORT,
})


class IndirectCostError(Exception):
    """Structured calculation failure carrying a machine-readable reason code."""

    def __init__(self, code: str, detail: str) -> None:
        self.code = code
        self.detail = detail
        super().__init__(f"{code}: {detail}")


@dataclass(frozen=True)
class CostElement:
    label: str
    element_type: CostElementType
    amount: Decimal


@dataclass
class MTDCResult:
    mtdc_base: Decimal
    included: List[Dict[str, str]] = field(default_factory=list)
    excluded: List[Dict[str, str]] = field(default_factory=list)


@dataclass
class IndirectResult:
    mtdc_base: Decimal
    applied_rate: Decimal
    indirect_recovery: Decimal
    breakdown: MTDCResult


class IndirectCostCalculator:
    """Compute 2 CFR §200.414 indirect recovery on a 2 CFR §200.1 MTDC base."""

    def __init__(self, ceiling_rate: Decimal, subaward_cap: Decimal = SUBAWARD_CAP) -> None:
        if ceiling_rate < Decimal("0"):
            raise IndirectCostError("ERR_INVALID_CEILING", "ceiling rate must be non-negative")
        self.ceiling_rate = ceiling_rate
        self.subaward_cap = subaward_cap

    def _classify(self, element: CostElement) -> Decimal:
        """Return the portion of one element that enters the MTDC base."""
        if element.amount < Decimal("0"):
            raise IndirectCostError("ERR_NEGATIVE_AMOUNT", f"{element.label} is negative")
        if element.element_type in MTDC_EXCLUSIONS:
            return Decimal("0")
        if element.element_type is CostElementType.SUBAWARD:
            # Only the first $50,000 of each subaward is indirect-bearing.
            return min(element.amount, self.subaward_cap)
        return element.amount

    def compute_mtdc(self, elements: List[CostElement]) -> MTDCResult:
        base = Decimal("0")
        result = MTDCResult(mtdc_base=Decimal("0"))
        for element in elements:
            included_amount = self._classify(element)
            base += included_amount
            record = {
                "label": element.label,
                "type": element.element_type.value,
                "amount": str(element.amount),
                "included_in_base": str(included_amount),
            }
            if included_amount == Decimal("0"):
                record["rationale"] = "excluded from MTDC per 2 CFR §200.1"
                result.excluded.append(record)
            else:
                if included_amount < element.amount:
                    record["rationale"] = "subaward capped at $50,000 per 2 CFR §200.1"
                result.included.append(record)
        result.mtdc_base = base.quantize(CENTS, rounding=ROUND_HALF_UP)
        logger.info(
            "mtdc_computed | base=%s | included=%d | excluded=%d | reg=2_CFR_200.1",
            result.mtdc_base, len(result.included), len(result.excluded),
        )
        return result

    def apply_rate(self, elements: List[CostElement], requested_rate: Decimal) -> IndirectResult:
        if requested_rate < Decimal("0"):
            raise IndirectCostError("ERR_NEGATIVE_RATE", "indirect rate must be non-negative")
        if requested_rate > self.ceiling_rate:
            logger.warning(
                "rate_rejected | requested=%s | ceiling=%s | reg=2_CFR_200.414",
                requested_rate, self.ceiling_rate,
            )
            raise IndirectCostError(
                "ERR_RATE_EXCEEDS_CEILING",
                f"requested {requested_rate} exceeds ceiling {self.ceiling_rate}",
            )
        breakdown = self.compute_mtdc(elements)
        recovery = (breakdown.mtdc_base * requested_rate).quantize(CENTS, rounding=ROUND_HALF_UP)
        logger.info(
            "indirect_computed | base=%s | rate=%s | recovery=%s | reg=2_CFR_200.414",
            breakdown.mtdc_base, requested_rate, recovery,
        )
        return IndirectResult(
            mtdc_base=breakdown.mtdc_base,
            applied_rate=requested_rate,
            indirect_recovery=recovery,
            breakdown=breakdown,
        )

from decimal import Decimal

# Canonical inbound record shape from Field Mapping & Normalization.
RAW_ELEMENT_ALIASES: dict = {
    "Personnel": "salaries_wages",
    "Salaries": "salaries_wages",
    "Fringe": "fringe_benefits",
    "Supplies": "materials_supplies",
    "Equipment": "equipment",
    "Capital Outlay": "equipment",
    "Participant Costs": "participant_support",
    "Stipends to Trainees": "participant_support",
    "Consortium/Subaward": "subaward",
}


def coerce_element(raw_label: str, raw_amount: str) -> CostElement:
    """Map a funder label to a canonical CostElementType with a Decimal amount."""
    canonical = RAW_ELEMENT_ALIASES.get(raw_label)
    if canonical is None:
        raise IndirectCostError("ERR_UNCLASSIFIED_ELEMENT", f"no mapping for {raw_label!r}")
    cleaned = raw_amount.replace("$", "").replace(",", "").strip()
    return CostElement(
        label=raw_label,
        element_type=CostElementType(canonical),
        amount=Decimal(cleaned),
    )

from decimal import Decimal

import pytest

from indirect_cost.calculator import (
    CostElement, CostElementType, IndirectCostCalculator, IndirectCostError,
)


@pytest.fixture
def elements() -> list:
    return [
        CostElement("Personnel", CostElementType.SALARIES_WAGES, Decimal("200000")),
        CostElement("Fringe", CostElementType.FRINGE_BENEFITS, Decimal("60000")),
        CostElement("Equipment", CostElementType.EQUIPMENT, Decimal("85000")),
        CostElement("Participant Costs", CostElementType.PARTICIPANT_SUPPORT, Decimal("40000")),
        CostElement("Consortium", CostElementType.SUBAWARD, Decimal("120000")),
    ]


def test_exclusions_removed_from_base(elements: list) -> None:
    calc = IndirectCostCalculator(ceiling_rate=Decimal("0.50"))
    mtdc = calc.compute_mtdc(elements)
    # 200000 + 60000 + first 50000 of the subaward; equipment & participant support excluded.
    assert mtdc.mtdc_base == Decimal("310000.00")


def test_subaward_capped_not_dropped(elements: list) -> None:
    calc = IndirectCostCalculator(ceiling_rate=Decimal("0.50"))
    mtdc = calc.compute_mtdc(elements)
    subaward = next(r for r in (mtdc.included + mtdc.excluded) if r["type"] == "subaward")
    assert subaward["included_in_base"] == "50000"


def test_de_minimis_recovery_is_exact(elements: list) -> None:
    calc = IndirectCostCalculator(ceiling_rate=Decimal("0.10"))
    result = calc.apply_rate(elements, requested_rate=Decimal("0.10"))
    assert result.indirect_recovery == Decimal("31000.00")


def test_rate_above_ceiling_rejected(elements: list) -> None:
    calc = IndirectCostCalculator(ceiling_rate=Decimal("0.155"))
    with pytest.raises(IndirectCostError) as exc:
        calc.apply_rate(elements, requested_rate=Decimal("0.20"))
    assert exc.value.code == "ERR_RATE_EXCEEDS_CEILING"

from decimal import Decimal

from hypothesis import given, strategies as st

from indirect_cost.calculator import CostElement, CostElementType, IndirectCostCalculator

money = st.decimals(min_value=0, max_value=1_000_000, places=2, allow_nan=False, allow_infinity=False)


@given(salary=money, supplies=money)
def test_recovery_never_exceeds_base_times_ceiling(salary: Decimal, supplies: Decimal) -> None:
    calc = IndirectCostCalculator(ceiling_rate=Decimal("0.10"))
    elements = [
        CostElement("Personnel", CostElementType.SALARIES_WAGES, salary),
        CostElement("Supplies", CostElementType.MATERIALS_SUPPLIES, supplies),
    ]
    result = calc.apply_rate(elements, requested_rate=Decimal("0.10"))
    assert result.indirect_recovery <= (result.mtdc_base * Decimal("0.10")).quantize(Decimal("0.01"))