238 lines
8.1 KiB
Python
238 lines
8.1 KiB
Python
|
"""Versioned fin-hub/resource-control exchange records."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import date, datetime
|
||
|
|
from decimal import Decimal
|
||
|
|
from typing import Annotated, Literal
|
||
|
|
|
||
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
|
||
|
|
|
||
|
|
from fin_hub.money import currency_code, money, reporting_month
|
||
|
|
|
||
|
|
|
||
|
|
class ExchangeRecord(BaseModel):
|
||
|
|
model_config = ConfigDict(extra="forbid")
|
||
|
|
|
||
|
|
schema_version: Literal["0.1"] = "0.1"
|
||
|
|
record_id: str = Field(min_length=1, max_length=256)
|
||
|
|
revision_of: str | None = None
|
||
|
|
resource_id: str | None = None
|
||
|
|
service_id: str | None = None
|
||
|
|
workload_id: str | None = None
|
||
|
|
tenant_id: str | None = None
|
||
|
|
environment: str | None = None
|
||
|
|
cost_attribution_key: str | None = None
|
||
|
|
period_start: date
|
||
|
|
period_end: date
|
||
|
|
source_evidence: list[str]
|
||
|
|
created_at: datetime
|
||
|
|
|
||
|
|
@model_validator(mode="after")
|
||
|
|
def validate_period(self):
|
||
|
|
if self.period_end < self.period_start:
|
||
|
|
raise ValueError("period_end cannot precede period_start")
|
||
|
|
return self
|
||
|
|
|
||
|
|
|
||
|
|
class MonetaryRecord(ExchangeRecord):
|
||
|
|
currency: str
|
||
|
|
|
||
|
|
@field_validator("currency")
|
||
|
|
@classmethod
|
||
|
|
def validate_currency(cls, value: str) -> str:
|
||
|
|
return currency_code(value)
|
||
|
|
|
||
|
|
class BookedCostEvidence(BaseModel):
|
||
|
|
"""Authoritative fin-hub fact; amounts serialize as decimal strings."""
|
||
|
|
|
||
|
|
model_config = ConfigDict(extra="forbid")
|
||
|
|
|
||
|
|
schema_version: Literal["0.1"] = "0.1"
|
||
|
|
record_type: Literal["booked_cost"] = "booked_cost"
|
||
|
|
financial_fact_id: str = Field(min_length=1, max_length=256)
|
||
|
|
correction_of: str | None = None
|
||
|
|
adjustment_kind: Literal["charge", "credit", "refund", "reversal", "correction"]
|
||
|
|
source_type: str = Field(min_length=1)
|
||
|
|
source_document_id: str = Field(min_length=1)
|
||
|
|
source_line_id: str = Field(min_length=1)
|
||
|
|
content_fingerprint: str = Field(min_length=1)
|
||
|
|
provider: str = Field(min_length=1)
|
||
|
|
provider_account_ref: str | None = None
|
||
|
|
accounting_period: str
|
||
|
|
service_period_start: date | None = None
|
||
|
|
service_period_end: date | None = None
|
||
|
|
currency: str
|
||
|
|
net_amount: Decimal = Decimal("0.00")
|
||
|
|
discount_amount: Decimal = Decimal("0.00")
|
||
|
|
tax_status: Literal["known", "unknown", "not_applicable"]
|
||
|
|
tax_amount: Decimal | None = None
|
||
|
|
gross_amount: Decimal
|
||
|
|
adjustment_amount: Decimal = Decimal("0.00")
|
||
|
|
effective_amount: Decimal
|
||
|
|
resource_id: str | None = None
|
||
|
|
service_id: str | None = None
|
||
|
|
workload_id: str | None = None
|
||
|
|
tenant_id: str | None = None
|
||
|
|
environment: str | None = None
|
||
|
|
cost_attribution_key: str | None = None
|
||
|
|
source_evidence_ref: str = Field(min_length=1)
|
||
|
|
recorded_at: datetime
|
||
|
|
|
||
|
|
@field_validator(
|
||
|
|
"net_amount", "discount_amount", "gross_amount", mode="before"
|
||
|
|
)
|
||
|
|
@classmethod
|
||
|
|
def validate_nonnegative_money(cls, value):
|
||
|
|
return money(value, non_negative=True)
|
||
|
|
|
||
|
|
@field_validator("tax_amount", "adjustment_amount", "effective_amount", mode="before")
|
||
|
|
@classmethod
|
||
|
|
def validate_signed_money(cls, value):
|
||
|
|
return None if value is None else money(value)
|
||
|
|
|
||
|
|
@field_validator("currency")
|
||
|
|
@classmethod
|
||
|
|
def validate_currency(cls, value: str) -> str:
|
||
|
|
return currency_code(value)
|
||
|
|
|
||
|
|
@field_validator("accounting_period")
|
||
|
|
@classmethod
|
||
|
|
def validate_accounting_period(cls, value: str) -> str:
|
||
|
|
return reporting_month(value)
|
||
|
|
|
||
|
|
@model_validator(mode="after")
|
||
|
|
def validate_accounting_relationships(self):
|
||
|
|
if self.tax_status == "unknown" and self.tax_amount is not None:
|
||
|
|
raise ValueError("tax_amount must be null when tax_status is unknown")
|
||
|
|
if self.tax_status != "unknown" and self.tax_amount is None:
|
||
|
|
raise ValueError("tax_amount is required when tax status is known")
|
||
|
|
tax = self.tax_amount or Decimal("0.00")
|
||
|
|
if self.tax_status != "unknown":
|
||
|
|
expected_gross = money(self.net_amount - self.discount_amount + tax)
|
||
|
|
if self.gross_amount != expected_gross:
|
||
|
|
raise ValueError("gross_amount must equal net - discount + tax")
|
||
|
|
if self.effective_amount != money(self.gross_amount + self.adjustment_amount):
|
||
|
|
raise ValueError("effective_amount must equal gross + adjustment")
|
||
|
|
if self.adjustment_kind != "charge" and self.correction_of is None:
|
||
|
|
raise ValueError("credit/refund/reversal/correction requires correction_of")
|
||
|
|
if self.adjustment_kind in {"credit", "refund", "reversal"} and self.adjustment_amount >= 0:
|
||
|
|
raise ValueError("credit/refund/reversal adjustment_amount must be negative")
|
||
|
|
if (
|
||
|
|
self.service_period_start is not None
|
||
|
|
and self.service_period_end is not None
|
||
|
|
and self.service_period_end < self.service_period_start
|
||
|
|
):
|
||
|
|
raise ValueError("service period end cannot precede start")
|
||
|
|
return self
|
||
|
|
|
||
|
|
|
||
|
|
class CostBreakdown(BaseModel):
|
||
|
|
model_config = ConfigDict(extra="forbid")
|
||
|
|
infrastructure: Decimal = Decimal("0.00")
|
||
|
|
internal_labor: Decimal = Decimal("0.00")
|
||
|
|
external_services: Decimal = Decimal("0.00")
|
||
|
|
setup: Decimal = Decimal("0.00")
|
||
|
|
other: Decimal = Decimal("0.00")
|
||
|
|
|
||
|
|
@field_validator("*", mode="before")
|
||
|
|
@classmethod
|
||
|
|
def validate_money(cls, value):
|
||
|
|
return money(value, non_negative=True)
|
||
|
|
|
||
|
|
@property
|
||
|
|
def total(self) -> Decimal:
|
||
|
|
return money(
|
||
|
|
sum(
|
||
|
|
(getattr(self, name) for name in self.__class__.model_fields),
|
||
|
|
Decimal("0"),
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class ForecastEvidence(MonetaryRecord):
|
||
|
|
record_type: Literal["forecast"] = "forecast"
|
||
|
|
scenario: Literal["low", "base", "high"]
|
||
|
|
forecast_version: str = Field(min_length=1)
|
||
|
|
costs: CostBreakdown
|
||
|
|
uncertainty: str | None = None
|
||
|
|
assumptions: list[str]
|
||
|
|
|
||
|
|
|
||
|
|
class UsageMeasure(BaseModel):
|
||
|
|
model_config = ConfigDict(extra="forbid")
|
||
|
|
name: str = Field(min_length=1)
|
||
|
|
value: Decimal
|
||
|
|
unit: str = Field(min_length=1)
|
||
|
|
|
||
|
|
|
||
|
|
class UsageObservation(ExchangeRecord):
|
||
|
|
record_type: Literal["usage_observation"] = "usage_observation"
|
||
|
|
measures: list[UsageMeasure]
|
||
|
|
|
||
|
|
|
||
|
|
class AllocationShare(BaseModel):
|
||
|
|
model_config = ConfigDict(extra="forbid")
|
||
|
|
target_key: str = Field(min_length=1)
|
||
|
|
share: Decimal = Field(ge=0, le=1)
|
||
|
|
|
||
|
|
|
||
|
|
class AllocationEvidence(MonetaryRecord):
|
||
|
|
record_type: Literal["allocation"] = "allocation"
|
||
|
|
financial_fact_ids: list[str] = Field(min_length=1)
|
||
|
|
method: str = Field(min_length=1)
|
||
|
|
allocated_amount: Decimal
|
||
|
|
shares: list[AllocationShare]
|
||
|
|
residual_share: Decimal = Field(ge=0, le=1)
|
||
|
|
|
||
|
|
@field_validator("allocated_amount", mode="before")
|
||
|
|
@classmethod
|
||
|
|
def validate_amount(cls, value):
|
||
|
|
return money(value, non_negative=True)
|
||
|
|
|
||
|
|
@model_validator(mode="after")
|
||
|
|
def validate_shares(self):
|
||
|
|
total = sum((share.share for share in self.shares), Decimal("0"))
|
||
|
|
if total + self.residual_share != Decimal("1"):
|
||
|
|
raise ValueError("allocation shares plus residual_share must equal 1")
|
||
|
|
return self
|
||
|
|
|
||
|
|
|
||
|
|
class OptimizationEvidence(MonetaryRecord):
|
||
|
|
record_type: Literal["optimization"] = "optimization"
|
||
|
|
baseline: CostBreakdown
|
||
|
|
alternative: CostBreakdown
|
||
|
|
one_time_cost: Decimal
|
||
|
|
expected_period_savings: Decimal
|
||
|
|
assumptions: list[str]
|
||
|
|
|
||
|
|
@field_validator("one_time_cost", "expected_period_savings", mode="before")
|
||
|
|
@classmethod
|
||
|
|
def validate_amount(cls, value):
|
||
|
|
return money(value, non_negative=True)
|
||
|
|
|
||
|
|
|
||
|
|
class CommitmentCandidate(MonetaryRecord):
|
||
|
|
record_type: Literal["commitment_candidate"] = "commitment_candidate"
|
||
|
|
setup_cost: Decimal
|
||
|
|
recurring_cost: Decimal
|
||
|
|
cadence: Literal["monthly", "quarterly", "annual", "one_time"]
|
||
|
|
term_start: date
|
||
|
|
term_end: date | None = None
|
||
|
|
approval_status: Literal["candidate"] = "candidate"
|
||
|
|
|
||
|
|
@field_validator("setup_cost", "recurring_cost", mode="before")
|
||
|
|
@classmethod
|
||
|
|
def validate_amount(cls, value):
|
||
|
|
return money(value, non_negative=True)
|
||
|
|
|
||
|
|
|
||
|
|
PlanningEvidence = Annotated[
|
||
|
|
ForecastEvidence
|
||
|
|
| UsageObservation
|
||
|
|
| AllocationEvidence
|
||
|
|
| OptimizationEvidence
|
||
|
|
| CommitmentCandidate,
|
||
|
|
Field(discriminator="record_type"),
|
||
|
|
]
|