Files
adaptive-pricing/projects/coulomb-pricing/observatory/pricing_context.py
2026-06-22 23:05:05 +02:00

93 lines
3.1 KiB
Python

from __future__ import annotations
from decimal import Decimal, ROUND_HALF_UP
from .economics import active_pricing_model
from .models import EconomicsSnapshot, PricingModel, Product
TWOPLACES = Decimal("0.01")
def _quantize(value: Decimal) -> Decimal:
return value.quantize(TWOPLACES, rounding=ROUND_HALF_UP)
def build_cost_floor(
snapshot: EconomicsSnapshot,
models: list[PricingModel],
) -> dict:
active = next((m for m in models if m.status == "active"), None)
return {
"period": snapshot.period,
"currency": snapshot.currency,
"monthly_infrastructure_cost": snapshot.monthly_infrastructure_cost,
"monthly_payment_processing_cost": snapshot.monthly_payment_processing_cost,
"monthly_total_platform_cost": snapshot.monthly_total_platform_cost,
"cost_per_member": snapshot.cost_per_member,
"active_members": snapshot.active_members,
"monthly_revenue": snapshot.monthly_revenue,
"gross_margin": snapshot.gross_margin,
"gross_margin_pct": snapshot.gross_margin_pct,
"active_price": active.access_fee_amount if active else Decimal("0"),
"active_model_id": active.id if active else None,
"active_model_name": active.name if active else None,
}
def build_value_range_view(
raw: dict,
snapshot: EconomicsSnapshot,
product: Product,
models: list[PricingModel],
) -> dict:
model = active_pricing_model(models, product)
current = model.access_fee_amount
segments = []
lows: list[Decimal] = []
highs: list[Decimal] = []
for item in raw.get("segments", []):
low = Decimal(str(item["low_eur"]))
high = Decimal(str(item["high_eur"]))
lows.append(low)
highs.append(high)
segments.append(
{
**item,
"headroom_to_high_eur": _quantize(high - current),
"below_floor": current < low,
}
)
aggregate_low = min(lows) if lows else current
aggregate_high = max(highs) if highs else current
return {
"currency": raw.get("currency", snapshot.currency),
"product_id": raw.get("product_id", product.id),
"current_price_eur": current,
"aggregate_low_eur": aggregate_low,
"aggregate_high_eur": aggregate_high,
"cost_per_member_eur": snapshot.cost_per_member,
"segments": segments,
"value_drivers": raw.get("value_drivers", []),
"notes": raw.get("notes", ""),
}
def build_market_price_view(raw: dict) -> dict:
alternatives = list(raw.get("alternatives", []))
prices = [
Decimal(str(item["price_monthly_eur"]))
for item in alternatives
if item.get("price_monthly_eur") not in (None, "", "0", "0.00")
]
return {
"currency": raw.get("currency", "EUR"),
"last_reviewed": raw.get("last_reviewed"),
"alternative_count": len(alternatives),
"priced_alternative_count": len(prices),
"market_low_eur": min(prices) if prices else None,
"market_high_eur": max(prices) if prices else None,
"alternatives": alternatives,
"notes": raw.get("notes", ""),
}