Add engagement margin reporting

src/fin_hub/services/ledger.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import json
 import sqlite3
+import uuid
 from dataclasses import asdict, dataclass
 from datetime import date, datetime, timezone
 from pathlib import Path
@@ -53,6 +54,38 @@ class ImportResult:
         return asdict(self)
 
 
+@dataclass(frozen=True)
+class EngagementPriceRecord:
+    id: str
+    client_id: str
+    application_id: str
+    app_instance_id: str
+    cost_attribution_key: str
+    period_month: str
+    amount: float
+    currency: str
+    source: str
+    revision_of: str | None
+    recorded_at: str
+
+
+@dataclass(frozen=True)
+class ClientMargin:
+    cost_attribution_key: str
+    client_id: str
+    application_id: str
+    app_instance_id: str
+    period_month: str
+    currency: str
+    revenue: float
+    attributed_cost: float
+    margin: float
+    price_id: str
+
+    def as_dict(self) -> dict:
+        return asdict(self)
+
+
 def default_ledger_path() -> Path:
     return DEFAULT_LEDGER_PATH
 
@@ -94,6 +127,23 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
             key TEXT PRIMARY KEY,
             value TEXT NOT NULL
         );
+
+        CREATE TABLE IF NOT EXISTS engagement_prices (
+            id TEXT PRIMARY KEY,
+            client_id TEXT NOT NULL,
+            application_id TEXT NOT NULL,
+            app_instance_id TEXT NOT NULL,
+            cost_attribution_key TEXT NOT NULL,
+            period_month TEXT NOT NULL,
+            amount REAL NOT NULL,
+            currency TEXT NOT NULL,
+            source TEXT NOT NULL,
+            revision_of TEXT REFERENCES engagement_prices(id),
+            recorded_at TEXT NOT NULL
+        );
+
+        CREATE INDEX IF NOT EXISTS ix_engagement_prices_basis
+        ON engagement_prices (cost_attribution_key, period_month, currency);
         """
     )
     columns = {row["name"] for row in conn.execute("PRAGMA table_info(ledger_entries)")}
@@ -304,6 +354,134 @@ def monthly_burn_series(
     return burns
 
 
+def record_engagement_price(
+    *,
+    client_id: str,
+    application_id: str,
+    app_instance_id: str,
+    period_month: str,
+    amount: float,
+    currency: str,
+    source: str,
+    ledger_path: Path | None = None,
+    price_id: str | None = None,
+    revision_of: str | None = None,
+) -> EngagementPriceRecord:
+    from fin_hub.attribution import ClientAttribution
+
+    attribution = ClientAttribution(client_id, application_id, app_instance_id)
+    if len(period_month) != 7 or period_month[4] != "-":
+        raise ValueError("period_month must use YYYY-MM")
+    try:
+        date.fromisoformat(f"{period_month}-01")
+    except ValueError as exc:
+        raise ValueError("period_month must use YYYY-MM") from exc
+    normalized_currency = currency.strip().upper()
+    if len(normalized_currency) != 3 or not normalized_currency.isalpha():
+        raise ValueError("currency must be a three-letter code")
+    if amount < 0:
+        raise ValueError("engagement price amount cannot be negative")
+    if not source.strip():
+        raise ValueError("source is required")
+
+    ledger = ledger_path or default_ledger_path()
+    identifier = price_id or str(uuid.uuid4())
+    recorded_at = _utc_now()
+    with _connect(ledger) as conn:
+        basis = (attribution.key, period_month, normalized_currency)
+        current = conn.execute(
+            "SELECT id FROM engagement_prices WHERE cost_attribution_key = ? "
+            "AND period_month = ? AND currency = ? "
+            "ORDER BY recorded_at DESC, rowid DESC LIMIT 1",
+            basis,
+        ).fetchone()
+        if current is not None and revision_of != current["id"]:
+            raise ValueError(f"revision_of must reference current price {current['id']}")
+        if current is None and revision_of is not None:
+            raise ValueError("revision_of cannot be used without an existing price")
+        conn.execute(
+            "INSERT INTO engagement_prices ("
+            "id, client_id, application_id, app_instance_id, cost_attribution_key, "
+            "period_month, amount, currency, source, revision_of, recorded_at"
+            ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (
+                identifier,
+                attribution.client_id,
+                attribution.application_id,
+                attribution.app_instance_id,
+                attribution.key,
+                period_month,
+                amount,
+                normalized_currency,
+                source.strip(),
+                revision_of,
+                recorded_at,
+            ),
+        )
+        conn.commit()
+    return EngagementPriceRecord(
+        id=identifier,
+        client_id=attribution.client_id,
+        application_id=attribution.application_id,
+        app_instance_id=attribution.app_instance_id,
+        cost_attribution_key=attribution.key,
+        period_month=period_month,
+        amount=amount,
+        currency=normalized_currency,
+        source=source.strip(),
+        revision_of=revision_of,
+        recorded_at=recorded_at,
+    )
+
+
+def client_margin_report(*, ledger_path: Path | None = None) -> list[ClientMargin]:
+    ledger = ledger_path or default_ledger_path()
+    with _connect(ledger) as conn:
+        rows = conn.execute(
+            """
+            WITH latest_prices AS (
+                SELECT p.*
+                FROM engagement_prices p
+                WHERE NOT EXISTS (
+                    SELECT 1 FROM engagement_prices revision
+                    WHERE revision.revision_of = p.id
+                )
+            ), attributed_costs AS (
+                SELECT cost_attribution_key, period_month, currency,
+                       SUM(amount) AS attributed_cost
+                FROM ledger_entries
+                WHERE cost_attribution_key IS NOT NULL
+                GROUP BY cost_attribution_key, period_month, currency
+            )
+            SELECT p.id AS price_id, p.client_id, p.application_id,
+                   p.app_instance_id, p.cost_attribution_key, p.period_month,
+                   p.currency, p.amount AS revenue,
+                   COALESCE(c.attributed_cost, 0) AS attributed_cost
+            FROM latest_prices p
+            LEFT JOIN attributed_costs c
+              ON c.cost_attribution_key = p.cost_attribution_key
+             AND c.period_month = p.period_month
+             AND c.currency = p.currency
+            ORDER BY p.period_month, p.client_id, p.application_id, p.app_instance_id
+            """
+        ).fetchall()
+    return [
+        ClientMargin(
+            cost_attribution_key=row["cost_attribution_key"],
+            client_id=row["client_id"],
+            application_id=row["application_id"],
+            app_instance_id=row["app_instance_id"],
+            period_month=row["period_month"],
+            currency=row["currency"],
+            revenue=float(row["revenue"]),
+            attributed_cost=float(row["attributed_cost"]),
+            margin=float(row["revenue"]) - float(row["attributed_cost"]),
+            price_id=row["price_id"],
+        )
+        for row in rows
+    ]
+
+
 def ledger_stats(*, ledger_path: Path | None = None) -> dict:
     ledger = ledger_path or default_ledger_path()
     balance, balance_currency = get_opening_balance(ledger)