"""Repository-backed Modelo 100 Renta expense aggregation.
This is the annual first-slice expense projection behind the
``ledger_renta_gastos_estimacion_directa_aggregation`` source. It loads both a
:class:`~domain.transactions.models.TransactionCatalogue` and a
:class:`~domain.invoices.models.InvoiceCatalogue` from the application-owned catalogue
read capabilities, uses purchase-invoice evidence to validate deductible-expense
facts, and returns
binding-ready :class:`~domain.renta.ledger_expenses.RentaDeductibleExpenseObservation`
records.
The source-mesh resolver in :mod:`~.modelo_bindings` applies the target
:class:`~domain.calculations.registry.schema.ModeloRevision`, resolves registry
bindings, and reports source issues or unrouted expenses on its
:class:`~.source_mesh.CalculationSourceResolution`. The M130 quarterly gasto
projection is intentionally separate in :mod:`~.renta_gasto_ledger`.
Related: :mod:`~.iva_ledger` and :mod:`~.renta_income_ledger` for sibling
ledger projections.
"""
from __future__ import annotations
from collections.abc import Mapping, Sequence
from datetime import date
from decimal import Decimal
from enum import StrEnum
from typing import TYPE_CHECKING, overload
from pydantic import BaseModel, Field
from ...core.casilla_id import CasillaId
from ...core.decimal.constants import ZERO
from ...core.filing_year import FilingYear
from ...core.i18n.translatable import Translatable as tr
from ...core.identity.transaction_ids import TransactionId
from ...core.modelo import Modelo
from ...core.models import STRICT_FROZEN_CONFIG as _STRICT_FROZEN
from ...core.period import Period, PeriodKind
from ...core.prose_elision import IssueDetail
from ...domain.calculations.registry.authority import PinnedAuthorityOperation, bundled_indexed_authority
from ...domain.calculations.registry.facts.resolution import MappingFactQuery, ResolvedMappingFact
from ...domain.calculations.registry.iva_schema_vocabulary import iva_regime_exento_token, require_iva_regime
from ...domain.calculations.registry.prorrata_register_catalogue import regime_apportions_deduction
from ...domain.calculations.registry.schema_base import DateAxis
from ...domain.categories.profile import CategoryProfile
from ...domain.categories.registry import resolve_category_profiles
from ...domain.categories.spending_category import SpendingCategory
from ...domain.contribuyente.ccaa import CCAA
from ...domain.contribuyente.errors import ForalRegimeError, TaxResidenceProfileError
from ...domain.contribuyente.seguro_enfermedad_insured import (
seguro_enfermedad_insured_counts_from_facts,
)
from ...domain.contribuyente.tax_residence import parse_tax_region
from ...domain.invoices.models import Invoice, InvoiceCatalogue
from ...domain.iva.classification import InvoiceKind
from ...domain.prorrata_register.protocols import ProrrataRegisterRepositoryProtocol
from ...domain.renta.ledger_expenses import (
RentaDeductibilityContext,
RentaDeductibilityStatus,
RentaDeductibleExpenseFact,
RentaDeductibleExpenseObservation,
RentaExpenseDirection,
build_renta_deductible_expense_observation,
evaluate_renta_deductibility,
normalize_spending_category,
resolve_region_category_profiles,
select_deductibility_profile,
)
from ...domain.transactions.enums import BusinessClassification, TransactionDirection, TransactionLifecycleState
from ...domain.transactions.models import Transaction, TransactionCatalogue
from ...domain.user_profile.errors import ProfileNotFoundError
from ...domain.user_profile.values import UserProfileRecord
from ..invoices.catalogue_reads_ports import InvoiceCatalogueReadPorts
from ..prorrata_register.service import require_prorrata_register_coordinates_current
from ..user_profile.profile_record_repository import ProfileRecordRepository
from ..user_profile.projections import fact_value, profile_fact_index
from . import _shared_issue_reasons
from ._models import CasillaAggregation, CasillaProvenance, LedgerAggregationResultBase
from ._renta_business_eligibility import (
relies_on_activity_marker,
renta_expense_business_proportion,
)
from .catalogue_read_binding import require_catalogue_reads_bound_to
from .currency_predicates import (
effective_eur_amount,
effective_eur_iva_amount,
effective_eur_taxable_base,
is_non_eur_without_conversion,
)
from .errors import AggregationPeriodError
if TYPE_CHECKING:
from ...domain.calculations.registry.authority_artifact import ProfileDecodeContext
_LEDGER_CATALOGUE_ID = "ledger"
# Registry authority: selected renta-ledger insurance variants, category-profile
# routing, issue/applicability, and binding declarations are consumed through the dated mapping
def _registry_renta_ledger_declarations(
*,
modelo: str,
filing_year: int,
operation: PinnedAuthorityOperation,
) -> dict[str, str]:
"""Resolve the selected renta-ledger declaration mapping without fallbacks."""
effective_date = date(filing_year, 12, 31)
directory = operation.modelo_directory(modelo)
if not any(metadata.contains_date(effective_date) for metadata in directory.revisions):
raise ValueError(f"{modelo} has no registry revision in force on {effective_date.isoformat()}")
resolved = operation.resolve_governed_fact(
MappingFactQuery(
fact_id="renta-ledger-category-variant-routing-mapping",
date_axis=DateAxis.FILING_PERIOD,
effective_date=effective_date,
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise TypeError("renta-ledger declarations must resolve as a mapping fact")
entries: dict[str, str] = {}
for entry in resolved.payload.entries:
if not isinstance(entry.key, str) or not isinstance(entry.value, str):
raise TypeError("renta-ledger declaration entries must be string-to-string")
if entry.key in entries:
raise ValueError(f"duplicate renta-ledger declaration key {entry.key!r}")
entries[entry.key] = entry.value
return entries
def _required_renta_ledger_declaration(entries: Mapping[str, str], key: str) -> str:
value = entries.get(key)
if value is None or not value.strip():
raise ValueError(f"renta-ledger mapping is missing {key!r}")
return value
def _registry_renta_iva_ratio_declarations(
*,
filing_year: int,
operation: PinnedAuthorityOperation,
) -> dict[str, str]:
"""Resolve the dated IVA-ratio policy used by the Renta ledger."""
resolved = operation.resolve_governed_fact(
MappingFactQuery(
fact_id="renta-iva-deduction-ratio-policy",
date_axis=DateAxis.FILING_PERIOD,
effective_date=date(filing_year, 12, 31),
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise TypeError("Renta IVA ratio policy must resolve as a mapping fact")
entries: dict[str, str] = {}
for entry in resolved.payload.entries:
if not isinstance(entry.key, str) or not isinstance(entry.value, str):
raise TypeError("Renta IVA ratio policy entries must be string-to-string")
if entry.key in entries:
raise ValueError(f"duplicate Renta IVA ratio policy key {entry.key!r}")
entries[entry.key] = entry.value
return entries
[docs]
class RentaLedgerAggregationIssueReason(StrEnum):
"""Machine-readable reasons why a ledger row did not produce an observation.
The five upstream filter rejections (``UNSUPPORTED_DIRECTION``,
``UNSUPPORTED_CURRENCY``, ``UNCLASSIFIED_BUSINESS_STATE``,
``PERSONAL_TRANSACTION``, ``OUTSIDE_PERIOD``) are shared with
:class:`~application.aggregation.iva_ledger.IvaLedgerAggregationIssueReason`
through :mod:`._shared_issue_reasons`. ``UNSUPPORTED_PERIOD`` is a
Renta-only refusal raised against quarter-level requests; the
remaining values describe Renta-specific deductibility checks.
"""
UNSUPPORTED_PERIOD = "unsupported_period"
UNSUPPORTED_DIRECTION = _shared_issue_reasons.UNSUPPORTED_DIRECTION
UNSUPPORTED_CURRENCY = _shared_issue_reasons.UNSUPPORTED_CURRENCY
UNCLASSIFIED_BUSINESS_STATE = _shared_issue_reasons.UNCLASSIFIED_BUSINESS_STATE
ACTIVITY_MARKED_PENDING_ANNUAL_REVIEW = "activity_marked_pending_annual_review"
"""An actividad-marked expense the quarterly M130 accepts but the annual declaration holds for review.
Narrower than ``UNCLASSIFIED_BUSINESS_STATE``: the row carries an explicit
``actividad_economica`` IRPF category, so it is not unclassified -- it is
awaiting the business-classification review the filing-grade annual
projection requires. Naming it separately tells the operator that the same
row already fed a pago fraccionado and what action clears it.
"""
PERSONAL_TRANSACTION = _shared_issue_reasons.PERSONAL_TRANSACTION
MISSING_CATEGORY = "missing_category"
UNKNOWN_CATEGORY = "unknown_category"
MISSING_CATEGORY_PROFILE = "missing_category_profile"
OUTSIDE_PERIOD = _shared_issue_reasons.OUTSIDE_PERIOD
MISSING_PURCHASE_INVOICE_EVIDENCE = "missing_purchase_invoice_evidence"
UNSUPPORTED_PURCHASE_INVOICE_EVIDENCE_KIND = "unsupported_purchase_invoice_evidence_kind"
PURCHASE_INVOICE_EVIDENCE_BUCKET_MISMATCH = "purchase_invoice_evidence_bucket_mismatch"
PURCHASE_INVOICE_EVIDENCE_LINK_MISMATCH = "purchase_invoice_evidence_link_mismatch"
PARTIAL_OR_MULTI_TRANSACTION_PURCHASE_INVOICE_EVIDENCE = "partial_or_multi_transaction_purchase_invoice_evidence"
AMOUNT_MISMATCH = "amount_mismatch"
INVALID_LEDGER_FACT = "invalid_ledger_fact"
INELIGIBLE_DEDUCTIBILITY = "ineligible_deductibility"
REGION_UNDECLARED_FOR_OVERRIDE = "region_undeclared_for_override"
#: The traceable-exclusion ``detail`` annotation: elides rather than refusing.
#:
#: These issues explain why a ledger row was excluded, so refusing one over its
#: length would drop the explanation for the exclusion AND fail the aggregation
#: that produced it -- a silent under-declaration dressed as a validation error.
#: Shortening the sentence is strictly the lesser loss.
[docs]
class RentaLedgerAggregationIssue(BaseModel):
"""Traceable exclusion emitted while aggregating ledger rows."""
model_config = _STRICT_FROZEN
transaction_id: TransactionId
purchase_invoice_evidence_id: str | None = Field(default=None, min_length=1, max_length=128)
category_id: str | None = Field(default=None, min_length=1, max_length=128)
reason: RentaLedgerAggregationIssueReason
detail: IssueDetail
class _PurchaseInvoiceEvidencePayload(BaseModel):
"""Typed enrichment fields copied from a reconciled linked invoice."""
model_config = _STRICT_FROZEN
invoice_issue_date: date | None = None
taxable_base: Decimal | None = None
iva_amount: Decimal | None = None
[docs]
class RentaLedgerExpenseAggregation(
LedgerAggregationResultBase[RentaDeductibleExpenseObservation, RentaLedgerAggregationIssue],
):
"""First-slice Renta observations plus binding-ready casilla totals.
Carries no ``out_of_window_summary``: this annual first-slice aggregation
has no repository-backed date partition, unlike the base's four other
known subclasses -- deliberately not declared here (see
:class:`~._models.LedgerAggregationResultBase`).
"""
profile_year: FilingYear
@property
def casilla_values(self) -> Mapping[CasillaId, Decimal]:
"""Return the frozen mapping of binding-ready casilla totals."""
return self.casilla_aggregation.casilla_values
def _seguro_enfermedad_person_counts(
*,
bucket_id: str,
profile_record: UserProfileRecord | None,
filing_year: int,
modelo: str,
profile_decode_context: ProfileDecodeContext | None,
operation: PinnedAuthorityOperation,
) -> dict[str, int]:
"""Resolve insured-person variant counts using selected registry declarations."""
if profile_decode_context is None:
profile_decode_context = operation.profile_decode_context()
declarations = _registry_renta_ledger_declarations(
modelo=modelo,
filing_year=filing_year,
operation=operation,
)
general_variant = _required_renta_ledger_declaration(declarations, "insurance.variant.general")
disability_variant = _required_renta_ledger_declaration(declarations, "insurance.variant.discapacidad")
general_field = _required_renta_ledger_declaration(declarations, "insurance.count_field.general")
disability_field = _required_renta_ledger_declaration(declarations, "insurance.count_field.discapacidad")
record = profile_record
profile_schema = profile_decode_context.schema
if record is None:
try:
repository = ProfileRecordRepository.for_current_session(
bucket_id,
profile_decode_context=profile_decode_context,
)
record = repository.load(bucket_id)
profile_schema = repository.session.profile_decode_context.schema
except ProfileNotFoundError:
return {}
counts = seguro_enfermedad_insured_counts_from_facts(
profile_fact_index(record, profile_schema),
filing_year=filing_year,
)
return {
general_variant: getattr(counts, general_field),
disability_variant: getattr(counts, disability_field),
}
def _resolve_residence_ccaa(
*,
bucket_id: str,
profile_record: UserProfileRecord | None = None,
profile_decode_context: ProfileDecodeContext | None = None,
operation: PinnedAuthorityOperation,
) -> CCAA | None:
"""Derive the ordinary-residence comunidad autonoma from the bucket's profile.
Reads the ``tax_residence.ccaa`` fact from the bucket's user profile and
parses it to the closed :class:`CCAA` enum. Returns ``None`` -- the D4
fail-closed outcome that falls the aggregation through to the state year
profile -- when no profile is selected, the fact is absent, or the recorded
region is a foral regime or otherwise unparseable. The value is inert while
:func:`resolve_region_category_profiles` returns an empty mapping (general
expense deductibility is state base-imponible law, invariant across
comunidades); it is derived here so a future territorial-regime override
selects by residence with no further caller wiring.
Args:
bucket_id: Stable bucket identifier used to load the user profile.
profile_record: Optional :class:`UserProfileRecord` override for testing;
when ``None`` the record is loaded from the bucket.
profile_decode_context: Optional decode context from the caller-held
authority operation used when the record is loaded.
operation: Pinned authority operation supplying the profile decode context.
"""
if profile_decode_context is None:
profile_decode_context = operation.profile_decode_context()
record = profile_record
if record is None:
try:
record = ProfileRecordRepository.for_current_session(
bucket_id,
profile_decode_context=profile_decode_context,
).load(bucket_id)
except ProfileNotFoundError:
return None
raw = fact_value(record, "tax_residence.ccaa")
if raw is None:
return None
try:
return parse_tax_region(raw)
except (ForalRegimeError, TaxResidenceProfileError):
return None
[docs]
def resolve_iva_deduction_ratio(
*,
bucket_id: str,
ejercicio: int,
profile_record: UserProfileRecord | None = None,
prorrata_register_repository: ProrrataRegisterRepositoryProtocol,
profile_decode_context: ProfileDecodeContext | None = None,
operation: PinnedAuthorityOperation | None = None,
) -> Decimal | None:
"""Resolve the activity's IVA-deduction fraction for :attr:`RentaDeductibilityContext.iva_deduction_ratio`.
The selected registry ratio policy supplies the exempt-regime result and
percentage-unit conversion; the prorrata register supplies the period
value for all other regimes. An absent or unresolved register value falls
through to ``None``.
Returns ``None`` -- the historic base-only fallback -- when neither fact
resolves a ratio (no profile, an unparseable regime, no register entry, or
an unresolved percentage). Never fabricates a percentage.
Args:
bucket_id: Stable bucket identifier used to load the profile and register.
ejercicio: Filing year to resolve the register entry for.
profile_record: Optional :class:`UserProfileRecord` override for testing;
when ``None`` the record is loaded from the bucket.
profile_decode_context: Optional decode context from the caller-held
authority operation used when the record is loaded.
prorrata_register_repository: Canonical repository for the bucket's
prorrata register. The caller owns its store selection so a
non-active bucket cannot be shadowed by a process-global default.
operation: Existing generation-pinned authority operation. When omitted,
one indexed operation is opened at this composition boundary.
"""
if operation is None:
with bundled_indexed_authority().operation() as indexed_operation:
return resolve_iva_deduction_ratio(
bucket_id=bucket_id,
ejercicio=ejercicio,
profile_record=profile_record,
prorrata_register_repository=prorrata_register_repository,
profile_decode_context=profile_decode_context,
operation=indexed_operation,
)
if profile_decode_context is None:
profile_decode_context = operation.profile_decode_context()
ratio_policy = _registry_renta_iva_ratio_declarations(filing_year=ejercicio, operation=operation)
exempt_ratio = Decimal(
_required_renta_ledger_declaration(ratio_policy, "exempt_regime.deduction_ratio"),
)
percentage_divisor = Decimal(
_required_renta_ledger_declaration(ratio_policy, "percentage.unit_divisor"),
)
record = profile_record
if record is None:
try:
record = ProfileRecordRepository.for_current_session(
bucket_id,
profile_decode_context=profile_decode_context,
).load(bucket_id)
except ProfileNotFoundError:
record = None
if record is not None:
raw_regime = fact_value(record, "iva.regime")
if raw_regime is not None:
regime = require_iva_regime(str(raw_regime).strip().upper())
if regime == iva_regime_exento_token():
return exempt_ratio
register = require_prorrata_register_coordinates_current(prorrata_register_repository.load(), operation=operation)
entry = register.entry_for(ejercicio, sector_id=None)
if entry is None or not regime_apportions_deduction(entry.regime):
return None
resolution = register.resolve_provisional(ejercicio, sector_id=None)
if resolution.percentage is None:
return None
return resolution.percentage / percentage_divisor
[docs]
def aggregate_renta_ledger_expenses_from_repositories(
*,
bucket_id: str,
period: Period,
ports: InvoiceCatalogueReadPorts,
profile_year: int | None = None,
usage_ratios: Mapping[SpendingCategory, Decimal] | None = None,
activity_key: str = "default",
modelo: str = Modelo("100").value,
profile_record: UserProfileRecord | None = None,
region_category_overrides: Mapping[CCAA, Mapping[SpendingCategory, CategoryProfile]] | None = None,
prorrata_register_repository: ProrrataRegisterRepositoryProtocol,
profile_decode_context: ProfileDecodeContext | None = None,
operation: PinnedAuthorityOperation | None = None,
) -> RentaLedgerExpenseAggregation:
"""Load persisted catalogues and aggregate first-slice Renta expenses.
Derives the ordinary-residence comunidad autonoma from the bucket's active
profile (``tax_residence.ccaa``) and threads it into the aggregation so a
territorial-regime deductibility override selects by residence; the axis is
fail-closed (an undeclared or unparseable region resolves to the state year
profile) and byte-identical while the override layer is empty. Pass
``profile_record`` to supply the :class:`UserProfileRecord` directly (the
residence is otherwise loaded from the bucket), and ``region_category_overrides``
to supply the per-:class:`CCAA` override layer forwarded to
:func:`aggregate_renta_ledger_expenses` (defaulting to the empty
registry-provisioned layer, :func:`resolve_region_category_profiles`).
Also derives the activity's IVA-deduction ratio (:func:`resolve_iva_deduction_ratio`)
from the bucket's ``iva.regime`` profile fact and its
:class:`~domain.prorrata_register.register.ProrrataRegister`, so the non-recoverable
share of input IVA joins the IRPF-deductible cost basis (PGC NRV 12.ª) for
an exempt or prorrata-rationed activity. ``prorrata_register_repository``
supplies the canonical register store explicitly.
Returns a :class:`RentaLedgerExpenseAggregation`.
"""
if operation is None:
with bundled_indexed_authority().operation() as indexed_operation:
return aggregate_renta_ledger_expenses_from_repositories(
bucket_id=bucket_id,
period=period,
ports=ports,
profile_year=profile_year,
usage_ratios=usage_ratios,
activity_key=activity_key,
modelo=modelo,
profile_record=profile_record,
region_category_overrides=region_category_overrides,
prorrata_register_repository=prorrata_register_repository,
profile_decode_context=profile_decode_context,
operation=indexed_operation,
)
if profile_decode_context is None:
profile_decode_context = operation.profile_decode_context()
# NOT pre-filtered by date range: a transaction's OWN
# date can fall outside the requested annual window while its LINKED
# INVOICE's issue date (the actual ``fact.filing_date`` the classifier
# below tests) falls inside it, and the reverse — a transaction inside the
# window whose linked invoice puts it outside. Pre-filtering by the
# transaction's own date with ``load_for_date_range`` silently drops both
# shapes before ``_classify_renta_transaction`` ever runs, so the
# ``OUTSIDE_PERIOD`` diagnostic the operator needs to see never fires for a
# multi-year catalogue. Mirrors the same revert already applied to
# ``_iva_ledger`` / ``_renta_income_ledger`` / ``_renta_gasto_ledger`` /
# ``_impatriado_income_ledger`` for the identical reason.
require_catalogue_reads_bound_to(ports, bucket_id=bucket_id)
transactions = ports.transaction_reader.load()
invoices = ports.invoice_reader.load()
residence_ccaa = _resolve_residence_ccaa(
bucket_id=bucket_id,
profile_record=profile_record,
profile_decode_context=profile_decode_context,
operation=operation,
)
resolved_ejercicio = profile_year if profile_year is not None else period.filing_year
iva_deduction_ratio = resolve_iva_deduction_ratio(
bucket_id=bucket_id,
ejercicio=resolved_ejercicio,
profile_record=profile_record,
prorrata_register_repository=prorrata_register_repository,
profile_decode_context=profile_decode_context,
operation=operation,
)
return aggregate_renta_ledger_expenses(
transactions,
invoices,
bucket_id=bucket_id,
period=period,
profile_year=profile_year,
usage_ratios=usage_ratios,
activity_key=activity_key,
modelo=modelo,
residence_ccaa=residence_ccaa,
region_category_overrides=region_category_overrides,
iva_deduction_ratio=iva_deduction_ratio,
profile_record=profile_record,
profile_decode_context=profile_decode_context,
operation=operation,
)
[docs]
def aggregate_renta_ledger_expenses(
transactions: TransactionCatalogue,
invoices: InvoiceCatalogue,
*,
bucket_id: str,
period: Period,
profile_year: int | None = None,
usage_ratios: Mapping[SpendingCategory, Decimal] | None = None,
activity_key: str = "default",
modelo: str = Modelo("100").value,
residence_ccaa: CCAA | None = None,
region_category_overrides: Mapping[CCAA, Mapping[SpendingCategory, CategoryProfile]] | None = None,
iva_deduction_ratio: Decimal | None = None,
profile_record: UserProfileRecord | None = None,
profile_decode_context: ProfileDecodeContext | None = None,
operation: PinnedAuthorityOperation | None = None,
) -> RentaLedgerExpenseAggregation:
"""Aggregate classified ledger transactions into Renta expense observations.
Args:
transactions: The :class:`TransactionCatalogue` supplying active ledger entries.
invoices: The :class:`InvoiceCatalogue` used for invoice-linked deductibility checks.
bucket_id: Activity bucket identifier for scoping the aggregation.
period: Filing period as a typed :class:`Period` instance.
profile_year: Optional category-profile year override; defaults to the
``period.filing_year`` when ``None``.
usage_ratios: Optional per-:class:`SpendingCategory` business-use ratio
overrides applied before deductibility calculation.
activity_key: Activity identifier carried through to the produced
observations' provenance; defaults to ``"default"``.
modelo: Modelo identifier (``"100"`` IRPF or ``"130"`` pagos
fraccionados) selecting the per-modelo deductibility rules.
residence_ccaa: Optional ordinary residence comunidad autonoma, used only
to select a territorial-regime deductibility override; inert while the
override layer is empty and general expense deductibility is state law.
region_category_overrides: Optional per-:class:`CCAA` category-profile
override layer; defaults to :func:`resolve_region_category_profiles`
(empty) so every fact falls through to the state year profile.
iva_deduction_ratio: Optional activity-wide IVA-deduction fraction
(LIVA arts. 94/104), threaded onto :class:`RentaDeductibilityContext`
so the non-recoverable share of a fact's input IVA joins its
IRPF-deductible cost basis. ``None`` (the default) preserves the
historic base-only behaviour.
profile_record: Optional :class:`UserProfileRecord` override; when ``None``
the record is loaded from the bucket. Supplies the LIRPF art. 30.2.5.a
insured-person counts that split the seguro de enfermedad cap across
its 500 and 1.500 euro limbs.
profile_decode_context: Optional decode context from the caller-held
authority operation when ``profile_record`` is supplied directly.
operation: Existing generation-pinned authority operation. When omitted,
one indexed operation is opened at this composition boundary.
Returns a :class:`RentaLedgerExpenseAggregation` containing the accepted
observations, exclusion issues, and binding-ready casilla totals.
"""
if operation is None:
with bundled_indexed_authority().operation() as indexed_operation:
return aggregate_renta_ledger_expenses(
transactions,
invoices,
bucket_id=bucket_id,
period=period,
profile_year=profile_year,
usage_ratios=usage_ratios,
activity_key=activity_key,
modelo=modelo,
residence_ccaa=residence_ccaa,
region_category_overrides=region_category_overrides,
iva_deduction_ratio=iva_deduction_ratio,
profile_record=profile_record,
profile_decode_context=profile_decode_context,
operation=indexed_operation,
)
if profile_decode_context is None:
profile_decode_context = operation.profile_decode_context()
resolved_period = _resolve_annual_period(period)
resolved_profile_year = profile_year if profile_year is not None else resolved_period.filing_year
profiles = resolve_category_profiles(resolved_profile_year, operation=operation)
region_overrides = (
region_category_overrides
if region_category_overrides is not None
else resolve_region_category_profiles(resolved_profile_year)
)
context = RentaDeductibilityContext(
profile_year=resolved_profile_year,
usage_ratios=dict(usage_ratios or {}),
residence_ccaa=residence_ccaa,
iva_deduction_ratio=iva_deduction_ratio,
statutory_cap_variant_person_counts=_seguro_enfermedad_person_counts(
bucket_id=bucket_id,
profile_record=profile_record,
filing_year=resolved_profile_year,
modelo=modelo,
profile_decode_context=profile_decode_context,
operation=operation,
),
)
observations: list[RentaDeductibleExpenseObservation] = []
issues: list[RentaLedgerAggregationIssue] = []
for transaction in transactions.values():
if transaction.lifecycle_state is not TransactionLifecycleState.ACTIVE:
continue
if transaction.business_classification is BusinessClassification.REVIEWED_EXCLUDED:
# Operator reviewed and deliberately excluded this row from filing
# (a final disposition): omit it silently, no observation and no
# advisory-bearing issue.
continue
outcome = _classify_renta_transaction(
transaction,
invoices=invoices,
bucket_id=bucket_id,
resolved_period=resolved_period,
resolved_profile_year=resolved_profile_year,
profiles=profiles,
region_overrides=region_overrides,
context=context,
activity_key=activity_key,
)
if isinstance(outcome, RentaLedgerAggregationIssue):
issues.append(outcome)
else:
observations.append(outcome)
casilla_aggregation = _casilla_aggregation(resolved_period, observations, modelo=modelo)
return RentaLedgerExpenseAggregation(
modelo=modelo,
period=resolved_period,
profile_year=resolved_profile_year,
observations=tuple(observations),
issues=tuple(issues),
casilla_aggregation=casilla_aggregation,
)
def _classify_renta_transaction(
transaction: Transaction,
*,
invoices: InvoiceCatalogue,
bucket_id: str,
resolved_period: Period,
resolved_profile_year: int,
profiles: Mapping[SpendingCategory, CategoryProfile],
region_overrides: Mapping[CCAA, Mapping[SpendingCategory, CategoryProfile]],
context: RentaDeductibilityContext,
activity_key: str,
) -> RentaDeductibleExpenseObservation | RentaLedgerAggregationIssue:
"""Filter + classify one ledger transaction against the Renta-expense pipeline.
Returns either a typed :class:`RentaDeductibleExpenseObservation`
(the transaction survives every gate and is eligible) or a typed
:class:`RentaLedgerAggregationIssue` (any gate rejects it). The
twelve original early-return branches each project to a typed
issue with their dedicated reason code; the eligible path returns
the constructed observation.
"""
direction = _renta_direction_for(transaction.direction, transaction.purchase_invoice_evidence_id)
if direction is None:
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.UNSUPPORTED_DIRECTION,
f"transaction direction {transaction.direction.value!r} is not a Renta expense flow",
)
if is_non_eur_without_conversion(transaction):
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.UNSUPPORTED_CURRENCY,
f"transaction currency {transaction.raw.currency!r} is not supported for Renta expenses",
)
# Use the EUR projection after rejecting unconverted non-EUR rows, so converted
# rows contribute their EUR equivalent while EUR rows retain their raw amount.
eur_amount = effective_eur_amount(transaction)
proportion = _renta_business_proportion_or_issue(transaction)
if isinstance(proportion, RentaLedgerAggregationIssue):
return proportion
category_profile = _renta_category_profile_or_issue(
transaction,
profiles=profiles,
region_overrides=region_overrides,
context=context,
profile_year=resolved_profile_year,
)
if isinstance(category_profile, RentaLedgerAggregationIssue):
return category_profile
category, profile = category_profile
fact = _renta_expense_fact_or_issue(
transaction,
invoices=invoices,
bucket_id=bucket_id,
direction=direction,
category=category,
activity_key=activity_key,
eur_amount=eur_amount,
proportion=proportion,
)
if isinstance(fact, RentaLedgerAggregationIssue):
return fact
return _renta_observation_or_issue(
transaction,
fact=fact,
profile=profile,
context=context,
period=resolved_period,
)
def _renta_business_proportion_or_issue(
transaction: Transaction,
) -> Decimal | RentaLedgerAggregationIssue:
"""Return the annual filing business proportion or its precise refusal."""
# Annual filing deliberately refuses the actividad marker that quarterly M130
# accepts, through the shared predicate rather than a second policy copy.
proportion = renta_expense_business_proportion(transaction, accept_activity_marker=False)
if proportion is not None:
return proportion
if relies_on_activity_marker(transaction):
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.ACTIVITY_MARKED_PENDING_ANNUAL_REVIEW,
"actividad-marked expense needs business-classification review before the annual declaration",
)
reason = (
RentaLedgerAggregationIssueReason.PERSONAL_TRANSACTION
if transaction.business_classification is BusinessClassification.PERSONAL
else RentaLedgerAggregationIssueReason.UNCLASSIFIED_BUSINESS_STATE
)
return _renta_transaction_issue(
transaction,
reason,
f"business classification {transaction.business_classification.value!r} cannot feed Renta expenses",
)
def _renta_category_profile_or_issue(
transaction: Transaction,
*,
profiles: Mapping[SpendingCategory, CategoryProfile],
region_overrides: Mapping[CCAA, Mapping[SpendingCategory, CategoryProfile]],
context: RentaDeductibilityContext,
profile_year: int,
) -> tuple[SpendingCategory, CategoryProfile] | RentaLedgerAggregationIssue:
"""Resolve the first-slice category and the profile selected for residence."""
category_id = transaction.category_id
if category_id is None:
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.MISSING_CATEGORY,
"classified expense transaction has no ledger category",
)
try:
category = normalize_spending_category(category_id)
except ValueError:
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.UNKNOWN_CATEGORY,
f"ledger category {category_id!r} is not in the spending taxonomy",
)
profile = profiles.get(category)
if profile is None:
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.MISSING_CATEGORY_PROFILE,
f"category {category.value!r} has no profile for {profile_year}",
)
category_region_overrides = {
ccaa: overrides[category] for ccaa, overrides in region_overrides.items() if category in overrides
}
selected_profile = select_deductibility_profile(
state_profile=profile,
region_override_profiles=category_region_overrides,
context=context,
)
if selected_profile is None:
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.REGION_UNDECLARED_FOR_OVERRIDE,
(
f"category {category.value!r} carries a territorial-regime deductibility override for "
f"{profile_year} but the residence comunidad autonoma is undeclared"
),
)
return category, selected_profile
def _renta_expense_fact_or_issue(
transaction: Transaction,
*,
invoices: InvoiceCatalogue,
bucket_id: str,
direction: RentaExpenseDirection,
category: SpendingCategory,
activity_key: str,
eur_amount: Decimal,
proportion: Decimal,
) -> RentaDeductibleExpenseFact | RentaLedgerAggregationIssue:
"""Enrich a transaction with invoice evidence and construct its expense fact."""
evidence_payload = _purchase_invoice_evidence_payload(
invoices=invoices,
bucket_id=bucket_id,
transaction_id=transaction.transaction_id,
purchase_invoice_evidence_id=transaction.purchase_invoice_evidence_id,
category_id=transaction.category_id,
transaction_amount=eur_amount,
)
if isinstance(evidence_payload, RentaLedgerAggregationIssue):
return evidence_payload
try:
return RentaDeductibleExpenseFact(
transaction_id=transaction.transaction_id,
invoice_id=transaction.purchase_invoice_evidence_id,
catalogue_id=_LEDGER_CATALOGUE_ID,
operation_date=transaction.raw.value_date or transaction.raw.booked_date,
invoice_issue_date=evidence_payload.invoice_issue_date,
posting_date=transaction.raw.booked_date,
gross_amount=_business_fact_amount(eur_amount, proportion),
taxable_base=_business_fact_amount(_taxable_base_for(transaction, evidence_payload), proportion),
iva_amount=_business_fact_amount(_iva_amount_for(transaction, evidence_payload), proportion),
direction=direction,
category=category,
activity_key=activity_key,
)
except ValueError as exc:
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.INVALID_LEDGER_FACT,
str(exc),
)
def _renta_observation_or_issue(
transaction: Transaction,
*,
fact: RentaDeductibleExpenseFact,
profile: CategoryProfile,
context: RentaDeductibilityContext,
period: Period,
) -> RentaDeductibleExpenseObservation | RentaLedgerAggregationIssue:
"""Apply filing-period and deductibility gates to one validated fact."""
if not period.contains(fact.filing_date):
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.OUTSIDE_PERIOD,
f"filing date {fact.filing_date.isoformat()} is outside {period}",
)
result = evaluate_renta_deductibility(fact, profile, context)
if result.status is not RentaDeductibilityStatus.ELIGIBLE:
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.INELIGIBLE_DEDUCTIBILITY,
result.reason,
)
try:
return build_renta_deductible_expense_observation(fact, result, tax_year=period.filing_year)
except ValueError as exc:
return _renta_transaction_issue(
transaction,
RentaLedgerAggregationIssueReason.INVALID_LEDGER_FACT,
str(exc),
)
def _renta_transaction_issue(
transaction: Transaction,
reason: RentaLedgerAggregationIssueReason,
detail: str,
) -> RentaLedgerAggregationIssue:
return RentaLedgerAggregationIssue(
transaction_id=transaction.transaction_id,
purchase_invoice_evidence_id=transaction.purchase_invoice_evidence_id,
category_id=transaction.category_id,
reason=reason,
detail=detail,
)
def _resolve_annual_period(period: Period) -> Period:
resolved = period
if resolved.kind is not PeriodKind.ANNUAL:
raise AggregationPeriodError(
tr("aggregation.renta_ledger.errors.annual_period_required"),
context={"period": str(resolved)},
)
return resolved
def _renta_direction_for(
direction: TransactionDirection,
purchase_invoice_evidence_id: str | None,
) -> RentaExpenseDirection | None:
if direction is TransactionDirection.OUTGOING:
return RentaExpenseDirection.OUTGOING_EXPENSE
if direction is TransactionDirection.INCOMING and purchase_invoice_evidence_id is not None:
return RentaExpenseDirection.REFUND
return None
@overload
def _business_fact_amount(amount: None, proportion: Decimal) -> None: ...
@overload
def _business_fact_amount(amount: Decimal, proportion: Decimal) -> Decimal: ...
def _business_fact_amount(amount: Decimal | None, proportion: Decimal) -> Decimal | None:
if amount is None:
return None
if amount < ZERO:
raise ValueError("ledger amount must be a non-negative magnitude")
return amount * proportion
def _purchase_invoice_evidence_payload(
*,
invoices: InvoiceCatalogue,
bucket_id: str,
transaction_id: str,
purchase_invoice_evidence_id: str | None,
category_id: str | None,
transaction_amount: Decimal,
) -> _PurchaseInvoiceEvidencePayload | RentaLedgerAggregationIssue:
if purchase_invoice_evidence_id is None:
return _PurchaseInvoiceEvidencePayload()
invoice = invoices.get(purchase_invoice_evidence_id)
if invoice is None:
return _missing_purchase_invoice_issue(
transaction_id=transaction_id,
purchase_invoice_evidence_id=purchase_invoice_evidence_id,
category_id=category_id,
)
reference_issue = _purchase_invoice_reference_issue(
invoice,
bucket_id=bucket_id,
transaction_id=transaction_id,
purchase_invoice_evidence_id=purchase_invoice_evidence_id,
category_id=category_id,
)
if reference_issue is not None:
return reference_issue
return _purchase_invoice_payload_for_totals(
invoice,
transaction_id=transaction_id,
purchase_invoice_evidence_id=purchase_invoice_evidence_id,
category_id=category_id,
transaction_amount=transaction_amount,
)
def _missing_purchase_invoice_issue(
*,
transaction_id: str,
purchase_invoice_evidence_id: str,
category_id: str | None,
) -> RentaLedgerAggregationIssue:
"""Describe an evidence reference absent from the confirmed invoice catalogue."""
# The evidence reference addresses two id spaces and only the invoice
# catalogue carries the fiscal totals this fold-in needs, so a miss here is
# NOT proof the reference is broken: it is also what a legitimately
# registered, not-yet-confirmed evidence record looks like from this side.
# Naming both cases keeps the operator from hunting a phantom missing record.
return RentaLedgerAggregationIssue(
transaction_id=transaction_id,
purchase_invoice_evidence_id=purchase_invoice_evidence_id,
category_id=category_id,
reason=RentaLedgerAggregationIssueReason.MISSING_PURCHASE_INVOICE_EVIDENCE,
detail=(
"transaction references no confirmed invoice in the catalogue, so this expense carries no "
"invoice totals to fold in; if the reference names a registered evidence record, confirming "
"it into an invoice does NOT repoint this transaction, and no verb currently replaces an "
"already-set purchase-invoice-evidence reference -- so record the expense against the "
"confirmed invoice on a transaction that has no evidence reference yet"
),
)
def _purchase_invoice_reference_issue(
invoice: Invoice,
*,
bucket_id: str,
transaction_id: str,
purchase_invoice_evidence_id: str,
category_id: str | None,
) -> RentaLedgerAggregationIssue | None:
"""Validate bucket, kind, and reciprocal-link constraints for one invoice."""
if invoice.bucket_id != bucket_id:
return RentaLedgerAggregationIssue(
transaction_id=transaction_id,
purchase_invoice_evidence_id=purchase_invoice_evidence_id,
category_id=category_id,
reason=RentaLedgerAggregationIssueReason.PURCHASE_INVOICE_EVIDENCE_BUCKET_MISMATCH,
detail="transaction references purchase invoice evidence outside the active bucket",
)
if invoice.kind is not InvoiceKind.RECEIVED:
return RentaLedgerAggregationIssue(
transaction_id=transaction_id,
purchase_invoice_evidence_id=purchase_invoice_evidence_id,
category_id=category_id,
reason=RentaLedgerAggregationIssueReason.UNSUPPORTED_PURCHASE_INVOICE_EVIDENCE_KIND,
detail=f"purchase invoice evidence kind {invoice.kind.value!r} is not RECEIVED",
)
if transaction_id not in invoice.linked_transaction_ids:
return RentaLedgerAggregationIssue(
transaction_id=transaction_id,
purchase_invoice_evidence_id=purchase_invoice_evidence_id,
category_id=category_id,
reason=RentaLedgerAggregationIssueReason.PURCHASE_INVOICE_EVIDENCE_LINK_MISMATCH,
detail="transaction and purchase invoice evidence links are not reciprocal",
)
if len(invoice.linked_transaction_ids) != 1:
return RentaLedgerAggregationIssue(
transaction_id=transaction_id,
purchase_invoice_evidence_id=purchase_invoice_evidence_id,
category_id=category_id,
reason=RentaLedgerAggregationIssueReason.PARTIAL_OR_MULTI_TRANSACTION_PURCHASE_INVOICE_EVIDENCE,
detail="first-slice aggregation only accepts one transaction per purchase invoice evidence record",
)
return None
def _purchase_invoice_payload_for_totals(
invoice: Invoice,
*,
transaction_id: str,
purchase_invoice_evidence_id: str,
category_id: str | None,
transaction_amount: Decimal,
) -> _PurchaseInvoiceEvidencePayload | RentaLedgerAggregationIssue:
"""Reconcile EUR totals and project the linked invoice's evidence payload."""
# invoice.grand_total is denominated in invoice.currency (native), while
# transaction_amount is the caller's already-EUR-converted eur_amount (see
# effective_eur_amount). Comparing them directly mixes currencies: a
# legitimate foreign-currency invoice would compare its native total
# against a converted EUR figure and mismatch even when the two documents
# agree, so this compares grand_total_eur instead. An invoice that cannot
# itself resolve a EUR total (foreign currency, unconverted) has no EUR
# figure to compare or fold in, so it is refused the same way rather than
# silently treated as a domestic-currency match.
invoice_grand_total_eur = invoice.grand_total_eur
if invoice_grand_total_eur is None or transaction_amount != invoice_grand_total_eur:
return RentaLedgerAggregationIssue(
transaction_id=transaction_id,
purchase_invoice_evidence_id=purchase_invoice_evidence_id,
category_id=category_id,
reason=RentaLedgerAggregationIssueReason.AMOUNT_MISMATCH,
detail="linked transaction amount does not match invoice grand total",
)
# invoice.base_total / invoice.iva_total are native-currency (same
# currency as invoice.currency); the EUR-safe *_eur properties are the
# ones that already exist for exactly this purpose (used correctly in
# invoice_retencion.py). base_total_eur cannot be None here: the
# grand_total_eur check above already proved the invoice resolves to EUR.
base_total_eur = invoice.base_total_eur
iva_total_eur = invoice.iva_total_eur
if base_total_eur is None or iva_total_eur is None:
msg = (
f"purchase invoice {purchase_invoice_evidence_id} resolves a EUR grand total but no EUR "
"base or cuota; its EUR-equivalent totals disagree with one another"
)
raise ValueError(msg)
return _PurchaseInvoiceEvidencePayload(
invoice_issue_date=invoice.issued_at,
taxable_base=base_total_eur,
iva_amount=iva_total_eur,
)
def _taxable_base_for(transaction: Transaction, evidence_payload: _PurchaseInvoiceEvidencePayload) -> Decimal | None:
"""Return the EUR taxable_base: the linked invoice's (already EUR), else the transaction's own.
``transaction.taxable_base`` is native-currency (see
``domain.transactions.tests.test_gross_invariant``), so the fallback goes
through the EUR-equivalent accessor rather than the raw field.
"""
if evidence_payload.taxable_base is not None:
return evidence_payload.taxable_base
return effective_eur_taxable_base(transaction)
def _iva_amount_for(transaction: Transaction, evidence_payload: _PurchaseInvoiceEvidencePayload) -> Decimal | None:
"""Return the EUR iva_amount: the linked invoice's (already EUR), else the transaction's own.
Via the EUR-equivalent accessor -- see :func:`_taxable_base_for`.
"""
if evidence_payload.iva_amount is not None:
return evidence_payload.iva_amount
return effective_eur_iva_amount(transaction)
def _casilla_aggregation(
period: Period,
observations: Sequence[RentaDeductibleExpenseObservation],
*,
modelo: str,
) -> CasillaAggregation:
totals: dict[CasillaId, Decimal] = {}
provenance_rows: list[CasillaProvenance] = []
grouped: dict[tuple[CasillaId, SpendingCategory], list[RentaDeductibleExpenseObservation]] = {}
for observation in observations:
totals[observation.target_casilla_id] = (
totals.get(observation.target_casilla_id, Decimal("0")) + observation.deductible_amount
)
grouped.setdefault((observation.target_casilla_id, observation.category), []).append(observation)
for (casilla, category), rows in sorted(grouped.items()):
provenance_rows.append(
CasillaProvenance(
casilla_id=casilla,
category_id=category,
transaction_ids=tuple(sorted(row.transaction_id for row in rows)),
subtotal=sum((row.deductible_amount for row in rows), start=Decimal("0")),
),
)
return CasillaAggregation(
modelo=modelo,
period=period,
casilla_values=totals,
provenance=tuple(provenance_rows),
)
__all__ = [
"RentaLedgerAggregationIssue",
"RentaLedgerAggregationIssueReason",
"RentaLedgerExpenseAggregation",
"aggregate_renta_ledger_expenses",
"aggregate_renta_ledger_expenses_from_repositories",
]