"""Formula-runtime operation helpers for registry calculations.
The formula evaluator delegates arithmetic dispatch, dated parameter lookup,
rounding, and input validation here while executing
:class:`~domain.calculations.registry.schema.ModeloRevision` formula graphs.
Helpers raise :class:`~domain.calculations.registry.errors.RegistryValidationError`
so :func:`domain.calculations.registry.formula_runtime.calculate_registry_snapshot`
reports contract failures through the registry error channel.
See Also:
:mod:`domain.calculations.registry.formula_runtime`
Snapshot evaluator that calls these helpers while materialising
:class:`~domain.calculations.registry.schema.RegistrySnapshot` outputs.
:mod:`domain.calculations.registry.runtime_graph`
Formula graph walkers that discover the casilla, binding, relation, and
parameter refs consumed before operation dispatch starts.
:class:`domain.calculations.registry.authority.ValidatedRegistryAuthority`
Registry authority loaded by :func:`read_parameter` for ad hoc parameter
reads outside snapshot execution.
"""
from __future__ import annotations
from collections.abc import Mapping
from datetime import date
from decimal import ROUND_CEILING, ROUND_HALF_UP, Decimal
from enum import StrEnum
from typing import TYPE_CHECKING
from ....core.casilla_id import CasillaId, validated_casilla_id
from ....core.decimal.constants import ONE, ZERO
from ....core.money.rounding import round_to_cents as _round_to_cents
from ._formula_operator_contracts import require_formula_operator_arity
from .casilla_membership import undeclared_casilla_ids
from .errors import RegistrySnapshotError, RegistryValidationError
from .ids import RevisionId
from .schema_base import NUMERIC_CASILLA_DATA_TYPES
from .schema_formula import BracketEntry, DatedValue, ParameterDefinition
from .schema_rounding import RegistryRoundingCode
if TYPE_CHECKING:
from _typeshed import SupportsAllComparisons
from .authority import PinnedAuthorityOperation
from .formula_runtime import EvalContext as _EvalContext
from .schema import ModeloRevision
_COMPARISON_OPS = frozenset({"less_than", "less_equal", "greater_than", "greater_equal", "equal"})
_UNARY_PASSTHROUGH_OPS = frozenset({"copy", "lookup_parameter", "previous_period_value", "cross_model_sum"})
[docs]
class RegistryUnresolvedOutcomeReason(StrEnum):
"""Closed reason catalogue for typed formula outcomes with no Decimal value."""
M210_BASELINE_TIPO_DEFERRED = "m210-baseline-tipo-deferred"
M210_CONVENIO_RATE_MISSING = "m210-convenio-rate-missing"
[docs]
def numeric_casilla_value(casilla_id: CasillaId, ctx: _EvalContext) -> Decimal:
"""Read a resolved numeric casilla value from the evaluation context.
Generic accessor shared by the M210/IRNR, M131 módulos, and M303 módulos
IVA formula-op families; each raises :class:`UnresolvedFormulaDependencyError`
the same way for a casilla deferred by a non-blocking source gap.
"""
if casilla_id not in ctx.values:
if casilla_id in ctx.unresolved_casilla_ids:
raise UnresolvedFormulaDependencyError((casilla_id,))
raise RegistryValidationError(
f"casilla {casilla_id!r} referenced before evaluation",
translated_message="errors.calc.casilla_referenced_before_evaluation",
context={"casilla_id": casilla_id},
)
value = ctx.values[casilla_id]
ctx.operand_refs.append(casilla_id)
ctx.operand_casilla_refs.append(casilla_id)
ctx.operand_values.append(value)
return value
[docs]
def evaluate_args_op(op: str, args: list[Decimal]) -> Decimal:
"""Evaluate a resolved formula operation over decimal operands.
Operation names mirror
:class:`~domain.calculations.registry.schema_formula.FormulaExpression` ``op`` values
consumed by
:func:`domain.calculations.registry.formula_runtime.calculate_registry_snapshot`.
"""
require_formula_operator_arity(op, len(args))
if op in {"add", "sum", "previous_period_sum"}:
if op == "previous_period_sum":
_require_non_empty(op, args)
return sum(args, ZERO)
if op in _COMPARISON_OPS:
_require_arg_count(op, args, 2)
return ONE if _compare(op, args[0], args[1]) else ZERO
if op in _UNARY_PASSTHROUGH_OPS:
_require_arg_count(op, args, 1)
return args[0]
return _dispatch_named_arithmetic_op(op, args)
def _dispatch_named_arithmetic_op(op: str, args: list[Decimal]) -> Decimal:
"""Dispatch non-comparison arithmetic operations for :func:`evaluate_args_op`."""
if op == "multiply":
return _multiply_operands(args)
if op in {"subtract", "divide", "percent"}:
return _dispatch_binary_arithmetic_op(op, args)
if op in {"min", "max"}:
return _dispatch_extreme_arithmetic_op(op, args)
if op == "clamp":
_require_arg_count(op, args, 3)
return max(args[1], min(args[0], args[2]))
if op == "negate":
_require_arg_count(op, args, 1)
return -args[0]
raise RegistryValidationError(f"formula expression uses unsupported op {op!r}")
def _multiply_operands(args: list[Decimal]) -> Decimal:
"""Multiply operands in their declared order."""
result = ONE
for arg in args:
result *= arg
return result
def _dispatch_binary_arithmetic_op(op: str, args: list[Decimal]) -> Decimal:
"""Evaluate the binary arithmetic operations with their canonical checks."""
_require_arg_count(op, args, 2)
if op == "subtract":
return args[0] - args[1]
if op == "divide":
if args[1] == ZERO:
raise RegistryValidationError(
"formula expression divides by zero",
translated_message="errors.calc.divide_by_zero",
)
return args[0] / args[1]
if op == "percent":
return args[0] * args[1] / Decimal("100")
raise RegistryValidationError(f"formula expression uses unsupported op {op!r}")
def _dispatch_extreme_arithmetic_op(op: str, args: list[Decimal]) -> Decimal:
"""Select the minimum or maximum from a non-empty operand sequence."""
_require_non_empty(op, args)
if op == "min":
return min(args)
if op == "max":
return max(args)
raise RegistryValidationError(f"formula expression uses unsupported op {op!r}")
def _compare(op: str, left: Decimal, right: Decimal) -> bool:
"""Evaluate a comparison operation from a registry formula expression."""
if op == "less_than":
return left < right
if op == "less_equal":
return left <= right
if op == "greater_than":
return left > right
if op == "greater_equal":
return left >= right
if op == "equal":
return left == right
raise RegistryValidationError(f"formula expression uses unsupported comparison op {op!r}")
def _bracket_candidates(
parameter: ParameterDefinition,
date_context: Mapping[str, date],
) -> list[BracketEntry]:
if parameter.data_type != "bracket_table":
raise RegistryValidationError(
f"parameter {parameter.id!r} must declare data_type='bracket_table' to use lookup_bracket",
)
if parameter.bracket_axis is None:
raise RegistryValidationError(f"parameter {parameter.id!r} bracket_table requires bracket_axis")
if parameter.bracket_axis not in date_context:
raise RegistryValidationError(f"parameter {parameter.id!r} requires date axis {parameter.bracket_axis!r}")
selected = date_context[parameter.bracket_axis]
candidates: list[BracketEntry] = [
bracket
for bracket in parameter.brackets
if bracket.valid_from <= selected and (bracket.valid_to is None or selected <= bracket.valid_to)
]
if not candidates:
raise RegistryValidationError(
f"parameter {parameter.id!r} has no bracket valid for {selected.isoformat()}",
translated_message="errors.calc.bracket_no_window",
context={"parameter_id": parameter.id, "as_of": selected.isoformat()},
)
return candidates
def _resolve_bracket_entry(
parameter: ParameterDefinition,
candidates: list[BracketEntry],
base: Decimal,
) -> BracketEntry:
if base < Decimal("0"):
raise RegistryValidationError(
f"parameter {parameter.id!r} lookup_bracket received negative base {base}",
translated_message="errors.calc.bracket_negative_base",
context={"parameter_id": parameter.id, "base": str(base)},
)
for entry in sorted(candidates, key=lambda bracket: bracket.lower_bound):
if entry.lower_bound <= base and (entry.upper_bound is None or base <= entry.upper_bound):
return entry
raise RegistryValidationError(
f"parameter {parameter.id!r} has no bracket covering base {base}",
translated_message="errors.calc.bracket_no_coverage",
context={"parameter_id": parameter.id, "base": str(base)},
)
[docs]
def resolve_bracket(
parameter: ParameterDefinition,
base: Decimal,
date_context: Mapping[str, date],
) -> Decimal:
"""Resolve a bracket-table :class:`ParameterDefinition` for a base amount.
The parameter's bracket date axis must be present in ``date_context`` so
registry-authored validity windows select exactly one bracket row.
"""
candidates = _bracket_candidates(parameter, date_context)
base = Decimal(base)
selected_entry = _resolve_bracket_entry(parameter, candidates, base)
return selected_entry.fixed_addition + selected_entry.marginal_rate * (base - selected_entry.lower_bound)
[docs]
def resolve_dated_value(parameter: ParameterDefinition, date_context: Mapping[str, date]) -> DatedValue:
"""Resolve the one dated value a :class:`ParameterDefinition` selects.
Exactly one :class:`~domain.calculations.registry.schema_formula.DatedValue` must match
the selected date axes for the parameter lookup to be deterministic.
Returns the RECORD rather than its scalar, because a value can carry more
than the number: a threshold's comparison operator moves independently of its
amount, so a caller applying a threshold needs the value and the operator
that the same redaction fixed. :func:`resolve_parameter` delegates here so
the selection lives in one place and the two cannot drift.
Returns:
The single matching :class:`DatedValue`.
Raises:
RegistryValidationError: If the parameter declares no values, a required
date axis is absent from ``date_context``, or the match is not exactly
one.
"""
if not parameter.values:
raise RegistryValidationError(f"parameter {parameter.id!r} has no dated values")
matches: list[DatedValue] = []
for value in parameter.values:
if value.date_axis not in date_context:
raise RegistryValidationError(f"parameter {parameter.id!r} requires date axis {value.date_axis!r}")
selected = date_context[value.date_axis]
if value.valid_from <= selected and (value.valid_to is None or selected <= value.valid_to):
matches.append(value)
if len(matches) != 1:
raise RegistryValidationError(
f"parameter {parameter.id!r} expected exactly one dated value, found {len(matches)}",
)
return matches[0]
[docs]
def resolve_parameter(parameter: ParameterDefinition, date_context: Mapping[str, date]) -> Decimal:
"""Resolve one dated scalar from a :class:`ParameterDefinition`.
Returns:
The selected value's scalar.
"""
return resolve_dated_value(parameter, date_context).value
[docs]
def resolve_scalar_parameter(parameter_id: str, ctx: _EvalContext, *, op: str) -> Decimal:
"""Resolve a scalar parameter and record its runtime provenance.
Formula-operation families share the same scalar type contract and must
record the parameter id alongside the resolved Decimal in their common
evaluation context. Keeping both actions together prevents a successful
lookup from escaping the explainability surface.
"""
parameter = ctx.parameters.get(parameter_id)
if parameter is None:
raise RegistryValidationError(
f"parameter {parameter_id!r} not registered",
translated_message="errors.calc.parameter_unknown",
context={"parameter_id": parameter_id},
)
if parameter.data_type not in NUMERIC_CASILLA_DATA_TYPES:
raise RegistryValidationError(
f"parameter {parameter_id!r} must be scalar to be used by {op}",
translated_message="errors.calc.dispatch_parameter_kind",
context={"parameter_id": parameter_id, "op": op},
)
value = resolve_parameter(parameter, ctx.date_context)
ctx.operand_refs.append(parameter_id)
ctx.operand_values.append(value)
return value
[docs]
def resolve_keyed_bracket(
parameter: ParameterDefinition | None,
*,
key: str,
filing_year: int,
) -> Decimal | None:
"""Resolve one keyed-bracket row for a filing year.
A missing parameter or key is the caller's explicit "not tabled" signal;
overlapping matching windows are a contradictory registry definition and
therefore fail closed instead of selecting an arbitrary first row.
"""
if parameter is None:
return None
if parameter.data_type != "keyed_bracket_table":
raise RegistryValidationError(
f"parameter {parameter.id!r} must declare data_type='keyed_bracket_table' to resolve a keyed bracket",
translated_message="errors.calc.dispatch_parameter_kind",
context={"parameter_id": parameter.id, "op": "resolve_keyed_bracket"},
)
matches = [
entry
for entry in parameter.keyed_brackets
if entry.key == key
and entry.valid_from.year <= filing_year
and (entry.valid_to is None or filing_year <= entry.valid_to.year)
]
if not matches:
return None
if len(matches) != 1:
raise RegistryValidationError(
f"parameter {parameter.id!r} key {key!r} expected exactly one keyed bracket for filing year "
f"{filing_year}, found {len(matches)}",
context={
"parameter_id": parameter.id,
"key": key,
"filing_year": str(filing_year),
"match_count": str(len(matches)),
},
)
return matches[0].value
[docs]
def apply_rounding(value: Decimal, rounding: RegistryRoundingCode | None) -> Decimal:
"""Apply a registry rounding rule to a decimal formula result.
``money-2`` uses :func:`core.money.rounding.round_to_cents`; ``integer`` uses
half-up quantization for registry-authored integer targets;
``integer-ceiling`` quantizes with :data:`decimal.ROUND_CEILING` for
the targets whose governing provision takes the result to the next
unit up rather than to the nearest one (LIVA art. 104.Dos, "se
redondeará en la unidad superior"). ``ROUND_CEILING`` leaves an
already-integral value untouched, so a 50 % ratio stays 50.
``ROUND_CEILING`` moves toward positive infinity, which differs from
away-from-zero for a negative operand. Every ``integer-ceiling``
target today is a registry-constrained non-negative percentage
(``sign = "non_negative"``, ``min_value = "0"``), so the two readings
coincide; a future negative-capable target must state which reading
its provision means before enrolling here.
"""
if rounding is None:
return value
if rounding == RegistryRoundingCode.MONEY_2:
return _round_to_cents(value)
if rounding == RegistryRoundingCode.INTEGER:
return value.quantize(ONE, rounding=ROUND_HALF_UP)
if rounding == RegistryRoundingCode.INTEGER_CEILING:
return value.quantize(ONE, rounding=ROUND_CEILING)
raise RegistryValidationError(f"unsupported rounding rule {rounding!r}")
[docs]
def reject_non_decimal[Key](items: Mapping[Key, Decimal], label: str) -> None:
"""Reject non-decimal values before formula runtime consumption."""
del items, label
def _reject_non_string_input_keys[InputKey, InputValue](inputs: Mapping[InputKey, InputValue]) -> None:
"""Reject non-string keys before canonical casilla-id validation."""
invalid = tuple(repr(key) for key in inputs if not isinstance(key, str))
if invalid:
raise RegistryValidationError(
f"input keys must be canonical casilla.id strings: {sorted(invalid)!r}",
translated_message="errors.calc.unknown_input_casillas",
context={"casilla_ids": ",".join(sorted(invalid))},
)
def _canonicalize_input_keys[InputKey, InputValue](
inputs: Mapping[InputKey, InputValue],
) -> dict[CasillaId, InputValue]:
"""Validate input key shape while retaining the caller's value sequence."""
malformed: list[str] = []
canonical_inputs: dict[CasillaId, InputValue] = {}
for key in inputs:
try:
canonical_inputs[validated_casilla_id(key, surface="input casilla.id")] = inputs[key]
except ValueError:
malformed.append(str(key))
if malformed:
raise RegistryValidationError(
f"input keys must be canonical casilla.id strings: {sorted(malformed)!r}",
translated_message="errors.calc.unknown_input_casillas",
context={"casilla_ids": ",".join(sorted(malformed))},
)
return canonical_inputs
def _validated_decimal_inputs[InputValue](
canonical_inputs: Mapping[CasillaId, InputValue],
) -> dict[CasillaId, Decimal]:
"""Validate Decimal values after casilla key shape and membership checks."""
resolved_inputs: dict[CasillaId, Decimal] = {}
for key, value in canonical_inputs.items():
if isinstance(value, bool) or not isinstance(value, Decimal):
raise RegistryValidationError(f"input {key!r} must be a Decimal")
resolved_inputs[key] = value
return resolved_inputs
[docs]
def reject_non_string[Key](values: Mapping[Key, str], label: str) -> None:
"""Reject empty or non-string external values before registry validation."""
for key, value in values.items():
if not value:
raise RegistryValidationError(f"{label} {key!r} must be a non-empty string")
[docs]
def reject_unknown_external_values[Key: SupportsAllComparisons](
items: Mapping[Key, object],
known_ids: set[Key],
label: str,
) -> None:
"""Reject external ids not declared by the current registry snapshot.
Keyed on ids alone: the gate screens WHICH ids arrived, never what they
carry, so every external channel -- Decimal, truth value, date or enum --
is screened by this one function rather than by a per-channel copy.
"""
unknown = sorted(set(items).difference(known_ids))
if unknown:
raise RegistryValidationError(f"unknown registry {label} ids: {unknown!r}")
def _require_arg_count(op: str, args: list[Decimal], count: int) -> None:
"""Require an exact operand count for a formula operation."""
if len(args) != count:
raise RegistryValidationError(f"formula op {op!r} expects {count} args, got {len(args)}")
def _require_non_empty(op: str, args: list[Decimal]) -> None:
"""Require at least one operand for aggregate formula operations."""
if not args:
raise RegistryValidationError(f"formula op {op!r} expects at least one arg")
[docs]
def read_parameter(
modelo_id: str,
revision_id: RevisionId,
parameter_id: str,
*,
date_context: Mapping[str, date],
operation: PinnedAuthorityOperation | None = None,
) -> Decimal:
"""Read a registry parameter through :class:`ValidatedRegistryAuthority`.
The ad hoc public helper loads the same validated registry authority used by
snapshot callers, narrows to the selected
:class:`~domain.calculations.registry.schema.ModeloRevision`, and delegates the
dated value lookup to :func:`resolve_parameter`.
When no operation is supplied, the helper leases one indexed authority
generation for the point lookup; it never traverses a whole authority graph.
"""
if operation is None:
from .authority import bundled_indexed_authority
with bundled_indexed_authority().operation() as indexed_operation:
return read_parameter(
modelo_id,
revision_id,
parameter_id,
date_context=date_context,
operation=indexed_operation,
)
try:
revision = operation.revision(modelo_id, str(revision_id))
except RegistrySnapshotError as exc:
if modelo_id not in operation.modelo_ids():
raise RegistryValidationError(f"modelo {modelo_id!r} is not registered in the indexed authority") from exc
raise RegistryValidationError(
f"modelo {modelo_id!r} has no revision {str(revision_id)!r} in the indexed authority"
) from exc
parameter = next((p for p in revision.parameters if p.id == parameter_id), None)
if parameter is None:
raise RegistryValidationError(
f"parameter {parameter_id!r} not registered under modelo {modelo_id!r} revision {revision_id!r}",
)
return resolve_parameter(parameter, date_context)