"""Foreign-asset re-declaration advisory helpers for Modelo 720.
Every projection here reads a persisted :class:`CalculationRevision` against the
registry's :class:`ModeloRevision` for the same filing context, and emits its
result as :class:`CasillaObservation` rows so the advisory carries the same
per-casilla grounding the rest of the calculation chain does.
See Also:
:mod:`~application.foreign_asset_thresholds`
Resolves the per-bloque declaration floors and re-declaration deltas
from the effective registry revision.
:class:`~domain.calculations.registry.bindings.RegistryModeloObservation`
Registry-grounded observation envelope accepted as prior, current, and
current-declaration evidence.
``application.calculations.tests.test_modelo_720_prior_year_baseline_fidelity``
Exercises the Modelo 720 prior-year baseline advisory path with real
observations across two filing years.
"""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from urllib.parse import quote
from ...core.aggregation import BindingSourceKind, ForeignAssetClass
from ...core.casilla_id import CasillaId
from ...core.foreign_asset_obligation import (
MODELO_720_FOREIGN_ASSET_CLASS_CODES,
ForeignAssetObligationGroup,
)
from ...core.modelo import Modelo
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
from ...domain.calculations.registry.bindings import CasillaObservation, RegistryModeloObservation
from ...domain.calculations.registry.bindings_previous_filing import previous_filing_binding_source_casilla_ids
from ...domain.calculations.registry.detail_record_bindings import ForeignAssetProvider
from ...domain.calculations.registry.foreign_asset_obligation_catalogue import (
resolve_foreign_asset_obligation_catalogue,
)
from ...domain.calculations.registry.schema import ModeloRevision
from ...domain.calculations.registry.schema_base import CasillaDataType
from ...domain.modelos.calculation_revision import CalculationRevision
from ...domain.modelos.verification_report import (
ModeloVerificationFinding,
ModeloVerificationFindingKind,
ModeloVerificationFindingSeverity,
)
from ..foreign_asset_thresholds import foreign_asset_declaration_thresholds
@dataclass(frozen=True, slots=True)
class _RedeclarationPosition:
key: tuple[str, ...]
group: ForeignAssetObligationGroup
value_eur: Decimal
def _registry_valuation_casillas(
modelo_revision: ModeloRevision,
*,
operation: PinnedAuthorityOperation,
) -> dict[ForeignAssetObligationGroup, CasillaId]:
"""Discover valuation casillas and their obligation groups from a revision."""
result: dict[ForeignAssetObligationGroup, CasillaId] = {}
del operation
# TODO(authority-owner): resolve the foreign-asset taxonomy from the same
# operation. The legacy catalogue adapter has not yet accepted
# PinnedAuthorityOperation and must not be reopened here.
catalogue = resolve_foreign_asset_obligation_catalogue()
for casilla in modelo_revision.casillas:
parts = tuple(str(casilla.id).split("."))
if len(parts) < 2 or parts[-1].casefold() != "valoracion":
continue
namespace = parts[-2].replace("-", "_").casefold()
for definition in catalogue.groups:
group = definition.token
if namespace == group.value or namespace in group.value.split("_"):
result[group] = casilla.id
break
return result
def _foreign_asset_binding_ids(
revision: CalculationRevision,
modelo_revision: ModeloRevision,
) -> tuple[str | None, str | None]:
"""Discover class and valuation row bindings without local binding IDs."""
foreign_asset_bindings = tuple(
binding for binding in modelo_revision.bindings if isinstance(binding.provider, ForeignAssetProvider)
)
valuation_binding = next(
(
binding.id
for binding in foreign_asset_bindings
if isinstance(binding.provider, ForeignAssetProvider)
and binding.provider.data_type is CasillaDataType.MONEY
),
None,
)
class_binding = next(
(
binding.id
for binding in foreign_asset_bindings
if binding.id != valuation_binding
and any(
_modelo_720_asset_class(value) is not None
for value in revision.row_binding_values.get(binding.id, {}).values()
)
),
None,
)
return class_binding, valuation_binding
def _modelo_720_asset_class(raw_code: object) -> ForeignAssetClass | None:
"""Hydrate a registry row code through the canonical closed asset-class map."""
if not isinstance(raw_code, str):
return None
normalized = raw_code.strip().upper()
return next(
(
asset_class
for asset_class, code in MODELO_720_FOREIGN_ASSET_CLASS_CODES.items()
if str(code).upper() == normalized
),
None,
)
def _obligation_group_for_valuation_casilla(
casilla_id: CasillaId,
*,
operation: PinnedAuthorityOperation,
) -> ForeignAssetObligationGroup | None:
"""Infer the typed obligation group from a registry valuation casilla identity."""
parts = tuple(str(casilla_id).split("."))
if len(parts) < 2 or parts[-1].casefold() != "valoracion":
return None
namespace = parts[-2].replace("-", "_").casefold()
del operation
catalogue = resolve_foreign_asset_obligation_catalogue()
return next(
(
definition.token
for definition in catalogue.groups
if namespace == definition.token.value or namespace in definition.token.value.split("_")
),
None,
)
[docs]
def modelo_720_redeclaration_advisory_findings(
*,
prior_observation: RegistryModeloObservation,
current_observation: RegistryModeloObservation,
current_declaration_observation: RegistryModeloObservation | None = None,
operation: PinnedAuthorityOperation,
) -> tuple[ModeloVerificationFinding, ...]:
"""Return non-blocking M720 re-declaration advisories for omitted grown groups.
``current_observation`` is the operator's current-year valuation evidence;
``current_declaration_observation`` is the current filed/declarable row set.
When omitted, the current evidence is also treated as the current declaration,
so a correctly present row produces no advisory.
See Also:
:func:`~application.foreign_asset_thresholds.foreign_asset_declaration_thresholds`
Supplies the strict per-obligation-block re-declaration delta.
"""
return _redeclaration_advisory_findings(
modelo=Modelo("720").value,
prior_positions=_modelo_720_positions(prior_observation, operation=operation),
current_positions=_modelo_720_positions(current_observation, operation=operation),
declared_positions=_modelo_720_positions(
current_declaration_observation or current_observation,
operation=operation,
),
filing_year=current_observation.filing_year,
operation=operation,
)
def _redeclaration_advisory_findings(
*,
modelo: str,
prior_positions: Mapping[tuple[str, ...], _RedeclarationPosition],
current_positions: Mapping[tuple[str, ...], _RedeclarationPosition],
declared_positions: Mapping[tuple[str, ...], _RedeclarationPosition],
filing_year: int,
operation: PinnedAuthorityOperation,
) -> tuple[ModeloVerificationFinding, ...]:
thresholds = foreign_asset_declaration_thresholds(modelo=modelo, filing_year=filing_year, operation=operation)
findings: list[ModeloVerificationFinding] = []
for key in sorted(prior_positions):
prior = prior_positions[key]
current = current_positions.get(key)
if current is None or key in declared_positions:
continue
threshold = thresholds[prior.group]
delta_eur = current.value_eur - prior.value_eur
if delta_eur <= threshold.redeclaration_increase_delta_eur:
continue
findings.append(
ModeloVerificationFinding(
kind=ModeloVerificationFindingKind.ADVISORY,
severity=ModeloVerificationFindingSeverity.WARNING,
message_locale_key="application.modelo.findings.foreign_asset_redeclaration",
message_facts={
"modelo_code": modelo,
"filing_year": filing_year,
"position_key": _position_key_code(current.key),
"group_code": current.group.value,
"prior_value_eur": prior.value_eur,
"current_value_eur": current.value_eur,
"delta_value_eur": delta_eur,
"redeclaration_increase_threshold_eur": threshold.redeclaration_increase_delta_eur,
},
legal_refs=threshold.legal_refs,
source_refs=threshold.source_refs,
)
)
return tuple(findings)
def _modelo_720_positions(
observation: RegistryModeloObservation,
*,
operation: PinnedAuthorityOperation,
) -> Mapping[tuple[str, ...], _RedeclarationPosition]:
totals: dict[ForeignAssetObligationGroup, Decimal] = {}
for item in observation.observations:
group = _obligation_group_for_valuation_casilla(item.casilla_id, operation=operation)
if group is None:
continue
if not isinstance(item.value, Decimal):
continue
totals[group] = totals.get(group, Decimal("0")) + item.value
return {
(group.value,): _RedeclarationPosition(
key=(group.value,),
group=group,
value_eur=value,
)
for group, value in totals.items()
}
def _position_key_code(key: tuple[str, ...]) -> str:
"""Encode the complete source-position identity as a locale-neutral token."""
return "|".join(quote(part, safe="") for part in key)
[docs]
def modelo_720_declared_observation(
*,
revision: CalculationRevision,
modelo_revision: ModeloRevision,
filing_year: int,
period: str,
operation: PinnedAuthorityOperation,
) -> RegistryModeloObservation:
"""Project what the operator actually DECLARED on a Modelo 720 draft.
The declared set is read from
:attr:`~CalculationRevision.input_values_by_casilla_id` —
the operator-supplied inputs — and deliberately NOT from
``casilla_values``. The engine materialises every declared casilla in
``casilla_values``, emitting ``0`` for a valuation the operator never
supplied; a declared set derived from that mapping would therefore always
contain all three obligation blocks, and the omitted-position test in
:func:`modelo_720_redeclaration_advisory_findings` could never be
satisfied. An operator-entered zero remains a genuine declaration and is
kept.
Returns a :class:`RegistryModeloObservation` carrying only the valuation
casillas present in the operator's input set.
Args:
revision: The persisted :class:`CalculationRevision` whose operator
inputs are read as the declaration.
modelo_revision: The :class:`ModeloRevision` supplying each valuation
casilla's legal and source grounding.
filing_year: Devengo year the observation is stamped with.
period: Registry period token the observation is stamped with.
operation: The caller-owned generation-pinned authority operation used
for the selected foreign-asset declarations.
See Also:
:func:`modelo_720_evidence_observation`
The independent valuation counterpart this declaration is tested
against.
"""
valuation_casillas = _registry_valuation_casillas(modelo_revision, operation=operation)
refs_by_casilla = {casilla.id: (casilla.legal_refs, casilla.source_refs) for casilla in modelo_revision.casillas}
observations: list[CasillaObservation] = []
for casilla_id in valuation_casillas.values():
raw = revision.input_values_by_casilla_id.get(casilla_id)
refs = refs_by_casilla.get(casilla_id)
if raw is None or refs is None:
continue
value = _decimal_or_none(raw)
if value is None:
continue
observations.append(
CasillaObservation(
casilla_id=casilla_id,
value=value,
legal_refs=refs[0],
source_refs=refs[1],
),
)
return RegistryModeloObservation(
modelo=Modelo("720").value,
filing_year=filing_year,
period=period,
observations=tuple(observations),
)
[docs]
def modelo_720_evidence_observation(
*,
revision: CalculationRevision,
modelo_revision: ModeloRevision,
filing_year: int,
period: str,
operation: PinnedAuthorityOperation,
) -> RegistryModeloObservation:
"""Project the per-asset-row valuation EVIDENCE on a Modelo 720 draft.
Joins the revision's persisted ``row_binding_values`` for the two
``foreign_asset`` row bindings that carry the asset class clave and the
row valuation, resolves each clave to its RGAT obligation bloque, and sums
the rows per bloque onto that bloque's valuation casilla.
This is independent of what the operator typed into the declaration: the
rows come from the foreign-asset source mesh, the declaration comes from
the operator's manual inputs. That independence is what makes the omitted
position detectable at all.
The two binding ids are discovered from ``modelo_revision`` by their
selector ``row_field`` rather than hardcoded, so a registry rename does not
silently empty the evidence side.
Returns an empty observation when the revision carries no asset rows.
Args:
revision: The persisted :class:`CalculationRevision` whose
``row_binding_values`` carry the per-asset rows.
modelo_revision: The :class:`ModeloRevision` the two ``foreign_asset``
row bindings are discovered from by selector.
filing_year: Devengo year the observation is stamped with.
period: Registry period token the observation is stamped with.
operation: The caller-owned generation-pinned authority operation used
for the selected foreign-asset declarations.
"""
class_binding, valuation_binding = _foreign_asset_binding_ids(revision, modelo_revision)
totals = _modelo_720_evidence_totals(
revision,
class_binding=class_binding,
valuation_binding=valuation_binding,
operation=operation,
)
observations = _modelo_720_evidence_observations(totals, modelo_revision=modelo_revision, operation=operation)
return RegistryModeloObservation(
modelo=Modelo("720").value,
filing_year=filing_year,
period=period,
observations=observations,
)
def _modelo_720_evidence_totals(
revision: CalculationRevision,
*,
class_binding: str | None,
valuation_binding: str | None,
operation: PinnedAuthorityOperation,
) -> dict[ForeignAssetObligationGroup, Decimal]:
if class_binding is None or valuation_binding is None:
return {}
class_rows = revision.row_binding_values.get(class_binding, {})
valuation_rows = revision.row_binding_values.get(valuation_binding, {})
totals: dict[ForeignAssetObligationGroup, Decimal] = {}
for row_index, raw_code in class_rows.items():
row = _modelo_720_evidence_row(raw_code, valuation_rows.get(row_index))
if row is None:
continue
asset_class, value = row
group = resolve_foreign_asset_obligation_catalogue().group_for_asset_class(asset_class)
totals[group] = totals.get(group, Decimal("0")) + value
return totals
def _modelo_720_evidence_row(
raw_code: str,
raw_valuation: object,
) -> tuple[ForeignAssetClass, Decimal] | None:
asset_class = _modelo_720_asset_class(raw_code)
if asset_class is None or raw_valuation is None:
return None
value = _decimal_or_none(raw_valuation)
if value is None:
return None
return asset_class, value
def _modelo_720_evidence_observations(
totals: Mapping[ForeignAssetObligationGroup, Decimal],
*,
modelo_revision: ModeloRevision,
operation: PinnedAuthorityOperation,
) -> tuple[CasillaObservation, ...]:
valuation_casillas = _registry_valuation_casillas(modelo_revision, operation=operation)
refs_by_casilla = {casilla.id: (casilla.legal_refs, casilla.source_refs) for casilla in modelo_revision.casillas}
observations: list[CasillaObservation] = []
for group, total in totals.items():
observation = _modelo_720_evidence_observation(
group,
total,
refs_by_casilla=refs_by_casilla,
valuation_casillas=valuation_casillas,
)
if observation is not None:
observations.append(observation)
return tuple(observations)
def _modelo_720_evidence_observation(
group: ForeignAssetObligationGroup,
total: Decimal,
*,
refs_by_casilla: Mapping[CasillaId, tuple[tuple[str, ...], tuple[str, ...]]],
valuation_casillas: Mapping[ForeignAssetObligationGroup, CasillaId],
) -> CasillaObservation | None:
casilla_id = valuation_casillas.get(group)
refs = refs_by_casilla.get(casilla_id) if casilla_id is not None else None
if casilla_id is None or refs is None:
return None
return CasillaObservation(
casilla_id=casilla_id,
value=total,
legal_refs=refs[0],
source_refs=refs[1],
)
[docs]
def modelo_720_prior_baseline_observation(
*,
binding_values: Mapping[str, Decimal],
modelo_revision: ModeloRevision,
filing_year: int,
period: str,
operation: PinnedAuthorityOperation,
) -> RegistryModeloObservation:
"""Project the prior-year declared baseline resolved by the previous-filing carry.
``binding_values`` is the resolved mapping produced by
:func:`~._binding_prefill.resolve_bindings_from_local_store`, which has
already re-confirmed each carried observation's stamped revision against
the law-determined revision for its source context. Only the
``previous_filing`` bindings whose selector names a Modelo 720 valuation
casilla are projected; every other resolved binding is ignored.
Returns an empty observation when no prior baseline resolved, which leaves
the re-declaration advisory silent rather than comparing against a
fabricated zero baseline.
Args:
binding_values: Resolved ``previous_filing`` binding values, already
revision-confirmed against their source context.
modelo_revision: The :class:`ModeloRevision` supplying each projected
casilla's legal and source grounding.
filing_year: Devengo year the observation is stamped with.
period: Registry period token the observation is stamped with.
operation: The caller-owned generation-pinned authority operation used
for the selected foreign-asset declarations.
"""
valuation_casillas = _registry_valuation_casillas(modelo_revision, operation=operation)
refs_by_casilla = {casilla.id: (casilla.legal_refs, casilla.source_refs) for casilla in modelo_revision.casillas}
observations: list[CasillaObservation] = []
for binding in modelo_revision.bindings:
if binding.source != BindingSourceKind.PREVIOUS_FILING:
continue
value = binding_values.get(binding.id)
if value is None:
continue
source_casilla_ids = previous_filing_binding_source_casilla_ids(binding)
# binding_values resolves ONE value per binding id, so this observation
# only applies when the binding targets exactly one casilla -- a
# multi-casilla (plural source_casilla_ids) binding has no single value
# to attribute to which of its several targets, and is out of scope
# here, same as before this typed read replaced the raw singular-key
# lookup (which only ever matched the singular shape in the first
# place).
if len(source_casilla_ids) != 1:
continue
(source_casilla_id,) = source_casilla_ids
if source_casilla_id not in valuation_casillas.values():
continue
refs = refs_by_casilla.get(source_casilla_id)
if refs is None:
continue
observations.append(
CasillaObservation(
casilla_id=source_casilla_id,
value=value,
legal_refs=refs[0],
source_refs=refs[1],
),
)
return RegistryModeloObservation(
modelo=Modelo("720").value,
filing_year=filing_year,
period=period,
observations=tuple(observations),
)
def _decimal_or_none(raw: object) -> Decimal | None:
if isinstance(raw, Decimal):
return raw
try:
return Decimal(str(raw).strip())
except (InvalidOperation, ValueError):
return None
__all__ = [
"modelo_720_declared_observation",
"modelo_720_evidence_observation",
"modelo_720_prior_baseline_observation",
"modelo_720_redeclaration_advisory_findings",
]