cadrumo.application.aggregation._grouping module

Shared group-by, name-cache, casilla-fold, period-window, and rollup-total helpers for per-modelo aggregators.

Used by: retenciones, counterpart to bucket observations and cache canonical names; foreign_assets to bucket observations by (source_kind, asset_class) with no name cache; renta_income_ledger, renta_gasto_ledger, impatriado_income_ledger, and irnr_income_ledger to fold their observations into a CasillaAggregation; renta_income_ledger and renta_gasto_ledger to resolve the pago-fraccionado year-to-date window both halves of the Modelo 130 base must share; retenciones, counterpart, and foreign_assets to prove a declared rollup total against the sum of its rollups.

Treat that list as a claim to re-check, not a guarantee. It is accurate for the callers it names and says nothing about the ones it does not: a module that needs one of these mechanisms and never found it will not appear here, and its absence is exactly what an inventory like this hides. Before adding a bucket-by-key loop, a casilla fold, a period window, or a rollup-total check anywhere in this package, look here first.

The name-cache consumers implement the same shape of aggregation: bucket observations by a composite key, then roll up each bucket. They additionally need to resolve a stable human-readable name per (source_kind, identity_nif) pair across multiple observations.

The ledger-projection consumers share a second shape: fold a sequence of single-casilla observations into per-casilla totals plus the CasillaProvenance rows tracing them, differing only in the modelo they target and the amount each observation contributes.

This module extracts both shared mechanisms. The per-domain aggregators retain their domain-specific rollup composition (e.g. counterpart adds country + readiness fields) — only the group-and-name-cache and casilla-fold steps are shared. A projection that groups on more than the casilla axis (the Modelo 100 first-slice expense path, which buckets by casilla AND spending category and emits a populated category_id) keeps its own fold: that is a different grouping shape, not this one under another name.

class CumulativeWindow(period, start, end)[source]

Bases: NamedTuple

The resolved quarter plus the year-to-date span it accumulates over.

Parameters:
period: Period

Alias for field number 0

start: date

Alias for field number 1

end: date

Alias for field number 2

cumulative_year_to_date_window(period)[source]

Return the year-to-date window an IRPF pago fraccionado accumulates over.

RD 439/2007 art. 110.2 computes the Modelo 130 payment on income and expenses accumulated from 1 January of the filing year through the last day of the declared quarter. Both halves of that base – ingresos and gastos – must read the same span for the same quarter, so the rule lives here rather than in either half: it was derived independently in each, and two copies of one legal rule agree only until someone edits one of them.

Raises:

AggregationPeriodError – When period is not quarterly. A pago fraccionado has no meaning outside a quarter, so this refuses rather than inventing a span.

Return type:

CumulativeWindow

Parameters:

period (Period)

group_observations(observations, *, group_key_fn)[source]

Bucket observations by a composite group key, preserving iteration order.

The plain bucket-by-key shape underlying group_and_collect_names() (which calls this internally), for a consumer that only needs the grouping and not the per-identity name cache.

Parameters:
Return type:

dict[TypeVar(GroupKey, bound= tuple[object, ...]), list[TypeVar(T)]]

Returns:

A mapping of each group_key_fn(obs) to the list of observations sharing that key, in insertion order.

group_and_collect_names(observations, *, group_key_fn, identity_key_fn, name_fn)[source]

Bucket observations by group key and cache a canonical name per identity.

Behaviour invariants (shared by both per-modelo aggregators):

  • Iteration order of observations is preserved within each bucket.

  • The first non-empty name_fn(obs) per identity_key_fn(obs) wins; later non-empty names for the same identity are discarded.

  • An empty / falsy name is skipped (does not overwrite a prior win).

Parameters:
Return type:

tuple[dict[TypeVar(GroupKey, bound= tuple[object, ...]), list[TypeVar(T)]], dict[TypeVar(IdentityKey, bound= tuple[object, ...]), str]]

Returns:

A two-tuple (grouped, names) where grouped maps each group_key_fn(obs) to the list of observations sharing that key (insertion order), and names maps each identity_key_fn(obs) to the first non-empty name observed.

filter_observations_for_modelo(observations, *, modelo, catalogue, attribute_fn, aggregator_label)[source]

Keep observations whose classifying attribute is in-scope for modelo.

Shared by both per-modelo aggregators: counterpart filters on operation_kind against the registry’s Modelo 347 / 349 operation catalogue; retenciones filters on scheme against a RetencionScheme catalogue. The only per-domain inputs are the catalogue, the attribute getter, and the label used in the unsupported-modelo error.

Parameters:

  • observations (tuple[TypeVar(T), ...]) – Typed observation records to filter.

  • modelo (str) – The requested modelo code; must key into catalogue.

  • catalogue (Mapping[str, Container[TypeVar(AttrValue)]]) – Maps each supported modelo code to the container of eligible attribute values.

  • attribute_fn (Callable[[TypeVar(T)], TypeVar(AttrValue)]) – Extracts the classifying attribute from each observation.

  • aggregator_label (str) – Human-readable aggregator name for the AggregationUnsupportedModeloError message.

Raises:

AggregationUnsupportedModeloError – When modelo is not a key in catalogue.

Return type:

tuple[TypeVar(T), ...]

Returns:

The observations whose classifying attribute is eligible for modelo, in input order.

class LedgerCasillaObservation(*args, **kwargs)[source]

Bases: Protocol

The read surface a ledger observation exposes to fold_casilla_observations().

Every ledger projection declares its own strict observation model carrying domain-specific fields (gross amount, deductible amount, withheld amount, source jurisdiction). Only these two are load-bearing for the fold: the casilla the row feeds, and the transaction id recorded in its provenance.

Domain-qualified to stay distinct from CasillaObservation, the concrete formula-runtime carrier: this is a structural read surface over the ledger-projection models, keyed on target_casilla_id, not that model’s casilla_id value shape.

property target_casilla_id: CasillaId

The canonical casilla id this observation contributes to.

property transaction_id: str

The originating ledger transaction id, recorded in provenance.

fold_casilla_observations(observations, *, modelo, period, amount_fn)[source]

Fold single-casilla observations into per-casilla totals and provenance.

Behaviour invariants (shared by every ledger-projection consumer):

  • Each observation contributes amount_fn(observation) to the total for its target_casilla_id.

  • Exactly one CasillaProvenance row is emitted per contributing casilla, in sorted casilla order, with category_id unset — this fold groups on the casilla axis alone.

  • Each provenance row carries its contributing transaction ids sorted, and a subtotal summed through the same amount_fn, so a row’s subtotal and the casilla total cannot diverge.

Parameters:

  • observations (Sequence[TypeVar(ObservationT, bound= LedgerCasillaObservation)]) – The eligible observations to fold, in any order.

  • modelo (str) – The modelo identifier the aggregation belongs to, as Modelo.<member>.value.

  • period (Period) – The Period the aggregation covers.

  • amount_fn (Callable[[TypeVar(ObservationT, bound= LedgerCasillaObservation)], Decimal]) – Extracts the fiscally computable amount one observation contributes. Callers pass their own accessor, so an IVA-exclusive base and a gross amount stay distinct concepts.

Return type:

CasillaAggregation

Returns:

The CasillaAggregation carrying the summed casilla values and their provenance rows.

assert_rollup_totals_match(rollups, *, checks)[source]

Raise ValueError for the first declared total that disagrees with its rollups.

Shared by every declarative rollup aggregation (347/349, retenciones, Modelo 720): each declares one or more grand totals that must equal the sum of the corresponding field across its own rollups. checks is an ordered sequence of (label, declared_total, extractor) triples; for each, sum(extractor(row) for row in rollups) must equal declared_total. The first mismatch raises, naming the diverging field, its declared value, and the value the rollups actually sum to, so an operator reading the error knows exactly which magnitude is wrong.

This proves only the arithmetic axis. A rollup-uniqueness invariant (a distinct-NIF count, a no-duplicate-cohort-key rule) is a DIFFERENT kind of check and stays on each aggregation’s own model: the three declaring families genuinely disagree on what “unique” means for their rollups (distinct counterparty NIF, distinct perceptor NIF, distinct (source_kind, asset_class) cohort), so that clause is not shared here.

Parameters:

  • rollups (Sequence[TypeVar(RollupT)]) – The aggregation’s own rollup rows.

  • checks (Sequence[tuple[str, Decimal | int, Callable[[TypeVar(RollupT)], Decimal | int]]]) – Ordered (label, declared_total, extractor) triples to verify. extractor reads the comparable field off one rollup row; the sum starts from type(declared_total)(0) so a Decimal total and an int total both accumulate at their own type.

Raises:

ValueError – The first declared_total that does not equal the sum of extractor(row) over rollups.

Return type:

None