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

from __future__ import annotations

from decimal import Decimal, ROUND_HALF_UP

from .models import (
    Budget,
    EconomicsSnapshot,
    LiquidityStatus,
    LiquiditySummary,
    MembershipRecord,
    MonthlyPlatformCost,
    PaymentRecord,
    PricingModel,
    Product,
)

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 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 payment_for_period(period: str, payments: list[PaymentRecord]) -> PaymentRecord | None:
    for record in payments:
        if record.period == period:
            return record
    return None


def estimate_monthly_revenue(
    period: str,
    product: Product,
    models: list[PricingModel],
    members: list[MembershipRecord],
    payments: list[PaymentRecord],
    monthly_costs: list[MonthlyPlatformCost],
) -> tuple[Decimal, Decimal, str]:
    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_ledger"

    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,
    payments: list[PaymentRecord],
    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_infrastructure = Decimal("0")
    cumulative_processing = Decimal("0")
    for period in tracked:
        month = cost_by_period[period]
        cumulative_infrastructure += month.infrastructure_cost
        cumulative_processing += month.payment_processing_cost
        payment = payment_for_period(period, payments)
        cumulative_payments += payment.net_amount if payment else Decimal("0")

    cumulative_net = _quantize(cumulative_payments - cumulative_infrastructure)
    remaining = _quantize(budget.initial_budget + cumulative_net)
    cumulative_total = _quantize(cumulative_infrastructure + cumulative_processing)

    return LiquiditySummary(
        currency=budget.currency,
        through_period=through_period,
        initial_budget=budget.initial_budget,
        cumulative_member_payments=_quantize(cumulative_payments),
        cumulative_infrastructure_cost=_quantize(cumulative_infrastructure),
        cumulative_payment_processing_cost=_quantize(cumulative_processing),
        cumulative_total_platform_cost=cumulative_total,
        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],
    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, payments, monthly_costs
    )
    infrastructure = month.infrastructure_cost
    processing = month.payment_processing_cost
    total_platform = month.total_platform_cost
    cost_per_member = _quantize(total_platform / count) if count else Decimal("0.00")
    gross_margin = _quantize(gross_revenue - total_platform)
    margin_pct = (
        _quantize((gross_margin / gross_revenue) * Decimal("100"), PCTPLACES)
        if gross_revenue
        else Decimal("-100.0") if total_platform else Decimal("0.0")
    )
    period_net = _quantize(net_revenue - infrastructure)

    return EconomicsSnapshot(
        period=period,
        currency=product.currency,
        active_members=count,
        monthly_revenue=_quantize(gross_revenue),
        monthly_infrastructure_cost=infrastructure,
        monthly_payment_processing_cost=processing,
        monthly_total_platform_cost=total_platform,
        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),
    )