coulomb/adaptive-pricing
2
0
Fork 0
generated from coulomb/repo-seed
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement governance workflows and close WP-0008

Browse Source
This commit is contained in:
codex
2026-07-03 01:27:37 +02:00
parent a76e57ba89
commit a9a55e19f1
11 changed files with 1177 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
projects/coulomb-pricing/observatory/recommendations.py
View File

@@ -1,68 +1,33 @@
from __future__ import annotations
from decimal import Decimal
from typing import Any
from .governance import build_pricing_recommendation_workflow, build_governance_policy
def build_pricing_recommendations(
cost_floor: dict,
value_range: dict,
market_price: dict,
simulations: dict,
usage_summary: dict,
) -> list[dict]:
recommendations: list[dict] = []
margin_pct = Decimal(str(cost_floor.get("gross_margin_pct", "0")))
ai_spend = Decimal(str(usage_summary.get("total_ai_spend_eur", "0")))
active_price = Decimal(str(value_range.get("current_price_eur", "0")))
cost_per_member = Decimal(str(cost_floor.get("cost_per_member", "0")))
if margin_pct < Decimal("10"):
recommendations.append(
{
"id": "margin-pressure",
"priority": "high",
"title": "Margin below 10%",
"rationale": f"Gross margin is {margin_pct}% at current pricing.",
"suggested_action": "Review infrastructure cost or test a higher access fee within value-range bands.",
}
)
if ai_spend > Decimal("0") and cost_per_member > Decimal("0"):
ai_ratio = (ai_spend / cost_per_member) * Decimal("100")
if ai_ratio > Decimal("15"):
best = simulations.get("best_ltv_scenario_id") or simulations.get("best_margin_scenario_id")
recommendations.append(
{
"id": "usage-pricing-signal",
"priority": "medium",
"title": "AI cost becoming material",
"rationale": f"AI spend is {ai_ratio:.1f}% of cost-per-member this period.",
"suggested_action": f"Evaluate hybrid model '{best}' in the pricing simulator before customer-visible credits.",
}
)
if market_price.get("market_high_eur") and active_price < Decimal(str(market_price["market_high_eur"])):
headroom = Decimal(str(value_range.get("aggregate_high_eur", active_price))) - active_price
if headroom > Decimal("5"):
recommendations.append(
{
"id": "value-headroom",
"priority": "low",
"title": "Value headroom above list price",
"rationale": f"Aggregate value band high is €{value_range['aggregate_high_eur']} vs €{active_price} list.",
"suggested_action": "Run a staged price experiment within the solo-builder segment band.",
}
)
if not recommendations:
recommendations.append(
{
"id": "hold-course",
"priority": "low",
"title": "Hold current pricing",
"rationale": "No urgent margin, usage, or competitive signals detected.",
"suggested_action": "Continue observatory tracking; re-run after next ledger period.",
}
)
return recommendations
cost_floor: dict[str, Any],
value_range: dict[str, Any],
market_price: dict[str, Any],
simulations: dict[str, Any],
usage_summary: dict[str, Any],
*,
boundary_validation: dict[str, Any] | None = None,
customer_tuning: dict[str, Any] | None = None,
provider_publication: dict[str, Any] | None = None,
governance_policy: dict[str, Any] | None = None,
product: Any | None = None,
) -> list[dict[str, Any]]:
policy = build_governance_policy(governance_policy or {})
return build_pricing_recommendation_workflow(
cost_floor,
value_range,
market_price,
simulations,
usage_summary,
boundary_validation=boundary_validation,
customer_tuning=customer_tuning or {},
provider_publication=provider_publication or {},
governance_policy=policy,
product=product,
)