Refactor economics to expense-record ledger with correct Bubble cost

Replace pre-aggregated costs.json with expense_records.json (48 line-item
records) and payment_records.json. All monthly and cumulative totals are
computed deterministically in observatory/ledger.py. Correct Bubble.io to
$32/mo (since Feb 2025) — infrastructure €69.44/mo not €72.20.

projects/coulomb-pricing/observatory/dashboard.py

@@ -9,26 +9,27 @@ from .load import (
     default_data_dir,
     latest_period,
     load_budget,
+    load_expense_records,
     load_membership,
-    load_monthly_platform_costs,
+    load_monthly_ledger,
+    load_payment_records,
     load_pricing_models,
     load_product,
-    load_revenue,
 )
-from .models import EconomicsSnapshot, LiquiditySummary, MonthlyPlatformCost, PricingModel, Product, RevenueEntry
+from .models import EconomicsSnapshot, LiquiditySummary, MonthlyPlatformCost, PaymentRecord, PricingModel, Product
 
 
 def _history_rows(
     monthly_costs: list[MonthlyPlatformCost],
-    revenue_entries: list[RevenueEntry],
+    payments: list[PaymentRecord],
     through_period: str,
 ) -> str:
-    revenue_by_period = {entry.period: entry for entry in revenue_entries}
+    payment_by_period = {record.period: record for record in payments}
     lines: list[str] = []
     for month in sorted(monthly_costs, key=lambda item: item.period):
         if month.period > through_period:
             continue
-        payment = revenue_by_period.get(month.period)
+        payment = payment_by_period.get(month.period)
         net_payment = payment.net_amount if payment else Decimal("0")
         period_net = net_payment - month.infrastructure_cost
         lines.append(
@@ -45,14 +46,15 @@ def render_dashboard(
     snapshot: EconomicsSnapshot,
     liquidity: LiquiditySummary,
     monthly_costs: list[MonthlyPlatformCost],
-    revenue_entries: list[RevenueEntry],
+    payments: list[PaymentRecord],
+    expense_count: int,
 ) -> str:
     active = next(m for m in models if m.id == product.active_pricing_model_id)
     registry_lines = "\n".join(
         f"| {model.id} | {model.name} | {model.model_type} | {model.status} |"
         for model in models
     )
-    history_lines = _history_rows(monthly_costs, revenue_entries, snapshot.period)
+    history_lines = _history_rows(monthly_costs, payments, snapshot.period)
     budget_state = (
         "over budget"
         if liquidity.remaining_budget < Decimal("0")
@@ -66,7 +68,9 @@ def render_dashboard(
 **Active pricing model:** {active.name} ({active.access_fee_amount} {active.currency}/{active.access_fee_cadence})
 
 > Platform costs accrue to the operator. Customer cost-pass-through billing is
-> **not active** in MVP — members pay subscription only.
+> **not active** in MVP — members pay subscription only. All totals are computed
+> programmatically from expense and payment record ledgers ({expense_count} expense
+> records).
 
 ## Key Metrics (current period)
 
@@ -114,9 +118,8 @@ _Revenue source: {snapshot.revenue_source}_
 
 - Product model (`data/product.json`)
 - Budget (`data/budget.json`)
-- Pricing model registry (`data/pricing-models.json`)
-- Platform costs (`data/costs.json`)
-- Member payments (`data/revenue.json`)
+- Expense records (`data/expense_records.json`) — source of truth for costs
+- Payment records (`data/payment_records.json`)
 - Membership (`data/membership.json`)
 - Requirements (`REQUIREMENTS.md`)
 """
@@ -128,13 +131,16 @@ def generate_dashboard(data_dir: Path | None = None, period: str | None = None)
     budget = load_budget(root)
     models = load_pricing_models(root)
     members = load_membership(root)
-    revenue = load_revenue(root)
-    monthly_costs = load_monthly_platform_costs(root)
+    payments = load_payment_records(root)
+    expenses = load_expense_records(root)
+    monthly_costs = load_monthly_ledger(root)
     target_period = period or latest_period(monthly_costs)
 
-    snapshot = build_snapshot(target_period, product, models, members, revenue, monthly_costs)
-    liquidity = build_liquidity_summary(budget, revenue, monthly_costs, target_period)
-    return render_dashboard(product, models, snapshot, liquidity, monthly_costs, revenue)
+    snapshot = build_snapshot(target_period, product, models, members, payments, monthly_costs)
+    liquidity = build_liquidity_summary(budget, payments, monthly_costs, target_period)
+    return render_dashboard(
+        product, models, snapshot, liquidity, monthly_costs, payments, len(expenses)
+    )
 
 
 def main(argv: list[str] | None = None) -> int:

projects/coulomb-pricing/observatory/economics.py

@@ -4,15 +4,14 @@ from decimal import Decimal, ROUND_HALF_UP
 
 from .models import (
     Budget,
-    CostEntry,
     EconomicsSnapshot,
     LiquidityStatus,
     LiquiditySummary,
     MembershipRecord,
     MonthlyPlatformCost,
+    PaymentRecord,
     PricingModel,
     Product,
-    RevenueEntry,
 )
 
 TWOPLACES = Decimal("0.01")
@@ -23,16 +22,6 @@ def _quantize(value: Decimal, exp: Decimal = TWOPLACES) -> Decimal:
     return value.quantize(exp, rounding=ROUND_HALF_UP)
 
 
-def _convert_to_eur(amount: Decimal, currency: str, fx_rates: dict[str, Decimal]) -> Decimal:
-    if currency == "EUR":
-        return amount
-    if currency == "USD":
-        return amount * fx_rates.get("USD/EUR", Decimal("1"))
-    if currency == "ratio":
-        return amount
-    raise ValueError(f"unsupported currency for conversion: {currency}")
-
-
 def liquidity_status_for(net: Decimal) -> LiquidityStatus:
     if net < Decimal("0"):
         return "burning"
@@ -52,27 +41,10 @@ def active_pricing_model(models: list[PricingModel], product: Product) -> Pricin
     raise ValueError(f"active pricing model not found: {product.active_pricing_model_id}")
 
 
-def monthly_cost_from_rate_card(
-    cost_entries: list[CostEntry],
-    fx_rates: dict[str, Decimal],
-    gross_revenue: Decimal,
-    member_count: int,
-) -> Decimal:
-    total = Decimal("0")
-    for entry in cost_entries:
-        if entry.allocation == "flat":
-            total += _convert_to_eur(entry.amount, entry.currency, fx_rates)
-        elif entry.allocation == "percent_of_gross_revenue":
-            total += gross_revenue * entry.amount
-        elif entry.allocation == "per_active_member":
-            total += _convert_to_eur(entry.amount, entry.currency, fx_rates) * member_count
-    return _quantize(total)
-
-
-def revenue_for_period(period: str, revenue_entries: list[RevenueEntry]) -> RevenueEntry | None:
-    for entry in revenue_entries:
-        if entry.period == period:
-            return entry
+def payment_for_period(period: str, payments: list[PaymentRecord]) -> PaymentRecord | None:
+    for record in payments:
+        if record.period == period:
+            return record
     return None
 
 
@@ -81,16 +53,16 @@ def estimate_monthly_revenue(
     product: Product,
     models: list[PricingModel],
     members: list[MembershipRecord],
-    revenue_entries: list[RevenueEntry],
+    payments: list[PaymentRecord],
     monthly_costs: list[MonthlyPlatformCost],
 ) -> tuple[Decimal, Decimal, str]:
-    recorded = revenue_for_period(period, revenue_entries)
+    recorded = payment_for_period(period, payments)
     if recorded:
         return recorded.gross_amount, recorded.net_amount, recorded.source
 
     month = next((item for item in monthly_costs if item.period == period), None)
     if month and month.gross_revenue > 0:
-        return month.gross_revenue, month.gross_revenue, "derived_from_platform_history"
+        return month.gross_revenue, month.gross_revenue, "derived_from_ledger"
 
     model = active_pricing_model(models, product)
     count = active_members(members)
@@ -104,7 +76,7 @@ def periods_through(target: str, monthly_costs: list[MonthlyPlatformCost]) -> li
 
 def build_liquidity_summary(
     budget: Budget,
-    revenue_entries: list[RevenueEntry],
+    payments: list[PaymentRecord],
     monthly_costs: list[MonthlyPlatformCost],
     through_period: str,
 ) -> LiquiditySummary:
@@ -118,11 +90,9 @@ def build_liquidity_summary(
         month = cost_by_period[period]
         cumulative_infrastructure += month.infrastructure_cost
         cumulative_processing += month.payment_processing_cost
-        payment = revenue_for_period(period, revenue_entries)
+        payment = payment_for_period(period, payments)
         cumulative_payments += payment.net_amount if payment else Decimal("0")
 
-    # Liquidity uses net member payments vs infrastructure cash out only.
-    # Payment-processing fees are already deducted from net payments.
     cumulative_net = _quantize(cumulative_payments - cumulative_infrastructure)
     remaining = _quantize(budget.initial_budget + cumulative_net)
     cumulative_total = _quantize(cumulative_infrastructure + cumulative_processing)
@@ -147,13 +117,13 @@ def build_snapshot(
     product: Product,
     models: list[PricingModel],
     members: list[MembershipRecord],
-    revenue_entries: list[RevenueEntry],
+    payments: list[PaymentRecord],
     monthly_costs: list[MonthlyPlatformCost],
 ) -> EconomicsSnapshot:
     month = next(item for item in monthly_costs if item.period == period)
     count = month.active_members if month.active_members else active_members(members)
     gross_revenue, net_revenue, revenue_source = estimate_monthly_revenue(
-        period, product, models, members, revenue_entries, monthly_costs
+        period, product, models, members, payments, monthly_costs
     )
     infrastructure = month.infrastructure_cost
     processing = month.payment_processing_cost

projects/coulomb-pricing/observatory/ledger.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+from decimal import Decimal, ROUND_HALF_UP
+
+from .models import (
+    Budget,
+    ExpenseRecord,
+    MembershipRecord,
+    MonthlyPlatformCost,
+    PaymentRecord,
+)
+
+TWOPLACES = Decimal("0.01")
+
+
+def _quantize(value: Decimal) -> Decimal:
+    return value.quantize(TWOPLACES, rounding=ROUND_HALF_UP)
+
+
+def convert_to_eur(amount: Decimal, currency: str, fx_rates: dict[str, Decimal]) -> Decimal:
+    if currency == "EUR":
+        return amount
+    if currency == "USD":
+        return amount * fx_rates.get("USD/EUR", Decimal("1"))
+    raise ValueError(f"unsupported expense currency: {currency}")
+
+
+def aggregate_infrastructure_by_period(
+    expenses: list[ExpenseRecord],
+    fx_rates: dict[str, Decimal],
+    reporting_currency: str = "EUR",
+) -> dict[str, Decimal]:
+    if reporting_currency != "EUR":
+        raise ValueError("only EUR reporting currency is supported")
+
+    totals: dict[str, Decimal] = {}
+    for record in expenses:
+        if record.cost_class != "infrastructure":
+            continue
+        totals[record.period] = totals.get(record.period, Decimal("0")) + convert_to_eur(
+            record.amount, record.currency, fx_rates
+        )
+    return {period: _quantize(total) for period, total in totals.items()}
+
+
+def payment_processing_by_period(payments: list[PaymentRecord]) -> dict[str, Decimal]:
+    totals: dict[str, Decimal] = {}
+    for record in payments:
+        totals[record.period] = totals.get(record.period, Decimal("0")) + record.fees_amount
+    return {period: _quantize(total) for period, total in totals.items()}
+
+
+def _period_sort_key(period: str) -> tuple[int, int]:
+    year, month = period.split("-")
+    return int(year), int(month)
+
+
+def periods_from_budget_through_latest(
+    budget: Budget,
+    expenses: list[ExpenseRecord],
+    payments: list[PaymentRecord],
+) -> list[str]:
+    candidates = {budget.started}
+    candidates.update(record.period for record in expenses)
+    candidates.update(record.period for record in payments)
+    latest = max(candidates, key=_period_sort_key)
+    periods: list[str] = []
+    year, month = map(int, budget.started.split("-"))
+    end_year, end_month = map(int, latest.split("-"))
+    while (year, month) <= (end_year, end_month):
+        periods.append(f"{year}-{month:02d}")
+        month += 1
+        if month > 12:
+            month = 1
+            year += 1
+    return periods
+
+
+def active_members_for_period(period: str, members: list[MembershipRecord]) -> int:
+    period_start = f"{period}-01"
+    count = 0
+    for member in members:
+        if member.joined_at[:7] > period:
+            continue
+        if member.churned_at and member.churned_at[:7] < period:
+            continue
+        if member.status == "active" or (
+            member.churned_at and member.churned_at[:7] >= period
+        ):
+            count += 1
+    return count
+
+
+def build_monthly_ledger(
+    budget: Budget,
+    expenses: list[ExpenseRecord],
+    payments: list[PaymentRecord],
+    members: list[MembershipRecord],
+    fx_rates: dict[str, Decimal],
+) -> list[MonthlyPlatformCost]:
+    infrastructure = aggregate_infrastructure_by_period(expenses, fx_rates)
+    processing = payment_processing_by_period(payments)
+    payment_by_period = {record.period: record for record in payments}
+    rows: list[MonthlyPlatformCost] = []
+
+    for period in periods_from_budget_through_latest(budget, expenses, payments):
+        payment = payment_by_period.get(period)
+        rows.append(
+            MonthlyPlatformCost(
+                period=period,
+                infrastructure_cost=infrastructure.get(period, Decimal("0.00")),
+                payment_processing_cost=processing.get(period, Decimal("0.00")),
+                active_members=(
+                    payment.member_count
+                    if payment and payment.member_count
+                    else active_members_for_period(period, members)
+                ),
+                gross_revenue=payment.gross_amount if payment else Decimal("0.00"),
+            )
+        )
+    return rows

projects/coulomb-pricing/observatory/load.py

@@ -4,16 +4,15 @@ import json
 from decimal import Decimal
 from pathlib import Path
 
-_ZERO = Decimal("0")
-
+from .ledger import build_monthly_ledger
 from .models import (
     Budget,
-    CostEntry,
+    ExpenseRecord,
     MembershipRecord,
     MonthlyPlatformCost,
+    PaymentRecord,
     PricingModel,
     Product,
-    RevenueEntry,
 )
 
 
@@ -67,55 +66,38 @@ def load_pricing_models(data_dir: Path | None = None) -> list[PricingModel]:
     ]
 
 
-def _parse_cost_entries(items: list[dict]) -> list[CostEntry]:
+def load_fx_rates(data_dir: Path | None = None) -> dict[str, Decimal]:
+    raw = _read_json((data_dir or default_data_dir()) / "expense_records.json")
+    return {pair: _money(rate) for pair, rate in raw.get("fx_rates", {}).items()}
+
+
+def load_expense_records(data_dir: Path | None = None) -> list[ExpenseRecord]:
+    raw = _read_json((data_dir or default_data_dir()) / "expense_records.json")
     return [
-        CostEntry(
+        ExpenseRecord(
             id=item["id"],
-            name=item["name"],
-            category=item["category"],
+            period=item["period"],
+            vendor=item["vendor"],
+            description=item["description"],
+            cost_class=item["cost_class"],
             amount=_money(item["amount"]),
             currency=item["currency"],
-            cadence=item["cadence"],
-            allocation=item["allocation"],
+            source=item["source"],
         )
-        for item in items
+        for item in raw["records"]
     ]
 
 
-def load_cost_rate_card(data_dir: Path | None = None) -> tuple[list[CostEntry], dict[str, Decimal]]:
-    raw = _read_json((data_dir or default_data_dir()) / "costs.json")
-    items = raw.get("rate_card", raw.get("entries", []))
-    fx = {pair: _money(rate) for pair, rate in raw.get("fx_rates", {}).items()}
-    return _parse_cost_entries(items), fx
-
-
-def load_monthly_platform_costs(data_dir: Path | None = None) -> list[MonthlyPlatformCost]:
-    raw = _read_json((data_dir or default_data_dir()) / "costs.json")
-    rows: list[MonthlyPlatformCost] = []
-    for item in raw.get("monthly_history", []):
-        if "infrastructure_cost" in item:
-            infrastructure = _money(item["infrastructure_cost"])
-            processing = _money(item.get("payment_processing_cost", "0"))
-        else:
-            # Legacy single platform_cost field treated as infrastructure only.
-            infrastructure = _money(item["platform_cost"])
-            processing = _ZERO
-        rows.append(
-            MonthlyPlatformCost(
-                period=item["period"],
-                infrastructure_cost=infrastructure,
-                payment_processing_cost=processing,
-                active_members=item["active_members"],
-                gross_revenue=_money(item["gross_revenue"]),
-            )
-        )
-    return rows
-
-
-def load_revenue(data_dir: Path | None = None) -> list[RevenueEntry]:
-    raw = _read_json((data_dir or default_data_dir()) / "revenue.json")
+def load_payment_records(data_dir: Path | None = None) -> list[PaymentRecord]:
+    root = data_dir or default_data_dir()
+    path = root / "payment_records.json"
+    if not path.exists():
+        # Backward compatibility with legacy revenue.json
+        path = root / "revenue.json"
+    raw = _read_json(path)
+    items = raw.get("records", raw.get("entries", []))
     return [
-        RevenueEntry(
+        PaymentRecord(
             id=item["id"],
             period=item["period"],
             gross_amount=_money(item["gross_amount"]),
@@ -124,8 +106,9 @@ def load_revenue(data_dir: Path | None = None) -> list[RevenueEntry]:
             net_amount=_money(item["net_amount"]),
             currency=item["currency"],
             source=item["source"],
+            member_count=item.get("member_count", 0),
         )
-        for item in raw["entries"]
+        for item in items
     ]
 
 
@@ -143,5 +126,16 @@ def load_membership(data_dir: Path | None = None) -> list[MembershipRecord]:
     ]
 
 
+def load_monthly_ledger(data_dir: Path | None = None) -> list[MonthlyPlatformCost]:
+    root = data_dir or default_data_dir()
+    return build_monthly_ledger(
+        load_budget(root),
+        load_expense_records(root),
+        load_payment_records(root),
+        load_membership(root),
+        load_fx_rates(root),
+    )
+
+
 def latest_period(monthly_costs: list[MonthlyPlatformCost]) -> str:
     return max(item.period for item in monthly_costs)

projects/coulomb-pricing/observatory/models.py

@@ -4,7 +4,7 @@ from dataclasses import dataclass
 from decimal import Decimal
 from typing import Literal
 
-CostCategory = Literal["fixed", "variable"]
+ExpenseClass = Literal["infrastructure", "payment_processing"]
 MemberStatus = Literal["active", "churned", "paused"]
 PricingModelStatus = Literal["active", "candidate", "retired"]
 LiquidityStatus = Literal["burning", "neutral", "generating"]
@@ -33,14 +33,28 @@ class PricingModel:
 
 
 @dataclass(frozen=True)
-class CostEntry:
+class ExpenseRecord:
     id: str
-    name: str
-    category: CostCategory
+    period: str
+    vendor: str
+    description: str
+    cost_class: ExpenseClass
     amount: Decimal
     currency: str
-    cadence: str
-    allocation: str
+    source: str
+
+
+@dataclass(frozen=True)
+class PaymentRecord:
+    id: str
+    period: str
+    gross_amount: Decimal
+    fees_amount: Decimal
+    refunds_amount: Decimal
+    net_amount: Decimal
+    currency: str
+    source: str
+    member_count: int = 0
 
 
 @dataclass(frozen=True)
@@ -63,18 +77,6 @@ class Budget:
     started: str
 
 
-@dataclass(frozen=True)
-class RevenueEntry:
-    id: str
-    period: str
-    gross_amount: Decimal
-    fees_amount: Decimal
-    refunds_amount: Decimal
-    net_amount: Decimal
-    currency: str
-    source: str
-
-
 @dataclass(frozen=True)
 class MembershipRecord:
     id: str