adaptive-pricing/projects/coulomb-pricing/observatory/economics.py

from __future__ import annotations

from decimal import Decimal, ROUND_HALF_UP

from .models import (
    Budget,
    CostEntry,
    EconomicsSnapshot,
    LiquidityStatus,
    LiquiditySummary,
    MembershipRecord,
    MonthlyPlatformCost,
    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 liquidity_status_for(net: Decimal) -> LiquidityStatus:
    if net < Decimal("0"):
        return "burning"
    if net > Decimal("0"):
        return "generating"
    return "neutral"


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 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
    return None


def estimate_monthly_revenue(
    period: str,
    product: Product,
    models: list[PricingModel],
    members: list[MembershipRecord],
    revenue_entries: list[RevenueEntry],
    monthly_costs: list[MonthlyPlatformCost],
) -> tuple[Decimal, Decimal, str]:
    recorded = revenue_for_period(period, revenue_entries)
    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"

    model = active_pricing_model(models, product)
    count = active_members(members)
    gross = model.access_fee_amount * count
    return gross, gross, "derived_from_membership"


def periods_through(target: str, monthly_costs: list[MonthlyPlatformCost]) -> list[str]:
    return sorted(item.period for item in monthly_costs if item.period <= target)


def build_liquidity_summary(
    budget: Budget,
    revenue_entries: list[RevenueEntry],
    monthly_costs: list[MonthlyPlatformCost],
    through_period: str,
) -> LiquiditySummary:
    tracked = periods_through(through_period, monthly_costs)
    cost_by_period = {item.period: item for item in monthly_costs}

    cumulative_payments = Decimal("0")
    cumulative_cost = Decimal("0")
    for period in tracked:
        month = cost_by_period[period]
        cumulative_cost += month.platform_cost
        payment = revenue_for_period(period, revenue_entries)
        cumulative_payments += payment.net_amount if payment else Decimal("0")

    cumulative_net = _quantize(cumulative_payments - cumulative_cost)
    remaining = _quantize(budget.initial_budget + cumulative_net)

    return LiquiditySummary(
        currency=budget.currency,
        through_period=through_period,
        initial_budget=budget.initial_budget,
        cumulative_member_payments=_quantize(cumulative_payments),
        cumulative_platform_cost=_quantize(cumulative_cost),
        cumulative_net_liquidity=cumulative_net,
        remaining_budget=remaining,
        liquidity_status=liquidity_status_for(cumulative_net),
        months_tracked=len(tracked),
    )


def build_snapshot(
    period: str,
    product: Product,
    models: list[PricingModel],
    members: list[MembershipRecord],
    revenue_entries: list[RevenueEntry],
    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
    )
    platform_cost = month.platform_cost
    cost_per_member = _quantize(platform_cost / count) if count else Decimal("0.00")
    gross_margin = _quantize(gross_revenue - platform_cost)
    margin_pct = (
        _quantize((gross_margin / gross_revenue) * Decimal("100"), PCTPLACES)
        if gross_revenue
        else Decimal("-100.0") if platform_cost else Decimal("0.0")
    )
    period_net = _quantize(net_revenue - platform_cost)

    return EconomicsSnapshot(
        period=period,
        currency=product.currency,
        active_members=count,
        monthly_revenue=_quantize(gross_revenue),
        monthly_platform_cost=platform_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,
        period_net_liquidity=period_net,
        liquidity_status=liquidity_status_for(period_net),
    )