coulomb/fin-hub
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Advance resource cost evidence contract

Browse source
This commit is contained in:
tegwick 2026-08-11 15:49:16 +02:00
parent d1528944c3
commit 49519a715a
7 changed files with 631 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

67
src/fin_hub/schemas/exchange.py
View file

@ -227,6 +227,73 @@ class CommitmentCandidate(MonetaryRecord):
return money(value, non_negative=True)
class FinancialConstraintSignal(BaseModel):
"""Bounded fin-hub authority signal for procurement ranking."""
model_config = ConfigDict(extra="forbid")
schema_version: Literal["0.1"] = "0.1"
record_type: Literal["financial_constraint"] = "financial_constraint"
signal_id: str = Field(min_length=1, max_length=256)
signal_kind: Literal[
"budget_ceiling", "active_commitment", "burn_pressure", "runway_pressure"
]
classification: Literal[
"policy_constraint", "informational_warning", "informational"
]
domain_slug: str = Field(min_length=1, max_length=64)
resource_id: str | None = None
service_id: str | None = None
environment: str | None = None
currency: str
period_start: date
period_end: date
amount: Decimal | None = None
metric_value: Decimal | None = None
metric_unit: Literal["currency_per_month", "ratio", "months"] | None = None
commitment_state: Literal["active"] | None = None
source_evidence: list[str] = Field(min_length=1)
generated_at: datetime
summary: str = Field(min_length=1, max_length=512)
@field_validator("currency")
@classmethod
def validate_currency(cls, value: str) -> str:
return currency_code(value)
@field_validator("amount", mode="before")
@classmethod
def validate_amount(cls, value):
return None if value is None else money(value, non_negative=True)
@field_validator("metric_value", mode="before")
@classmethod
def validate_metric(cls, value):
if value is None:
return None
normalized = Decimal(str(value))
if not normalized.is_finite() or normalized < 0:
raise ValueError("metric_value must be a finite non-negative decimal")
return normalized
@model_validator(mode="after")
def validate_signal(self):
if self.period_end < self.period_start:
raise ValueError("period_end cannot precede period_start")
if self.signal_kind in {"budget_ceiling", "active_commitment"}:
if self.amount is None:
raise ValueError(f"{self.signal_kind} requires amount")
if self.classification != "policy_constraint":
raise ValueError(f"{self.signal_kind} must be a policy_constraint")
if self.signal_kind == "active_commitment" and self.commitment_state != "active":
raise ValueError("active_commitment requires commitment_state=active")
if self.signal_kind == "burn_pressure" and self.metric_unit != "ratio":
raise ValueError("burn_pressure requires a ratio metric")
if self.signal_kind == "runway_pressure" and self.metric_unit != "months":
raise ValueError("runway_pressure requires a months metric")
return self
PlanningEvidence = Annotated[
ForecastEvidence
| UsageObservation

325
src/fin_hub/services/exchange.py
View file

@ -3,20 +3,196 @@
from __future__ import annotations
import json
import hashlib
import sqlite3
from calendar import monthrange
from datetime import date
from dataclasses import asdict, dataclass
from datetime import date, datetime, timezone
from decimal import Decimal
from pathlib import Path
from typing import Mapping
from pydantic import TypeAdapter
from pydantic import TypeAdapter, ValidationError
from fin_hub.money import minor_money, money
from fin_hub.schemas.exchange import BookedCostEvidence, PlanningEvidence
from fin_hub.money import currency_code, minor_money, money
from fin_hub.schemas.exchange import (
BookedCostEvidence,
FinancialConstraintSignal,
PlanningEvidence,
)
from fin_hub.services.ledger import _connect, default_ledger_path
_PLANNING_ADAPTER = TypeAdapter(PlanningEvidence)
@dataclass(frozen=True)
class FinancialConstraintExport:
schema_version: str
artifact_type: str
signals: tuple[FinancialConstraintSignal, ...]
disclaimer: str
def as_dict(self) -> dict:
return asdict(self)
@dataclass(frozen=True)
class ExchangeQualityIssue:
code: str
severity: str
record_id: str | None
detail: str
@dataclass(frozen=True)
class ExchangeHealthReport:
schema_version: str
artifact_type: str
booked_fact_count: int
current_planning_record_count: int
rejected_delivery_count: int
issues: tuple[ExchangeQualityIssue, ...]
def as_dict(self) -> dict:
return asdict(self)
def _signal_id(*parts: object) -> str:
canonical = "\x1f".join(str(part) for part in parts)
return "financial-constraint:" + hashlib.sha256(canonical.encode()).hexdigest()
def financial_constraint_projection(
*,
domain_slug: str,
period_start: date,
period_end: date,
currency: str,
source_evidence: tuple[str, ...],
budget_ceiling: Decimal | str | int | float | None = None,
active_commitment: Decimal | str | int | float | None = None,
spent: Decimal | str | int | float | None = None,
monthly_burn: Decimal | str | int | float | None = None,
runway_months: Decimal | str | int | float | None = None,
runway_threshold_months: Decimal | str | int | float | None = None,
resource_id: str | None = None,
service_id: str | None = None,
environment: str | None = None,
generated_at: datetime | None = None,
) -> FinancialConstraintExport:
"""Create the minimal affordability view without exposing ledger details."""
if not domain_slug.strip():
raise ValueError("domain_slug is required")
if period_end < period_start:
raise ValueError("period_end cannot precede period_start")
if not source_evidence or any(not item.strip() for item in source_evidence):
raise ValueError("at least one non-empty source_evidence reference is required")
normalized_currency = currency_code(currency)
timestamp = generated_at or datetime.now(timezone.utc)
common = {
"domain_slug": domain_slug.strip(),
"resource_id": resource_id,
"service_id": service_id,
"environment": environment,
"currency": normalized_currency,
"period_start": period_start,
"period_end": period_end,
"source_evidence": list(source_evidence),
"generated_at": timestamp,
}
signals: list[FinancialConstraintSignal] = []
def add(kind: str, classification: str, summary: str, **values: object) -> None:
identity_values = tuple((name, values[name]) for name in sorted(values))
signals.append(
FinancialConstraintSignal(
signal_id=_signal_id(
domain_slug,
period_start,
period_end,
normalized_currency,
resource_id,
service_id,
environment,
kind,
identity_values,
source_evidence,
),
signal_kind=kind,
classification=classification,
summary=summary,
**common,
**values,
)
)
ceiling = None if budget_ceiling is None else money(budget_ceiling, non_negative=True)
committed = (
None if active_commitment is None else money(active_commitment, non_negative=True)
)
spent_amount = None if spent is None else money(spent, non_negative=True)
burn = None if monthly_burn is None else money(monthly_burn, non_negative=True)
if ceiling is not None:
add(
"budget_ceiling",
"policy_constraint",
"Applicable budget ceiling for the effective period.",
amount=ceiling,
)
if committed is not None and committed > 0:
add(
"active_commitment",
"policy_constraint",
"Funds already committed during the effective period.",
amount=committed,
commitment_state="active",
)
if ceiling is not None and spent_amount is not None and ceiling > 0:
utilisation = spent_amount / ceiling
add(
"burn_pressure",
"informational_warning" if utilisation >= Decimal("0.8") else "informational",
"Current spend as a ratio of the applicable budget ceiling.",
amount=spent_amount,
metric_value=utilisation,
metric_unit="ratio",
)
if runway_months is not None:
months = Decimal(str(runway_months))
if not months.is_finite() or months < 0:
raise ValueError("runway_months must be a finite non-negative decimal")
threshold = (
None
if runway_threshold_months is None
else Decimal(str(runway_threshold_months))
)
if threshold is not None and (not threshold.is_finite() or threshold < 0):
raise ValueError(
"runway_threshold_months must be a finite non-negative decimal"
)
under_pressure = threshold is not None and months < threshold
add(
"runway_pressure",
"informational_warning" if under_pressure else "informational",
"Projected runway compared with the fin-hub viability threshold.",
amount=burn,
metric_value=months,
metric_unit="months",
)
return FinancialConstraintExport(
schema_version="0.1",
artifact_type="financial_constraint_projection",
signals=tuple(signals),
disclaimer=(
"Bounded affordability signals only. Not payment authority, procurement "
"approval, a provider commitment, or a replacement for fin-hub calculations."
),
)
def _ensure_planning_schema(conn: sqlite3.Connection) -> None:
conn.execute(
"""
@ -34,6 +210,18 @@ def _ensure_planning_schema(conn: sqlite3.Connection) -> None:
"CREATE INDEX IF NOT EXISTS ix_planning_evidence_type_current "
"ON planning_evidence (record_type, is_current)"
)
conn.execute(
"""
CREATE TABLE IF NOT EXISTS planning_evidence_rejections (
payload_fingerprint TEXT PRIMARY KEY,
claimed_record_id TEXT,
claimed_record_type TEXT,
error_json TEXT NOT NULL,
delivery_count INTEGER NOT NULL DEFAULT 1,
last_received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
)
"""
)
conn.commit()
@ -44,11 +232,32 @@ def ingest_planning_evidence(
) -> PlanningEvidence:
"""Validate and idempotently retain planning evidence outside booked spend."""
record = _PLANNING_ADAPTER.validate_python(payload)
canonical = record.model_dump_json()
ledger = ledger_path or default_ledger_path()
with _connect(ledger) as conn:
_ensure_planning_schema(conn)
try:
record = _PLANNING_ADAPTER.validate_python(payload)
except ValidationError as error:
payload_fingerprint = hashlib.sha256(
json.dumps(payload, sort_keys=True, default=str).encode()
).hexdigest()
safe_errors = error.errors(include_url=False, include_input=False)
conn.execute(
"INSERT INTO planning_evidence_rejections "
"(payload_fingerprint, claimed_record_id, claimed_record_type, error_json) "
"VALUES (?, ?, ?, ?) "
"ON CONFLICT(payload_fingerprint) DO UPDATE SET "
"delivery_count = delivery_count + 1, last_received_at = CURRENT_TIMESTAMP",
(
payload_fingerprint,
str(payload.get("record_id", ""))[:256] or None,
str(payload.get("record_type", ""))[:64] or None,
json.dumps(safe_errors, sort_keys=True, default=str),
),
)
conn.commit()
raise
canonical = record.model_dump_json()
conn.execute("BEGIN IMMEDIATE")
existing = conn.execute(
"SELECT payload_json FROM planning_evidence WHERE record_id = ?",
@ -81,7 +290,11 @@ def ingest_planning_evidence(
return record
def booked_cost_projection(*, ledger_path: Path | None = None) -> list[BookedCostEvidence]:
def booked_cost_projection(
*,
ledger_path: Path | None = None,
fact_resource_ids: Mapping[str, str] | None = None,
) -> list[BookedCostEvidence]:
"""Project current authoritative facts without exposing raw invoice content."""
ledger = ledger_path or default_ledger_path()
@ -90,6 +303,16 @@ def booked_cost_projection(*, ledger_path: Path | None = None) -> list[BookedCos
"SELECT * FROM ledger_entries WHERE is_current = 1 ORDER BY id"
).fetchall()
projected: list[BookedCostEvidence] = []
bindings = fact_resource_ids or {}
current_fact_ids = {row["financial_fact_id"] for row in rows}
unknown_bindings = set(bindings) - current_fact_ids
if unknown_bindings:
raise ValueError(
"resource bindings reference unknown current financial facts: "
+ ", ".join(sorted(unknown_bindings))
)
if any(not value.strip() for value in bindings.values()):
raise ValueError("resource binding IDs must be non-empty")
for row in rows:
amount = minor_money(int(row["amount_minor"]))
adjustment_kind = row["adjustment_kind"]
@ -126,6 +349,7 @@ def booked_cost_projection(*, ledger_path: Path | None = None) -> list[BookedCos
gross_amount=gross_amount,
adjustment_amount=adjustment_amount,
effective_amount=amount,
resource_id=bindings.get(row["financial_fact_id"]),
service_id=row["category"],
environment=row["environment"],
cost_attribution_key=row["cost_attribution_key"],
@ -136,6 +360,93 @@ def booked_cost_projection(*, ledger_path: Path | None = None) -> list[BookedCos
return projected
def exchange_health(
*,
ledger_path: Path | None = None,
fact_resource_ids: Mapping[str, str] | None = None,
as_of: date | None = None,
) -> ExchangeHealthReport:
"""Summarize actionable exchange-quality failures without raw payloads."""
ledger = ledger_path or default_ledger_path()
today = as_of or datetime.now(timezone.utc).date()
booked = booked_cost_projection(
ledger_path=ledger, fact_resource_ids=fact_resource_ids
)
issues: list[ExchangeQualityIssue] = []
for fact in booked:
if fact.resource_id is None and fact.cost_attribution_key is None:
issues.append(
ExchangeQualityIssue(
code="unattributed_booked_cost",
severity="warning",
record_id=fact.financial_fact_id,
detail=(
f"{fact.effective_amount} {fact.currency} has neither an external "
"resource binding nor a cost attribution key"
),
)
)
with _connect(ledger) as conn:
_ensure_planning_schema(conn)
planning_rows = conn.execute(
"SELECT record_id, record_type, payload_json FROM planning_evidence "
"WHERE is_current = 1 ORDER BY record_id"
).fetchall()
rejection_rows = conn.execute(
"SELECT claimed_record_id, delivery_count FROM planning_evidence_rejections "
"ORDER BY claimed_record_id"
).fetchall()
for row in planning_rows:
payload = json.loads(row["payload_json"])
if row["record_type"] == "forecast" and date.fromisoformat(
payload["period_end"]
) < today:
issues.append(
ExchangeQualityIssue(
code="stale_forecast",
severity="warning",
record_id=row["record_id"],
detail=f"forecast period ended before {today.isoformat()}",
)
)
if row["record_type"] == "allocation":
known = {fact.financial_fact_id for fact in booked}
missing = sorted(set(payload["financial_fact_ids"]) - known)
if missing:
issues.append(
ExchangeQualityIssue(
code="allocation_missing_financial_fact",
severity="error",
record_id=row["record_id"],
detail="missing current facts: " + ", ".join(missing),
)
)
for row in rejection_rows:
issues.append(
ExchangeQualityIssue(
code="rejected_planning_delivery",
severity="error",
record_id=row["claimed_record_id"],
detail=f"rejected delivery observed {row['delivery_count']} time(s)",
)
)
return ExchangeHealthReport(
schema_version="0.1",
artifact_type="resource_cost_exchange_health",
booked_fact_count=len(booked),
current_planning_record_count=len(planning_rows),
rejected_delivery_count=sum(int(row["delivery_count"]) for row in rejection_rows),
issues=tuple(
sorted(issues, key=lambda issue: (issue.code, issue.record_id or ""))
),
)
def ingest_resource_forecast(
payload: dict,
*,