cadrumo.domain.calculations.registry.ledger_oss_bindings module

Canonical ledger aggregation binding family.

class OssIossLedgerObservation(**data)[source]

Bases: BaseModel

One factual ledger line tagged with substrate-grounded classification.

Modelo 369 binding selectors filter these observations by the four classification axes (regime, destination Member State, rate tier, invoice direction) plus the optional transaction kind set; the runtime aggregates the matched lines through the binding’s aggregation operator.

Variables:

  • ledger_id – Stable id of the source ledger line.

  • transaction_date – When the supply takes place.

  • regime – OSS / IOSS Esquema the line is filed under.

  • destination_member_state – Member State of consumption (the destination MS for the supply, which determines the applicable IVA rate per the OSS / IOSS rules).

  • rate_kind – Substrate rate tier (general / reduced / etc.).

  • invoice_direction – Whether the autónomo issued or received the invoice.

  • transaction_kind – Substrate cadrumo.domain.iva.classification.TransactionKind the line resolves to.

  • base_amount – Taxable base in EUR.

  • iva_amount – IVA amount in EUR (already applied at the destination MS rate per OSS / IOSS rules).

Parameters:

data (Any)

ledger_id: str
transaction_date: date
regime: OssIossRegime
destination_member_state: EUMemberState
rate_kind: IvaRateKind
invoice_direction: InvoiceKind
transaction_kind: TransactionKind
base_amount: Decimal
iva_amount: Decimal
validate_ledger_oss_aggregation_binding_definition(binding)[source]

Validate a ledger_oss_aggregation binding’s selector and aggregation.

Parameters:

binding (BindingDefinition) – The BindingDefinition to validate. Must have source == "ledger_oss_aggregation".

Raises:

RegistryValidationError – If the selector is malformed (unknown regime / member state / rate kind / invoice direction / transaction kind), or if the aggregation operator is inconsistent with the declared fact.

Return type:

None

resolve_ledger_oss_aggregation_binding_values(revision, observations)[source]

Resolve every ledger_oss_aggregation binding on revision.

For each binding, observations are filtered by the four classification axes plus the transaction-kind set; matched observations are aggregated through the binding’s declared fact (iva_amount_sum defaults; base_amount_sum selects the base). The resolver is deterministic and side-effect-free. Delegates the filter/aggregate skeleton to resolve_ledger_family_binding_values(), shared by every ledger family resolver; the per-selector transaction_kinds set is built once when the matcher closure is constructed for a binding, not once per observation.

Parameters:
Return type:

dict[TypeAliasType, Decimal]

Returns:

Mapping of binding id to the aggregated Decimal value. Empty match sets resolve to Decimal("0").

unsupported_ledger_oss_observations(revision, observations)[source]

Return the OssIossLedgerObservation rows no binding on revision can consume.

Delegates the screen to 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’s own contribution is narrow: the compound regime/destination/rate/ direction/transaction-kind match predicate (reused from the resolver’s _oss_build_matcher, so the transaction_kinds set is built once per binding rather than once per (observation, binding) pair) and a false-fire guard excluding an observation carrying neither base nor IVA (both zero). No extra_exclusion.

Parameters:
Return type:

tuple[OssIossLedgerObservation, ...]

Returns:

Tuple of observations whose non-zero base/cuota is selected by no ledger_oss_aggregation binding.

validate_ledger_oss_aggregation_binding(binding)[source]

Validate a ledger_oss_aggregation binding at registry-build time.

Accumulating list[str] validator: validates the selector shape against LedgerOssProvider (preserving the underlying pydantic field error) then runs the fact/aggregation-op invariant through invariant_diagnostics(), whose raise-style body is validate_ledger_oss_aggregation_binding_definition(). This validator is that body’s only caller, so the invariant is enforced at registry-build time only; resolve_ledger_oss_aggregation_binding_values() re-parses the selector independently through _ledger_oss_selector().

Return type:

list[str]

Parameters:

binding (BindingDefinition)

class LedgerOssProvider(**data)[source]

Bases: BaseModel

Validated form of a ledger_oss_aggregation binding selector.

The selector is expressed in TOML as a mapping of string-valued keys (the registry binding selector contract); this record coerces those strings into substrate enum members at validation time so downstream consumers see typed values.

Parameters:

data (Any)

kind: Literal[BindingSourceKind.LEDGER_OSS_AGGREGATION]
regime: OssIossRegime
destination_member_state: Annotated[EUMemberState, BeforeValidator(_coerce_registry_eu_member_state)]
rate_kind: Annotated[IvaRateKind, BeforeValidator(coerce_enum_member(IvaRateKind))]
invoice_direction: Annotated[InvoiceKind, BeforeValidator(coerce_enum_member(InvoiceKind))]
transaction_kinds: Annotated[tuple[TransactionKind, ...], BeforeValidator(coerce_enum_tuple(TransactionKind))]
fact: OssIossLedgerFact