"""Advisory projection for the annual prorrata-general regularización (LIVA arts. 104-105).
Builds the projection and live source resolver for Modelo 303 casilla 44
(Regularización prorrata por porcentaje definitivo - Cuota) and the Modelo 390
annual regularización field when a taxpayer under prorrata general has
exempt-without-right operations in the year and a provisional percentage was
applied. The calculate path still keeps the advisory surface so missing
provisional/current-year inputs never collapse into a silent zero.
This is a pure function over the two prorrata percentages and the year's
deductible input IVA, plus a resolver that consumes the governed prorrata
register or a stamped prior-year settlement observation for the provisional
percentage. The definitive percentage itself comes from the full-year volume
rollup fed to
:func:`~domain.iva.prorrata.compute_prorrata_definitiva_anual`; deriving it from a
single quarter is a correctness defect (a silent zero base).
The resolver reads the target :class:`ModeloRevision`'s bindings to locate the
casillas that carry the provisional/definitive percentages, and consumes a
:class:`RegistrySnapshot` to resolve those bindings' legal and source refs.
See Also:
:func:`~domain.iva.prorrata.compute_regularizacion_prorrata_anual`
Pure LIVA art. 105.Cuatro computation consumed by this projection.
:mod:`~application.modelo.prorrata_regularizacion_advisory`
Calculate-path collector that calls this advisory projection from
Modelo 303 values and prior-year observations.
:mod:`~application.aggregation.iva_ledger`
Source of typed IVA ledger observations used for declared-volume
divergence checks.
:mod:`~domain.prorrata_register`
Cross-period carry home for provisional and definitive prorrata
percentages.
``application.calculations.tests.test_prorrata_regularizacion``
Focused regressions for live-feed and ledger-divergence behavior.
"""
from __future__ import annotations
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from typing import ClassVar, Final
from pydantic import BaseModel
from ...core.aggregation import BindingSourceKind, CalculationSourceLineageRole
from ...core.casilla_id import CasillaId
from ...core.decimal.constants import MONEY_ZERO
from ...core.json_contract import Notice, NoticeSeverity
from ...core.models import STRICT_FROZEN_CONFIG
from ...core.period import Period
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
from ...domain.calculations.registry.binding_targets import casillas_by_binding
from ...domain.calculations.registry.binding_terminal_origin import TerminalOriginClass
from ...domain.calculations.registry.errors import RegistryValidationError
from ...domain.calculations.registry.facts.resolution import MappingFactQuery, ResolvedMappingFact
from ...domain.calculations.registry.ids import (
BindingId,
LegalRefId,
SourceRefId,
)
from ...domain.calculations.registry.prorrata_register_catalogue import (
especial_prorrata_register_regime,
regime_apportions_deduction,
)
from ...domain.calculations.registry.prorrata_regularizacion_bindings import (
ProrrataRegularizacionOutput,
ProrrataRegularizacionProvider,
prorrata_source_casilla_ids,
)
from ...domain.calculations.registry.schema import (
ModeloRevision,
RegistrySnapshot,
)
from ...domain.calculations.registry.schema_base import DateAxis
from ...domain.iva.prorrata import (
RegularizacionProrrataDireccion,
RegularizacionProrrataResult,
compute_regularizacion_prorrata_anual,
especial_mandatory_rule,
is_especial_mandatory,
)
from ...domain.iva.prorrata_especial_parameters import ProrrataEspecialMandatoryParameters
from ...domain.prorrata_register.protocols import ProrrataRegisterRepositoryProtocol
from ...domain.prorrata_register.register import (
ProrrataProvisionalResolution,
ProrrataRegister,
ProrrataRegisterEntry,
ProrrataRegisterError,
)
from ..aggregation.source_mesh import (
CalculationSourceContext,
CalculationSourceDiagnostic,
CalculationSourceProvenance,
CalculationSourceResolution,
)
from ..aggregation.source_resolution_operations import storage_degradation_resolution
from ..persistence_errors import PersistenceDegradationError
from ..prorrata_register.service import require_prorrata_register_coordinates_current
from .observations_repository import CalculationObservationRepositoryProtocol
from .revision_carry_gate import revision_carry_outcome
# Registry-owned prorrata declarations are read from the selected Modelo 303/390
# revision through the query boundary below. This module retains aggregation and
# evidence mechanics only.
_SOURCE_KIND: Final = BindingSourceKind.PRORRATA_REGULARIZACION
# Registry-owned target declarations remain in the selected binding fragments.
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
#
def _prorrata_source_id(revision: ModeloRevision, position: int) -> CasillaId:
"""Read one role by the registry selector's reviewed source order."""
ids = prorrata_source_casilla_ids(revision.bindings)
try:
return ids[position]
except IndexError as exc:
raise ValueError("selected prorrata binding does not declare the required source casillas") from exc
def _prorrata_target_casilla(revision: ModeloRevision, output: str) -> CasillaId | None:
"""Resolve a printed target number from the selected revision's casillas."""
marker = "_casilla_"
if marker not in output:
return None
number = output.rsplit(marker, 1)[-1]
return next((casilla.id for casilla in revision.casillas if casilla.number == number), None)
@dataclass(frozen=True, slots=True)
class _PriorDefinitivaCarry:
percentage: Decimal
source_filing_year: int
source_period: str
@dataclass(frozen=True, slots=True)
class _CurrentYearSourcePeriodFeed:
values: Mapping[CasillaId, Decimal]
source_periods: tuple[str, ...]
missing_source_periods: tuple[str, ...] = ()
@property
def complete(self) -> bool:
return not self.missing_source_periods
@dataclass(frozen=True, slots=True)
class _ProrrataProvisionalSource:
percentage: Decimal | None
provenance: tuple[CalculationSourceProvenance, ...]
[docs]
class ProrrataRegularizacionFeedProjection(BaseModel):
"""Structured proposed feeds for annual prorrata-general regularización.
The proposed value feeds the live source resolver and the operator-facing
advisory. The selected registry revision supplies the output destinations;
this carrier intentionally contains no filing-model or casilla facts.
See Also:
:func:`project_prorrata_regularizacion_feed`
Constructs this carrier from the pure annual regularización result.
"""
model_config = STRICT_FROZEN_CONFIG
result: RegularizacionProrrataResult
operaciones_sin_derecho_deduccion: Decimal
proposed_value: Decimal | None = None
[docs]
class ProrrataDeclaredVolumeLedgerRollup(BaseModel):
"""Ledger-side annual volume rollup used only as a divergence advisory.
Declared annual volume casillas remain the filing authority. This projection
records the currently classifiable ledger output-volume view so settlement
can warn when it contradicts those declared values.
"""
model_config = STRICT_FROZEN_CONFIG
declared_volume_total: Decimal
declared_volume_con_derecho: Decimal
declared_volume_sin_derecho: Decimal
ledger_volume_total: Decimal
ledger_volume_con_derecho: Decimal
ledger_volume_sin_derecho: Decimal
included_ledger_ids: tuple[str, ...] = ()
#: Ledger ids skipped from the rollup because they carry an operator-declared
#: LIVA art. 104.Tres judgment exclusion (foreign PE, non-habitual
#: inmobiliario/financiero). Recorded so the exclusion is auditable and the
#: proposal is never a silent substitution of the declared volumes.
art_104_tres_excluded_ledger_ids: tuple[str, ...] = ()
[docs]
class ProrrataApplicabilityProjection(BaseModel):
"""Fail-closed-to-visible prorrata applicability evidence for one ejercicio.
Prorrata applies when the taxpayer has an active register entry or the
ejercicio shows exempt-without-right operations through declared annual
volumes or the ledger rollup. This projection is deliberately pure; later
steps turn an applicable-but-unresolved state into calculate/verify
diagnostics.
"""
model_config = STRICT_FROZEN_CONFIG
applies: bool
register_active: bool
declared_volume_sin_derecho: Decimal
ledger_volume_sin_derecho: Decimal
evidence_kinds: tuple[str, ...] = ()
[docs]
def derive_prorrata_applicability(
*,
register_entries: Iterable[ProrrataRegisterEntry] = (),
declared_volume_total: Decimal | None = None,
declared_volume_con_derecho: Decimal | None = None,
ledger_rollup: ProrrataDeclaredVolumeLedgerRollup | None = None,
) -> ProrrataApplicabilityProjection:
"""Derive whether prorrata applies for an ejercicio.
The rule is intentionally fail-closed-to-visible: any active register entry
(``general`` or ``especial``), any positive declared sin-derecho annual
volume, or any positive ledger-projected sin-derecho annual volume means
prorrata applies and later steps must not silently assume a full-deduction
default.
"""
if (declared_volume_total is None) != (declared_volume_con_derecho is None):
raise ValueError("declared_volume_total and declared_volume_con_derecho must be supplied together")
entries = tuple(register_entries)
register_active = any(regime_apportions_deduction(entry.regime) for entry in entries)
declared_volume_sin_derecho = (
Decimal("0")
if declared_volume_total is None or declared_volume_con_derecho is None
else declared_volume_total - declared_volume_con_derecho
)
ledger_volume_sin_derecho = Decimal("0") if ledger_rollup is None else ledger_rollup.ledger_volume_sin_derecho
evidence: list[str] = []
if register_active:
evidence.append("register_active")
if declared_volume_sin_derecho > Decimal("0"):
evidence.append("declared_sin_derecho_volume")
if ledger_volume_sin_derecho > Decimal("0"):
evidence.append("ledger_sin_derecho_volume")
return ProrrataApplicabilityProjection(
applies=bool(evidence),
register_active=register_active,
declared_volume_sin_derecho=declared_volume_sin_derecho,
ledger_volume_sin_derecho=ledger_volume_sin_derecho,
evidence_kinds=tuple(evidence),
)
[docs]
def build_prorrata_missing_provisional_advisory(
*,
applicability: ProrrataApplicabilityProjection,
provisional_resolution: ProrrataProvisionalResolution,
ejercicio: int,
revision: ModeloRevision,
first_ejercicio: bool = False,
) -> CalculationSourceDiagnostic | None:
"""Build the visible advisory for an applicable prorrata with no provisional percentage.
This is the per-period no-silent-under-declaration warning. It does not
fabricate a percentage: an applicable-but-unresolved prorrata tells the
operator to record the inicio-de-actividad percentage for a first ejercicio
or seed/record the prior definitive percentage for subsequent ejercicios.
Core types:
:class:`~cadrumo.domain.calculations.registry.schema.ModeloRevision`.
"""
if not applicability.applies or provisional_resolution.resolved:
return None
operator_action = (
"registre el porcentaje provisional de inicio de actividad"
if first_ejercicio
else "siembre o registre la prorrata definitiva del ejercicio anterior"
)
evidence = ", ".join(applicability.evidence_kinds) or "prorrata_applicability"
target = _modelo_303_regularizacion_target(revision)
message = (
f"Prorrata aplicable en {ejercicio} sin porcentaje provisional resuelto "
f"({evidence}); {_regularizacion_target_label(revision, target)} no puede regularizarse. "
f"{operator_action}; no se aplica un porcentaje por defecto."
)
return CalculationSourceDiagnostic(
reason="source_issue",
source_kind=BindingSourceKind.PRORRATA_REGULARIZACION.value,
message=message,
casilla_id=target,
)
def _modelo_303_regularizacion_target(revision: ModeloRevision) -> CasillaId | None:
"""Return the casilla the revision declares for the Modelo 303 prorrata regularisation."""
targets = casillas_by_binding(revision)
candidates = {
casilla_id
for binding in revision.bindings
if isinstance(binding.provider, ProrrataRegularizacionProvider)
and binding.provider.regularizacion_output is ProrrataRegularizacionOutput.MODELO_303_CASILLA_44
for casilla_id in targets.get(binding.id, ())
}
return next(iter(candidates)) if len(candidates) == 1 else None
def _regularizacion_target_label(revision: ModeloRevision, casilla_id: CasillaId | None) -> str:
"""Name the registry-declared target casilla by its printed number for operator messages."""
number = next((casilla.number for casilla in revision.casillas if casilla.id == casilla_id), None)
return f"casilla {number}" if number else "la casilla de regularización declarada"
[docs]
def project_prorrata_regularizacion_feed(
*,
cuotas_soportadas_deducibles: Decimal,
prorrata_provisional_pct: Decimal,
prorrata_definitiva_pct: Decimal,
operaciones_sin_derecho_deduccion: Decimal,
) -> ProrrataRegularizacionFeedProjection:
"""Project the annual regularización onto selected registry targets.
``prorrata_definitiva_pct`` is supplied by the registry-computed annual
volume casillas. This helper deliberately does not recompute that
percentage; it turns the pure result into one generic proposed value. The
live resolver fans that value out only to destinations declared by selected
binding rows.
See Also:
:class:`ProrrataRegularizacionFeedProjection`
Structured carrier for the proposed registry value.
"""
result = compute_regularizacion_prorrata_anual(
cuotas_soportadas_deducibles=cuotas_soportadas_deducibles,
prorrata_provisional_pct=prorrata_provisional_pct,
prorrata_definitiva_pct=prorrata_definitiva_pct,
)
proposed_value = (
result.importe
if operaciones_sin_derecho_deduccion > Decimal("0")
and result.direccion is not RegularizacionProrrataDireccion.NINGUNA
else None
)
return ProrrataRegularizacionFeedProjection(
result=result,
operaciones_sin_derecho_deduccion=operaciones_sin_derecho_deduccion,
proposed_value=proposed_value,
)
def _binding_legal_refs(revision: ModeloRevision) -> tuple[LegalRefId, ...]:
refs: list[LegalRefId] = []
for binding in revision.bindings:
if binding.source != _SOURCE_KIND:
continue
for ref in getattr(binding, "legal_refs", ()):
if ref not in refs:
refs.append(ref)
return tuple(refs)
def _binding_source_refs(revision: ModeloRevision) -> tuple[SourceRefId, ...]:
refs: list[SourceRefId] = []
for binding in revision.bindings:
if binding.source != _SOURCE_KIND:
continue
for ref in getattr(binding, "source_refs", ()):
if ref not in refs:
refs.append(ref)
return tuple(refs)
def _prorrata_bindings_by_output(revision: ModeloRevision) -> dict[str, BindingId]:
bindings: dict[str, BindingId] = {}
for binding in revision.bindings:
if not isinstance(binding.provider, ProrrataRegularizacionProvider):
continue
bindings[binding.provider.regularizacion_output] = binding.id
return bindings
def _prorrata_declared_binding_ids(revision: ModeloRevision) -> tuple[BindingId, ...]:
return tuple(sorted(binding.id for binding in revision.bindings if binding.source == _SOURCE_KIND))
def _prorrata_source_periods(revision: ModeloRevision) -> tuple[str, ...]:
periods: list[str] = []
for binding in revision.bindings:
if not isinstance(binding.provider, ProrrataRegularizacionProvider):
continue
for period in binding.provider.source_periods:
if period not in periods:
periods.append(period)
return tuple(periods)
def _prorrata_source_modelo(revision: ModeloRevision) -> str:
"""Return the one modelo the revision's prorrata bindings declare as their source."""
modelos = {
str(binding.provider.source_modelo)
for binding in revision.bindings
if isinstance(binding.provider, ProrrataRegularizacionProvider)
}
if len(modelos) != 1:
raise ValueError("selected prorrata bindings must declare exactly one source modelo")
return next(iter(modelos))
def _missing_current_year_casillas(
current_year_values: Mapping[CasillaId, Decimal],
*,
revision: ModeloRevision,
) -> tuple[CasillaId, ...]:
return tuple(
casilla_id
for casilla_id in prorrata_source_casilla_ids(revision.bindings)
if casilla_id not in current_year_values
)
def _unresolved_binding_diagnostics(
*,
binding_ids: tuple[BindingId, ...],
resolver_id: str,
message: str,
) -> tuple[CalculationSourceDiagnostic, ...]:
return tuple(
CalculationSourceDiagnostic(
reason="unresolved_binding",
source_kind=_SOURCE_KIND.value,
resolver_id=resolver_id,
binding_id=binding_id,
message=message,
)
for binding_id in binding_ids
)
def _current_year_values_provenance(
*,
context: CalculationSourceContext,
revision: ModeloRevision,
source_periods: tuple[str, ...] | None = None,
) -> CalculationSourceProvenance:
periods = source_periods or _prorrata_source_periods(revision)
period_ref = ",".join(periods)
return CalculationSourceProvenance(
resolver_id=ProrrataRegularizacionSourceResolver.resolver_id,
resolved_binding_source=_SOURCE_KIND,
contributor_source_kind=_SOURCE_KIND.value,
contributor_binding_source=_SOURCE_KIND,
lineage_role=CalculationSourceLineageRole.PRIMARY,
source_ref=f"{context.modelo}:{context.filing_year}:{period_ref}:prorrata-current-year-values",
parent_source_ref=None,
terminal_origin=TerminalOriginClass.DERIVED_CALCULATION,
source_modelo=context.modelo,
source_filing_year=context.filing_year,
source_periods=periods,
source_casilla_ids=prorrata_source_casilla_ids(revision.bindings),
legal_refs=_binding_legal_refs(revision),
source_refs=_binding_source_refs(revision),
)
def _entry_for_register_provenance(
register: ProrrataRegister,
*,
ejercicio: int,
) -> ProrrataRegisterEntry | None:
entry = register.entry_for(ejercicio)
if entry is not None:
return entry
entries = register.entries_for_ejercicio(ejercicio)
return entries[0] if entries else None
def _register_provenance(
*,
context: CalculationSourceContext,
entry: ProrrataRegisterEntry | None,
provisional_resolution: ProrrataProvisionalResolution,
revision: ModeloRevision,
) -> CalculationSourceProvenance:
provenance = provisional_resolution.provenance
provenance_token = provenance.value if provenance is not None else "resolved"
suffix = provenance_token
if entry is not None and entry.source_observation_ref is not None:
suffix = f"{provenance_token}:{entry.source_observation_ref}"
elif entry is not None and entry.authorisation_reference is not None:
suffix = f"{provenance_token}:{entry.authorisation_reference}"
return CalculationSourceProvenance(
resolver_id=ProrrataRegularizacionSourceResolver.resolver_id,
resolved_binding_source=_SOURCE_KIND,
contributor_source_kind=_SOURCE_KIND.value,
contributor_binding_source=_SOURCE_KIND,
lineage_role=CalculationSourceLineageRole.PRIMARY,
source_ref=f"prorrata-register:{context.filing_year}:{suffix}",
parent_source_ref=None,
terminal_origin=TerminalOriginClass.DERIVED_CALCULATION,
source_filing_year=context.filing_year,
legal_refs=_binding_legal_refs(revision),
source_refs=_binding_source_refs(revision),
)
def _prior_definitiva_provenance(
*,
carry: _PriorDefinitivaCarry,
revision: ModeloRevision,
source_modelo: str,
) -> CalculationSourceProvenance:
return CalculationSourceProvenance(
resolver_id=ProrrataRegularizacionSourceResolver.resolver_id,
resolved_binding_source=_SOURCE_KIND,
contributor_source_kind=_SOURCE_KIND.value,
contributor_binding_source=_SOURCE_KIND,
lineage_role=CalculationSourceLineageRole.PRIMARY,
source_ref=(
f"{source_modelo}:{carry.source_filing_year}:{carry.source_period}:{_prorrata_source_id(revision, 3)}"
),
parent_source_ref=None,
terminal_origin=TerminalOriginClass.DERIVED_CALCULATION,
source_modelo=source_modelo,
source_filing_year=carry.source_filing_year,
source_periods=(carry.source_period,),
source_casilla_ids=(_prorrata_source_id(revision, 3),),
legal_refs=_binding_legal_refs(revision),
source_refs=_binding_source_refs(revision),
)
def _stamped_prior_year_definitiva(
repository: CalculationObservationRepositoryProtocol,
*,
operation: PinnedAuthorityOperation,
filing_year: int,
modelo: str,
revision: ModeloRevision,
) -> _PriorDefinitivaCarry | None:
prior_year = filing_year - 1
candidates: list[tuple[datetime, _PriorDefinitivaCarry]] = []
for payload in repository.iter_modelo(modelo):
observation = payload.observation
if observation.filing_year != prior_year:
continue
percentage = observation.casilla_values.get(_prorrata_source_id(revision, 3))
if percentage is None:
continue
refused = revision_carry_outcome(payload.registry_snapshot_ref, operation=operation).refused
if refused:
continue
candidates.append(
(
payload.captured_at,
_PriorDefinitivaCarry(
percentage=percentage,
source_filing_year=observation.filing_year,
source_period=observation.period,
),
)
)
if not candidates:
return None
return max(candidates, key=lambda item: item[0])[1]
def _observed_source_period_values(
repository: CalculationObservationRepositoryProtocol,
*,
operation: PinnedAuthorityOperation,
modelo: str,
periods: tuple[str, ...],
filing_year: int,
) -> tuple[dict[str, Mapping[CasillaId, Decimal]], tuple[str, ...]]:
observed_by_period: dict[str, Mapping[CasillaId, Decimal]] = {}
missing_periods: list[str] = []
for period in periods:
payload = repository.load_observation(modelo, Period.from_year_and_code(filing_year, period))
if payload is None:
missing_periods.append(period)
continue
observation = payload.observation
if revision_carry_outcome(payload.registry_snapshot_ref, operation=operation).refused:
missing_periods.append(period)
continue
observed_by_period[period] = observation.casilla_values
return observed_by_period, tuple(missing_periods)
def _regularised_cuota_value(
observed_by_period: Mapping[str, Mapping[CasillaId, Decimal]],
periods: tuple[str, ...],
*,
cuota_id: CasillaId,
) -> Decimal | None:
regularised_periods = periods[:-1] if len(periods) > 1 else periods
cuota_values = [
period_values[cuota_id]
for period in regularised_periods
if (period_values := observed_by_period.get(period)) is not None and cuota_id in period_values
]
if len(cuota_values) != len(regularised_periods):
return None
return sum(cuota_values, Decimal("0"))
def _settlement_source_values(
observed_by_period: Mapping[str, Mapping[CasillaId, Decimal]],
periods: tuple[str, ...],
*,
source_ids: tuple[CasillaId, ...],
) -> dict[CasillaId, Decimal]:
settlement_period = next(reversed(periods), None)
settlement_values = observed_by_period.get(settlement_period) if settlement_period is not None else None
if settlement_values is None:
return {}
return {
casilla_id: value for casilla_id in source_ids[1:] if (value := settlement_values.get(casilla_id)) is not None
}
def _project_source_period_values(
observed_by_period: Mapping[str, Mapping[CasillaId, Decimal]],
periods: tuple[str, ...],
*,
source_ids: tuple[CasillaId, ...],
) -> dict[CasillaId, Decimal]:
values: dict[CasillaId, Decimal] = {}
cuota_value = _regularised_cuota_value(observed_by_period, periods, cuota_id=source_ids[0])
if cuota_value is not None:
values[source_ids[0]] = cuota_value
values.update(_settlement_source_values(observed_by_period, periods, source_ids=source_ids))
return values
def _source_period_feed_from_observations(
repository: CalculationObservationRepositoryProtocol,
*,
operation: PinnedAuthorityOperation,
modelo: str,
revision: ModeloRevision,
filing_year: int,
) -> _CurrentYearSourcePeriodFeed:
periods = _prorrata_source_periods(revision)
if not periods:
return _CurrentYearSourcePeriodFeed(values={}, source_periods=())
observed_by_period, missing_periods = _observed_source_period_values(
repository,
operation=operation,
modelo=modelo,
periods=periods,
filing_year=filing_year,
)
values = _project_source_period_values(
observed_by_period,
periods,
source_ids=prorrata_source_casilla_ids(revision.bindings),
)
return _CurrentYearSourcePeriodFeed(
values=values,
source_periods=periods,
missing_source_periods=missing_periods,
)
def _resolve_prorrata_regularizacion_binding_values(
revision: ModeloRevision,
*,
current_year_values: Mapping[CasillaId, Decimal],
provisional_percentage: Decimal,
) -> dict[BindingId, Decimal]:
binding_by_output = _prorrata_bindings_by_output(revision)
if not binding_by_output:
return {}
source_ids = prorrata_source_casilla_ids(revision.bindings)
volumen_total = current_year_values[source_ids[2]]
volumen_con_derecho = current_year_values[source_ids[1]]
projection = project_prorrata_regularizacion_feed(
cuotas_soportadas_deducibles=current_year_values[source_ids[0]],
prorrata_provisional_pct=provisional_percentage,
prorrata_definitiva_pct=current_year_values[source_ids[3]],
operaciones_sin_derecho_deduccion=volumen_total - volumen_con_derecho,
)
values_by_output: dict[str, Decimal] = {}
for output in binding_by_output:
values_by_output[output] = projection.proposed_value or MONEY_ZERO
return {
binding_id: values_by_output[output]
for output, binding_id in binding_by_output.items()
if output in values_by_output
}
def _target_inputs(
revision: ModeloRevision,
*,
binding_values: Mapping[BindingId, Decimal],
modelo: str,
) -> dict[CasillaId, Decimal]:
binding_by_output = _prorrata_bindings_by_output(revision)
model_prefix = f"modelo_{modelo}_"
output = next(
(candidate for candidate in binding_by_output if candidate.startswith(model_prefix)),
None,
)
if output is None:
return {}
binding_id = binding_by_output[output]
target_casilla = _prorrata_target_casilla(revision, output)
if target_casilla is None or binding_id not in binding_values:
return {}
return {target_casilla: binding_values[binding_id]}
def _revision_for_context(
context: CalculationSourceContext,
registry_snapshot: RegistrySnapshot | None,
) -> ModeloRevision:
revision = registry_snapshot.revision if registry_snapshot is not None else None
if revision is not None:
return revision
# The source mesh already selected this exact revision before invoking the
# resolver. Re-selecting through a process-wide catalogue could cross a
# generation boundary and would make a persisted calculation depend on a
# later authority edition.
return context.revision
def _merge_current_year_values(
source_period_feed: _CurrentYearSourcePeriodFeed,
*,
revision: ModeloRevision,
supplied_values: Mapping[CasillaId, Decimal],
missing_casilla_ids: Iterable[CasillaId],
unresolved_casilla_ids: Iterable[CasillaId],
) -> tuple[dict[CasillaId, Decimal], tuple[CasillaId, ...]]:
current_year_values = {
**supplied_values,
**dict(source_period_feed.values),
}
missing_current = tuple(
dict.fromkeys(
(
*_missing_current_year_casillas(current_year_values, revision=revision),
*(casilla_id for casilla_id in missing_casilla_ids if casilla_id not in current_year_values),
*(casilla_id for casilla_id in unresolved_casilla_ids if casilla_id not in current_year_values),
)
)
)
return current_year_values, missing_current
def _missing_current_year_resolution(
*,
resolver_id: str,
owned_sources: tuple[BindingSourceKind, ...],
declared_binding_ids: tuple[BindingId, ...],
missing_current: tuple[CasillaId, ...],
) -> CalculationSourceResolution | None:
if not missing_current:
return None
message = (
f"prorrata_regularizacion binding requires current-year registry casillas "
f"{tuple(str(casilla_id) for casilla_id in missing_current)} before it can resolve"
)
return CalculationSourceResolution(
resolver_id=resolver_id,
owned_sources=owned_sources,
unresolved_binding_ids=declared_binding_ids,
diagnostics=_unresolved_binding_diagnostics(
binding_ids=declared_binding_ids,
resolver_id=resolver_id,
message=message,
),
)
def _resolve_prorrata_provisional_source(
register: ProrrataRegister,
prior_definitiva: _PriorDefinitivaCarry | None,
*,
context: CalculationSourceContext,
revision: ModeloRevision,
current_provenance: CalculationSourceProvenance,
) -> _ProrrataProvisionalSource:
provisional = register.resolve_provisional(context.filing_year)
if provisional.resolved:
register_entry = _entry_for_register_provenance(register, ejercicio=context.filing_year)
provenance = (
current_provenance,
_register_provenance(
context=context,
entry=register_entry,
provisional_resolution=provisional,
revision=revision,
),
)
if provisional.percentage is None:
raise ValueError(
"prorrata register reported a resolved provisional percentage without a value",
)
return _ProrrataProvisionalSource(percentage=provisional.percentage, provenance=provenance)
if prior_definitiva is not None:
return _ProrrataProvisionalSource(
percentage=prior_definitiva.percentage,
provenance=(
current_provenance,
_prior_definitiva_provenance(
carry=prior_definitiva,
revision=revision,
source_modelo=_prorrata_source_modelo(revision),
),
),
)
return _ProrrataProvisionalSource(percentage=None, provenance=(current_provenance,))
def _missing_provisional_resolution(
*,
context: CalculationSourceContext,
resolver_id: str,
owned_sources: tuple[BindingSourceKind, ...],
declared_binding_ids: tuple[BindingId, ...],
provenance: tuple[CalculationSourceProvenance, ...],
) -> CalculationSourceResolution:
message = (
f"prorrata_regularizacion binding for {context.filing_year} requires a resolved provisional "
"percentage from the prorrata register or a stamped prior-year Modelo 303 settlement observation"
)
return CalculationSourceResolution(
resolver_id=resolver_id,
owned_sources=owned_sources,
unresolved_binding_ids=declared_binding_ids,
diagnostics=_unresolved_binding_diagnostics(
binding_ids=declared_binding_ids,
resolver_id=resolver_id,
message=message,
),
provenance=provenance,
)
def _zero_prorrata_resolution(
*,
revision: ModeloRevision,
context: CalculationSourceContext,
resolver_id: str,
owned_sources: tuple[BindingSourceKind, ...],
declared_binding_ids: tuple[BindingId, ...],
current_provenance: CalculationSourceProvenance,
) -> CalculationSourceResolution:
zero_values = {binding_id: MONEY_ZERO for binding_id in declared_binding_ids}
return CalculationSourceResolution(
resolver_id=resolver_id,
owned_sources=owned_sources,
binding_values=zero_values,
bound_inputs_by_casilla_id=_target_inputs(
revision,
binding_values=zero_values,
modelo=context.modelo,
),
provenance=(current_provenance,),
)
def _resolved_prorrata_resolution(
*,
revision: ModeloRevision,
context: CalculationSourceContext,
resolver_id: str,
owned_sources: tuple[BindingSourceKind, ...],
declared_binding_ids: tuple[BindingId, ...],
current_year_values: Mapping[CasillaId, Decimal],
provisional_percentage: Decimal,
provenance: tuple[CalculationSourceProvenance, ...],
) -> CalculationSourceResolution:
binding_values = _resolve_prorrata_regularizacion_binding_values(
revision,
current_year_values=current_year_values,
provisional_percentage=provisional_percentage,
)
unresolved = tuple(binding_id for binding_id in declared_binding_ids if binding_id not in binding_values)
diagnostics = _unresolved_binding_diagnostics(
binding_ids=unresolved,
resolver_id=resolver_id,
message="prorrata_regularizacion binding selector did not map to a resolver output",
)
return CalculationSourceResolution(
resolver_id=resolver_id,
owned_sources=owned_sources,
binding_values=binding_values,
bound_inputs_by_casilla_id=_target_inputs(
revision,
binding_values=binding_values,
modelo=context.modelo,
),
unresolved_binding_ids=unresolved,
diagnostics=diagnostics,
provenance=provenance,
)
[docs]
class ProrrataRegularizacionSourceResolver:
"""Resolve annual prorrata-general regularisation bindings from governed carries."""
resolver_id: ClassVar[str] = _SOURCE_KIND.value
owned_sources: ClassVar[tuple[BindingSourceKind, ...]] = (_SOURCE_KIND,)
def __init__(
self,
*,
current_year_values: Mapping[CasillaId, Decimal] | None = None,
missing_current_year_casilla_ids: Iterable[CasillaId] = (),
unresolved_current_year_casilla_ids: Iterable[CasillaId] = (),
prorrata_register_repository: ProrrataRegisterRepositoryProtocol,
observation_repository: CalculationObservationRepositoryProtocol,
operation: PinnedAuthorityOperation,
registry_snapshot: RegistrySnapshot | None = None,
) -> None:
"""Bind the current-year prorrata inputs and repositories used to resolve carries.
``current_year_values`` supplies the casillas already known for the active
ejercicio, while ``missing_current_year_casilla_ids`` and
``unresolved_current_year_casilla_ids`` preserve the distinction between an
input that is absent and one that is present but not yet computable. The
prorrata register and observation repositories back the annual regularisation
lookup and its diagnostics; ``registry_snapshot`` pins the compiled revision
when the caller already holds one, avoiding a redundant registry load in
:meth:`resolve`.
"""
self._current_year_values = dict(current_year_values or {})
self._missing_current_year_casilla_ids = tuple(missing_current_year_casilla_ids)
self._unresolved_current_year_casilla_ids = tuple(unresolved_current_year_casilla_ids)
self._prorrata_register_repository = prorrata_register_repository
self._observation_repository = observation_repository
self._operation = operation
self._registry_snapshot = registry_snapshot
[docs]
def resolve(self, context: CalculationSourceContext) -> CalculationSourceResolution:
"""Resolve the annual prorrata regularisation casilla for the active revision.
Loads the compiled revision (reusing ``registry_snapshot`` when set) to
discover the declared prorrata binding ids, then delegates to the governed
carry lookup to produce the bound value or an explicit missing/unresolved
diagnostic.
"""
revision = _revision_for_context(context, self._registry_snapshot)
declared_binding_ids = _prorrata_declared_binding_ids(revision)
if not declared_binding_ids:
return CalculationSourceResolution(resolver_id=self.resolver_id, owned_sources=self.owned_sources)
try:
source_period_feed = _source_period_feed_from_observations(
self._observation_repository,
operation=self._operation,
modelo=_prorrata_source_modelo(revision),
revision=revision,
filing_year=context.filing_year,
)
except (PersistenceDegradationError, ProrrataRegisterError) as exc:
return storage_degradation_resolution(
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
source_kinds=self.owned_sources,
error=exc,
)
current_year_values, missing_current = _merge_current_year_values(
source_period_feed,
revision=revision,
supplied_values=self._current_year_values,
missing_casilla_ids=self._missing_current_year_casilla_ids,
unresolved_casilla_ids=self._unresolved_current_year_casilla_ids,
)
missing_resolution = _missing_current_year_resolution(
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
declared_binding_ids=declared_binding_ids,
missing_current=missing_current,
)
if missing_resolution is not None:
return missing_resolution
try:
register = require_prorrata_register_coordinates_current(
self._prorrata_register_repository.load(),
operation=self._operation,
)
prior_definitiva = _stamped_prior_year_definitiva(
self._observation_repository,
operation=self._operation,
filing_year=context.filing_year,
modelo=_prorrata_source_modelo(revision),
revision=revision,
)
except (PersistenceDegradationError, ProrrataRegisterError) as exc:
return storage_degradation_resolution(
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
source_kinds=self.owned_sources,
error=exc,
)
current_provenance = _current_year_values_provenance(
context=context,
revision=revision,
source_periods=source_period_feed.source_periods,
)
register_entries = register.entries_for_ejercicio(context.filing_year)
applicability = derive_prorrata_applicability(
register_entries=register_entries,
declared_volume_total=current_year_values[_prorrata_source_id(revision, 2)],
declared_volume_con_derecho=current_year_values[_prorrata_source_id(revision, 1)],
)
if not applicability.applies:
return _zero_prorrata_resolution(
revision=revision,
context=context,
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
declared_binding_ids=declared_binding_ids,
current_provenance=current_provenance,
)
provisional_source = _resolve_prorrata_provisional_source(
register,
prior_definitiva,
context=context,
revision=revision,
current_provenance=current_provenance,
)
if provisional_source.percentage is None:
return _missing_provisional_resolution(
context=context,
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
declared_binding_ids=declared_binding_ids,
provenance=provisional_source.provenance,
)
return _resolved_prorrata_resolution(
revision=revision,
context=context,
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
declared_binding_ids=declared_binding_ids,
current_year_values=current_year_values,
provisional_percentage=provisional_source.percentage,
provenance=provisional_source.provenance,
)
[docs]
def buildprorrata_regularizacion_advisory(
*,
cuotas_soportadas_deducibles: Decimal,
prorrata_provisional_pct: Decimal,
prorrata_definitiva_pct: Decimal,
operaciones_sin_derecho_deduccion: Decimal,
regularizacion_year: int,
revision: ModeloRevision,
) -> tuple[RegularizacionProrrataResult, CalculationSourceDiagnostic | None]:
"""Compute the annual regularización and build the fallback advisory.
Returns the pure :class:`RegularizacionProrrataResult` plus a non-blocking
:class:`~application.aggregation.source_mesh.CalculationSourceDiagnostic` when the taxpayer
has exempt-without-right operations in the year (``operaciones_sin_derecho_
deduccion > 0`` — prorrata applies) and the definitive percentage differs from
the provisional one applied across the quarters (a regularización is due). In
that case a taxpayer who leaves casilla 44 blank is alerted rather than
silently under- or over-declaring. When prorrata does not apply, or the two
percentages coincide, the diagnostic is ``None`` (nothing to regularise, no
noise).
The diagnostic ``message`` names the provisional and definitive percentages,
the direction (deducción complementaria vs ingreso), and the proposed
casilla-44 value.
Args:
cuotas_soportadas_deducibles: The year's total deductible input IVA
(LIVA art. 105.Seis).
prorrata_provisional_pct: Provisional deduction percentage applied during
the year (LIVA art. 105.Uno — the prior-year definitive).
prorrata_definitiva_pct: Definitive deduction percentage for the year
(LIVA art. 104, computed from full-year volumes).
operaciones_sin_derecho_deduccion: The year's exempt-without-right
operation volume. When zero, prorrata does not apply and no
regularización is proposed.
regularizacion_year: The year being calculated (for the message).
revision: The selected Modelo 303 revision whose bindings declare the
regularisation target casilla.
Returns:
``(result, diagnostic)`` where ``result`` is the
:class:`RegularizacionProrrataResult`; the diagnostic is ``None`` when
there is nothing to regularise.
Core types:
:class:`~cadrumo.domain.calculations.registry.schema.ModeloRevision`.
"""
projection = project_prorrata_regularizacion_feed(
cuotas_soportadas_deducibles=cuotas_soportadas_deducibles,
prorrata_provisional_pct=prorrata_provisional_pct,
prorrata_definitiva_pct=prorrata_definitiva_pct,
operaciones_sin_derecho_deduccion=operaciones_sin_derecho_deduccion,
)
result = projection.result
if projection.proposed_value is None:
return result, None
sentido = "deducción complementaria" if result.direccion is RegularizacionProrrataDireccion.DEDUCCION else "ingreso"
target = _modelo_303_regularizacion_target(revision)
message = (
f"Regularización de prorrata por porcentaje definitivo (LIVA arts. 104-105) "
f"para {regularizacion_year}: prorrata provisional {prorrata_provisional_pct}% "
f"→ definitiva {prorrata_definitiva_pct}% ({sentido}). "
f"Regularización propuesta para {_regularizacion_target_label(revision, target)}: "
f"{projection.proposed_value}. Confirme el valor antes de presentar."
)
diagnostic = CalculationSourceDiagnostic(
reason="official_box_unpopulated",
source_kind=BindingSourceKind.PRORRATA_REGULARIZACION.value,
message=message,
casilla_id=target,
)
return result, diagnostic
_PRORRATA_SPECIAL_LEGAL_REF_KEY: Final = "prorrata_especial_mandatory.legal_ref"
def _resolve_prorrata_special_legal_ref(*, ejercicio: int, operation: PinnedAuthorityOperation) -> str:
"""Resolve the mandatory-special advisory's legal reference from registry data."""
resolved = operation.resolve_governed_fact(
MappingFactQuery(
fact_id="renta-iva-deduction-ratio-policy",
date_axis=DateAxis.FILING_PERIOD,
effective_date=date(ejercicio, 12, 31),
),
)
if not isinstance(resolved, ResolvedMappingFact):
raise RegistryValidationError("Renta IVA ratio policy must resolve as a mapping fact")
for entry in resolved.payload.entries:
if entry.key == _PRORRATA_SPECIAL_LEGAL_REF_KEY:
value = str(entry.value).strip()
if value:
return value
raise RegistryValidationError(
f"Renta IVA ratio policy is missing {_PRORRATA_SPECIAL_LEGAL_REF_KEY!r}",
)
[docs]
def build_prorrata_especial_mandatory_advisory(
*,
deduction_under_general: Decimal,
deduction_under_especial: Decimal,
ejercicio: int,
parameters: ProrrataEspecialMandatoryParameters,
operation: PinnedAuthorityOperation,
) -> Notice | None:
"""Build the LIVA art. 103.Dos.2.º mandatory-especial settlement advisory.
At settlement (4T / 0A), once the ejercicio's deducción computed under the
general regime and under the especial regime are both known, art. 103.Dos.2.º
makes prorrata especial OBLIGATORY when the general-regime deduction exceeds
the especial-regime deduction by the margin in force for that ejercicio —
ten percent or more from 2015, more than twenty percent before it
(:func:`~domain.iva.prorrata.is_especial_mandatory`). This surfaces that obligation as
a NON-BLOCKING warning :class:`~core.json_contract.Notice` so the operator
elects and records especial before filing; it NEVER refuses the in-progress
filing (the especial election is a filed taxpayer decision and the
classification data may still be incomplete). Both compared totals ride on
``Notice.context`` alongside the ejercicio and the binding legal reference.
Returns ``None`` when especial is not obligatory (no noise); the two amounts
must be non-negative (:func:`is_especial_mandatory` refuses negatives).
Args:
deduction_under_general: The ejercicio's total deducible IVA under the
prorrata general regime (single whole-entity percentage).
deduction_under_especial: The ejercicio's total deducible IVA under the
prorrata especial regime (per-input art. 106 routing).
ejercicio: The filing year being settled (for the message and context).
parameters: The registry-resolved LIVA art-103.Dos.2 margin and its
comparison direction for that ejercicio.
operation: Caller-owned generation-pinned authority operation used to
resolve the advisory's legal reference.
"""
if not is_especial_mandatory(
deduction_under_general,
deduction_under_especial,
year=ejercicio,
parameters=parameters,
):
return None
rule = especial_mandatory_rule(ejercicio, parameters=parameters)
exceso = f"en un {rule.margin_percentage}% o más" if rule.inclusive else f"en más de un {rule.margin_percentage}%"
message = (
f"Prorrata especial obligatoria para {ejercicio} (LIVA art. 103.Dos.2.º): la deducción por "
f"prorrata general ({deduction_under_general}) supera {exceso} la deducción por prorrata "
f"especial ({deduction_under_especial}). Aplique y registre la prorrata especial del ejercicio; "
"este aviso no bloquea la presentación."
)
return Notice(
severity=NoticeSeverity.WARNING,
code="modelo.work.calculate.prorrata_especial_obligatoria",
message=message,
context={
"ejercicio": str(ejercicio),
"regime": especial_prorrata_register_regime().value,
"deduction_under_general": str(deduction_under_general),
"deduction_under_especial": str(deduction_under_especial),
# The art. 103.Dos.2.º margin actually applied, and whether reaching
# it sufficed. Both are year-dependent (twenty percent exclusive
# before 2015, ten percent inclusive after), so an operator reading
# the envelope can tell which redaction produced the obligation.
"margin_percentage": str(rule.margin_percentage),
"margin_inclusive": "true" if rule.inclusive else "false",
"legal_refs": _resolve_prorrata_special_legal_ref(ejercicio=ejercicio, operation=operation),
},
)
__all__ = [
"ProrrataApplicabilityProjection",
"ProrrataDeclaredVolumeLedgerRollup",
"ProrrataRegularizacionFeedProjection",
"ProrrataRegularizacionSourceResolver",
"build_prorrata_especial_mandatory_advisory",
"build_prorrata_missing_provisional_advisory",
"buildprorrata_regularizacion_advisory",
"derive_prorrata_applicability",
"project_prorrata_regularizacion_feed",
]