cadrumo.domain.contribuyente.inventory.records module

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 LIFOForbiddenError.

Public functions:

parse_valuation_method() — coerce user input into a ValuationMethod, refusing LIFO. The valuation and Anexo D projection engines live in valuation.

resolve_inventory_authoritative_closing(ledger, *, decision, physical_observation, prior_closing_link, authority=None)

Resolve closing authority while retaining any physical/movement conflict.

Return type:

InventoryClosingResolution

Parameters:
exception AmortizacionLedgerError(message=None, *, context=None, translated_message=None)[source]

Bases: CadrumoError

Raised when an amortizacion ledger operation is invalid.

Parameters:

  • message (str | None)

  • context (Mapping[str, object] | None)

  • translated_message (str | None)

Return type:

None

code
exception InventoryLedgerError(message=None, *, context=None, translated_message=None)[source]

Bases: CadrumoError

Raised when an inventory ledger operation is invalid.

Parameters:

  • message (str | None)

  • context (Mapping[str, object] | None)

  • translated_message (str | None)

Return type:

None

code
exception InventoryValidationError(message=None, *, context=None, translated_message=None)[source]

Bases: 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.

Parameters:

  • message (str | None)

  • context (Mapping[str, object] | None)

  • translated_message (str | None)

Return type:

None

code
exception LIFOForbiddenError(method='lifo')[source]

Bases: 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.

Parameters:

method (str)

Return type:

None

code
exception BasisCapExceededError(message=None, *, context=None, translated_message=None)[source]

Bases: AmortizacionLedgerError

Raised when cumulative amortization would exceed cost basis.

Parameters:

  • message (str | None)

  • context (Mapping[str, object] | None)

  • translated_message (str | None)

Return type:

None

code
INVENTORY_SCHEMA_VERSION

Forward-compatible schema version stamped onto every record in this module.

class MovementKind(*values)[source]

Bases: StrEnum

Supported inventory movement kinds.

Variables:

  • 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
PURCHASE
COGS
COUNT
class InventoryAcquisitionEvidenceKind(*values)[source]

Bases: StrEnum

Closed evidence authorities admitted for inventory acquisition facts.

PURCHASE_INVOICE
TRANSPORT_DOCUMENT
INSURANCE_DOCUMENT
CUSTOMS_DECLARATION
CONTRACT
OTHER_ACQUISITION_EVIDENCE
ATTRIBUTABLE_COST_REVIEW
IVA_RECOVERABILITY_REVIEW
class InventoryAttributableCostKind(*values)[source]

Bases: StrEnum

Closed directly-attributable cost vocabulary for inventory purchases.

FREIGHT
INSURANCE
CUSTOMS_DUTY
HANDLING
PROFESSIONAL_FEE
OTHER_DIRECTLY_ATTRIBUTABLE
require_inventory_cents(value, *, field_name)
Return type:

Decimal

Parameters:
class InventoryAcquisitionEvidence(**data)[source]

Bases: BaseModel

Nominal evidence reference plus its immutable content digest.

Parameters:

data (Any)

reference: FilingEvidenceReference
evidence_kind: InventoryAcquisitionEvidenceKind
content_digest: ContentDigest
class InventoryAttributableCostComponent(**data)[source]

Bases: BaseModel

One evidenced cost directly attributable to an inventory purchase.

Parameters:

data (Any)

component_id: str
kind: InventoryAttributableCostKind
taxable_base: Decimal
iva_amount: Decimal
deductible_iva_ratio: Decimal
evidence_references: tuple[FilingEvidenceReference, ...]
property recoverable_iva: Decimal

Return IVA excluded from inventory acquisition cost.

class InventoryAcquisitionCompleteness(**data)[source]

Bases: BaseModel

Evidence-backed attestations that all three cost reviews occurred.

Parameters:

data (Any)

consideration_evidence: FilingEvidenceReference
attributable_cost_review_evidence: FilingEvidenceReference
iva_recoverability_review_evidence: FilingEvidenceReference
class InventoryAcquisitionCost(**data)[source]

Bases: BaseModel

Complete, evidenced acquisition-cost decomposition for one purchase.

Parameters:

data (Any)

consideration_excluding_iva: Decimal
consideration_iva_amount: Decimal
consideration_deductible_iva_ratio: Decimal
attributable_cost_components: tuple[InventoryAttributableCostComponent, ...]
evidence: tuple[InventoryAcquisitionEvidence, ...]
completeness: InventoryAcquisitionCompleteness
directly_attributable_cost_total: Decimal
nonrecoverable_iva_included: Decimal
recoverable_iva_excluded: Decimal
total_acquisition_cost: Decimal
class ValuationMethod(*values)[source]

Bases: StrEnum

Supported inventory valuation methods.

FIFO
PMP
COSTE_MEDIO
class InventoryClosingValuationBasis(*values)[source]

Bases: StrEnum

Grounded acquisition-price basis used by a physical year-end count.

FIFO_ACQUISITION_PRICE
PMP_ACQUISITION_PRICE
COSTE_MEDIO_ACQUISITION_PRICE
class PhysicalClosingEvidenceRole(*values)[source]

Bases: StrEnum

Closed evidence roles required to ground a physical closing.

PHYSICAL_COUNT
ACQUISITION_PRICE_VALUATION
class PhysicalClosingEvidence(**data)[source]

Bases: BaseModel

Digest-bound opaque evidence supporting a physical closing.

Parameters:

data (Any)

reference: FilingEvidenceReference
role: PhysicalClosingEvidenceRole
content_digest: ContentDigest
class InventoryClosingAuthority(*values)[source]

Bases: StrEnum

Closed authority choices for inventory closing valuation.

MOVEMENT_DERIVED
PHYSICAL_OBSERVATION
class InventoryClosingDecisionEvidenceRole(*values)[source]

Bases: StrEnum

Closed evidence role grounding an authority reconciliation decision.

AUTHORITY_RECONCILIATION
class InventoryClosingDecisionEvidence(**data)[source]

Bases: BaseModel

Digest-bound reconciliation evidence used by one authority decision.

Parameters:

data (Any)

reference: FilingEvidenceReference
role: InventoryClosingDecisionEvidenceRole
content_digest: ContentDigest
class PriorClosingContinuityEvidence(**data)[source]

Bases: BaseModel

Digest-bound evidence of the immediately prior authoritative closing.

Parameters:

data (Any)

reference: FilingEvidenceReference
content_digest: ContentDigest
fingerprint_prior_authoritative_closing(*, actividad_id, filing_year, authoritative_closing_value, authoritative_source_fingerprint, evidence)[source]

Derive immutable identity for one prior authoritative closing fact.

Return type:

TypeAliasType

Parameters:
class PhysicalClosingObservation(**data)[source]

Bases: BaseModel

Immutable evidenced physical closing valuation for one activity and year.

Parameters:

data (Any)

observation_id: str
observed_on: date
as_of_date: date
actividad_id: str
filing_year: FilingYear
closing_value: Decimal
valuation_basis: InventoryClosingValuationBasis
evidence: tuple[PhysicalClosingEvidence, ...]
property fingerprint: ContentDigest

Return canonical economic/evidence identity for the observation.

class MovementRecord(**data)[source]

Bases: BaseModel

One inventory movement for an activity/year.

Variables:

  • movement_id – Stable natural key for the movement.

  • movement_date – Date the movement applies on.

  • kind – 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".

Parameters:

data (Any)

movement_id: str
movement_date: date
kind: MovementKind
sku: str
quantity: Decimal
unit_cost: Decimal | None
taxable_base: Decimal | None
iva_rate: Percentage
iva_amount: Decimal | None
deductible_iva_ratio: UnitProportion
acquisition_cost: InventoryAcquisitionCost | None
schema_version: str
classmethod from_purchase_acquisition(*, movement_id, movement_date, quantity, acquisition_cost, sku='default')[source]

Project one complete acquisition into its canonical purchase movement.

Return type:

MovementRecord

Parameters:
property value: Decimal

Return the IVA-exclusive movement value.

property capitalized_value: Decimal

Return the sole value capitalized by inventory valuation.

property resolved_unit_cost: Decimal

Return capitalized unit cost, falling back to consideration for openings.

class StockLayer(**data)[source]

Bases: BaseModel

Remaining inventory quantity at one IVA-exclusive unit cost.

Variables:

  • 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 MovementRecord (or a synthetic id for opening stock and weighted-average pools).

Parameters:

data (Any)

sku: str
quantity: Decimal
unit_cost: Decimal
source_movement_id: str
class InventoryLedger(**data)[source]

Bases: BaseModel

Per-activity inventory ledger for one tax year.

Variables:

  • 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 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".

Parameters:
actividad_id: str
year: InventoryYear
valuation_method: ValuationMethod
opening_stock: Decimal
opening_layers: tuple[StockLayer, ...]
period_movements: tuple[MovementRecord, ...]
closing_authority_record: InventoryClosingAuthorityRecord | None
schema_version: str
class InventoryLedgerDocument(**data)[source]

Bases: 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.

Variables:

  • schema_version – Forward-compatible schema version. "3".

  • ledgers – Tuple of InventoryLedger rows, each with a distinct (actividad_id, year) pair.

Parameters:

data (Any)

schema_version: str
ledgers: tuple[InventoryLedger, ...]
parse_valuation_method(raw)[source]

Parse a user-supplied valuation method and refuse LIFO explicitly.

Parameters:

raw (str) – User input (case- and separator-insensitive).

Return type:

ValuationMethod

Returns:

The matching ValuationMethod member.

Raises:
class InventoryValuationResult(**data)[source]

Bases: BaseModel

Computed valuation outcome for an inventory ledger.

Variables:

  • closing_layers – Tuple of 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.

Parameters:

data (Any)

closing_layers: tuple[StockLayer, ...]
closing_value: Decimal
cogs_value: Decimal
purchase_value: Decimal
class InventoryAnexoDResult(**data)

Bases: BaseModel

Complete source-owned inventory projection for one activity.

Parameters:

data (Any)

property expected_projection_fingerprint: ContentDigest

Derive the versioned identity of the complete projection envelope.

source_ledger: InventoryLedger
source_ledger_fingerprint: ContentDigest
actividad_id: str
filing_year: FilingYear
opening_value: Decimal
movement_derived_closing_value: Decimal
authoritative_closing_value: Decimal
selected_authority: InventoryClosingAuthority
authority_record_fingerprint: ContentDigest
decision_id: str
decision_fingerprint: ContentDigest
physical_observation_id: str | None
physical_observation_fingerprint: ContentDigest | None
physical_observed_closing_value: Decimal | None
complete_acquisition_total: Decimal
acquisition_fingerprints: tuple[ContentDigest, ...]
variation_increase_value: Decimal
variation_decrease_value: Decimal
closing_conflict: InventoryClosingConflictDiagnostic | None
issues: tuple[Literal['physical_closing_conflict'], ...]
projection_fingerprint: ContentDigest
class InventoryClosingAuthorityDecision(**data)

Bases: BaseModel

Explicit evidenced choice between movement and physical closing authority.

Parameters:

data (Any)

property fingerprint: ContentDigest

Return canonical tamper-sensitive decision identity.

decision_id: str
actividad_id: str
filing_year: FilingYear
authority: InventoryClosingAuthority
physical_observation_id: str | None
physical_observation_fingerprint: ContentDigest | None
reason: str
actor: str
source_command: str
decided_at: UtcInstant
evidence: tuple[InventoryClosingDecisionEvidence, ...]
class InventoryClosingAuthorityRecord(**data)

Bases: BaseModel

Ledger-owned immutable inputs for one closing-authority resolution.

Parameters:

data (Any)

property fingerprint: ContentDigest

Return canonical identity for the complete persisted authority input set.

decision: InventoryClosingAuthorityDecision
physical_observation: PhysicalClosingObservation | None
class InventoryClosingConflictDiagnostic(**data)

Bases: BaseModel

Retained conflict between movement-derived and physical closing values.

Parameters:

data (Any)

actividad_id: str
filing_year: FilingYear
movement_derived_value: Decimal
physical_observed_value: Decimal
physical_observation_fingerprint: ContentDigest
class InventoryClosingResolution(**data)

Bases: BaseModel

Auditable authoritative closing resolution with retained conflict.

Parameters:

data (Any)

actividad_id: str
filing_year: FilingYear
authority: InventoryClosingAuthority
authoritative_value: Decimal
movement_derived_value: Decimal
physical_observed_value: Decimal | None
physical_observation_fingerprint: ContentDigest | None
decision_id: str
decision_fingerprint: ContentDigest
physical_observation_id: str | None
conflict: InventoryClosingConflictDiagnostic | None

Bases: BaseModel

Continuity link from the immediately prior authoritative closing.

Parameters:

data (Any)

property expected_prior_closing_fingerprint: ContentDigest

Derive the fingerprint binding the claimed prior authoritative closing.

property fingerprint: ContentDigest

Return current-link identity including the opening-side coordinate.

actividad_id: str
current_filing_year: int
prior_filing_year: FilingYear
prior_authoritative_closing_value: Decimal
current_opening_value: Decimal
prior_authoritative_source_fingerprint: ContentDigest
prior_authoritative_closing_fingerprint: ContentDigest
evidence: tuple[PriorClosingContinuityEvidence, ...]