"""Repository-backed Modelo 151 impatriado Spanish-source income aggregation.
This is the ledger projection behind the
``ledger_impatriado_income_aggregation`` source for Modelo 151 (régimen
especial de trabajadores desplazados, "Ley Beckham", art. 93 LIRPF). The annual
entry point :func:`aggregate_impatriado_income_ledger_from_repositories` loads a
:class:`~domain.transactions.TransactionCatalogue` from the active bucket
through :class:`~domain.transactions.TransactionCatalogueRepository` and
delegates to :func:`aggregate_impatriado_income_ledger`.
Unlike the Modelo 130 / Modelo 100 actividad-económica income pipeline
(:mod:`~._renta_income_ledger`), which admits worldwide income into the
resident-IRPF base (LIRPF art. 8), the impatriado base is legally
source-scoped: art. 93.2 taxes the impatriado by the IRNR scope rules, so its
``impatriado.base-liquidable-general`` casilla admits ONLY Spanish-source
income. The declared per-row ``source_jurisdiction`` axis — which the CLI
create-boundary gate compels an impatriado profile to supply on every ledger
row — is finally consumed here:
- an INCOMING row whose ``source_jurisdiction`` resolves to ``ES`` folds into
the impatriado base;
- a foreign-source row (``source_jurisdiction`` set to any non-``ES`` code) is
segregated out of the base and surfaced as a typed
:attr:`ImpatriadoIncomeLedgerAggregationIssueReason.BECKHAM_FOREIGN_SOURCE_SEGREGATED`
issue carrying the rejected jurisdiction code (art. 93.2 / art. 25.1.f
TRLIRNR segregation);
- a jurisdiction-unresolved row (``source_jurisdiction is None``) is NEVER
silently coerced to ``ES``: it fails loud as the same segregation issue with
an unresolved-jurisdiction detail (``no-silent-under-declaration``).
The impatriado base admits ``trabajo`` income — the exact income class the M130
pipeline routes OUT — because the Beckham base is predominantly rendimientos
del trabajo (nómina). The two pipelines are complementary, not shared.
The savings escala (art. 93.2.e.2º → art. 25.1.f TRLIRNR: the parte del ahorro)
is out of scope here and blocked on a separate corpus ingest; the base casilla
is labelled "excluida la parte del ahorro" to keep that deferral honest.
"""
from __future__ import annotations
from collections.abc import Sequence
from datetime import date
from decimal import Decimal
from enum import StrEnum
from typing import Self
from pydantic import BaseModel, Field, field_serializer, field_validator, model_validator
from ...adapters.persistence.profile.transactions import TransactionCatalogueRepository
from ...core import STRICT_FROZEN_CONFIG as _STRICT_FROZEN
from ...core import Modelo, Period, PeriodKind
from ...domain.calculations.registry import CasillaId, validated_casilla_id
from ...domain.transactions import (
BusinessClassification,
OutOfWindowTransactionSummary,
Transaction,
TransactionCatalogue,
TransactionCatalogueRepositoryProtocol,
TransactionDirection,
TransactionLifecycleState,
)
from . import _shared_issue_reasons
from ._business_proportion import business_proportion
from ._currency_predicates import is_non_eur_without_conversion
from ._errors import AggregationPeriodError, AggregationValidationError, t
from ._models import CasillaAggregation, CasillaProvenance
# The Modelo 151 base liquidable general (régimen impatriados, excluida la parte
# del ahorro). The impatriado income aggregation folds Spanish-source income into
# this single base casilla; the flat 24/47 escala (art. 93.2.e.1º) then computes
# the cuota íntegra from it.
_TARGET_CASILLA_IMPATRIADO_BASE: CasillaId = validated_casilla_id(
"impatriado.base-liquidable-general",
surface="_TARGET_CASILLA_IMPATRIADO_BASE",
)
# ISO 3166-1 alpha-2 code for Spain. The impatriado base admits only rows whose
# declared source jurisdiction equals this code (art. 93.2 IRNR scope).
_SPANISH_SOURCE_JURISDICTION: str = "ES"
_IRPF_CATEGORY_ACTIVIDAD_ECONOMICA: str = "actividad_economica"
_IRPF_CATEGORY_TRABAJO: str = "trabajo"
[docs]
class ImpatriadoIncomeLedgerAggregationIssueReason(StrEnum):
"""Machine-readable reasons why a ledger row did not fold into the impatriado base."""
UNSUPPORTED_DIRECTION = _shared_issue_reasons.UNSUPPORTED_DIRECTION
UNSUPPORTED_CURRENCY = _shared_issue_reasons.UNSUPPORTED_CURRENCY
UNCLASSIFIED_BUSINESS_STATE = _shared_issue_reasons.UNCLASSIFIED_BUSINESS_STATE
PERSONAL_TRANSACTION = _shared_issue_reasons.PERSONAL_TRANSACTION
OUTSIDE_PERIOD = _shared_issue_reasons.OUTSIDE_PERIOD
UNSUPPORTED_PERIOD = "unsupported_period"
# art. 93.2 LIRPF / art. 25.1.f TRLIRNR: the impatriado is taxed by IRNR
# scope rules, so foreign-source income is segregated OUT of the base
# liquidable general. This reason fires for a row whose declared
# source_jurisdiction is a non-ES code (foreign-source) OR is unresolved
# (None) — an unresolved jurisdiction is NEVER silently coerced to ES.
BECKHAM_FOREIGN_SOURCE_SEGREGATED = "beckham_foreign_source_segregated"
[docs]
class ImpatriadoIncomeLedgerAggregationIssue(BaseModel):
"""Traceable exclusion emitted while aggregating impatriado income ledger rows."""
model_config = _STRICT_FROZEN
transaction_id: str = Field(min_length=1, max_length=128)
reason: ImpatriadoIncomeLedgerAggregationIssueReason
detail: str = Field(min_length=1, max_length=512)
# The rejected ISO 3166-1 alpha-2 source-jurisdiction code for a
# BECKHAM_FOREIGN_SOURCE_SEGREGATED row; ``None`` when the row carried no
# declared jurisdiction (the unresolved case), so an auditor can tell a
# foreign-source segregation apart from an unresolved-provenance one.
rejected_source_jurisdiction: str | None = None
[docs]
class ImpatriadoIncomeObservation(BaseModel):
"""One eligible INCOMING Spanish-source income ledger row for the impatriado base.
Carries the typed gross amount and the target casilla id it feeds
(``impatriado.base-liquidable-general``). The domain registry resolver sums
the fiscally computable ingreso (``taxable_base_amount`` when the row carries
an explicit IVA tagging, else ``gross_amount``) across all observations for
that casilla.
``source_jurisdiction`` is retained on the observation for provenance and is
``"ES"`` by construction: a foreign or unresolved jurisdiction is segregated
into an issue before an observation is ever emitted.
"""
model_config = _STRICT_FROZEN
transaction_id: str = Field(min_length=1, max_length=128)
target_casilla_id: CasillaId
gross_amount: Decimal = Field(ge=Decimal("0"))
taxable_base_amount: Decimal | None = Field(default=None, ge=Decimal("0"))
filing_date: date
source_jurisdiction: str = Field(min_length=2, max_length=2)
[docs]
class ImpatriadoIncomeLedgerAggregation(BaseModel):
"""Annual Spanish-source income observations for one Modelo 151 ejercicio.
``out_of_window_summary`` is populated by repository-backed date partitions.
Full-catalogue aggregation keeps row-level issues because every transaction
is already loaded for classification.
"""
model_config = _STRICT_FROZEN
modelo: str = Field(min_length=1, max_length=16)
period: Period
observations: Sequence[ImpatriadoIncomeObservation] = Field(default_factory=tuple)
issues: Sequence[ImpatriadoIncomeLedgerAggregationIssue] = Field(default_factory=tuple)
out_of_window_summary: OutOfWindowTransactionSummary | None = None
casilla_aggregation: CasillaAggregation
@field_validator("observations")
@classmethod
def _freeze_observations(
cls,
value: Sequence[ImpatriadoIncomeObservation],
) -> tuple[ImpatriadoIncomeObservation, ...]:
return tuple(value)
@field_validator("issues")
@classmethod
def _freeze_issues(
cls,
value: Sequence[ImpatriadoIncomeLedgerAggregationIssue],
) -> tuple[ImpatriadoIncomeLedgerAggregationIssue, ...]:
return tuple(value)
@model_validator(mode="after")
def _validate_casilla_period(self) -> Self:
if self.casilla_aggregation.modelo != self.modelo:
raise AggregationValidationError(t("aggregation.renta_ledger.errors.modelo_mismatch"))
if self.casilla_aggregation.period != self.period:
raise AggregationValidationError(t("aggregation.renta_ledger.errors.period_mismatch"))
return self
@field_serializer("observations")
def _serialize_observations(
self,
value: Sequence[ImpatriadoIncomeObservation],
) -> tuple[ImpatriadoIncomeObservation, ...]:
return tuple(value)
@field_serializer("issues")
def _serialize_issues(
self,
value: Sequence[ImpatriadoIncomeLedgerAggregationIssue],
) -> tuple[ImpatriadoIncomeLedgerAggregationIssue, ...]:
return tuple(value)
[docs]
def aggregate_impatriado_income_ledger_from_repositories(
*,
bucket_id: str,
period: Period,
transaction_repository: TransactionCatalogueRepositoryProtocol | None = None,
) -> ImpatriadoIncomeLedgerAggregation:
"""Load the transaction catalogue and aggregate annual impatriado Spanish-source income.
When no protocol-compatible repository override is supplied, this loader uses
:class:`~domain.transactions.TransactionCatalogueRepository` scoped to
``bucket_id``.
Returns an :class:`ImpatriadoIncomeLedgerAggregation`.
"""
repository = transaction_repository or TransactionCatalogueRepository(bucket_id=bucket_id)
if repository.bucket_id != bucket_id:
raise AggregationValidationError(
t("aggregation.renta_ledger.errors.bucket_mismatch"),
context={"bucket_id": bucket_id, "repository_bucket_id": repository.bucket_id},
)
# Only the in-window ejercicio subset is decrypted and classified. The
# out-of-window remainder comes from the plaintext date index and is
# reported uniformly as ``OUTSIDE_PERIOD``. Non-annual periods fall back to
# the unfiltered load so the aggregation's own period validation still
# raises the same error.
if period.kind is not PeriodKind.ANNUAL:
return aggregate_impatriado_income_ledger(repository.load(), bucket_id=bucket_id, period=period)
partition = repository.partition_by_date_range(period.start_date, period.end_date)
result = aggregate_impatriado_income_ledger(partition.in_window, bucket_id=bucket_id, period=period)
out_of_window_summary = partition.out_of_window_summary or OutOfWindowTransactionSummary.from_stubs(
partition.out_of_window,
)
return result.model_copy(
update={"out_of_window_summary": out_of_window_summary},
)
[docs]
def aggregate_impatriado_income_ledger(
transactions: TransactionCatalogue,
*,
bucket_id: str,
period: Period,
) -> ImpatriadoIncomeLedgerAggregation:
"""Aggregate INCOMING Spanish-source income into Modelo 151 ``impatriado.base-liquidable-general``.
Applies the impatriado source scope over the FULL ejercicio (Jan 1 to Dec 31
of ``period.year``): only INCOMING, EUR-denominated rows whose declared
``source_jurisdiction`` resolves to ``ES`` fold into the base. Foreign-source
and jurisdiction-unresolved rows are segregated into
:attr:`ImpatriadoIncomeLedgerAggregationIssueReason.BECKHAM_FOREIGN_SOURCE_SEGREGATED`
issues rather than silently entering (or silently dropping from) the base.
Args:
transactions: The :class:`TransactionCatalogue` to aggregate.
bucket_id: Bucket identifier carried through to provenance so the
aggregation cannot be silently misattributed.
period: The annual :class:`Period` whose year anchors the window.
Returns an :class:`ImpatriadoIncomeLedgerAggregation` for the ejercicio.
``period`` must be the annual period.
"""
if period.kind is not PeriodKind.ANNUAL:
raise AggregationPeriodError(
t("aggregation.renta_ledger.errors.unsupported_period"),
context={"period": str(period)},
)
window_start = date(period.year, 1, 1)
window_end = date(period.year, 12, 31)
observations: list[ImpatriadoIncomeObservation] = []
issues: list[ImpatriadoIncomeLedgerAggregationIssue] = []
for transaction in transactions.values():
if transaction.lifecycle_state is not TransactionLifecycleState.ACTIVE:
continue
outcome = _classify_impatriado_income_transaction(
transaction,
window_start=window_start,
window_end=window_end,
)
if outcome is None:
continue
if isinstance(outcome, ImpatriadoIncomeLedgerAggregationIssue):
issues.append(outcome)
else:
observations.append(outcome)
casilla_aggregation = _impatriado_base_casilla_aggregation(period, observations)
return ImpatriadoIncomeLedgerAggregation(
modelo=Modelo.M151.value,
period=period,
observations=tuple(observations),
issues=tuple(issues),
casilla_aggregation=casilla_aggregation,
)
def _classify_impatriado_income_transaction(
transaction: Transaction,
*,
window_start: date,
window_end: date,
) -> ImpatriadoIncomeObservation | ImpatriadoIncomeLedgerAggregationIssue | None:
"""Filter one ledger transaction against the impatriado Spanish-source income scope.
Returns an :class:`ImpatriadoIncomeObservation` for an eligible ES-source
receipt, an :class:`ImpatriadoIncomeLedgerAggregationIssue` for a row that
fails a gate (currency, source-jurisdiction segregation, personal, window),
or ``None`` for a row this base pipeline does not own (OUTGOING / internal
transfer / operator-excluded).
The source-jurisdiction gate is the load-bearing art. 93.2 scope: it runs
BEFORE the amount/eligibility gates so a foreign-source or unresolved row is
always segregated as a typed issue and can never be silently admitted or
silently dropped.
"""
transaction_id = transaction.transaction_id
if transaction.business_classification is BusinessClassification.REVIEWED_EXCLUDED:
# Operator reviewed and deliberately excluded this row from filing.
return None
if transaction.direction is not TransactionDirection.INCOMING:
# Only INCOMING income folds into the impatriado base. OUTGOING and
# internal-transfer rows are out of scope for the base.
return None
if is_non_eur_without_conversion(transaction):
return ImpatriadoIncomeLedgerAggregationIssue(
transaction_id=transaction_id,
reason=ImpatriadoIncomeLedgerAggregationIssueReason.UNSUPPORTED_CURRENCY,
detail=f"transaction currency {transaction.raw.currency!r} is not supported for impatriado income",
)
# art. 93.2 source-scope gate (art. 25.1.f TRLIRNR segregation). The
# impatriado base admits ONLY Spanish-source income. A None jurisdiction is
# an unresolved provenance, NOT a resident-general ES default: it fails loud
# as a segregation issue (no-silent-under-declaration).
declared_jurisdiction = transaction.source_jurisdiction
if declared_jurisdiction is None:
return ImpatriadoIncomeLedgerAggregationIssue(
transaction_id=transaction_id,
reason=ImpatriadoIncomeLedgerAggregationIssueReason.BECKHAM_FOREIGN_SOURCE_SEGREGATED,
detail=(
"source_jurisdiction is unresolved (None) on an impatriado income row; "
"art. 93.2 LIRPF admits only Spanish-source income into the base liquidable "
"general and an unresolved jurisdiction is never coerced to ES"
),
rejected_source_jurisdiction=None,
)
normalized_jurisdiction = declared_jurisdiction.strip().upper()
if normalized_jurisdiction != _SPANISH_SOURCE_JURISDICTION:
return ImpatriadoIncomeLedgerAggregationIssue(
transaction_id=transaction_id,
reason=ImpatriadoIncomeLedgerAggregationIssueReason.BECKHAM_FOREIGN_SOURCE_SEGREGATED,
detail=(
f"source_jurisdiction {normalized_jurisdiction!r} is foreign-source; "
"art. 93.2 LIRPF / art. 25.1.f TRLIRNR segregate it out of the impatriado "
"base liquidable general (taxed by IRNR scope rules, not the art. 8 worldwide base)"
),
rejected_source_jurisdiction=normalized_jurisdiction,
)
gross_amount = _impatriado_income_amount(transaction)
if gross_amount is None:
reason = (
ImpatriadoIncomeLedgerAggregationIssueReason.PERSONAL_TRANSACTION
if transaction.business_classification is BusinessClassification.PERSONAL
else ImpatriadoIncomeLedgerAggregationIssueReason.UNCLASSIFIED_BUSINESS_STATE
)
return ImpatriadoIncomeLedgerAggregationIssue(
transaction_id=transaction_id,
reason=reason,
detail=(
f"business classification {transaction.business_classification.value!r} cannot feed the impatriado base"
),
)
filing_date = transaction.raw.value_date or transaction.raw.booked_date
if filing_date is None or not (window_start <= filing_date <= window_end):
return ImpatriadoIncomeLedgerAggregationIssue(
transaction_id=transaction_id,
reason=ImpatriadoIncomeLedgerAggregationIssueReason.OUTSIDE_PERIOD,
detail=f"filing date {filing_date} is outside the annual impatriado income window",
)
taxable_base_amount: Decimal | None = None
if transaction.taxable_base is not None:
raw_tb = transaction.taxable_base
if transaction.business_classification is BusinessClassification.MIXED and transaction.business_pct is not None:
taxable_base_amount = raw_tb * transaction.business_pct
else:
taxable_base_amount = raw_tb
return ImpatriadoIncomeObservation(
transaction_id=transaction_id,
target_casilla_id=_TARGET_CASILLA_IMPATRIADO_BASE,
gross_amount=gross_amount,
taxable_base_amount=taxable_base_amount,
filing_date=filing_date,
source_jurisdiction=_SPANISH_SOURCE_JURISDICTION,
)
def _impatriado_income_amount(transaction: Transaction) -> Decimal | None:
"""Return the income amount that folds into the impatriado base, or None if ineligible.
The impatriado base admits both ``trabajo`` (rendimientos del trabajo — the
predominant Beckham base, the class the M130 income pipeline routes OUT) and
``actividad_economica`` income at their full magnitude; any other row is
admitted only through its business proportion, so a genuinely personal
transfer contributes nothing.
"""
amount = abs(transaction.raw.amount)
if transaction.irpf_category in {_IRPF_CATEGORY_TRABAJO, _IRPF_CATEGORY_ACTIVIDAD_ECONOMICA}:
# The explicit IRPF income category is the authoritative eligibility gate
# for the impatriado base.
return amount
proportion = business_proportion(transaction.business_classification, transaction.business_pct)
if proportion is None:
return None
return amount * proportion
def _computable_impatriado_income_amount(observation: ImpatriadoIncomeObservation) -> Decimal:
"""Return the fiscally computable ingreso for one observation.
IVA-exclusive ``taxable_base_amount`` when the row carries an explicit IVA
tagging, falling back to ``gross_amount`` when no base is declared — the same
ingresos-íntegros convention the M130 / M100 income aggregation uses, so the
projection and the binding resolver agree per the one-aggregation-path
discipline.
"""
if observation.taxable_base_amount is not None:
return observation.taxable_base_amount
return observation.gross_amount
def _impatriado_base_casilla_aggregation(
period: Period,
observations: Sequence[ImpatriadoIncomeObservation],
) -> CasillaAggregation:
totals: dict[CasillaId, Decimal] = {}
grouped: dict[CasillaId, list[ImpatriadoIncomeObservation]] = {}
for observation in observations:
totals[observation.target_casilla_id] = totals.get(
observation.target_casilla_id,
Decimal("0"),
) + _computable_impatriado_income_amount(observation)
grouped.setdefault(observation.target_casilla_id, []).append(observation)
provenance_rows = [
CasillaProvenance(
casilla_id=casilla,
category_id=None,
transaction_ids=tuple(sorted(row.transaction_id for row in rows)),
subtotal=sum((_computable_impatriado_income_amount(row) for row in rows), start=Decimal("0")),
)
for casilla, rows in sorted(grouped.items())
]
return CasillaAggregation(
modelo=Modelo.M151.value,
period=period,
casilla_values=totals,
provenance=tuple(provenance_rows),
)
__all__ = [
"ImpatriadoIncomeLedgerAggregation",
"ImpatriadoIncomeLedgerAggregationIssue",
"ImpatriadoIncomeLedgerAggregationIssueReason",
"ImpatriadoIncomeObservation",
"aggregate_impatriado_income_ledger",
"aggregate_impatriado_income_ledger_from_repositories",
]