generated from coulomb/repo-seed
tegwick
a1a4aa972f
Add Coulomb observatory package with JSON registries (product, pricing models, costs, revenue, membership), economics snapshot engine, Economics Dashboard v1 CLI, sample report, and pytest coverage. Complete T01 and queue Sprint 2 Bubble.io integration.
109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
|
|
from .models import (
|
|
CostEntry,
|
|
EconomicsSnapshot,
|
|
MembershipRecord,
|
|
PricingModel,
|
|
Product,
|
|
RevenueEntry,
|
|
)
|
|
|
|
TWOPLACES = Decimal("0.01")
|
|
PCTPLACES = Decimal("0.1")
|
|
|
|
|
|
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 active_members(members: list[MembershipRecord]) -> int:
|
|
return sum(1 for member in members if member.status == "active")
|
|
|
|
|
|
def active_pricing_model(models: list[PricingModel], product: Product) -> PricingModel:
|
|
for model in models:
|
|
if model.id == product.active_pricing_model_id:
|
|
return model
|
|
raise ValueError(f"active pricing model not found: {product.active_pricing_model_id}")
|
|
|
|
|
|
def estimate_monthly_revenue(
|
|
period: str,
|
|
product: Product,
|
|
models: list[PricingModel],
|
|
members: list[MembershipRecord],
|
|
revenue_entries: list[RevenueEntry],
|
|
) -> tuple[Decimal, str]:
|
|
for entry in revenue_entries:
|
|
if entry.period == period and entry.currency == product.currency:
|
|
return entry.gross_amount, entry.source
|
|
|
|
model = active_pricing_model(models, product)
|
|
count = active_members(members)
|
|
return model.access_fee_amount * count, "derived_from_membership"
|
|
|
|
|
|
def monthly_cost_total(
|
|
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 build_snapshot(
|
|
period: str,
|
|
product: Product,
|
|
models: list[PricingModel],
|
|
members: list[MembershipRecord],
|
|
revenue_entries: list[RevenueEntry],
|
|
cost_entries: list[CostEntry],
|
|
fx_rates: dict[str, Decimal],
|
|
) -> EconomicsSnapshot:
|
|
count = active_members(members)
|
|
gross_revenue, revenue_source = estimate_monthly_revenue(
|
|
period, product, models, members, revenue_entries
|
|
)
|
|
monthly_cost = monthly_cost_total(cost_entries, fx_rates, gross_revenue, count)
|
|
cost_per_member = _quantize(monthly_cost / count) if count else Decimal("0.00")
|
|
gross_margin = _quantize(gross_revenue - monthly_cost)
|
|
margin_pct = (
|
|
_quantize((gross_margin / gross_revenue) * Decimal("100"), PCTPLACES)
|
|
if gross_revenue
|
|
else Decimal("0.0")
|
|
)
|
|
|
|
return EconomicsSnapshot(
|
|
period=period,
|
|
currency=product.currency,
|
|
active_members=count,
|
|
monthly_revenue=_quantize(gross_revenue),
|
|
monthly_cost=monthly_cost,
|
|
cost_per_member=cost_per_member,
|
|
gross_margin=gross_margin,
|
|
gross_margin_pct=margin_pct,
|
|
pricing_model_count=len(models),
|
|
revenue_source=revenue_source,
|
|
) |