cadrumo.domain.calculations.registry._ledger_binding_resolution module

Shared skeleton for the ledger-family binding value resolvers.

Every ledger_<family>_aggregation source (OSS, IVA, renta income, renta gastos estimación directa, renta gastos pago fraccionado, IRNR, impatriado) resolves its bindings with the same shape: filter revision.bindings by source kind, parse each binding’s selector, build a per-selector match predicate over the ledger observations, and fold the matched set through a fact-dispatched Decimal aggregation. The fail-closed sibling (unsupported_ledger_*_observations) shares the same source-filter and predicate machinery, screening for observations no binding selects.

resolve_ledger_family_binding_values() and unsupported_ledger_family_observations() factor that shape out once. Each family module supplies its own selector parser, a build_matcher that closes over the selector’s per-axis membership sets — precomputed once per binding (or once per binding-selector in the fail-closed screen), never rebuilt once per observation — a fact-dispatch aggregate, and, for the fail-closed sibling, an is_declarable false-fire guard plus an optional extra_exclusion for a documented family-specific carve-out (the IVA family’s cuota-less-category exclusion is the sole instance).

Both of those key on the ROW. unrouted_ledger_family_quantities() is the third shape and keys on the QUANTITY: a row a binding DOES consume can carry a second, independent quantity that no binding draws, and the two row-keyed screens are silent on it by construction. Its family adapters supply a fact vocabulary, the subset of that vocabulary that measures one quantity by different rules, and a per-fact reader.

resolve_ledger_family_binding_values(revision, observations, *, source_kind, parse_selector, build_matcher, aggregate)

Resolve every source_kind binding on revision into a Decimal aggregate.

Shared core for the ledger-family scalar resolvers (OSS, IVA, renta income/gastos, IRNR, impatriado). For each matching binding: parse its selector, build that selector’s match predicate once (build_matcher closes over any per-axis membership sets so they are computed once per binding, not once per observation), filter the available observations, and fold the matched set through the family’s fact-dispatch aggregate.

Parameters:

• revision (ModeloRevision) – The ModeloRevision whose bindings to resolve.

• observations (Iterable[TypeVar(ObservationT)]) – The family’s typed ledger observations to aggregate.

• source_kind (BindingSourceKind) – The BindingSourceKind this family’s bindings declare.

• parse_selector (Callable[[BindingDefinition], TypeVar(SelectorT)]) – The family’s selector parser, raising RegistryValidationError on a malformed selector.

• build_matcher (Callable[[TypeVar(SelectorT)], Callable[[TypeVar(ObservationT)], bool]]) – Builds a per-observation match predicate from a parsed selector, closing over any membership sets once.

• aggregate (Callable[[Sequence[TypeVar(ObservationT)], TypeVar(SelectorT)], Decimal]) – Folds the matched observations for one binding, per the selector’s declared fact, into the resolved Decimal.

Return type:

dict[TypeAliasType, Decimal]

Returns:

Mapping of binding id to its resolved Decimal aggregate. A binding whose selector matches no observation resolves to Decimal("0").

unsupported_ledger_family_observations(revision, observations, *, source_kind, parse_selector, build_matcher, is_declarable, extra_exclusion=None)

Return the observations no source_kind binding on revision can consume.

Fail-closed counterpart to resolve_ledger_family_binding_values(), shared by every ledger family’s unsupported_ledger_*_observations. An observation matched by no declared binding has its value silently dropped from the filing — a modelling gap, not a legitimate zero (no-silent-under-declaration).

Parameters:

• revision (ModeloRevision) – The ModeloRevision whose bindings define the supported selector set.

• observations (Iterable[TypeVar(ObservationT)]) – The family’s typed ledger observations to screen.

• source_kind (BindingSourceKind) – The BindingSourceKind this family’s bindings declare.

• parse_selector (Callable[[BindingDefinition], TypeVar(SelectorT)]) – The family’s selector parser.

• build_matcher (Callable[[TypeVar(SelectorT)], Callable[[TypeVar(ObservationT)], bool]]) – Builds a per-observation match predicate from a parsed selector; matchers are built once per binding before screening observations, never rebuilt per (observation, binding) pair.

• is_declarable (Callable[[TypeVar(ObservationT)], bool]) – False-fire guard excluding observations that contribute nothing whether or not they are routed (e.g. a zero-amount row). Every family except IVA declares one; IVA has no such guard and MUST pass lambda observation: True to preserve that documented asymmetry.

• extra_exclusion (Callable[[TypeVar(ObservationT)], bool] | None) – Optional family-specific carve-out evaluated before is_declarable. The sole documented instance is the IVA family’s CUOTA_LESS_M303_IVA_CATEGORIES exclusion; omit for every other family.

Return type:

tuple[TypeVar(ObservationT), ...]

Returns:

Tuple of observations selected by no binding, in input order.

class UnroutedLedgerQuantity(fact, total, observations)

Bases: NamedTuple, Generic

An independent quantity the observations carry that no binding draws.

fact is the selector fact that would have drawn it, total the sum the rows carry, and observations the contributing rows in input order.

Parameters:

• fact (str)

• total (Decimal)

• observations (tuple[ObservationT, ...])

fact: str

Alias for field number 0

total: Decimal

Alias for field number 1

observations: tuple[TypeVar(ObservationT), ...]

Alias for field number 2

unrouted_ledger_family_quantities(revision, observations, *, source_kind, parse_selector, build_matcher, read_fact, independent_facts, readers)

Return independent quantities the rows carry that no binding on revision draws.

The third ledger-family screen, watching an axis the two row-keyed screens cannot see. unsupported_ledger_family_observations() asks whether a ROW is selected by some binding; this one asks, PER ROW AND PER FACT, whether the quantity that row carries is drawn by a binding that actually reaches it.

The distinction is load-bearing wherever one observation carries several quantities: a row consumed for one of them reads as routed while a second, independent quantity it carries reaches nothing at all, so the row-keyed screens stay clean while the value disappears from the filing (no-silent-under-declaration).

The coverage question is asked per (row, fact) rather than per fact, and that is the whole of the difference between this screen working and this screen being decorative. A flat “does any binding declare this fact” set passes the PARTITIONED case: a fact declared by bindings whose selectors reach only OTHER rows counts as drawn for every row, so a quantity carried by the rows those bindings do not select vanishes with both screens clean — the same defect this screen exists to catch, one level in. Modelo 303 is a live instance: its base bindings cover the domestic tiers, intra-community supplies and exports, while its cuota bindings additionally cover the reverse-charge and import categories, so an intra-community acquisition’s base imponible is reached by no base binding at all.

Reports nothing when the rows carry nothing: a zero total is a legitimate zero rather than a modelling gap, and this screen must not manufacture a finding from it.

Parameters:

• revision (ModeloRevision) – The ModeloRevision whose bindings decide which facts are drawn, and which rows each drawing binding reaches.

• observations (Iterable[TypeVar(ObservationT)]) – The family’s typed ledger observations to screen.

• source_kind (BindingSourceKind) – The BindingSourceKind this family’s bindings declare.

• parse_selector (Callable[[BindingDefinition], TypeVar(SelectorT)]) – The family’s selector parser.

• build_matcher (Callable[[TypeVar(SelectorT)], Callable[[TypeVar(ObservationT)], bool]]) – Builds a per-observation match predicate from a parsed selector — the same collaborator the other two shapes take, so “does this binding reach this row” has one definition across all three screens. Matchers are built once per binding, never per (observation, binding) pair.

• read_fact (Callable[[TypeVar(SelectorT)], str]) – Reads the declared fact off a parsed selector.

• independent_facts (frozenset[str]) – The screened set, from independent_quantity_facts().

• readers (Mapping[str, Callable[[TypeVar(ObservationT)], Decimal]]) – Per-fact readers, keyed on the same selector-fact vocabulary the family’s resolver dispatches on so a fact cannot be screened under one reading and resolved under another.

Return type:

tuple[UnroutedLedgerQuantity[TypeVar(ObservationT)], ...]

Returns:

One UnroutedLedgerQuantity per fact with uncovered non-zero rows, ordered by fact name, each carrying only the rows that fact’s bindings fail to reach. Empty when every quantity the rows carry is drawn by a binding selecting them.

Raises:

RegistryValidationError – If a screened fact declares no reader. The import-time gate (assert_quantity_readers_cover_independent_facts()) pins the two together, so reaching here means that gate was bypassed.