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

from __future__ import annotations

import argparse
from decimal import Decimal
from pathlib import Path

from .economics import build_liquidity_summary, build_snapshot
from .load import (
    default_data_dir,
    latest_period,
    load_budget,
    load_membership,
    load_monthly_platform_costs,
    load_pricing_models,
    load_product,
    load_revenue,
)
from .models import EconomicsSnapshot, LiquiditySummary, MonthlyPlatformCost, PricingModel, Product, RevenueEntry


def _history_rows(
    monthly_costs: list[MonthlyPlatformCost],
    revenue_entries: list[RevenueEntry],
    through_period: str,
) -> str:
    revenue_by_period = {entry.period: entry for entry in revenue_entries}
    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)
        net_payment = payment.net_amount if payment else Decimal("0")
        period_net = net_payment - month.infrastructure_cost
        lines.append(
            f"| {month.period} | {month.active_members} | {month.gross_revenue} | "
            f"{month.infrastructure_cost} | {month.payment_processing_cost} | "
            f"{month.total_platform_cost} | {period_net:.2f} |"
        )
    return "\n".join(lines)


def render_dashboard(
    product: Product,
    models: list[PricingModel],
    snapshot: EconomicsSnapshot,
    liquidity: LiquiditySummary,
    monthly_costs: list[MonthlyPlatformCost],
    revenue_entries: list[RevenueEntry],
) -> 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)
    budget_state = (
        "over budget"
        if liquidity.remaining_budget < Decimal("0")
        else "within budget"
    )

    return f"""# Economics Dashboard v1 — {product.name}

**Period:** {snapshot.period}  
**Lifecycle phase:** {product.lifecycle_phase}  
**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.

## Key Metrics (current period)

| Metric | Value |
|--------|------:|
| Active members | {snapshot.active_members} |
| Member payments (gross) | {snapshot.monthly_revenue} {snapshot.currency} |
| Infrastructure cost | {snapshot.monthly_infrastructure_cost} {snapshot.currency} |
| Payment processing cost | {snapshot.monthly_payment_processing_cost} {snapshot.currency} |
| Total platform cost | {snapshot.monthly_total_platform_cost} {snapshot.currency} |
| Platform cost per member | {snapshot.cost_per_member} {snapshot.currency} |
| Period gross margin | {snapshot.gross_margin} {snapshot.currency} |
| Period gross margin % | {snapshot.gross_margin_pct}% |
| Period net liquidity | {snapshot.period_net_liquidity} {snapshot.currency} ({snapshot.liquidity_status}) |

_Period net liquidity = net member payments − infrastructure cost (processing fees already netted from payments)._  
_Revenue source: {snapshot.revenue_source}_

## Liquidity & Budget (through {liquidity.through_period})

| Metric | Value |
|--------|------:|
| Initial budget | {liquidity.initial_budget} {liquidity.currency} |
| Cumulative member payments (net) | {liquidity.cumulative_member_payments} {liquidity.currency} |
| Cumulative infrastructure cost | {liquidity.cumulative_infrastructure_cost} {liquidity.currency} |
| Cumulative payment processing | {liquidity.cumulative_payment_processing_cost} {liquidity.currency} |
| Cumulative total platform cost | {liquidity.cumulative_total_platform_cost} {liquidity.currency} |
| Cumulative net liquidity | {liquidity.cumulative_net_liquidity} {liquidity.currency} ({liquidity.liquidity_status}) |
| Remaining budget | {liquidity.remaining_budget} {liquidity.currency} ({budget_state}) |
| Months tracked | {liquidity.months_tracked} |

## Monthly History

| Period | Members | Gross revenue | Infrastructure | Processing | Total platform | Net liquidity |
|--------|--------:|--------------:|---------------:|-----------:|---------------:|--------------:|
{history_lines}

## Pricing Model Registry

| ID | Name | Type | Status |
|----|------|------|--------|
{registry_lines}

## Registries Loaded

- 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`)
- Membership (`data/membership.json`)
- Requirements (`REQUIREMENTS.md`)
"""


def generate_dashboard(data_dir: Path | None = None, period: str | None = None) -> str:
    root = data_dir or default_data_dir()
    product = load_product(root)
    budget = load_budget(root)
    models = load_pricing_models(root)
    members = load_membership(root)
    revenue = load_revenue(root)
    monthly_costs = load_monthly_platform_costs(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)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Coulomb Social Economics Dashboard v1")
    parser.add_argument("--data-dir", type=Path, default=None, help="Registry data directory")
    parser.add_argument("--period", default=None, help="Reporting period (YYYY-MM)")
    parser.add_argument(
        "--output",
        type=Path,
        default=None,
        help="Write Markdown report to this path (default: stdout only)",
    )
    args = parser.parse_args(argv)

    report = generate_dashboard(args.data_dir, args.period)
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(report, encoding="utf-8")
        print(f"Wrote {args.output}")
    else:
        print(report)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())