"""Resolve recargo de equivalencia pairings from governed registry data.
The canonical ``iva-recargo-by-applied-rate`` mapping fact owns the legal
pairing, date window, classification, and provenance. This module retains
only typed projection, date selection, and fail-closed lookup mechanics.
"""
from __future__ import annotations
from datetime import date
from decimal import Decimal
from typing import TYPE_CHECKING
from pydantic import BaseModel, Field, model_validator
from ...core.errors.hierarchy import pydantic_validation_boundary
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.unit_proportion import UnitProportion
from ..calculations.registry.facts.resolution import MappingFactQuery, ResolvedMappingFact
from ..calculations.registry.facts.schema import FactSelector, MappingFactPayload
from ..calculations.registry.schema_base import DateAxis
from .errors import IvaCatalogueError, IvaValidationError
if TYPE_CHECKING:
from ..calculations.registry.authority import PinnedAuthorityOperation
IVA_RECARGO_FACT_ID = "iva-recargo-by-applied-rate"
[docs]
class RecargoRateRecord(BaseModel):
"""One governed recargo pairing projected into the public record.
The applied rate is the selector because one classification can have more
than one dated pairing. The authority supplies the legal references and
date window; this record carries them without defining their values.
Attributes:
iva_rate: The IVA rate this recargo accompanies, as a fraction.
recargo_rate: The recargo rate itself, as a fraction. Zero is a
legitimate value and means a rate of zero, not an absent one.
effective_from: First date the pairing applies.
effective_until: Last date it applies, or ``None`` for open-ended.
legal_refs: Registry legal-reference identities establishing the value.
notes: Authoring note; carries no runtime meaning.
"""
model_config = STRICT_FROZEN_CONFIG
iva_rate: UnitProportion
recargo_rate: Decimal = Field(ge=Decimal("0"), lt=Decimal("1"))
effective_from: date
effective_until: date | None = None
legal_refs: tuple[str, ...] = Field(min_length=1)
notes: str = ""
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_window(self) -> RecargoRateRecord:
if self.effective_until is not None and self.effective_from > self.effective_until:
raise IvaValidationError(
f"RecargoRateRecord[iva_rate={self.iva_rate}]: "
f"effective_from {self.effective_from} is after effective_until {self.effective_until}",
)
return self
[docs]
def covers(self, on_date: date) -> bool:
"""Report whether this record's window contains ``on_date``."""
if on_date < self.effective_from:
return False
return self.effective_until is None or on_date <= self.effective_until
[docs]
def load_recargo_rate_table(
*,
operation: PinnedAuthorityOperation,
) -> tuple[RecargoRateRecord, ...]:
"""Return published recargo pairings from the authority artifact.
Runtime callers must provide the operation that owns the published
authority generation; an unsigned authoring tree is never a fallback.
"""
fact = operation.governed_fact(IVA_RECARGO_FACT_ID)
records: list[RecargoRateRecord] = []
for variant in fact.variants:
if not isinstance(variant.payload, MappingFactPayload):
raise IvaCatalogueError("IVA recargo fact contains a non-mapping variant")
selectors = {selector.name: selector.value for selector in variant.selectors}
payload = {str(entry.key): entry.value for entry in variant.payload.entries}
if variant.valid_from is None:
raise IvaCatalogueError("IVA recargo variant has no effective start date")
records.append(
RecargoRateRecord(
iva_rate=Decimal(str(selectors["applied_rate"])),
recargo_rate=Decimal(str(payload["recargo_rate"])),
effective_from=variant.valid_from,
effective_until=variant.valid_to,
legal_refs=variant.legal_refs,
notes=str(payload["notes"]),
)
)
return tuple(records)
[docs]
def recargo_rate_for_applied_rate(
applied_rate: Decimal,
on_date: date,
*,
operation: PinnedAuthorityOperation,
) -> Decimal | None:
"""Return the recargo rate paired with ``applied_rate`` on ``on_date``.
Date-scoped lookup is keyed by the applied rate rather than a broader
category, so overlapping dated pairings remain distinguishable without
reproducing the authority's classifications here.
Args:
applied_rate: The IVA rate the line actually carried, as a fraction.
on_date: The operation date, which selects among windowed pairings.
operation: The pinned authority operation supplying dated recargo pairings.
Returns:
The paired recargo rate, which may legitimately be zero. ``None`` when
the table models no pairing for that rate on that date -- an unmodelled
combination, which callers must not read as "no recargo applies".
"""
if not _recargo_fact_candidate_exists(applied_rate, on_date, operation=operation):
return None
return recargo_rate_record_from_fact(
resolve_recargo_rate_for_applied_rate(
applied_rate,
on_date,
operation=operation,
),
).recargo_rate
[docs]
def resolve_recargo_rate_for_applied_rate(
applied_rate: Decimal,
on_date: date,
*,
operation: PinnedAuthorityOperation,
) -> ResolvedMappingFact:
"""Resolve the dated recargo pairing with its complete governed-fact provenance."""
resolved = operation.resolve_governed_fact(
MappingFactQuery(
fact_id=IVA_RECARGO_FACT_ID,
date_axis=DateAxis.DEVENGO_DATE,
effective_date=on_date,
selectors=(FactSelector(name="applied_rate", value=applied_rate),),
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise IvaCatalogueError("IVA recargo fact must resolve as a mapping fact")
return resolved
def _recargo_fact_candidate_exists(
applied_rate: Decimal,
on_date: date,
*,
operation: PinnedAuthorityOperation,
) -> bool:
"""Return false only for an unmodelled applied-rate/date pairing.
A candidate still resolves through the authority afterwards, so any overlap
or other invalid exact selection remains a loud fail-closed error rather
than being mistaken for the public ``None`` sentinel.
"""
fact = operation.governed_fact(IVA_RECARGO_FACT_ID)
return any(
variant.date_axis is DateAxis.DEVENGO_DATE
and variant.valid_from is not None
and variant.valid_from <= on_date
and (variant.valid_to is None or on_date <= variant.valid_to)
and {selector.name: selector.value for selector in variant.selectors}.get("applied_rate") == applied_rate
for variant in fact.variants
)
[docs]
def recargo_rate_record_from_fact(resolved: ResolvedMappingFact) -> RecargoRateRecord:
"""Project a provenance-bearing authority result onto the retained public record."""
selectors = {selector.name: selector.value for selector in resolved.matched_selectors}
payload = {str(entry.key): entry.value for entry in resolved.payload.entries}
return RecargoRateRecord(
iva_rate=Decimal(str(selectors["applied_rate"])),
recargo_rate=Decimal(str(payload["recargo_rate"])),
effective_from=resolved.valid_from,
effective_until=resolved.valid_to,
legal_refs=resolved.legal_refs,
notes=str(payload["notes"]),
)
__all__ = [
"IVA_RECARGO_FACT_ID",
"RecargoRateRecord",
"load_recargo_rate_table",
"recargo_rate_for_applied_rate",
"recargo_rate_record_from_fact",
"resolve_recargo_rate_for_applied_rate",
]