"""Modelo 720 foreign-assets aggregation and source-mesh resolver.
Modelo 720 is an informativa declaration for assets and rights abroad. This
module groups per-asset observations by ``(source_kind, asset_class)`` and
returns :class:`ForeignAssetsAggregation` for
:mod:`application.aggregation._service`; the repository-free
:class:`ForeignAssetsAggregationSourceResolver` adapts the same observations to
the calculation source-mesh envelope when a caller supplies them.
Declarability is per regulatory obligation block. The aggregate keeps raw class
rollups, and :func:`declarable_asset_classes_720` applies each block's threshold
to the sum of every present class in that block. Observation construction accepts
only the four canonical source-kind values ``ledger_transaction``,
``purchase_invoice_evidence``, ``payable_invoice``, and
``collectible_invoice``; bare ``invoice`` is rejected.
"""
from __future__ import annotations
from collections.abc import Iterable
from decimal import Decimal
from pydantic import BaseModel, Field, InstanceOf, field_validator, model_validator
from ...core import (
MODELO_720_FOREIGN_ASSET_CLASS_CODES,
STRICT_FROZEN_CONFIG,
BindingSourceKind,
ForeignAssetObligationGroup,
Modelo,
Period,
foreign_asset_declaration_threshold,
foreign_asset_obligation_group,
)
from ...core.aggregation import ForeignAssetClass
from ...core.parsing import parse_iso8601_date
from ...domain.calculations.registry import Modelo720RowObservation, resolve_foreign_asset_binding_row_values
from ._source_mesh import CalculationSourceContext, CalculationSourceProvenance, CalculationSourceResolution
_CANONICAL_SOURCE_KINDS: frozenset[BindingSourceKind] = frozenset(
{
BindingSourceKind.LEDGER_TRANSACTION,
BindingSourceKind.PURCHASE_INVOICE_EVIDENCE,
BindingSourceKind.PAYABLE_INVOICE,
BindingSourceKind.COLLECTIBLE_INVOICE,
},
)
_OWNED_SOURCES: tuple[BindingSourceKind, ...] = (BindingSourceKind.FOREIGN_ASSET,)
def _foreign_asset_source_kind(value: object) -> BindingSourceKind:
if isinstance(value, BindingSourceKind):
source_kind = value
elif isinstance(value, str):
try:
source_kind = BindingSourceKind(value)
except ValueError as exc:
raise ValueError(f"foreign asset source_kind {value!r} is not a BindingSourceKind") from exc
else:
raise ValueError("foreign asset source_kind must be a BindingSourceKind or source-kind string")
if source_kind not in _CANONICAL_SOURCE_KINDS:
allowed = ", ".join(kind.value for kind in _CANONICAL_SOURCE_KINDS)
raise ValueError(
f"unsupported source_kind {source_kind.value!r}; use one of {allowed}",
)
return source_kind
def _validate_country(value: str) -> str:
if len(value) != 2 or any(char < "A" or char > "Z" for char in value):
raise ValueError(f"country must be uppercase ISO-3166 alpha-2, got {value!r}")
return value
[docs]
class ForeignAssetIngestObservation(BaseModel):
"""One asset observation for a Modelo 720 aggregator pass."""
model_config = STRICT_FROZEN_CONFIG
source_kind: BindingSourceKind
source_object_id: str = Field(min_length=1)
asset_class: ForeignAssetClass
asset_external_id: str = Field(min_length=1, max_length=128)
country: str = Field(min_length=2, max_length=2)
issuer_or_institution: str = Field(default="", max_length=200)
valuation_eur: Decimal = Field(ge=Decimal("0"))
acquisition_date: str = Field(min_length=10, max_length=10)
held_at_year_end: bool = True
@field_validator("source_kind", mode="before")
@classmethod
def _source_kind_is_canonical(cls, value: object) -> BindingSourceKind:
return _foreign_asset_source_kind(value)
@field_validator("country")
@classmethod
def _country_is_uppercase(cls, value: str) -> str:
return _validate_country(value)
[docs]
class ForeignAssetClassRollup(BaseModel):
"""Per-source-kind and per-asset-class rollup row."""
model_config = STRICT_FROZEN_CONFIG
source_kind: BindingSourceKind
asset_class: ForeignAssetClass
assets_count: int = Field(ge=0)
held_at_year_end_count: int = Field(ge=0)
total_valuation_eur: Decimal = Field(ge=Decimal("0"))
countries: tuple[str, ...] = Field(default_factory=tuple)
@field_validator("countries")
@classmethod
def _countries_are_uppercase_ascii_alpha2(cls, value: tuple[str, ...]) -> tuple[str, ...]:
for country in value:
_validate_country(country)
return value
@field_validator("source_kind", mode="before")
@classmethod
def _source_kind_is_canonical(cls, value: object) -> BindingSourceKind:
return _foreign_asset_source_kind(value)
@model_validator(mode="after")
def _held_count_within_total(self) -> ForeignAssetClassRollup:
if self.held_at_year_end_count > self.assets_count:
raise ValueError(
f"held_at_year_end_count {self.held_at_year_end_count} > assets_count {self.assets_count}",
)
return self
[docs]
class ForeignAssetsAggregation(BaseModel):
"""720 aggregation output: per-class rollups + cross-class totals."""
model_config = STRICT_FROZEN_CONFIG
modelo: str = Field(min_length=1)
period: InstanceOf[Period]
rollups: tuple[ForeignAssetClassRollup, ...] = Field(default_factory=tuple)
total_assets: int = Field(ge=0)
total_valuation_eur: Decimal = Field(ge=Decimal("0"))
@model_validator(mode="after")
def _totals_match_rollups(self) -> ForeignAssetsAggregation:
computed_assets = sum(row.assets_count for row in self.rollups)
computed_valuation = sum(
(row.total_valuation_eur for row in self.rollups),
Decimal("0"),
)
if computed_assets != self.total_assets:
raise ValueError(
f"total_assets {self.total_assets} != sum of rollups {computed_assets}",
)
if computed_valuation != self.total_valuation_eur:
raise ValueError(
f"total_valuation_eur {self.total_valuation_eur} != sum of rollups {computed_valuation}",
)
cohorts = [(row.source_kind, row.asset_class) for row in self.rollups]
if len(cohorts) != len(set(cohorts)):
raise ValueError("each source_kind and ForeignAssetClass cohort may appear at most once in rollups")
return self
[docs]
def declarable_asset_classes_720(aggregation: ForeignAssetsAggregation) -> frozenset[ForeignAssetClass]:
"""Return present asset classes whose obligation block exceeds its 720 declaration floor.
Returns a frozenset of :class:`ForeignAssetClass` members.
"""
group_totals: dict[ForeignAssetObligationGroup, Decimal] = {}
asset_classes_by_group: dict[ForeignAssetObligationGroup, set[ForeignAssetClass]] = {}
for rollup in aggregation.rollups:
group = foreign_asset_obligation_group(rollup.asset_class)
group_totals[group] = group_totals.get(group, Decimal("0")) + rollup.total_valuation_eur
asset_classes_by_group.setdefault(group, set()).add(rollup.asset_class)
return frozenset(
asset_class
for group, total in group_totals.items()
if total > foreign_asset_declaration_threshold(group).initial_declaration_floor_eur
for asset_class in asset_classes_by_group[group]
)
[docs]
def declarable_class(aggregation: ForeignAssetsAggregation, *, asset_class: ForeignAssetClass) -> bool:
"""Return True iff an asset class's obligation block crosses the 720 declaration floor."""
return asset_class in declarable_asset_classes_720(aggregation)
[docs]
def aggregate_foreign_assets_720(
observations: tuple[ForeignAssetIngestObservation, ...],
*,
period: Period,
) -> ForeignAssetsAggregation:
"""Aggregate Modelo 720 observations into per-class rollups.
Returns a :class:`ForeignAssetsAggregation` grouping observations
by asset class. Pure function: identical observation input + period
yields identical output. Rollups are sorted by asset_class.value so
two equal aggregations serialise to identical bytes.
No threshold gate is applied here; callers use
:func:`declarable_class` to filter rollups by obligation block before binding to
Modelo 720 casillas.
"""
grouped: dict[tuple[BindingSourceKind, ForeignAssetClass], list[ForeignAssetIngestObservation]] = {}
for obs in observations:
grouped.setdefault((obs.source_kind, obs.asset_class), []).append(obs)
rollups: list[ForeignAssetClassRollup] = []
for source_kind, asset_class in sorted(grouped, key=lambda c: (c[0], c[1].value)):
group = grouped[(source_kind, asset_class)]
countries = tuple(sorted({obs.country for obs in group}))
rollups.append(
ForeignAssetClassRollup(
source_kind=source_kind,
asset_class=asset_class,
assets_count=len(group),
held_at_year_end_count=sum(1 for o in group if o.held_at_year_end),
total_valuation_eur=sum(
(obs.valuation_eur for obs in group),
Decimal("0"),
),
countries=countries,
),
)
return ForeignAssetsAggregation(
modelo=Modelo.M720.value,
period=period,
rollups=tuple(rollups),
total_assets=sum(row.assets_count for row in rollups),
total_valuation_eur=sum(
(row.total_valuation_eur for row in rollups),
Decimal("0"),
),
)
[docs]
class ForeignAssetsAggregationSourceResolver:
"""Resolve Modelo 720 foreign-asset rows from operator-supplied observations.
The resolver is deliberately repository-free, mirroring the existing
``aggregate_foreign_assets_720`` shape-C surface: callers supply typed
observations, the resolver delegates to the aggregate function for threshold
semantics, then validates the declarable rows against the live M720 registry
row-producer bindings.
"""
resolver_id = "foreign_assets_aggregation"
owned_sources = _OWNED_SOURCES
def __init__(self, *, observations: Iterable[ForeignAssetIngestObservation] = ()) -> None:
self._observations = tuple(observations)
[docs]
def resolve(self, context: CalculationSourceContext) -> CalculationSourceResolution:
if not _foreign_asset_source_for_revision(context):
return CalculationSourceResolution(resolver_id=self.resolver_id, owned_sources=self.owned_sources)
aggregation = aggregate_foreign_assets_720(self._observations, period=context.period)
selected_observations = _selected_foreign_asset_observations(aggregation, self._observations)
row_observations = _registry_observations_from_foreign_assets_aggregation(
aggregation,
selected_observations,
)
row_binding_values = resolve_foreign_asset_binding_row_values(context.revision, row_observations)
return CalculationSourceResolution(
resolver_id=self.resolver_id,
owned_sources=self.owned_sources,
row_binding_values=row_binding_values,
source_transaction_ids=tuple(
sorted(
observation.source_object_id
for observation in selected_observations
if observation.source_kind is BindingSourceKind.LEDGER_TRANSACTION
),
),
provenance=tuple(
CalculationSourceProvenance(
source_kind=observation.source_kind.value,
source_ref=f"{observation.source_kind.value}:{observation.source_object_id}",
)
for observation in selected_observations
),
)
def _foreign_asset_source_for_revision(context: CalculationSourceContext) -> bool:
return any(binding.source == BindingSourceKind.FOREIGN_ASSET for binding in context.revision.bindings)
def _registry_observations_from_foreign_assets_aggregation(
aggregation: ForeignAssetsAggregation,
observations: Iterable[ForeignAssetIngestObservation],
) -> tuple[Modelo720RowObservation, ...]:
declarable_classes = declarable_asset_classes_720(aggregation)
return tuple(
_registry_observation_from_foreign_asset(observation)
for observation in observations
if observation.asset_class in declarable_classes
)
def _selected_foreign_asset_observations(
aggregation: ForeignAssetsAggregation,
observations: Iterable[ForeignAssetIngestObservation],
) -> tuple[ForeignAssetIngestObservation, ...]:
declarable_classes = declarable_asset_classes_720(aggregation)
return tuple(observation for observation in observations if observation.asset_class in declarable_classes)
def _registry_observation_from_foreign_asset(
observation: ForeignAssetIngestObservation,
) -> Modelo720RowObservation:
acquisition_date = parse_iso8601_date(observation.acquisition_date)
if acquisition_date is None:
raise ValueError(f"acquisition_date {observation.acquisition_date!r} is not a valid ISO-8601 date")
return Modelo720RowObservation(
source_id=f"{observation.source_kind.value}:{observation.source_object_id}",
asset_class_code=_asset_class_code(observation.asset_class),
country_code=observation.country,
asset_identifier=observation.asset_external_id,
acquisition_date=acquisition_date,
valuation_amount=observation.valuation_eur,
)
def _asset_class_code(asset_class: ForeignAssetClass) -> str:
try:
return MODELO_720_FOREIGN_ASSET_CLASS_CODES[asset_class]
except KeyError as exc:
raise ValueError(f"{asset_class.value!r} is not a Modelo 720 foreign-asset class") from exc
__all__ = [
"ForeignAssetClass",
"ForeignAssetClassRollup",
"ForeignAssetIngestObservation",
"ForeignAssetsAggregation",
"ForeignAssetsAggregationSourceResolver",
"aggregate_foreign_assets_720",
"declarable_asset_classes_720",
"declarable_class",
]