"""Inventory ledgers for actividad economica stock valuation.
Defines strict pydantic v2 records for tracking opening stock,
period movements (purchases, COGS, counts), and closing stock per
activity / year, plus the FIFO and weighted-average (PMP / coste
medio) valuation engines required by LIS art. 17.1. LIFO is rejected
explicitly via :class:`LIFOForbiddenError`.
Public functions:
:func:`parse_valuation_method` — coerce user input into a
:class:`ValuationMethod`, refusing LIFO.
The valuation and Anexo D projection engines live in :mod:`.valuation`.
"""
from __future__ import annotations
from collections.abc import Callable
from datetime import date
from decimal import Decimal
from enum import StrEnum
from typing import TYPE_CHECKING, Annotated, Any, cast
from pydantic import BaseModel, Field, ValidationInfo, field_validator, model_validator
from ....core.decimal.constants import HUNDRED, MONEY_ZERO, ONE
from ....core.errors.hierarchy import CadrumoError as _CadrumoError
from ....core.errors.hierarchy import CoreValidationError as _CoreValidationError
from ....core.errors.hierarchy import pydantic_validation_boundary
from ....core.filing_year import FilingYear
from ....core.hashing import content_hash_hex as _content_hash_hex
from ....core.identity.digest import ContentDigest
from ....core.models import STRICT_FROZEN_CONFIG as _STRICT_FROZEN_CONFIG
from ....core.models import STRICT_FROZEN_HIDDEN_INPUT_CONFIG
from ....core.money.rounding import round_to_cents as _quantize
from ....core.percentage import Percentage
from ....core.unit_proportion import UnitProportion
from ...filing_evidence import FilingEvidenceReference
from ...identifiers import canonical_decimal_string as _canonical_decimal_string
if TYPE_CHECKING:
from ._anexo_d_records import InventoryAnexoDResult as _InventoryAnexoDResult
from .closing_authority_records import InventoryClosingAuthorityDecision as _InventoryClosingAuthorityDecision
from .closing_authority_records import InventoryClosingAuthorityRecord as _InventoryClosingAuthorityRecord
from .closing_authority_records import InventoryClosingConflictDiagnostic as _InventoryClosingConflictDiagnostic
from .closing_authority_records import InventoryClosingResolution as _InventoryClosingResolution
from .closing_authority_records import PriorAuthoritativeClosingLink as _PriorAuthoritativeClosingLink
type InventoryAnexoDResult = _InventoryAnexoDResult
type InventoryClosingAuthorityDecision = _InventoryClosingAuthorityDecision
type InventoryClosingAuthorityRecord = _InventoryClosingAuthorityRecord
type InventoryClosingConflictDiagnostic = _InventoryClosingConflictDiagnostic
type InventoryClosingResolution = _InventoryClosingResolution
type PriorAuthoritativeClosingLink = _PriorAuthoritativeClosingLink
resolve_inventory_authoritative_closing: Callable[..., InventoryClosingResolution]
[docs]
class AmortizacionLedgerError(_CadrumoError):
"""Raised when an amortizacion ledger operation is invalid."""
[docs]
class InventoryLedgerError(_CadrumoError):
"""Raised when an inventory ledger operation is invalid."""
[docs]
class InventoryValidationError(InventoryLedgerError, _CoreValidationError):
"""Raised when an inventory ledger fails Pydantic validation.
Inherits from CoreValidationError (which itself inherits from CoreError
and ValueError) to participate in the shared CoreValidationError catch
surface and remain compatible with pydantic validators.
"""
[docs]
class LIFOForbiddenError(InventoryLedgerError):
"""Raised when a caller attempts LIFO inventory valuation.
LIS art. 17.1 does not admit LIFO for tax-purpose stock valuation
in this regime; the message routes the operator to FIFO, PMP, or
coste medio.
"""
def __init__(self, method: str = "lifo") -> None:
"""Construct a refusal citing the LIS art. 17 valuation boundary.
Args:
method: User-supplied valuation method.
"""
super().__init__(
"LIFO valuation is not admitted for this tax ledger; use FIFO, PMP, or coste_medio per LIS art. 17.1.",
context={"method": method, "legal_basis": "LIS art. 17.1"},
)
[docs]
class BasisCapExceededError(AmortizacionLedgerError):
"""Raised when cumulative amortization would exceed cost basis."""
INVENTORY_SCHEMA_VERSION = "3"
"""Forward-compatible schema version stamped onto every record in this module."""
[docs]
class MovementKind(StrEnum):
"""Supported inventory movement kinds.
Attributes:
OPENING: Stock present at the start of the period.
PURCHASE: Inbound inventory acquisition.
COGS: Cost-of-goods-sold consumption.
COUNT: Adjustment to the physical count, modelled as a
synthetic COGS movement against the discrepancy.
"""
OPENING = "opening"
PURCHASE = "purchase"
COGS = "cogs"
COUNT = "count"
[docs]
class InventoryAcquisitionEvidenceKind(StrEnum):
"""Closed evidence authorities admitted for inventory acquisition facts."""
PURCHASE_INVOICE = "purchase_invoice"
TRANSPORT_DOCUMENT = "transport_document"
INSURANCE_DOCUMENT = "insurance_document"
CUSTOMS_DECLARATION = "customs_declaration"
CONTRACT = "contract"
OTHER_ACQUISITION_EVIDENCE = "other_acquisition_evidence"
ATTRIBUTABLE_COST_REVIEW = "attributable_cost_review"
IVA_RECOVERABILITY_REVIEW = "iva_recoverability_review"
[docs]
class InventoryAttributableCostKind(StrEnum):
"""Closed directly-attributable cost vocabulary for inventory purchases."""
FREIGHT = "freight"
INSURANCE = "insurance"
CUSTOMS_DUTY = "customs_duty"
HANDLING = "handling"
PROFESSIONAL_FEE = "professional_fee"
OTHER_DIRECTLY_ATTRIBUTABLE = "other_directly_attributable"
def _require_cents(value: Decimal, *, field_name: str) -> Decimal:
if value != _quantize(value):
raise InventoryValidationError(f"{field_name} must be quantised to cents")
return value
require_inventory_cents = _require_cents
[docs]
class InventoryAcquisitionEvidence(BaseModel):
"""Nominal evidence reference plus its immutable content digest."""
model_config = _STRICT_FROZEN_CONFIG
reference: FilingEvidenceReference
evidence_kind: InventoryAcquisitionEvidenceKind
content_digest: ContentDigest
[docs]
class InventoryAttributableCostComponent(BaseModel):
"""One evidenced cost directly attributable to an inventory purchase."""
model_config = _STRICT_FROZEN_CONFIG
component_id: str = Field(min_length=1, max_length=128)
kind: InventoryAttributableCostKind
taxable_base: Decimal = Field(gt=MONEY_ZERO)
iva_amount: Decimal = Field(ge=MONEY_ZERO)
deductible_iva_ratio: Decimal = Field(ge=MONEY_ZERO, le=ONE)
evidence_references: tuple[FilingEvidenceReference, ...] = Field(min_length=1)
@field_validator("component_id")
@classmethod
@pydantic_validation_boundary
def _trim_component_id(cls, value: str) -> str:
trimmed = value.strip()
if not trimmed:
raise InventoryValidationError("component_id must not be blank")
return trimmed
@field_validator("taxable_base", "iva_amount")
@classmethod
@pydantic_validation_boundary
def _monetary_fields_are_cents(cls, value: Decimal, info: ValidationInfo) -> Decimal:
return _require_cents(value, field_name=info.field_name or "amount")
@field_validator("evidence_references")
@classmethod
@pydantic_validation_boundary
def _evidence_references_are_distinct(
cls,
value: tuple[FilingEvidenceReference, ...],
) -> tuple[FilingEvidenceReference, ...]:
identities = tuple(item.reference for item in value)
if len(set(identities)) != len(identities):
raise InventoryValidationError("component evidence references must be unique")
return value
@property
def recoverable_iva(self) -> Decimal:
"""Return IVA excluded from inventory acquisition cost."""
return _quantize(self.iva_amount * self.deductible_iva_ratio)
[docs]
class InventoryAcquisitionCompleteness(BaseModel):
"""Evidence-backed attestations that all three cost reviews occurred."""
model_config = _STRICT_FROZEN_CONFIG
consideration_evidence: FilingEvidenceReference
attributable_cost_review_evidence: FilingEvidenceReference
iva_recoverability_review_evidence: FilingEvidenceReference
[docs]
class InventoryAcquisitionCost(BaseModel):
"""Complete, evidenced acquisition-cost decomposition for one purchase."""
model_config = _STRICT_FROZEN_CONFIG
consideration_excluding_iva: Decimal = Field(ge=MONEY_ZERO)
consideration_iva_amount: Decimal = Field(ge=MONEY_ZERO)
consideration_deductible_iva_ratio: Decimal = Field(ge=MONEY_ZERO, le=ONE)
attributable_cost_components: tuple[InventoryAttributableCostComponent, ...]
evidence: tuple[InventoryAcquisitionEvidence, ...] = Field(min_length=1)
completeness: InventoryAcquisitionCompleteness
directly_attributable_cost_total: Decimal = Field(ge=MONEY_ZERO)
nonrecoverable_iva_included: Decimal = Field(ge=MONEY_ZERO)
recoverable_iva_excluded: Decimal = Field(ge=MONEY_ZERO)
total_acquisition_cost: Decimal = Field(ge=MONEY_ZERO)
@field_validator(
"consideration_excluding_iva",
"consideration_iva_amount",
"directly_attributable_cost_total",
"nonrecoverable_iva_included",
"recoverable_iva_excluded",
"total_acquisition_cost",
)
@classmethod
@pydantic_validation_boundary
def _monetary_fields_are_cents(cls, value: Decimal, info: ValidationInfo) -> Decimal:
return _require_cents(value, field_name=info.field_name or "amount")
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_complete_decomposition(self) -> InventoryAcquisitionCost:
evidence_ids = tuple(item.reference.reference for item in self.evidence)
evidence_by_reference = _validate_acquisition_evidence_links(self, evidence_ids)
_validate_acquisition_evidence_roles(self, evidence_by_reference)
_validate_acquisition_amount_decomposition(self)
return self
def _validate_acquisition_evidence_links(
acquisition: InventoryAcquisitionCost,
evidence_ids: tuple[str, ...],
) -> dict[str, InventoryAcquisitionEvidenceKind]:
if len(set(evidence_ids)) != len(evidence_ids):
raise InventoryValidationError("inventory acquisition evidence references must be unique")
component_ids = tuple(item.component_id for item in acquisition.attributable_cost_components)
if len(set(component_ids)) != len(component_ids):
raise InventoryValidationError("inventory acquisition component identities must be unique")
declared_refs = {
acquisition.completeness.consideration_evidence.reference,
acquisition.completeness.attributable_cost_review_evidence.reference,
acquisition.completeness.iva_recoverability_review_evidence.reference,
*(
reference.reference
for component in acquisition.attributable_cost_components
for reference in component.evidence_references
),
}
missing = sorted(declared_refs - set(evidence_ids))
if missing:
raise InventoryValidationError(f"inventory acquisition evidence references are unresolved: {missing!r}")
return {item.reference.reference: item.evidence_kind for item in acquisition.evidence}
def _validate_acquisition_evidence_roles(
acquisition: InventoryAcquisitionCost,
evidence_by_reference: dict[str, InventoryAcquisitionEvidenceKind],
) -> None:
if (
evidence_by_reference[acquisition.completeness.attributable_cost_review_evidence.reference]
is not InventoryAcquisitionEvidenceKind.ATTRIBUTABLE_COST_REVIEW
):
raise InventoryValidationError("attributable-cost completeness requires attributable-cost review evidence")
if (
evidence_by_reference[acquisition.completeness.iva_recoverability_review_evidence.reference]
is not InventoryAcquisitionEvidenceKind.IVA_RECOVERABILITY_REVIEW
):
raise InventoryValidationError("IVA completeness requires IVA-recoverability review evidence")
consideration_kind = evidence_by_reference[acquisition.completeness.consideration_evidence.reference]
if consideration_kind in {
InventoryAcquisitionEvidenceKind.ATTRIBUTABLE_COST_REVIEW,
InventoryAcquisitionEvidenceKind.IVA_RECOVERABILITY_REVIEW,
}:
raise InventoryValidationError("purchase consideration requires acquisition evidence, not review evidence")
def _validate_acquisition_amount_decomposition(acquisition: InventoryAcquisitionCost) -> None:
attributable = sum((item.taxable_base for item in acquisition.attributable_cost_components), MONEY_ZERO)
recoverable = _quantize(
acquisition.consideration_iva_amount * acquisition.consideration_deductible_iva_ratio,
) + sum((item.recoverable_iva for item in acquisition.attributable_cost_components), MONEY_ZERO)
total_iva = acquisition.consideration_iva_amount + sum(
(item.iva_amount for item in acquisition.attributable_cost_components),
MONEY_ZERO,
)
nonrecoverable = total_iva - recoverable
total = acquisition.consideration_excluding_iva + attributable + nonrecoverable
expected = {
"directly_attributable_cost_total": _quantize(attributable),
"recoverable_iva_excluded": _quantize(recoverable),
"nonrecoverable_iva_included": _quantize(nonrecoverable),
"total_acquisition_cost": _quantize(total),
}
for field_name, expected_value in expected.items():
if getattr(acquisition, field_name) != expected_value:
raise InventoryValidationError(f"{field_name} does not match the acquisition decomposition")
[docs]
class ValuationMethod(StrEnum):
"""Supported inventory valuation methods."""
FIFO = "fifo"
PMP = "pmp"
COSTE_MEDIO = "coste_medio"
[docs]
class InventoryClosingValuationBasis(StrEnum):
"""Grounded acquisition-price basis used by a physical year-end count."""
FIFO_ACQUISITION_PRICE = "fifo_acquisition_price"
PMP_ACQUISITION_PRICE = "pmp_acquisition_price"
COSTE_MEDIO_ACQUISITION_PRICE = "coste_medio_acquisition_price"
[docs]
class PhysicalClosingEvidenceRole(StrEnum):
"""Closed evidence roles required to ground a physical closing."""
PHYSICAL_COUNT = "physical_count"
ACQUISITION_PRICE_VALUATION = "acquisition_price_valuation"
[docs]
class PhysicalClosingEvidence(BaseModel):
"""Digest-bound opaque evidence supporting a physical closing."""
model_config = _STRICT_FROZEN_CONFIG
reference: FilingEvidenceReference
role: PhysicalClosingEvidenceRole
content_digest: ContentDigest
[docs]
class InventoryClosingAuthority(StrEnum):
"""Closed authority choices for inventory closing valuation."""
MOVEMENT_DERIVED = "movement_derived"
PHYSICAL_OBSERVATION = "physical_observation"
[docs]
class InventoryClosingDecisionEvidenceRole(StrEnum):
"""Closed evidence role grounding an authority reconciliation decision."""
AUTHORITY_RECONCILIATION = "authority_reconciliation"
[docs]
class InventoryClosingDecisionEvidence(BaseModel):
"""Digest-bound reconciliation evidence used by one authority decision."""
model_config = _STRICT_FROZEN_CONFIG
reference: FilingEvidenceReference
role: InventoryClosingDecisionEvidenceRole
content_digest: ContentDigest
[docs]
class PriorClosingContinuityEvidence(BaseModel):
"""Digest-bound evidence of the immediately prior authoritative closing."""
model_config = _STRICT_FROZEN_CONFIG
reference: FilingEvidenceReference
content_digest: ContentDigest
[docs]
def fingerprint_prior_authoritative_closing(
*,
actividad_id: str,
filing_year: int,
authoritative_closing_value: Decimal,
authoritative_source_fingerprint: ContentDigest,
evidence: tuple[PriorClosingContinuityEvidence, ...],
) -> ContentDigest:
"""Derive immutable identity for one prior authoritative closing fact."""
return _content_hash_hex(
{
"fingerprint_schema_version": "1",
"actividad_id": actividad_id,
"filing_year": filing_year,
"authoritative_closing_value": _canonical_decimal_string(authoritative_closing_value),
"authoritative_source_fingerprint": authoritative_source_fingerprint,
"evidence": [
{
"reference": item.reference.reference,
"content_digest": item.content_digest,
}
for item in sorted(evidence, key=lambda item: item.reference.reference)
],
},
)
[docs]
class PhysicalClosingObservation(BaseModel):
"""Immutable evidenced physical closing valuation for one activity and year."""
model_config = _STRICT_FROZEN_CONFIG
observation_id: str = Field(min_length=1, max_length=128)
observed_on: date
as_of_date: date
actividad_id: str = Field(min_length=1)
filing_year: FilingYear
closing_value: Decimal = Field(ge=MONEY_ZERO)
valuation_basis: InventoryClosingValuationBasis
evidence: tuple[PhysicalClosingEvidence, ...] = Field(min_length=2)
@field_validator("closing_value")
@classmethod
@pydantic_validation_boundary
def _closing_value_is_cents(cls, value: Decimal) -> Decimal:
return _require_cents(value, field_name="physical closing_value")
@field_validator("evidence")
@classmethod
@pydantic_validation_boundary
def _evidence_is_unique(
cls,
value: tuple[PhysicalClosingEvidence, ...],
) -> tuple[PhysicalClosingEvidence, ...]:
identities = tuple(item.reference.reference for item in value)
if len(set(identities)) != len(identities):
raise InventoryValidationError("physical closing evidence references must be unique")
roles = {item.role for item in value}
required = {
PhysicalClosingEvidenceRole.PHYSICAL_COUNT,
PhysicalClosingEvidenceRole.ACQUISITION_PRICE_VALUATION,
}
if not required.issubset(roles):
raise InventoryValidationError("physical closing requires count and acquisition-price valuation evidence")
if len(roles) != len(value):
raise InventoryValidationError("physical closing evidence roles must be unique")
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _observation_dates_match_year_end(self) -> PhysicalClosingObservation:
expected_as_of = date(self.filing_year, 12, 31)
if self.as_of_date != expected_as_of:
raise InventoryValidationError("physical closing as_of_date must be filing-year end")
if self.observed_on < self.as_of_date:
raise InventoryValidationError("physical closing cannot be observed before its as-of date")
return self
@property
def fingerprint(self) -> ContentDigest:
"""Return canonical economic/evidence identity for the observation."""
return _content_hash_hex(
{
"fingerprint_schema_version": "1",
"observation_id": self.observation_id,
"observed_on": self.observed_on.isoformat(),
"as_of_date": self.as_of_date.isoformat(),
"actividad_id": self.actividad_id,
"filing_year": self.filing_year,
"closing_value": _canonical_decimal_string(self.closing_value),
"valuation_basis": self.valuation_basis.value,
"evidence": [
{
"reference": item.reference.reference,
"role": item.role.value,
"content_digest": item.content_digest,
}
for item in sorted(self.evidence, key=lambda evidence: evidence.reference.reference)
],
},
)
def _load_closing_authority_models() -> tuple[Any, ...]:
from . import closing_authority_records as _closing
models = (
_closing.InventoryClosingAuthorityDecision,
_closing.PriorAuthoritativeClosingLink,
_closing.InventoryClosingConflictDiagnostic,
_closing.InventoryClosingResolution,
_closing.InventoryClosingAuthorityRecord,
)
for model in models:
model.__module__ = __name__
globals().update(
{
"InventoryClosingAuthorityDecision": models[0],
"PriorAuthoritativeClosingLink": models[1],
"InventoryClosingConflictDiagnostic": models[2],
"InventoryClosingResolution": models[3],
"InventoryClosingAuthorityRecord": models[4],
},
)
return models
_load_closing_authority_models()
[docs]
class MovementRecord(BaseModel):
"""One inventory movement for an activity/year.
Attributes:
movement_id: Stable natural key for the movement.
movement_date: Date the movement applies on.
kind: :class:`MovementKind`.
sku: SKU / item identifier; defaults to ``"default"`` for
single-SKU ledgers.
quantity: Movement quantity (strictly positive).
unit_cost: IVA-exclusive per-unit cost.
taxable_base: Invoice taxable base (IVA-exclusive).
iva_rate: IVA rate percentage (0-100).
iva_amount: Optional explicit IVA amount; when set must equal
``taxable_base * iva_rate / 100``.
deductible_iva_ratio: Fraction of input IVA the contribuyente
may deduct (0-1).
schema_version: Forward-compatible schema version. ``"3"``.
"""
model_config = _STRICT_FROZEN_CONFIG
movement_id: str = Field(min_length=1)
movement_date: date
kind: MovementKind = MovementKind.PURCHASE
sku: str = Field(default="default", min_length=1)
quantity: Decimal = Field(gt=Decimal("0"))
unit_cost: Decimal | None = Field(default=None, ge=Decimal("0"))
taxable_base: Decimal | None = Field(default=None, ge=Decimal("0"))
iva_rate: Percentage = Field(default=None, validate_default=True)
iva_amount: Decimal | None = Field(default=None, ge=Decimal("0"))
deductible_iva_ratio: UnitProportion = Decimal("1.00")
acquisition_cost: InventoryAcquisitionCost | None = None
schema_version: str = INVENTORY_SCHEMA_VERSION
@field_validator("iva_rate", mode="before")
@classmethod
@pydantic_validation_boundary
def _default_iva_rate_from_movement_devengo(
cls,
value: object,
info: ValidationInfo,
) -> object:
"""Resolve an omitted rate through the dated Spanish IVA facade."""
if isinstance(value, str) and info.mode == "json":
# A before-validator hands JSON text through unparsed, and the strict
# Decimal schema behind it accepts only an instance.
return Decimal(value)
if value is not None:
return value
movement_date = info.data.get("movement_date")
if not isinstance(movement_date, date):
raise InventoryValidationError("inventory IVA default requires a valid movement_date")
from ...calculations.registry.authority import PinnedAuthorityOperation
from ...calculations.registry.governed_fact_scope import governed_facts_in_scope
from ...calculations.registry.iva_category_catalogue import require_iva_category
from ...calculations.registry.iva_rate_kind_catalogue import resolve_iva_rate_kind_catalogue
from ...iva.lookup import lookup_rate
from ...iva.schema import spanish_eu_member_state
operation = governed_facts_in_scope()
if not isinstance(operation, PinnedAuthorityOperation):
raise InventoryValidationError(
"inventory IVA default requires a generation-pinned authority operation",
)
rate_kind = resolve_iva_rate_kind_catalogue(
effective_date=movement_date,
authority=operation,
).for_category(
require_iva_category(
"domestic_general",
effective_date=movement_date,
authority=operation,
),
)
return lookup_rate(
spanish_eu_member_state(effective_date=movement_date, authority=operation),
rate_kind,
movement_date,
operation=operation,
).pct
[docs]
@classmethod
def from_purchase_acquisition(
cls,
*,
movement_id: str,
movement_date: date,
quantity: Decimal,
acquisition_cost: InventoryAcquisitionCost,
sku: str = "default",
) -> MovementRecord:
"""Project one complete acquisition into its canonical purchase movement."""
consideration = acquisition_cost.consideration_excluding_iva
iva_amount = acquisition_cost.consideration_iva_amount
iva_rate = MONEY_ZERO if consideration == MONEY_ZERO else iva_amount * HUNDRED / consideration
return cls(
movement_id=movement_id,
movement_date=movement_date,
kind=MovementKind.PURCHASE,
sku=sku,
quantity=quantity,
taxable_base=consideration,
iva_rate=iva_rate,
iva_amount=iva_amount,
deductible_iva_ratio=acquisition_cost.consideration_deductible_iva_ratio,
acquisition_cost=acquisition_cost,
)
@property
def value(self) -> Decimal:
"""Return the IVA-exclusive movement value."""
if self.taxable_base is not None:
return self.taxable_base
if self.unit_cost is None:
return MONEY_ZERO
return self.quantity * self.unit_cost
@property
def capitalized_value(self) -> Decimal:
"""Return the sole value capitalized by inventory valuation."""
if self.kind is MovementKind.PURCHASE:
acquisition_cost = self.acquisition_cost
if acquisition_cost is None:
raise InventoryValidationError("a purchase movement capitalizes its acquisition cost and declares none")
return acquisition_cost.total_acquisition_cost
return self.value
@property
def resolved_unit_cost(self) -> Decimal:
"""Return capitalized unit cost, falling back to consideration for openings."""
if self.kind is MovementKind.PURCHASE:
return self.capitalized_value / self.quantity
if self.unit_cost is not None:
return self.unit_cost
if self.taxable_base is None:
return MONEY_ZERO
return self.taxable_base / self.quantity
@field_validator("schema_version")
@classmethod
@pydantic_validation_boundary
def _schema_version_supported(cls, value: str) -> str:
"""Reject any schema_version other than the current :data:`INVENTORY_SCHEMA_VERSION`."""
if value != INVENTORY_SCHEMA_VERSION:
raise InventoryValidationError(f"unsupported MovementRecord schema_version {value!r}")
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _validate_movement_amounts(self) -> MovementRecord:
"""Enforce that opening / purchase movements carry a cost and IVA decomposes consistently."""
_require_movement_cost(self)
_validate_movement_taxable_amount(self)
_validate_purchase_movement_acquisition(self)
_validate_non_purchase_movement_acquisition(self)
return self
def _require_movement_cost(movement: MovementRecord) -> None:
needs_cost = movement.kind in {MovementKind.OPENING, MovementKind.PURCHASE}
if needs_cost and movement.unit_cost is None and movement.taxable_base is None:
raise InventoryValidationError("opening and purchase movements require unit_cost or taxable_base")
def _validate_movement_taxable_amount(movement: MovementRecord) -> None:
if movement.taxable_base is not None:
computed_iva = _quantize(movement.taxable_base * movement.iva_rate / HUNDRED)
if movement.iva_amount is not None and movement.iva_amount != computed_iva:
raise InventoryValidationError("iva_amount must equal taxable_base * iva_rate")
if (
movement.unit_cost is not None
and movement.taxable_base is not None
and _quantize(movement.quantity * movement.unit_cost) != movement.taxable_base
):
raise InventoryValidationError("taxable_base must equal quantity * unit_cost")
def _validate_purchase_movement_acquisition(movement: MovementRecord) -> None:
if movement.kind is not MovementKind.PURCHASE:
return
acquisition_cost = movement.acquisition_cost
if acquisition_cost is None:
raise InventoryValidationError("purchase movements require complete acquisition_cost")
if acquisition_cost.consideration_excluding_iva != _quantize(movement.value):
raise InventoryValidationError("acquisition consideration must equal the purchase consideration")
expected_iva = movement.iva_amount
if expected_iva is None:
expected_iva = _quantize(movement.value * movement.iva_rate / HUNDRED)
if acquisition_cost.consideration_iva_amount != expected_iva:
raise InventoryValidationError("acquisition consideration IVA must equal the purchase IVA")
if acquisition_cost.consideration_deductible_iva_ratio != movement.deductible_iva_ratio:
raise InventoryValidationError("acquisition IVA recoverability must equal the purchase ratio")
def _validate_non_purchase_movement_acquisition(movement: MovementRecord) -> None:
if movement.kind is not MovementKind.PURCHASE and movement.acquisition_cost is not None:
raise InventoryValidationError("acquisition_cost is permitted only for purchase movements")
[docs]
class StockLayer(BaseModel):
"""Remaining inventory quantity at one IVA-exclusive unit cost.
Attributes:
sku: SKU / item identifier.
quantity: Layer quantity (strictly positive).
unit_cost: Per-unit cost the layer was acquired at.
source_movement_id: Stable id of the originating
:class:`MovementRecord` (or a synthetic id for opening
stock and weighted-average pools).
"""
model_config = _STRICT_FROZEN_CONFIG
sku: str = Field(default="default", min_length=1)
quantity: Decimal = Field(gt=Decimal("0"))
unit_cost: Decimal = Field(ge=Decimal("0"))
source_movement_id: str = Field(min_length=1)
InventoryYear = Annotated[int, Field(ge=1900)]
"""The calendar year an inventory ledger or movement covers.
Deliberately NOT the filing-year window. An inventory record may reach back
further than the registry has authored revisions for, so it carries its own
floor and no ceiling.
"""
[docs]
class InventoryLedger(BaseModel):
"""Per-activity inventory ledger for one tax year.
Attributes:
actividad_id: Activity identifier the ledger is keyed by.
year: Calendar year the ledger covers.
valuation_method: FIFO, PMP, or coste medio; LIFO is forbidden.
opening_stock: Aggregate IVA-exclusive opening valuation.
opening_layers: Per-layer breakdown of opening stock; when
non-empty must value-balance with ``opening_stock``.
period_movements: Tuple of :class:`MovementRecord` rows
covering the period.
closing_authority_record: Required nullable persisted authority bundle;
``None`` states that no operator authority decision is recorded.
schema_version: Forward-compatible schema version. ``"3"``.
"""
model_config = STRICT_FROZEN_HIDDEN_INPUT_CONFIG
actividad_id: str = Field(min_length=1)
year: InventoryYear
valuation_method: ValuationMethod
opening_stock: Decimal = Field(ge=Decimal("0"))
opening_layers: tuple[StockLayer, ...] = ()
period_movements: tuple[MovementRecord, ...] = ()
closing_authority_record: InventoryClosingAuthorityRecord | None
schema_version: str = INVENTORY_SCHEMA_VERSION
@field_validator("actividad_id")
@classmethod
@pydantic_validation_boundary
def _actividad_id_is_canonical(cls, value: str) -> str:
if value != value.strip() or any(ord(character) < 32 for character in value):
raise InventoryValidationError("inventory actividad_id is not canonical")
return value
@field_validator("schema_version")
@classmethod
@pydantic_validation_boundary
def _schema_version_supported(cls, value: str) -> str:
"""Reject any schema_version other than the current :data:`INVENTORY_SCHEMA_VERSION`."""
if value != INVENTORY_SCHEMA_VERSION:
raise InventoryValidationError(f"unsupported InventoryLedger schema_version {value!r}")
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _opening_stock_matches_layers(self) -> InventoryLedger:
"""Enforce that ``opening_layers`` value-balances with ``opening_stock``."""
from .valuation import layers_value
movement_ids = tuple(movement.movement_id for movement in self.period_movements)
if len(set(movement_ids)) != len(movement_ids):
raise InventoryValidationError("inventory ledger movement_id values must be unique")
if self.opening_layers and _quantize(layers_value(self.opening_layers)) != _quantize(self.opening_stock):
raise InventoryValidationError("opening_stock must equal the value of opening_layers")
if self.closing_authority_record is not None:
record = self.closing_authority_record
if (record.decision.actividad_id, record.decision.filing_year) != (self.actividad_id, self.year):
raise InventoryValidationError("closing authority record must match the inventory ledger coordinate")
cast(Callable[..., InventoryClosingResolution], globals()["resolve_inventory_authoritative_closing"])(
self,
decision=record.decision,
physical_observation=record.physical_observation,
prior_closing_link=record.prior_closing_link,
)
return self
[docs]
class InventoryLedgerDocument(BaseModel):
"""JSON document containing inventory ledgers.
``(actividad_id, year)`` is the natural key: creation, movement recording,
and lookup all assume one canonical ledger per activity per year. The
uniqueness invariant lives here, on the document, because the repository
had two competing versions of it -- ``create`` and the application service
refused a second ledger for a pair, while the public
``InventoryLedgerRepository.save`` path accepted any document and wrote it without
validating, so a replacement could persist two ledgers for one pair while
every reader still assumed one.
Attributes:
schema_version: Forward-compatible schema version. ``"3"``.
ledgers: Tuple of :class:`InventoryLedger` rows, each with a distinct
``(actividad_id, year)`` pair.
"""
model_config = _STRICT_FROZEN_CONFIG
schema_version: str = INVENTORY_SCHEMA_VERSION
ledgers: tuple[InventoryLedger, ...] = ()
@field_validator("schema_version")
@classmethod
@pydantic_validation_boundary
def _schema_version_supported(cls, value: str) -> str:
"""Reject any schema_version other than the current :data:`INVENTORY_SCHEMA_VERSION`."""
if value != INVENTORY_SCHEMA_VERSION:
raise InventoryValidationError(f"unsupported InventoryLedgerDocument schema_version {value!r}")
return value
@model_validator(mode="after")
@pydantic_validation_boundary
def _reject_duplicate_actividad_year(self) -> InventoryLedgerDocument:
"""Refuse a document carrying two ledgers for one activity and year.
Enforced on the document rather than at a write method so creation,
bulk replacement, and the read boundary answer the same way. Applying
at the read boundary is deliberate: a stored document that already
holds a duplicate pair has no canonical ledger for it, and surfacing it
silently is what the repository used to do.
"""
keys = [(ledger.actividad_id, ledger.year) for ledger in self.ledgers]
duplicates = sorted({key for key in keys if keys.count(key) > 1})
if duplicates:
rendered = ", ".join(f"{actividad_id}/{year}" for actividad_id, year in duplicates)
raise InventoryValidationError(
f"InventoryLedgerDocument carries duplicate actividad/year ledgers: {rendered}",
)
return self
[docs]
def parse_valuation_method(raw: str) -> ValuationMethod:
"""Parse a user-supplied valuation method and refuse LIFO explicitly.
Args:
raw: User input (case- and separator-insensitive).
Returns:
The matching :class:`ValuationMethod` member.
Raises:
LIFOForbiddenError: When the input normalises to ``"lifo"``.
InventoryLedgerError: When the input matches no known method.
"""
normalized = raw.strip().lower().replace("-", "_")
if normalized == "lifo":
raise LIFOForbiddenError(raw)
try:
return ValuationMethod(normalized)
except ValueError as exc:
raise InventoryLedgerError(
f"unknown valuation method {raw!r}; use fifo, pmp, or coste_medio",
context={"method": raw},
) from exc
[docs]
class InventoryValuationResult(BaseModel):
"""Computed valuation outcome for an inventory ledger.
Attributes:
closing_layers: Tuple of :class:`StockLayer` rows surviving
after period movements were applied.
closing_value: Aggregate closing valuation.
cogs_value: Cost-of-goods-sold for the period.
purchase_value: Aggregate value of PURCHASE movements during
the period.
"""
model_config = _STRICT_FROZEN_CONFIG
closing_layers: tuple[StockLayer, ...]
closing_value: Decimal
cogs_value: Decimal
purchase_value: Decimal
def _load_anexo_d_models() -> tuple[Any, ...]:
from . import _anexo_d_records as _anexo
result = _anexo.InventoryAnexoDResult
resolve = _anexo.resolve_inventory_authoritative_closing
result.__module__ = __name__
resolve.__module__ = __name__
globals().update(
{
"InventoryAnexoDResult": result,
"resolve_inventory_authoritative_closing": resolve,
},
)
return result, resolve
_load_anexo_d_models()