Add per-client billing basis export

This commit is contained in:
tegwick 2026-08-11 14:39:48 +02:00
parent b75ad06b3a
commit fcec177ded
10 changed files with 506 additions and 11 deletions

View file

@ -2,7 +2,7 @@
# Custodian Brief — fin-hub
**Domain:** financials
**Last synced:** 2026-08-11 09:04 UTC
**Last synced:** 2026-08-11 12:39 UTC
**State Hub:** http://127.0.0.1:8000 *(adjust if running on a remote machine)*
## Active Workstreams
@ -16,11 +16,10 @@ Progress: 6/9 done | workplan_id: `67b6de6c-4820-4478-9789-f50260204c27`
- · T06 — Generalize and operate the contract `a1309d51`
### Client attribution and billing basis
Progress: 4/6 done | workplan_id: `ebc1d2de-ae11-4cde-b860-047922fc74b9`
Progress: 5/6 done | workplan_id: `ebc1d2de-ae11-4cde-b860-047922fc74b9`
**Open tasks:**
- ! Select the external invoicing system `d30b606f`
- · Export a per-client billing basis `b5886131`
---
## MCP Orientation (when available)

View file

@ -25,6 +25,7 @@ uv run finhub ledger import cloud tests/fixtures/cloud-costs.csv
uv run finhub ledger set-price --client acme --application portal --instance prod-01 --period 2026-07 --amount 100 --source agreement-2026-01
uv run finhub ledger margins
uv run finhub ledger allocations
uv run finhub ledger billing-basis
uv run finhub evaluate
uv run finhub evidence --seed-fixtures
uv run finhub serve
@ -78,6 +79,13 @@ and provenance remain visible. Deterministic largest-remainder rounding makes
all target amounts plus the explicit unattributed residual reconcile exactly
to booked cost.
`ledger billing-basis` produces an idempotent client-period reporting artifact.
Each record includes the current price and revision, direct and allocated cost,
margin, financial-fact/correction IDs, allocation/revision IDs, and non-secret
provenance. Residuals, non-client targets, and client costs without a price are
explicit exceptions. The artifact carries a mandatory disclaimer and never
assigns invoice numbers, performs bookkeeping, or requests/tracks payment.
## Related Workplans
- `the-custodian/workplans/CUST-WP-0025-fos-hub-bootstrap.md` — umbrella

View file

@ -27,7 +27,7 @@
| task | FIN-WP-0002-T01 | done | — | workplans/FIN-WP-0002-client-attribution-and-billing-basis.md |
| task | FIN-WP-0002-T02 | done | — | workplans/FIN-WP-0002-client-attribution-and-billing-basis.md |
| task | FIN-WP-0002-T03 | done | — | workplans/FIN-WP-0002-client-attribution-and-billing-basis.md |
| task | FIN-WP-0002-T04 | todo | — | workplans/FIN-WP-0002-client-attribution-and-billing-basis.md |
| task | FIN-WP-0002-T04 | done | — | workplans/FIN-WP-0002-client-attribution-and-billing-basis.md |
| task | FIN-WP-0002-T05 | wait | — | workplans/FIN-WP-0002-client-attribution-and-billing-basis.md |
| task | FIN-WP-0003-T01 | todo | — | workplans/FIN-WP-0003-fabric-authority-boundary.md |
| task | FIN-WP-0003-T02 | todo | — | workplans/FIN-WP-0003-fabric-authority-boundary.md |

View file

@ -57,3 +57,21 @@ def optional_attribution(
"client_id, application_id, and app_instance_id must be supplied together"
)
return ClientAttribution(client_id or "", application_id or "", app_instance_id or "")
def parse_client_attribution_key(value: str) -> ClientAttribution:
"""Parse only the canonical fin-hub client attribution namespace."""
parts = value.split("|")
if len(parts) != 3:
raise ValueError("invalid client attribution key")
expected = ("client:", "app:", "instance:")
values: list[str] = []
for part, prefix in zip(parts, expected, strict=True):
if not part.startswith(prefix):
raise ValueError("invalid client attribution key")
values.append(part.removeprefix(prefix))
attribution = ClientAttribution(*values)
if attribution.key != value:
raise ValueError("client attribution key is not canonical")
return attribution

View file

@ -15,6 +15,7 @@ from fin_hub.ingest.cloud import parse_cloud_cost_csv
from fin_hub.ingest.hosteurope import parse_hosteurope_csv
from fin_hub.services.alerts import evaluate_budget_alerts
from fin_hub.services.allocation import shared_cost_allocations
from fin_hub.services.billing import build_billing_basis
from fin_hub.services.evaluate import evaluate_runway
from fin_hub.services.evidence import write_runway_evidence
from fin_hub.services.ledger import (
@ -156,6 +157,12 @@ def _cmd_ledger_allocations(args: argparse.Namespace) -> int:
return 0
def _cmd_ledger_billing_basis(args: argparse.Namespace) -> int:
export = build_billing_basis(ledger_path=_ledger_path(args))
print(json.dumps(export.as_dict(), indent=2, default=str))
return 0
def _cmd_evaluate(args: argparse.Namespace) -> int:
report = evaluate_runway(
ledger_path=_ledger_path(args),
@ -275,6 +282,13 @@ def build_parser() -> argparse.ArgumentParser:
ledger_allocations.add_argument("--ledger", help="Ledger database path")
ledger_allocations.set_defaults(func=_cmd_ledger_allocations)
billing_basis = ledger_sub.add_parser(
"billing-basis",
help="Export deterministic per-client reporting basis (never an invoice)",
)
billing_basis.add_argument("--ledger", help="Ledger database path")
billing_basis.set_defaults(func=_cmd_ledger_billing_basis)
evaluate = sub.add_parser(
"evaluate",
help="Evaluate runway from ledger burns (cron/systemd friendly)",

View file

@ -8,6 +8,7 @@ from datetime import date
from decimal import Decimal, ROUND_FLOOR
from pathlib import Path
from fin_hub.attribution import parse_client_attribution_key
from fin_hub.money import MONEY_QUANTUM, minor_money, money_minor
from fin_hub.schemas.exchange import AllocationEvidence
from fin_hub.services.exchange import _ensure_planning_schema
@ -28,6 +29,7 @@ class AllocatedTarget:
@dataclass(frozen=True)
class AllocationReport:
allocation_id: str
revision_of: str | None
method: str
financial_fact_ids: tuple[str, ...]
period_start: date
@ -108,6 +110,15 @@ def shared_cost_allocations(*, ledger_path: Path | None = None) -> list[Allocati
).fetchone()
if fact is None:
raise ValueError(f"allocation references missing/current fact {fact_id}")
if fact["cost_attribution_key"] is not None:
try:
parse_client_attribution_key(fact["cost_attribution_key"])
except ValueError:
pass
else:
raise ValueError(
f"allocation cannot reallocate directly attributed fact {fact_id}"
)
if fact["currency"] != allocation.currency:
raise ValueError(f"allocation currency does not match fact {fact_id}")
if allocation.environment and fact["environment"] != allocation.environment:
@ -145,6 +156,7 @@ def shared_cost_allocations(*, ledger_path: Path | None = None) -> list[Allocati
reports.append(
AllocationReport(
allocation_id=allocation.record_id,
revision_of=allocation.revision_of,
method=allocation.method,
financial_fact_ids=tuple(allocation.financial_fact_ids),
period_start=allocation.period_start,

View file

@ -0,0 +1,288 @@
"""Idempotent per-client billing-basis reporting (never invoice generation)."""
from __future__ import annotations
import hashlib
from calendar import monthrange
from dataclasses import asdict, dataclass
from datetime import date
from decimal import Decimal
from pathlib import Path
from fin_hub.attribution import parse_client_attribution_key
from fin_hub.money import minor_money
from fin_hub.services.allocation import AllocationReport, shared_cost_allocations
from fin_hub.services.ledger import _connect, default_ledger_path
@dataclass(frozen=True)
class BillingBasisRecord:
billing_basis_id: str
cost_attribution_key: str
client_id: str
application_id: str
app_instance_id: str
period_month: str
currency: str
price_id: str
price_revision_of: str | None
price_source: str
revenue: Decimal
direct_cost: Decimal
allocated_cost: Decimal
total_cost: Decimal
margin: Decimal
financial_fact_ids: tuple[str, ...]
financial_fact_corrections: tuple[tuple[str, str], ...]
allocation_ids: tuple[str, ...]
allocation_revisions: tuple[tuple[str, str], ...]
provenance: tuple[str, ...]
def as_dict(self) -> dict:
return asdict(self)
@dataclass(frozen=True)
class BillingBasisException:
reason: str
cost_attribution_key: str | None
period_month: str
currency: str
amount: Decimal
evidence_ids: tuple[str, ...]
def as_dict(self) -> dict:
return asdict(self)
@dataclass(frozen=True)
class BillingBasisExport:
schema_version: str
artifact_type: str
records: tuple[BillingBasisRecord, ...]
exceptions: tuple[BillingBasisException, ...]
disclaimer: str
def as_dict(self) -> dict:
return asdict(self)
def _stable_id(*parts: object) -> str:
payload = "\x1f".join(str(part) for part in parts)
return "billing-basis:" + hashlib.sha256(payload.encode("utf-8")).hexdigest()
def _allocation_month(report: AllocationReport) -> str:
if report.period_start.year != report.period_end.year or report.period_start.month != report.period_end.month:
raise ValueError(
f"allocation {report.allocation_id} spans multiple billing months"
)
expected_end = date(
report.period_start.year,
report.period_start.month,
monthrange(report.period_start.year, report.period_start.month)[1],
)
if report.period_start.day != 1 or report.period_end != expected_end:
raise ValueError(
f"allocation {report.allocation_id} must cover a complete billing month"
)
return report.period_start.strftime("%Y-%m")
def build_billing_basis(*, ledger_path: Path | None = None) -> BillingBasisExport:
"""Build deterministic client-period reporting with explicit omissions."""
ledger = ledger_path or default_ledger_path()
allocations = shared_cost_allocations(ledger_path=ledger)
with _connect(ledger) as conn:
price_rows = conn.execute(
"SELECT * FROM engagement_prices WHERE is_current = 1 "
"ORDER BY period_month, cost_attribution_key, currency"
).fetchall()
fact_rows = conn.execute(
"SELECT * FROM ledger_entries WHERE is_current = 1 ORDER BY financial_fact_id"
).fetchall()
prices = {
(row["cost_attribution_key"], row["period_month"], row["currency"]): row
for row in price_rows
}
facts_by_id = {row["financial_fact_id"]: row for row in fact_rows}
direct: dict[tuple[str, str, str], list] = {}
for fact in fact_rows:
key = fact["cost_attribution_key"]
if key is None:
continue
try:
parse_client_attribution_key(key)
except ValueError:
continue
direct.setdefault((key, fact["period_month"], fact["currency"]), []).append(fact)
allocated: dict[tuple[str, str, str], list[tuple[AllocationReport, Decimal]]] = {}
exceptions: list[BillingBasisException] = []
for allocation in allocations:
period = _allocation_month(allocation)
for target in allocation.targets:
try:
parse_client_attribution_key(target.target_key)
except ValueError:
exceptions.append(
BillingBasisException(
reason="non_client_allocation_target",
cost_attribution_key=target.target_key,
period_month=period,
currency=allocation.currency,
amount=target.amount,
evidence_ids=(allocation.allocation_id,),
)
)
continue
allocated.setdefault(
(target.target_key, period, allocation.currency), []
).append((allocation, target.amount))
if allocation.residual_amount:
exceptions.append(
BillingBasisException(
reason="unattributed_allocation_residual",
cost_attribution_key=None,
period_month=period,
currency=allocation.currency,
amount=allocation.residual_amount,
evidence_ids=(allocation.allocation_id,),
)
)
cost_keys = set(direct) | set(allocated)
for key in sorted(cost_keys - set(prices)):
facts = direct.get(key, [])
allocation_parts = allocated.get(key, [])
amount_minor = sum(int(row["amount_minor"]) for row in facts) + sum(
int(amount / Decimal("0.01")) for _allocation, amount in allocation_parts
)
exceptions.append(
BillingBasisException(
reason="missing_price_basis",
cost_attribution_key=key[0],
period_month=key[1],
currency=key[2],
amount=minor_money(amount_minor),
evidence_ids=tuple(
sorted(
[row["financial_fact_id"] for row in facts]
+ [allocation.allocation_id for allocation, _amount in allocation_parts]
)
),
)
)
records: list[BillingBasisRecord] = []
for key, price in sorted(prices.items()):
attribution = parse_client_attribution_key(key[0])
facts = direct.get(key, [])
allocation_parts = allocated.get(key, [])
direct_minor = sum(int(row["amount_minor"]) for row in facts)
allocated_minor = sum(
int(amount / Decimal("0.01")) for _allocation, amount in allocation_parts
)
revenue_minor = int(price["amount_minor"])
allocated_fact_ids = {
fact_id
for allocation, _amount in allocation_parts
for fact_id in allocation.financial_fact_ids
}
all_evidence_facts = [
*facts,
*(facts_by_id[fact_id] for fact_id in sorted(allocated_fact_ids)),
]
fact_ids = tuple(
sorted({row["financial_fact_id"] for row in all_evidence_facts})
)
correction_pairs = tuple(
sorted(
(row["financial_fact_id"], row["correction_of"])
for row in all_evidence_facts
if row["correction_of"]
)
)
allocation_ids = tuple(
sorted(allocation.allocation_id for allocation, _amount in allocation_parts)
)
allocation_revisions = tuple(
sorted(
(allocation.allocation_id, allocation.revision_of)
for allocation, _amount in allocation_parts
if allocation.revision_of
)
)
provenance = tuple(
sorted(
{price["source"]}
| {
row["correction_source"] or row["source_document_id"]
for row in all_evidence_facts
}
| {
evidence
for allocation, _amount in allocation_parts
for evidence in allocation.source_evidence
}
)
)
total_minor = direct_minor + allocated_minor
basis_id = _stable_id(
key,
price["id"],
price["revision_of"],
fact_ids,
correction_pairs,
allocation_ids,
allocation_revisions,
direct_minor,
allocated_minor,
revenue_minor,
)
records.append(
BillingBasisRecord(
billing_basis_id=basis_id,
cost_attribution_key=key[0],
client_id=attribution.client_id,
application_id=attribution.application_id,
app_instance_id=attribution.app_instance_id,
period_month=key[1],
currency=key[2],
price_id=price["id"],
price_revision_of=price["revision_of"],
price_source=price["source"],
revenue=minor_money(revenue_minor),
direct_cost=minor_money(direct_minor),
allocated_cost=minor_money(allocated_minor),
total_cost=minor_money(total_minor),
margin=minor_money(revenue_minor - total_minor),
financial_fact_ids=fact_ids,
financial_fact_corrections=correction_pairs,
allocation_ids=allocation_ids,
allocation_revisions=allocation_revisions,
provenance=provenance,
)
)
return BillingBasisExport(
schema_version="0.1",
artifact_type="billing_basis_report",
records=tuple(records),
exceptions=tuple(
sorted(
exceptions,
key=lambda item: (
item.period_month,
item.currency,
item.reason,
item.cost_attribution_key or "",
),
)
),
disclaimer=(
"Reporting basis only; not an invoice, bookkeeping record, payment request, "
"or evidence of payment."
),
)

View file

@ -1,11 +1,16 @@
import pytest
from fin_hub.attribution import ClientAttribution, optional_attribution
from fin_hub.attribution import (
ClientAttribution,
optional_attribution,
parse_client_attribution_key,
)
def test_client_attribution_key_is_stable():
attribution = ClientAttribution(" acme ", "portal", "prod-01")
assert attribution.key == "client:acme|app:portal|instance:prod-01"
assert parse_client_attribution_key(attribution.key) == attribution
def test_empty_attribution_is_explicitly_unattributed():
@ -17,3 +22,8 @@ def test_empty_attribution_is_explicitly_unattributed():
def test_client_attribution_rejects_unsafe_parts(invalid: str):
with pytest.raises(ValueError):
ClientAttribution(invalid, "portal", "prod-01")
def test_client_attribution_parser_rejects_other_namespaces():
with pytest.raises(ValueError, match="invalid"):
parse_client_attribution_key("platform:shared-cluster")

138
tests/test_billing.py Normal file
View file

@ -0,0 +1,138 @@
import sqlite3
from decimal import Decimal
from pathlib import Path
from fin_hub.services.billing import build_billing_basis
from fin_hub.services.exchange import ingest_planning_evidence
from fin_hub.services.ledger import import_csv, record_engagement_price
CLIENT_KEY = "client:acme|app:portal|instance:prod-01"
def _seed_billing_ledger(tmp_path: Path):
ledger = tmp_path / "ledger.db"
direct = tmp_path / "direct.csv"
direct.write_text(
"product,amount,currency,invoice_date,environment,client_id,application_id,app_instance_id\n"
"Dedicated service,20.00,EUR,2026-07-01,production,acme,portal,prod-01\n",
encoding="utf-8",
)
shared = tmp_path / "shared.csv"
shared.write_text(
"product,amount,currency,invoice_date,environment\n"
"Shared cluster,100.00,EUR,2026-07-01,production\n",
encoding="utf-8",
)
import_csv(direct, "hosteurope", ledger_path=ledger)
import_csv(shared, "hosteurope", ledger_path=ledger)
with sqlite3.connect(ledger) as conn:
shared_fact_id = conn.execute(
"SELECT financial_fact_id FROM ledger_entries "
"WHERE cost_attribution_key IS NULL"
).fetchone()[0]
price = record_engagement_price(
client_id="acme",
application_id="portal",
app_instance_id="prod-01",
period_month="2026-07",
amount="200.00",
currency="EUR",
source="agreement-v1",
ledger_path=ledger,
)
allocation = {
"schema_version": "0.1",
"record_type": "allocation",
"record_id": "allocation:shared:2026-07:v1",
"revision_of": None,
"resource_id": "resource:shared_cluster",
"service_id": "shared-cluster",
"workload_id": None,
"tenant_id": None,
"environment": "production",
"cost_attribution_key": "platform:shared-cluster",
"period_start": "2026-07-01",
"period_end": "2026-07-31",
"currency": "EUR",
"source_evidence": ["resource-control:namespace-cpu-v1"],
"created_at": "2026-08-11T10:00:00Z",
"financial_fact_ids": [shared_fact_id],
"method": "namespace-cpu-v1",
"allocated_amount": "100.00",
"shares": [{"target_key": CLIENT_KEY, "share": "0.80"}],
"residual_share": "0.20",
}
ingest_planning_evidence(allocation, ledger_path=ledger)
return ledger, direct, price, allocation
def test_billing_basis_combines_direct_and_allocated_cost_idempotently(tmp_path: Path):
ledger, _direct, price, allocation = _seed_billing_ledger(tmp_path)
first = build_billing_basis(ledger_path=ledger)
second = build_billing_basis(ledger_path=ledger)
assert first == second
assert first.artifact_type == "billing_basis_report"
assert "not an invoice" in first.disclaimer
record = first.records[0]
assert record.price_id == price.id
assert record.revenue == Decimal("200.00")
assert record.direct_cost == Decimal("20.00")
assert record.allocated_cost == Decimal("80.00")
assert record.total_cost == Decimal("100.00")
assert record.margin == Decimal("100.00")
assert record.allocation_ids == (allocation["record_id"],)
assert record.financial_fact_ids
assert "agreement-v1" in record.provenance
assert first.exceptions[0].reason == "unattributed_allocation_residual"
assert first.exceptions[0].amount == Decimal("20.00")
def test_billing_basis_id_changes_and_exposes_correction_and_price_revision(tmp_path: Path):
ledger, direct, price, _allocation = _seed_billing_ledger(tmp_path)
before = build_billing_basis(ledger_path=ledger).records[0]
direct.write_text(
"product,amount,currency,invoice_date,environment,client_id,application_id,app_instance_id\n"
"Dedicated service,25.00,EUR,2026-07-01,production,acme,portal,prod-01\n",
encoding="utf-8",
)
import_csv(direct, "hosteurope", ledger_path=ledger, force=True)
revised_price = record_engagement_price(
client_id="acme",
application_id="portal",
app_instance_id="prod-01",
period_month="2026-07",
amount="210.00",
currency="EUR",
source="agreement-v2",
revision_of=price.id,
ledger_path=ledger,
)
after = build_billing_basis(ledger_path=ledger).records[0]
assert after.billing_basis_id != before.billing_basis_id
assert after.price_id == revised_price.id
assert after.price_revision_of == price.id
assert after.financial_fact_corrections
assert after.direct_cost == Decimal("25.00")
assert after.margin == Decimal("105.00")
def test_client_cost_without_price_is_an_explicit_exception(tmp_path: Path):
ledger = tmp_path / "ledger.db"
source = tmp_path / "direct.csv"
source.write_text(
"product,amount,currency,invoice_date,client_id,application_id,app_instance_id\n"
"Dedicated service,20.00,EUR,2026-07-01,acme,portal,prod-01\n",
encoding="utf-8",
)
import_csv(source, "hosteurope", ledger_path=ledger)
export = build_billing_basis(ledger_path=ledger)
assert export.records == ()
assert export.exceptions[0].reason == "missing_price_basis"
assert export.exceptions[0].cost_attribution_key == CLIENT_KEY
assert export.exceptions[0].amount == Decimal("20.00")

View file

@ -142,7 +142,7 @@ missing facts, mismatches, evidence revisions, and duplicate allocation.
```task
id: FIN-WP-0002-T04
status: todo
status: done
priority: medium
state_hub_task_id: "b5886131-82f6-49fb-a7e3-ba47d5b627a8"
```
@ -155,6 +155,14 @@ currency, corrections, and provenance.
The export must not issue invoices, assign legal invoice numbers, execute or
track payments, or represent itself as bookkeeping evidence.
Completed 2026-08-11: added a deterministic v0.1 billing-basis artifact and
`ledger billing-basis` command. Each client-period record contains the current
price/revision, direct and allocated cost, margin, financial-fact/correction
IDs, allocation/revision IDs, and non-secret provenance. Stable IDs change
only when an input basis changes. Residuals, non-client targets, and missing
prices are explicit exceptions. The artifact carries a mandatory reporting-
only disclaimer and has no invoice, bookkeeping, or payment behavior.
## Select the external invoicing system
```task
@ -172,12 +180,12 @@ not required to settle T00T03.
## Acceptance
- [ ] Repository scope explicitly allows the reporting and export boundary.
- [ ] Costs can be reported by client × application × instance without making
- [x] Repository scope explicitly allows the reporting and export boundary.
- [x] Costs can be reported by client × application × instance without making
fin-hub authoritative for client or resource identity.
- [ ] Revenue and margin are period-aware, currency-aware, and distinguishable
- [x] Revenue and margin are period-aware, currency-aware, and distinguishable
from invoices and payments.
- [ ] Shared-infrastructure allocations reconcile to authoritative booked
- [x] Shared-infrastructure allocations reconcile to authoritative booked
costs with assumptions and unattributed residuals visible.
- [ ] The billing-basis export is idempotent and contains no payment execution.
- [x] The billing-basis export is idempotent and contains no payment execution.
- [ ] A human records the external invoicing/bookkeeping system decision.