"""Canonical IVA ledger aggregation binding family."""
from __future__ import annotations
from collections.abc import Callable, Iterable, Mapping, Sequence
from datetime import date
from decimal import Decimal
from typing import TYPE_CHECKING, Annotated, Literal, NamedTuple, Protocol, cast
from pydantic import BaseModel, BeforeValidator, Field, StringConstraints, field_validator, model_validator
from ....core.aggregation import (
BindingAggregationOp,
BindingSourceKind,
)
from ....core.errors.hierarchy import pydantic_validation_boundary
from ....core.iva_deduction_fact import IvaDeductionFactKind
from ....core.models import STRICT_FROZEN_CONFIG
from ....core.time.clock import today_madrid
from ....core.unit_proportion import UnitProportion
from ...iva.components import registry_category_projection
from ...iva.deduction_facts import IvaDeductionClassificationProvenance, validate_iva_deduction_fact
from ...iva.flow import (
IvaFlowDirection,
is_deducible_flow,
is_standard_issued_or_received_flow,
issued_flow_direction,
received_flow_direction,
)
from ...iva.prorrata import InputClassification
from ...iva.schema import (
IvaCashAccountingTreatment,
IvaCategory,
IvaExemptionArticle,
IvaLedgerObservationRole,
IvaRateKind,
default_iva_cash_accounting_treatment,
)
from ._ledger_binding_resolution import (
UnroutedLedgerQuantity,
resolve_ledger_family_binding_values,
unrouted_ledger_family_quantities,
unsupported_ledger_family_observations,
)
from .binding_aggregation import binding_aggregation_op
from .binding_selector_utils import invariant_diagnostics, provider_member, selector_against_model
from .binding_targets import casillas_by_binding
from .errors import RegistryValidationError
from .ids import BindingId
from .iva_category_catalogue import resolve_iva_category_catalogue
from .iva_deduction_catalogue import require_iva_deduction_fact_kind
from .iva_flow_catalogue import require_iva_flow_direction, require_registry_declared_iva_flow_direction
from .iva_rate_kind_catalogue import (
require_iva_rate_kind,
require_registry_declared_iva_rate_kind,
resolve_iva_rate_kind_catalogue,
)
from .iva_schema_vocabulary import (
require_iva_cash_accounting_treatment,
require_iva_exemption_article,
require_registry_declared_iva_cash_accounting_treatment,
)
from .ledger_binding_selector_support import LedgerIvaFact, LedgerIvaFactValue
from .prorrata_vocabulary import require_input_classification
from .quantity_screen_enrolment import assert_quantity_readers_cover_independent_facts, independent_quantity_facts
from .schema_base import coerce_decimal_tuple, coerce_enum_tuple
if TYPE_CHECKING:
from .schema import BindingDefinition, ModeloRevision
IvaCashAccountingTreatmentCode = Annotated[
str,
StringConstraints(min_length=1, pattern=r"^[a-z][a-z0-9_]*$"),
]
"""Authored cash-accounting vocabulary token, validated against the IVA fact catalogue.
The runtime observation still uses :class:`IvaCashAccountingTreatment` because
that enum describes the application's operational routes. A registry binding
is different: its treatment vocabulary is an authored tax classification and
must be checked against the canonical IVA schema fact rather than duplicated
in this module.
"""
def _coerce_cash_accounting_treatment_codes(value: object) -> object:
"""Hydrate TOML arrays before strict tuple validation."""
if isinstance(value, (tuple, list)):
return tuple(cast("Sequence[object]", value))
return value
[docs]
class IvaLedgerObservation(BaseModel):
"""One factual ledger line tagged with the IVA classification triple.
Modelo 303 / 322 / 353 / 309 / 390 binding selectors filter these
observations by category, rate kind, and flow direction; the
runtime aggregates the matched lines through the binding's
aggregation operator.
Attributes:
ledger_id: Stable id of the source ledger line.
transaction_date: When the supply takes place.
category: Substrate :class:`IvaCategory` resolved by the
classifier.
rate_kind: Substrate :class:`IvaRateKind` rate tier.
flow_direction: Substrate :class:`IvaFlowDirection` (output /
input / self-assessed reverse charge).
base_amount: Taxable base in EUR.
iva_amount: IVA amount in EUR (cuota repercutida or soportada,
depending on flow direction).
"""
model_config = STRICT_FROZEN_CONFIG
ledger_id: str = Field(min_length=1, max_length=128)
transaction_date: date
category: IvaCategory
exemption_article: IvaExemptionArticle | None = None
rate_kind: IvaRateKind
flow_direction: IvaFlowDirection
base_amount: Decimal
iva_amount: Decimal
applied_rate: UnitProportion | None = None
"""The numeric IVA rate this line was charged at, as a fraction, when known.
Carried ALONGSIDE :attr:`rate_kind` rather than instead of it, because the
two answer different questions and the annual return asks both. ``rate_kind``
is the semantic tier (general / reducido / super-reducido / zero); this is the
rate that tier resolved to on this line's date. One tier can produce several
rates within a single filing year: the anti-inflation food measures stepped
the super-reducido tier 4 % to 2 % to 0 % and the reducido tier 10 % to 5 %
to 7.5 % across successive extensions, which is why Modelo 390 carries one box
per rate per window where Modelo 303 carries one per tier. Resolving to the
tier and discarding the value made those boxes unpopulatable, because two
lines at 4 % and 2 % arrived downstream indistinguishable.
``None`` where the rate is genuinely unknown rather than zero — a
pre-classified candidate supplied without one, or an exempt line that carries
no rate at all. A caller MUST NOT read ``None`` as 0 %: zero-rated and
rate-less are different declarations, and :class:`IvaRateKind` already
distinguishes ``ZERO`` from ``EXEMPT`` for exactly that reason.
"""
recargo_amount: Decimal = Decimal("0")
"""Recargo de equivalencia cuota the supplier charged on a repercutido sale to
a recargo-regime retailer, in EUR. ``Decimal("0")`` on every line that carries
no recargo. Modelo 303 recargo cuota casillas select these via the
``recargo_amount_sum`` fact, routed by the line's IVA rate tier."""
prorrata_reference_id: str | None = Field(default=None, min_length=1, max_length=128)
"""Stable id of the linked :class:`ProrrataLedgerReference` row, when set.
Populated by the aggregator only on ``SOPORTADO`` (input IVA) flows
that carry a validated prorrata reference. Downstream Modelo 303 /
390 binding selectors filter prorrata-linked observations without
a manual join against the parallel ``prorrata_references`` tuple.
"""
cash_accounting_treatment: IvaCashAccountingTreatment = Field(default_factory=default_iva_cash_accounting_treatment)
"""Independent criterio-de-caja affiliation for this projection."""
observation_role: IvaLedgerObservationRole
"""Whether this is a monetary settlement or an art. 75 information projection.
Cash-accounting operation and payment observations retain the same typed
treatment affiliation. Registry bindings select this orthogonal role so
informational boxes cannot consume monetary settlements, and ordinary IVA
boxes cannot consume the operation-date information projection.
"""
input_classification: InputClassification | None = None
"""Operator-declared LIVA art. 106 prorrata-especial per-input use class.
Carried from the source ledger transaction's
``input_classification``. Meaningful only for ``SOPORTADO`` (input IVA)
rows in a bucket under prorrata especial: the regime-aware ledger IVA
apportionment routes the deducible cuota by this classification (the
art. 106.Uno reglas 100%/0%/general). ``None`` for every row not under
especial or carrying no per-input use declaration; the general-regime
apportionment ignores it.
"""
prorrata_sector_id: str | None = Field(default=None, min_length=1, max_length=64)
"""Operator-declared LIVA arts. 9.1.c / 101 differentiated sector.
Carried from the source ledger transaction's ``prorrata_sector_id``.
Meaningful only for ``SOPORTADO`` (input IVA) rows in a sectorized bucket:
the sector-aware ledger IVA apportionment applies THAT sector's provisional
percentage to the deducible cuota. ``None`` is a common-use input in a
sectorized bucket (apportioned by the art. 104.Dos common percentage) and
the whole-entity default otherwise; the non-sectorized apportionment ignores
it.
"""
deduction_fact_kind: IvaDeductionFactKind | None = None
"""Exact, evidence-grounded differentiated-sector deduction family."""
deduction_provenance: IvaDeductionClassificationProvenance | None = None
investment_asset_id: str | None = Field(default=None, min_length=1, max_length=128)
rectifies_ledger_id: str | None = Field(default=None, min_length=1, max_length=128)
@field_validator("input_classification", mode="before")
@classmethod
def _input_classification_registry_declared(cls, value: object) -> object:
"""Project raw input-use tokens through the dated prorrata catalogue."""
if value is None or isinstance(value, InputClassification):
return value
return require_input_classification(value)
@field_validator("flow_direction", mode="before")
@classmethod
@pydantic_validation_boundary
def _flow_direction_registry_declared(cls, value: object) -> object:
"""Project raw flow tokens, such as a JSON round trip, through fact 0083."""
if isinstance(value, IvaFlowDirection):
return value
return require_iva_flow_direction(value, effective_date=today_madrid())
@field_validator("deduction_fact_kind", mode="before")
@classmethod
@pydantic_validation_boundary
def _deduction_fact_kind_registry_declared(cls, value: object) -> object:
"""Project raw deduction-kind tokens through fact 0085."""
if value is None or isinstance(value, IvaDeductionFactKind):
return value
return require_iva_deduction_fact_kind(value)
@model_validator(mode="after")
@pydantic_validation_boundary
def _enforce_exemption_article_category(self) -> IvaLedgerObservation:
require_iva_rate_kind(self.rate_kind, effective_date=self.transaction_date)
require_iva_cash_accounting_treatment(self.cash_accounting_treatment)
if self.exemption_article is not None:
require_iva_exemption_article(self.exemption_article)
category_catalogue = resolve_iva_category_catalogue(effective_date=self.transaction_date)
if self.exemption_article is not None and self.category != category_catalogue.require("domestic_exempt"):
raise RegistryValidationError(
"exemption_article is only valid when category is DOMESTIC_EXEMPT; "
f"got category {self.category.value!r}",
)
if not is_deducible_flow(self.flow_direction) or self.category == category_catalogue.require(
"recargo_equivalencia"
):
if self.deduction_fact_kind is not None or self.deduction_provenance is not None:
raise RegistryValidationError("output IVA facts cannot carry deduction authority")
return self
if self.deduction_fact_kind is None or self.deduction_provenance is None:
raise RegistryValidationError("input IVA facts require exact deduction authority")
validate_iva_deduction_fact(
kind=self.deduction_fact_kind,
provenance=self.deduction_provenance,
category=self.category,
rate_kind=self.rate_kind,
flow_direction=self.flow_direction,
base_amount=self.base_amount,
iva_amount=self.iva_amount,
investment_asset_id=self.investment_asset_id,
rectifies_ledger_id=self.rectifies_ledger_id,
)
return self
#: The closed fact vocabulary a ``ledger_iva_aggregation`` selector may declare.
#: Named rather than inlined at each check so the quantity screen below can
#: derive its screened set from the same vocabulary the validator enforces and
#: :func:`_iva_aggregate` dispatches on.
_IVA_SUPPORTED_FACTS: frozenset[str] = frozenset(
{"iva_amount_sum", "base_amount_sum", "recargo_amount_sum"},
)
[docs]
class LedgerIvaProvider(BaseModel):
"""Validated form of a ledger_iva_aggregation binding selector."""
model_config = STRICT_FROZEN_CONFIG
kind: Literal[BindingSourceKind.LEDGER_IVA_AGGREGATION] = BindingSourceKind.LEDGER_IVA_AGGREGATION
categories: Annotated[tuple[IvaCategory, ...], BeforeValidator(coerce_enum_tuple(IvaCategory))] = Field(
min_length=1,
)
exemption_articles: (
Annotated[tuple[IvaExemptionArticle, ...], BeforeValidator(coerce_enum_tuple(IvaExemptionArticle))] | None
) = Field(default=None, min_length=1)
rate_kinds: Annotated[tuple[IvaRateKind, ...], BeforeValidator(coerce_enum_tuple(IvaRateKind))] = Field(
min_length=1,
)
flow_direction: Annotated[
IvaFlowDirection,
BeforeValidator(pydantic_validation_boundary(require_registry_declared_iva_flow_direction)),
]
observation_roles: Annotated[
tuple[IvaLedgerObservationRole, ...],
BeforeValidator(coerce_enum_tuple(IvaLedgerObservationRole)),
] = Field(min_length=1)
cash_accounting_treatments: Annotated[
tuple[IvaCashAccountingTreatmentCode, ...],
BeforeValidator(_coerce_cash_accounting_treatment_codes),
] = Field(min_length=1)
applied_rates: Annotated[tuple[Decimal, ...], BeforeValidator(coerce_decimal_tuple)] | None = Field(
default=None,
min_length=1,
)
"""Numeric rates this binding accepts, when the box is rate-specific.
``None`` -- the default and the shape every quarterly binding uses -- means
the binding does not discriminate on the value and matches whatever rates
its ``rate_kinds`` tiers admit. Set it only where the FORM asks per rate
rather than per tier: Modelo 390 carries one box per rate per window because
a tier's rate can change inside a filing year, so its 2 % box must reject a
4 % line that shares the super-reducido tier with it.
An observation whose ``applied_rate`` is ``None`` (rate genuinely unknown --
an invoice-sourced line carries a rate slot, not a number) matches NO
rate-specific binding. That is deliberate: admitting it would put an
unmeasured line in a box that asserts a specific rate, and the annual return
is where that assertion is read.
"""
fact: LedgerIvaFactValue = LedgerIvaFact.IVA_AMOUNT_SUM
@field_validator("categories", mode="after")
@classmethod
@pydantic_validation_boundary
def _categories_registry_declared(cls, value: tuple[IvaCategory, ...]) -> tuple[IvaCategory, ...]:
"""Refuse category tokens absent from the candidate's governed catalogue."""
catalogue = resolve_iva_category_catalogue(effective_date=today_madrid())
return tuple(catalogue.require(str(category)) for category in value)
@field_validator("categories", mode="after")
@classmethod
@pydantic_validation_boundary
def _categories_unique(cls, value: tuple[IvaCategory, ...]) -> tuple[IvaCategory, ...]:
if len(set(value)) != len(value):
raise RegistryValidationError("categories entries must be unique")
return value
@field_validator("rate_kinds", mode="after")
@classmethod
@pydantic_validation_boundary
def _rate_kinds_registry_declared(cls, value: tuple[IvaRateKind, ...]) -> tuple[IvaRateKind, ...]:
"""Refuse binding rate tiers absent from the IVA facts being validated."""
return tuple(require_registry_declared_iva_rate_kind(kind, effective_date=today_madrid()) for kind in value)
@field_validator("rate_kinds", mode="after")
@classmethod
@pydantic_validation_boundary
def _rate_kinds_unique(cls, value: tuple[IvaRateKind, ...]) -> tuple[IvaRateKind, ...]:
if len(set(value)) != len(value):
raise RegistryValidationError("rate_kinds entries must be unique")
return value
@field_validator("cash_accounting_treatments", mode="after")
@classmethod
@pydantic_validation_boundary
def _cash_accounting_treatments_unique(
cls,
value: tuple[IvaCashAccountingTreatmentCode, ...],
) -> tuple[IvaCashAccountingTreatmentCode, ...]:
for token in value:
require_registry_declared_iva_cash_accounting_treatment(token, effective_date=today_madrid())
if len(set(value)) != len(value):
raise RegistryValidationError("cash_accounting_treatments entries must be unique")
return value
@field_validator("observation_roles", mode="after")
@classmethod
@pydantic_validation_boundary
def _observation_roles_unique(
cls,
value: tuple[IvaLedgerObservationRole, ...],
) -> tuple[IvaLedgerObservationRole, ...]:
if len(set(value)) != len(value):
raise RegistryValidationError("observation_roles entries must be unique")
return value
@field_validator("exemption_articles", mode="after")
@classmethod
@pydantic_validation_boundary
def _exemption_articles_unique(
cls,
value: tuple[IvaExemptionArticle, ...] | None,
) -> tuple[IvaExemptionArticle, ...] | None:
if value is not None:
for token in value:
require_iva_exemption_article(token)
if value is not None and len(set(value)) != len(value):
raise RegistryValidationError("exemption_articles entries must be unique")
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _exemption_article_filter_requires_domestic_exempt(self) -> LedgerIvaProvider:
if (
self.exemption_articles is not None
and resolve_iva_category_catalogue(effective_date=today_madrid()).require("domestic_exempt")
not in self.categories
):
raise RegistryValidationError(
"exemption_articles selector requires DOMESTIC_EXEMPT in categories",
)
return self
[docs]
def iva_ledger_selector(binding: BindingDefinition) -> LedgerIvaProvider:
"""Validate and parse a binding selector into a typed IVA selector."""
try:
return provider_member(binding, LedgerIvaProvider)
except (ValueError, TypeError) as exc:
raise RegistryValidationError(f"binding {binding.id!r} has malformed ledger_iva_aggregation selector") from exc
[docs]
class IvaLedgerScreenBinding(NamedTuple):
"""One selected-revision binding used by the invoice-versus-ledger screen.
The screen is deliberately selected from the typed IVA selector rather than
from binding IDs authored in application code. Keeping the parsed selector
alongside its ID lets callers retain the category/rate/flow provenance when
producing diagnostics without introducing a second selector representation.
"""
binding_id: BindingId
selector: LedgerIvaProvider
class _InvoiceLedgerScreenShape(NamedTuple):
"""The selector axes that identify one invoice-screen slot."""
categories: tuple[IvaCategory, ...]
rate_kinds: tuple[IvaRateKind, ...]
flow_direction: IvaFlowDirection
fact: LedgerIvaFact
_INVOICE_LEDGER_SCREEN_OBSERVATION_ROLES: tuple[IvaLedgerObservationRole, ...] = (IvaLedgerObservationRole.SETTLEMENT,)
_INVOICE_LEDGER_SCREEN_FACTS = frozenset(
{
LedgerIvaFact.IVA_AMOUNT_SUM,
LedgerIvaFact.RECARGO_AMOUNT_SUM,
},
)
_INVOICE_LEDGER_SCREEN_MODELOS = frozenset({"303", "390"})
def _invoice_ledger_screen_rate_slots(effective_date: date) -> tuple[_InvoiceLedgerScreenShape, ...]:
"""Project the screen's rate tiers from the governed rate-kind catalogue."""
category_catalogue = resolve_iva_category_catalogue(effective_date=effective_date)
general_category = category_catalogue.require("domestic_general")
reduced_category = category_catalogue.require("domestic_reduced")
super_reduced_category = category_catalogue.require("domestic_super_reduced")
catalogue = resolve_iva_rate_kind_catalogue(effective_date=effective_date)
general = catalogue.for_category(general_category)
reduced = catalogue.for_category(reduced_category)
super_reduced = catalogue.for_category(super_reduced_category)
return (
_InvoiceLedgerScreenShape(
(general_category,),
(general,),
issued_flow_direction(),
LedgerIvaFact.IVA_AMOUNT_SUM,
),
_InvoiceLedgerScreenShape(
(reduced_category,),
(reduced,),
issued_flow_direction(),
LedgerIvaFact.IVA_AMOUNT_SUM,
),
_InvoiceLedgerScreenShape(
(super_reduced_category,),
(super_reduced,),
issued_flow_direction(),
LedgerIvaFact.IVA_AMOUNT_SUM,
),
_InvoiceLedgerScreenShape(
(
general_category,
reduced_category,
super_reduced_category,
),
(general, reduced, super_reduced),
received_flow_direction(),
LedgerIvaFact.IVA_AMOUNT_SUM,
),
_InvoiceLedgerScreenShape(
(general_category,),
(general,),
issued_flow_direction(),
LedgerIvaFact.RECARGO_AMOUNT_SUM,
),
_InvoiceLedgerScreenShape(
(reduced_category,),
(reduced,),
issued_flow_direction(),
LedgerIvaFact.RECARGO_AMOUNT_SUM,
),
_InvoiceLedgerScreenShape(
(super_reduced_category,),
(super_reduced,),
issued_flow_direction(),
LedgerIvaFact.RECARGO_AMOUNT_SUM,
),
)
def _is_invoice_ledger_screen_candidate(
revision: ModeloRevision,
binding: BindingDefinition,
selector: LedgerIvaProvider,
*,
modelo: str,
effective_date: date,
cash_accounting_treatments: tuple[str, ...] | None = None,
) -> bool:
"""Return whether a binding has the screen's typed candidate envelope.
``applied_rates`` is an orthogonal rate-layer axis. M390's rate-specific
annual boxes are not screen candidates because its conceptual total rows
are rate-blind. M303's ordinary reduced/super-reduced rows are qualified
by a rate, while the same revision also declares supplemental transitional
rungs. The selected casilla continuity target distinguishes those layers:
a target carrying the revision's transitional qualifier is a separate rung,
not the conceptual screen slot. This keeps the distinction in authored
registry target identity rather than copying a rate fact into Python.
"""
if selector.applied_rates is not None:
if modelo != "303":
return False
target_ids = casillas_by_binding(revision).get(binding.id, ())
if not target_ids:
return False
if any("transitorio" in str(target_id).casefold() for target_id in target_ids):
return False
rate_kinds = frozenset(resolve_iva_rate_kind_catalogue(effective_date=effective_date).positive_kinds)
return (
selector.exemption_articles is None
and selector.observation_roles == _INVOICE_LEDGER_SCREEN_OBSERVATION_ROLES
and (cash_accounting_treatments is None or selector.cash_accounting_treatments == cash_accounting_treatments)
and is_standard_issued_or_received_flow(selector.flow_direction)
and selector.fact in _INVOICE_LEDGER_SCREEN_FACTS
and set(selector.categories).issubset(
{
resolve_iva_category_catalogue(effective_date=effective_date).require("domestic_general"),
resolve_iva_category_catalogue(effective_date=effective_date).require("domestic_reduced"),
resolve_iva_category_catalogue(effective_date=effective_date).require("domestic_super_reduced"),
},
)
and set(selector.rate_kinds).issubset(rate_kinds)
)
def _invoice_ledger_screen_shape(selector: LedgerIvaProvider) -> _InvoiceLedgerScreenShape:
return _InvoiceLedgerScreenShape(
categories=selector.categories,
rate_kinds=selector.rate_kinds,
flow_direction=selector.flow_direction,
fact=selector.fact,
)
[docs]
def invoice_ledger_screen_bindings(
revision: ModeloRevision,
*,
modelo: str,
) -> tuple[IvaLedgerScreenBinding, ...]:
"""Resolve the invoice screen's seven typed bindings from a revision.
The invoice-versus-ledger refusal applies only to the M303 and M390
revisions. For every other modelo the empty result preserves the previous
non-applicability contract. For an applicable modelo, all seven shapes
must be present exactly once; an incomplete, duplicate, cross-model, or
unknown screen-shaped binding raises a registry validation error rather than
silently weakening the screen.
Binding IDs are returned in a stable conceptual order (three IVA output
tiers, domestic input IVA, then three recargo tiers). The IDs are merely
selected outputs; the selector is the authority used to identify them.
Core types:
:class:`~cadrumo.domain.calculations.registry.schema.ModeloRevision`.
"""
if modelo not in _INVOICE_LEDGER_SCREEN_MODELOS:
return ()
rate_slots = _invoice_ledger_screen_rate_slots(revision.valid_from)
shape_set = frozenset(rate_slots)
expected_by_shape: dict[_InvoiceLedgerScreenShape, list[IvaLedgerScreenBinding]] = {
shape: [] for shape in rate_slots
}
expected_prefix = f"modelo-{modelo}-"
# The screen's treatment vocabulary is a revision declaration. Derive it
# from the first structural candidate instead of restating the allowed
# classifications in Python; authored compilation checks each declaration
# against the canonical IVA vocabulary fact.
screen_cash_accounting_treatments: tuple[str, ...] | None = None
for binding in revision.bindings:
if binding.source != BindingSourceKind.LEDGER_IVA_AGGREGATION:
continue
selector = iva_ledger_selector(binding)
if not _is_invoice_ledger_screen_candidate(
revision,
binding,
selector,
modelo=modelo,
effective_date=revision.valid_from,
cash_accounting_treatments=screen_cash_accounting_treatments,
):
continue
if screen_cash_accounting_treatments is None:
screen_cash_accounting_treatments = selector.cash_accounting_treatments
shape = _invoice_ledger_screen_shape(selector)
if shape not in shape_set:
raise RegistryValidationError(
f"revision {revision.id!r} modelo {modelo!r} has an unknown "
f"invoice IVA screen selector shape {shape!r} on binding {binding.id!r}",
)
if not str(binding.id).startswith(expected_prefix):
raise RegistryValidationError(
f"revision {revision.id!r} modelo {modelo!r} has cross-model invoice IVA screen binding {binding.id!r}",
)
expected_by_shape[shape].append(IvaLedgerScreenBinding(binding.id, selector))
selected: list[IvaLedgerScreenBinding] = []
for shape in rate_slots:
matches = expected_by_shape[shape]
if len(matches) != 1:
state = "missing" if not matches else "duplicate"
raise RegistryValidationError(
f"revision {revision.id!r} modelo {modelo!r} invoice IVA screen "
f"shape is {state}: expected exactly one, got {len(matches)} for {shape!r}",
)
selected.append(matches[0])
selected_ids = tuple(item.binding_id for item in selected)
if len(set(selected_ids)) != len(selected_ids):
raise RegistryValidationError(
f"revision {revision.id!r} modelo {modelo!r} invoice IVA screen resolved duplicate binding IDs",
)
return tuple(selected)
[docs]
def invoice_ledger_screen_binding_ids(
revision: ModeloRevision,
*,
modelo: str,
) -> tuple[BindingId, ...]:
"""Return the selected revision's stable invoice-screen binding IDs.
Core types:
:class:`~cadrumo.domain.calculations.registry.schema.ModeloRevision`.
"""
return tuple(item.binding_id for item in invoice_ledger_screen_bindings(revision, modelo=modelo))
[docs]
def validate_ledger_iva_aggregation_binding_definition(
binding: BindingDefinition,
) -> None:
"""Validate a ``ledger_iva_aggregation`` binding's selector and aggregation.
Args:
binding: The :class:`BindingDefinition` to validate. Must
have ``source == "ledger_iva_aggregation"``.
Raises:
RegistryValidationError: If the selector is malformed (unknown
category / rate kind / flow direction / fact, empty
tuple), if the aggregation operator is not "sum", or if
the binding source is not "ledger_iva_aggregation".
"""
if binding.source != BindingSourceKind.LEDGER_IVA_AGGREGATION:
raise RegistryValidationError(f"binding {binding.id!r} is not a ledger_iva_aggregation source")
diagnostics = selector_against_model(binding, LedgerIvaProvider)
if diagnostics:
raise RegistryValidationError(f"binding {binding.id!r} has malformed IVA selector: {'; '.join(diagnostics)}")
selector = iva_ledger_selector(binding)
if binding.aggregation is not None:
op = binding_aggregation_op(binding)
if op != BindingAggregationOp.SUM:
raise RegistryValidationError(
f"binding {binding.id!r} ledger_iva_aggregation supports only aggregation op 'sum', got {op.value!r}",
)
if selector.fact not in _IVA_SUPPORTED_FACTS:
raise RegistryValidationError(
f"binding {binding.id!r} ledger_iva_aggregation supports only "
f"facts {sorted(_IVA_SUPPORTED_FACTS)!r}, got {selector.fact!r}",
)
try:
_iva_reachability_probe(selector)
except RegistryValidationError as exc:
raise RegistryValidationError(f"binding {binding.id!r} {exc}") from exc
[docs]
class IvaSelectorAxesProtocol(Protocol):
"""The six axes the IVA selector matcher reads off an observation.
Declared so the matcher can state the shape it actually needs instead of
naming the full :class:`IvaLedgerObservation` record. Both that record and
the reachability probe below satisfy it structurally, which is what makes
the probe a legitimate stand-in rather than an unchecked substitution --
previously the probe was passed where the concrete record was declared and
only the absence of a checked seam kept that quiet.
Same shape as :class:`RentaIncomeObservationProtocol` serves for the renta
side of this module.
"""
@property
def category(self) -> IvaCategory:
"""Return the IVA category assigned to the observation."""
...
@property
def rate_kind(self) -> IvaRateKind:
"""Return the IVA rate tier assigned to the observation."""
...
@property
def flow_direction(self) -> IvaFlowDirection:
"""Return the observation's IVA flow direction."""
...
@property
def cash_accounting_treatment(self) -> str:
"""Return the observation's cash-accounting treatment."""
...
@property
def observation_role(self) -> IvaLedgerObservationRole:
"""Return the ledger observation role used by the selector."""
...
@property
def exemption_article(self) -> IvaExemptionArticle | None:
"""Return the exemption article when the observation is exempt."""
...
@property
def applied_rate(self) -> Decimal | None:
"""Return the applied IVA rate as a fraction, when known."""
...
class _IvaReachabilityProbeObservation(NamedTuple):
"""Minimal shape carrying only the six axes the IVA matcher reads.
Stands in for :class:`IvaLedgerObservation` so the probe never needs the
full record's unrelated required fields (ledger id, dates, amounts). The
matcher reads exactly these five attributes, which is the seam that makes
the substitution legitimate.
"""
category: IvaCategory
rate_kind: IvaRateKind
flow_direction: IvaFlowDirection
cash_accounting_treatment: str
observation_role: IvaLedgerObservationRole
exemption_article: IvaExemptionArticle | None
# The probe asks whether a binding is REACHABLE by some observation, so it
# supplies the rate the binding under test names rather than a fixed value:
# a rate-specific binding is reachable, just not by a differently-rated line.
applied_rate: Decimal | None = None
def _iva_reachability_probe(selector: LedgerIvaProvider) -> None:
"""Assert the selector matches at least one constructible observation shape.
Builds a synthetic observation from the selector's OWN declared values and
runs it through the real matcher this family's resolver builds
(:func:`_iva_build_matcher`), not a reimplementation of the match rule.
A selector whose matcher accepts no shape assembled from its own
declarations can never resolve a value: the binding aggregates to zero
forever, indistinguishable from a taxpayer with no IVA of that kind.
This family is where a selector-derived probe genuinely bites, and the
reason is worth stating because it is not true of every family. Every
multi-value axis is non-empty -- ``categories``, ``rate_kinds``,
``observation_roles`` and ``cash_accounting_treatments`` carry
``MinLen(1)``, while ``exemption_articles`` is either absent or non-empty
and ``flow_direction`` is a single enum. The probe therefore confirms the
declared combination itself can reach the matcher without inventing an
implicit observation role or cash-accounting policy.
What this cannot catch, stated so no reader over-trusts it:
* It never touches real ledger data, so it proves the selector CAN match a
shape, never that any real row DOES.
* It cannot catch a matcher that accepts the WRONG rows, only one that
accepts none.
* It cannot catch a resolver that aggregates correctly-matched rows
incorrectly. That residual blind spot needs a data-carrying check, and
no build-time data-free check can observe it.
"""
matcher = _iva_build_matcher(selector)
probe = _IvaReachabilityProbeObservation(
category=selector.categories[0],
rate_kind=selector.rate_kinds[0],
flow_direction=selector.flow_direction,
# Both role and treatment policies are deliberately explicit and
# non-empty; reachability may therefore exercise their first declared
# members without introducing an implicit default.
cash_accounting_treatment=selector.cash_accounting_treatments[0],
observation_role=selector.observation_roles[0],
exemption_article=(selector.exemption_articles[0] if selector.exemption_articles else None),
# Offer the first rate the selector names, so a rate-specific binding is
# asked whether ANY line can reach it rather than whether a rate-less one
# can. A None here would fail every rate-specific binding at build time
# and read as "unreachable" when the binding is simply particular.
applied_rate=(selector.applied_rates[0] if selector.applied_rates else None),
)
if not matcher(probe):
raise RegistryValidationError(
"ledger_iva_aggregation selector matches no constructible observation shape -- "
"the binding can never resolve a value from any ledger data",
)
def _iva_ledger_observation_matches_selector(
observation: IvaSelectorAxesProtocol,
selector: LedgerIvaProvider,
*,
categories: set[IvaCategory],
rate_kinds: set[IvaRateKind],
) -> bool:
if observation.category not in categories:
return False
if observation.rate_kind not in rate_kinds:
return False
if observation.flow_direction != selector.flow_direction:
return False
if observation.cash_accounting_treatment not in set(selector.cash_accounting_treatments):
return False
if observation.observation_role not in set(selector.observation_roles):
return False
if selector.applied_rates is not None and observation.applied_rate not in set(selector.applied_rates):
return False
if selector.exemption_articles is None:
return True
return observation.exemption_article in set(selector.exemption_articles)
def _iva_build_matcher(selector: LedgerIvaProvider) -> Callable[[IvaSelectorAxesProtocol], bool]:
categories = set(selector.categories)
rate_kinds = set(selector.rate_kinds)
def matcher(observation: IvaSelectorAxesProtocol) -> bool:
return _iva_ledger_observation_matches_selector(
observation,
selector,
categories=categories,
rate_kinds=rate_kinds,
)
return matcher
def _iva_aggregate(matched: Sequence[IvaLedgerObservation], selector: LedgerIvaProvider) -> Decimal:
if selector.fact == "iva_amount_sum":
return sum((observation.iva_amount for observation in matched), Decimal("0"))
if selector.fact == "recargo_amount_sum":
return sum((observation.recargo_amount for observation in matched), Decimal("0"))
return sum((observation.base_amount for observation in matched), Decimal("0"))
[docs]
def resolve_ledger_iva_aggregation_binding_values(
revision: ModeloRevision,
observations: Iterable[IvaLedgerObservation],
) -> dict[BindingId, Decimal]:
"""Resolve every ``ledger_iva_aggregation`` binding on ``revision``.
Filters observations by the three classification axes (category in
selector.categories, rate_kind in selector.rate_kinds, flow_direction
matches selector.flow_direction) and aggregates the matched lines'
iva_amount, recargo_amount, or base_amount per the declared fact.
Delegates the filter/aggregate skeleton to
:func:`resolve_ledger_family_binding_values`, shared by every ledger
family resolver; the per-selector ``categories`` / ``rate_kinds`` sets
are built once when the matcher closure is constructed for a binding,
not once per observation.
Args:
revision: The :class:`ModeloRevision` whose bindings to resolve.
observations: Iterable of substrate-classified ledger lines.
Returns:
Mapping of binding id to the aggregated Decimal value. Empty
match sets resolve to ``Decimal("0")``.
"""
return resolve_ledger_family_binding_values(
revision,
observations,
source_kind=BindingSourceKind.LEDGER_IVA_AGGREGATION,
parse_selector=iva_ledger_selector,
build_matcher=_iva_build_matcher,
aggregate=_iva_aggregate,
)
[docs]
def unsupported_ledger_iva_observations(
revision: ModeloRevision,
observations: Iterable[IvaLedgerObservation],
) -> tuple[IvaLedgerObservation, ...]:
"""Return IVA observations no binding on ``revision`` can consume.
Delegates the screen to :func:`unsupported_ledger_family_observations` —
see that function for the shared fail-closed contract (why an unmatched
observation is a modelling gap, not a legitimate zero). This family
carries the campaign's two documented, deliberate asymmetries against
its six siblings — both load-bearing, neither a gap to "fix":
1. **``extra_exclusion`` (no sibling has one).** Categories that bear no
Modelo 303 cuota *by law*
(the published 0084 cuota-less projection — exempt,
zero-rated, not-subject, exempt intra-community supplies/exports,
triangulation, régimen simplificado) are excluded before the binding
check: they correctly match no cuota binding, so flagging them would
be a false positive. After the M303 routing tail (domestic /
intra-community reverse-charge bindings, the import deducible
binding) landed, every cuota-bearing declarable category has a
consuming binding, so the residual unsupported set is empty for the
known declarable categories; the function still fail-closes on any
*new* declarable triple that no binding selects.
2. **``is_declarable`` is ``lambda observation: True`` — this family has
NO zero-amount false-fire guard, unlike every other ledger family.**
This is intentional, not an oversight: adding one would SUPPRESS
real findings the current code fires on
(``no-silent-under-declaration``). Do not add a zero-amount /
zero-cuota guard here to "match" the other six families.
Args:
revision: The :class:`ModeloRevision` whose bindings define the
supported IVA classification triples.
observations: Ledger lines to screen.
Returns:
Tuple of :class:`IvaLedgerObservation` instances not matched by any binding.
"""
return unsupported_ledger_family_observations(
revision,
observations,
source_kind=BindingSourceKind.LEDGER_IVA_AGGREGATION,
parse_selector=iva_ledger_selector,
build_matcher=_iva_build_matcher,
is_declarable=lambda observation: True,
extra_exclusion=lambda observation: observation.category in registry_category_projection("cuota_less_m303"),
)
# The IVA facts excluded from the quantity screen. There are NONE:
# ``base_amount_sum``, ``iva_amount_sum`` and ``recargo_amount_sum`` are three
# INDEPENDENT quantities carried on one line -- the taxable base, the cuota
# charged on it, and the recargo de equivalencia surcharged alongside. No one of
# them stands in for another, so every undrawn one is a real gap.
#
# Contrast the renta side, where three income measures ARE alternatives
# (:data:`_RENTA_INCOME_ALTERNATIVE_MEASURE_FACTS`) and screening all three
# would fire on every revision. The emptiness here is a measured property of
# this family, not an unfilled placeholder, and
# ``test_ledger_quantity_screen_partition`` asserts it so the claim is checked
# rather than left as an absence a later author could fill in unnoticed.
_IVA_ALTERNATIVE_MEASURE_FACTS: Mapping[str, str] = dict[str, str]()
#: DERIVED as the complement, exactly as the renta side derives its own, so the
#: two sets cannot drift apart.
_IVA_INDEPENDENT_QUANTITY_FACTS: frozenset[str] = independent_quantity_facts(
_IVA_SUPPORTED_FACTS,
_IVA_ALTERNATIVE_MEASURE_FACTS,
)
#: Per-fact readers for the independent quantities. Keyed on the same
#: selector-fact vocabulary :func:`_iva_aggregate` dispatches on, so a fact
#: cannot be screened under one reading and resolved under another.
_IVA_INDEPENDENT_QUANTITY_READERS: dict[str, Callable[[IvaLedgerObservation], Decimal]] = {
"base_amount_sum": lambda observation: observation.base_amount,
"iva_amount_sum": lambda observation: observation.iva_amount,
"recargo_amount_sum": lambda observation: observation.recargo_amount,
}
assert_quantity_readers_cover_independent_facts(
"ledger-IVA",
_IVA_INDEPENDENT_QUANTITY_FACTS,
_IVA_INDEPENDENT_QUANTITY_READERS,
)
[docs]
def unrouted_ledger_iva_quantities(
revision: ModeloRevision,
observations: Iterable[IvaLedgerObservation],
) -> tuple[UnroutedLedgerQuantity[IvaLedgerObservation], ...]:
"""Return IVA quantities the rows carry that no binding on ``revision`` draws.
The quantity-keyed sibling of :func:`unsupported_ledger_iva_observations`,
and it watches the axis that screen cannot. That one asks whether a ROW is
selected by some binding; every IVA row carries three independent
quantities, so a row selected for its cuota reads as consumed while its
base imponible reaches nothing at all.
The gap is live, not hypothetical, and it survives a revision declaring the
fact. Both Modelo 303 and Modelo 390 declare ``base_amount_sum`` bindings
covering the domestic tiers, so "is base imponible drawn" answers yes on
each. Their base bindings nonetheless reach no import or reverse-charge row:
``import_third_country``,
``intra_community_acquisition_reverse_charge`` and
``intra_community_service_acquisition_reverse_charge`` have their CUOTA
drawn and their base drawn by nothing, on both modelos. The rows are
consumed for that cuota, so the row screen is silent by construction, and a
coverage test keyed on the fact alone would be silent too.
That is why coverage is asked per row and per fact. The earlier worked
example here — Modelo 390 declaring no base binding at all — was closed by
the annual-form campaign, and closing it is precisely what would have
blinded a fact-keyed screen to the three categories above. A screen's value
is the mechanism, never the instance: this one keeps reporting whatever the
declared bindings fail to reach, and correctly falls silent on the four
domestic categories both modelos now cover.
This function reports; it does not close. Routing an import or
reverse-charge base imponible is registry work with its own casillas and
grounding.
Args:
revision: The :class:`ModeloRevision` whose IVA bindings decide which
facts are drawn, and which rows each drawing binding reaches.
observations: Ledger IVA observations to screen.
Returns:
One :class:`UnroutedLedgerQuantity` per fact with uncovered non-zero
rows, ordered by fact name, each carrying only the rows that fact's
bindings fail to reach.
"""
return unrouted_ledger_family_quantities(
revision,
observations,
source_kind=BindingSourceKind.LEDGER_IVA_AGGREGATION,
parse_selector=iva_ledger_selector,
build_matcher=_iva_build_matcher,
read_fact=lambda selector: selector.fact,
independent_facts=_IVA_INDEPENDENT_QUANTITY_FACTS,
readers=_IVA_INDEPENDENT_QUANTITY_READERS,
)
def _base_iva_selectors(revision: ModeloRevision) -> tuple[LedgerIvaProvider, ...]:
"""Return the revision's ``base_amount_sum`` IVA selectors in source order."""
selectors = [
iva_ledger_selector(binding)
for binding in revision.bindings
if binding.source == BindingSourceKind.LEDGER_IVA_AGGREGATION
]
return tuple(selector for selector in selectors if selector.fact == "base_amount_sum")
def _iva_reachability_probe_for_category(
category: IvaCategory,
selector: LedgerIvaProvider,
) -> _IvaReachabilityProbeObservation | None:
"""Build a probe from a selector when it declares ``category``."""
if category not in selector.categories:
return None
return _IvaReachabilityProbeObservation(
category=category,
rate_kind=selector.rate_kinds[0],
flow_direction=selector.flow_direction,
cash_accounting_treatment=selector.cash_accounting_treatments[0],
observation_role=selector.observation_roles[0],
exemption_article=selector.exemption_articles[0] if selector.exemption_articles else None,
applied_rate=selector.applied_rates[0] if selector.applied_rates else None,
)
def _iva_category_is_reachable(
category: IvaCategory,
matchers: Sequence[tuple[LedgerIvaProvider, Callable[[IvaSelectorAxesProtocol], bool]]],
) -> bool:
"""Return whether a real IVA matcher can accept a declared category probe."""
for selector, matcher in matchers:
probe = _iva_reachability_probe_for_category(category, selector)
if probe is not None and matcher(probe):
return True
return False
[docs]
def structurally_unroutable_iva_base_categories(
revision: ModeloRevision,
*,
out_of_scope: frozenset[IvaCategory] | None = None,
) -> tuple[IvaCategory, ...]:
"""Return :class:`IvaCategory` members no ``base_amount_sum`` binding on ``revision`` could ever reach.
A fourth axis beside :func:`unsupported_ledger_iva_observations` (is this
ROW selected) and :func:`unrouted_ledger_iva_quantities` (is this row's
FACT drawn). Both siblings are OBSERVATION-DEPENDENT: they cannot fire
until a taxpayer's ledger actually holds a row of the affected category,
and :func:`unrouted_ledger_iva_quantities` in particular reports only
*"uncovered non-zero rows"* -- its own docstring states it "must not
manufacture a finding from" a zero total, which is exactly why it needs no
category exclusion set: a cuota-less category's cuota is 0.00 BY LAW, so
it is filtered by the non-zero guard before any exclusion set could matter.
This screen has no rows to filter on, so it needs the opposite of that
guard -- an explicit declaration of what is out of scope, because
"unmentioned" would otherwise mean "unrouted." It answers a question
neither sibling can: *"could this revision's bindings EVER route this
category's base, independent of whether any taxpayer has fallen into the
hole yet?"* True or false from the registry alone, before a single ledger
row exists. For a filing-grade tool that distinction matters: the first
taxpayer with an unrouted reverse-charge base should not be the detector.
Three states result from combining this screen with its siblings, and a
caller must not collapse two of them into one "no finding":
- **unroutable**: this screen fires -- no binding on the revision could
ever draw this category's base, for any observation.
- **unrouted**: this screen is silent (some binding COULD draw it) but
:func:`unrouted_ledger_iva_quantities` fires for a real row -- the
binding exists but a particular row still reached nothing (e.g. a
binding scoped to one flow direction leaves a row of the same category
under a different flow direction uncovered).
- **routed**: both are silent.
``out_of_scope`` is deliberately NOT the published 0084 cuota-less
projection.
That set answers "does this category produce a cuota?"; this screen asks
"does this category's BASE reach some casilla?", and several by-law
cuota-less categories DO carry a real base by law --
the registry-declared zero-rated category is the proof: zero
cuota by definition, base-bearing by law. Reusing CUOTA_LESS here would
suppress exactly the population this screen exists to catch. The default
is M303's own published 0084 out-of-scope declaration; no
generic per-modelo scope mechanism exists in the registry today (a
registry-expressiveness gap in its own right), so a caller working a
different modelo must supply its own set rather than default into M303's.
Uses the real production selector parser (:func:`iva_ledger_selector`)
and matcher (:func:`_iva_build_matcher`) -- never a re-implementation of
the match rule. For each ``base_amount_sum`` binding whose declared
categories include the candidate, a probe observation is assembled from
values the selector's OWN declared axes already admit (one representative
rate kind, the declared flow direction, one representative cash-accounting
treatment, and one representative applied rate / exemption article where
the selector restricts them) and run through the real matcher. This never
guesses a value the selector did not itself declare, so it cannot invent a
false match.
Args:
revision: The :class:`ModeloRevision` whose ``ledger_iva_aggregation``
bindings decide which categories' base is drawn.
out_of_scope: Categories this screen must not evaluate at all, because
"unroutable base" is not a meaningful question for them on this
modelo. When omitted, the published 0084 M303 projection is
resolved fail-closed.
Returns:
Every :class:`IvaCategory` member not in ``out_of_scope`` for which no
``base_amount_sum`` binding on ``revision`` could ever match an
observation of that category, in enum declaration order.
"""
if out_of_scope is None:
out_of_scope = registry_category_projection("m303_base_out_of_scope")
base_selectors = _base_iva_selectors(revision)
matchers = tuple((selector, _iva_build_matcher(selector)) for selector in base_selectors)
return tuple(
category
for category in resolve_iva_category_catalogue(effective_date=revision.valid_from).all_categories
if category not in out_of_scope and not _iva_category_is_reachable(category, matchers)
)
# Ledger Renta estimación directa gastos aggregation source bindings.
#
# These bindings consume first-slice Modelo 100 gastos observations produced
# by the ledger/Renta aggregation layer. They deliberately aggregate already
# evaluated deductible amounts, so proportionality, legal category eligibility,
# invoice reconciliation, and period/date filtering stay outside the registry
# formula runtime.
#
# The registry accesses only four attributes on each observation. A Protocol
# avoids a cross-domain import (domain.calculations -> domain.renta) that
# would violate the hexagonal direction.
# Casilla IDs covered by the first Renta gastos slice (Modelo 100, period 0A).
# These must stay in sync with the binding selectors in the TOML and with
# domain-owned SpendingCategory -> casilla routing projection (which this
# registry-layer module cannot import directly without reversing the hexagonal
# dependency direction); they are validated at registry load time so
# mismatches surface before any calculation. Coverage is currently a subset of
# the full
[docs]
def validate_ledger_iva_aggregation_binding(binding: BindingDefinition) -> list[str]:
"""Validate a ``ledger_iva_aggregation`` binding at registry-build time.
Accumulating ``list[str]`` validator over :class:`LedgerIvaProvider`; runs
the fact/aggregation-op invariant at build time through
:func:`invariant_diagnostics`, whose raise-style body is
:func:`validate_ledger_iva_aggregation_binding_definition`.
"""
failures = selector_against_model(binding, LedgerIvaProvider)
if failures:
return failures
return invariant_diagnostics(binding, "ledger_iva_aggregation", validate_ledger_iva_aggregation_binding_definition)
LedgerIvaProvider = LedgerIvaProvider