|
| 1 | +"""Cost simulator — what-if scenarios for vstack diagnostics.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +from dataclasses import dataclass, field |
| 6 | +from typing import Any |
| 7 | + |
| 8 | + |
| 9 | +@dataclass |
| 10 | +class PerPatternCost: |
| 11 | + """Per-pattern × mode cost in USD.""" |
| 12 | + |
| 13 | + pattern: str |
| 14 | + quick_usd: float = 0.02 |
| 15 | + standard_usd: float = 0.05 |
| 16 | + forensic_usd: float = 0.50 |
| 17 | + |
| 18 | + |
| 19 | +# Default pricing table, calibrated against typical Anthropic flagship-model usage. |
| 20 | +# Per-call cost in USD; override via Scenario(custom_pricing=...) for your provider. |
| 21 | +_DEFAULT_PRICING: dict[str, PerPatternCost] = { |
| 22 | + "lewin": PerPatternCost("lewin", 0.02, 0.05, 0.55), |
| 23 | + "goleman_ei": PerPatternCost("goleman_ei", 0.02, 0.05, 0.50), |
| 24 | + "johari": PerPatternCost("johari", 0.02, 0.05, 0.45), |
| 25 | + "danva_emotion": PerPatternCost("danva_emotion", 0.02, 0.05, 0.35), |
| 26 | + "cognitive_reappraisal": PerPatternCost("cognitive_reappraisal", 0.02, 0.05, 0.35), |
| 27 | + "yerkes_dodson": PerPatternCost("yerkes_dodson", 0.02, 0.05, 0.55), |
| 28 | + "hexaco": PerPatternCost("hexaco", 0.02, 0.04, 0.40), |
| 29 | + "grant_strengths": PerPatternCost("grant_strengths", 0.02, 0.04, 0.40), |
| 30 | + "motivation_traps": PerPatternCost("motivation_traps", 0.02, 0.04, 0.40), |
| 31 | + "sdt_reward": PerPatternCost("sdt_reward", 0.02, 0.04, 0.40), |
| 32 | + "mcgregor": PerPatternCost("mcgregor", 0.02, 0.04, 0.40), |
| 33 | + "vroom_expectancy": PerPatternCost("vroom_expectancy", 0.02, 0.04, 0.40), |
| 34 | + "grpi": PerPatternCost("grpi", 0.03, 0.06, 0.55), |
| 35 | + "process_gain_loss": PerPatternCost("process_gain_loss", 0.03, 0.06, 0.55), |
| 36 | + "social_loafing": PerPatternCost("social_loafing", 0.02, 0.05, 0.40), |
| 37 | + "superflocks": PerPatternCost("superflocks", 0.02, 0.05, 0.40), |
| 38 | + "lencioni": PerPatternCost("lencioni", 0.03, 0.06, 0.55), |
| 39 | + "trust_triangle": PerPatternCost("trust_triangle", 0.03, 0.06, 0.55), |
| 40 | + "mcallister_trust": PerPatternCost("mcallister_trust", 0.02, 0.05, 0.40), |
| 41 | + "psych_safety": PerPatternCost("psych_safety", 0.02, 0.05, 0.45), |
| 42 | + "glaser_conversation": PerPatternCost("glaser_conversation", 0.02, 0.05, 0.40), |
| 43 | + "feedback_triggers": PerPatternCost("feedback_triggers", 0.02, 0.05, 0.55), |
| 44 | + "plus_delta": PerPatternCost("plus_delta", 0.02, 0.04, 0.35), |
| 45 | + "smart_goal": PerPatternCost("smart_goal", 0.02, 0.04, 0.35), |
| 46 | + "group_decision": PerPatternCost("group_decision", 0.02, 0.04, 0.35), |
| 47 | + "debate_pathology": PerPatternCost("debate_pathology", 0.02, 0.05, 0.55), |
| 48 | + "bias_stack": PerPatternCost("bias_stack", 0.02, 0.05, 0.55), |
| 49 | + "devils_advocate": PerPatternCost("devils_advocate", 0.02, 0.05, 0.40), |
| 50 | + "thomas_kilmann": PerPatternCost("thomas_kilmann", 0.02, 0.05, 0.40), |
| 51 | + "aar": PerPatternCost("aar", 0.02, 0.05, 0.55), |
| 52 | + "schein_culture": PerPatternCost("schein_culture", 0.03, 0.06, 0.55), |
| 53 | + "robbins_culture": PerPatternCost("robbins_culture", 0.02, 0.04, 0.40), |
| 54 | + "org_structure": PerPatternCost("org_structure", 0.02, 0.04, 0.40), |
| 55 | + "span_of_control": PerPatternCost("span_of_control", 0.02, 0.04, 0.35), |
| 56 | +} |
| 57 | + |
| 58 | + |
| 59 | +def baseline_pricing() -> dict[str, PerPatternCost]: |
| 60 | + """Return a copy of the baseline pricing table.""" |
| 61 | + return { |
| 62 | + k: PerPatternCost(v.pattern, v.quick_usd, v.standard_usd, v.forensic_usd) |
| 63 | + for k, v in _DEFAULT_PRICING.items() |
| 64 | + } |
| 65 | + |
| 66 | + |
| 67 | +@dataclass |
| 68 | +class Scenario: |
| 69 | + """A what-if cost scenario.""" |
| 70 | + |
| 71 | + name: str = "scenario" |
| 72 | + traces_per_day: int = 0 |
| 73 | + sample_rate: float = 1.0 |
| 74 | + """Fraction of traces actually diagnosed (0.0-1.0).""" |
| 75 | + |
| 76 | + mode: str = "standard" |
| 77 | + """quick / standard / forensic""" |
| 78 | + |
| 79 | + patterns: list[str] = field(default_factory=list) |
| 80 | + """Patterns to run per sampled trace.""" |
| 81 | + |
| 82 | + custom_pricing: dict[str, PerPatternCost] | None = None |
| 83 | + |
| 84 | + failure_upgrade: bool = False |
| 85 | + """If True, failed traces (10% assumed) get re-run in forensic mode.""" |
| 86 | + |
| 87 | + failure_rate_assumed: float = 0.10 |
| 88 | + """Assumed rate of failures that trigger forensic upgrade.""" |
| 89 | + |
| 90 | + def get_pricing(self) -> dict[str, PerPatternCost]: |
| 91 | + return self.custom_pricing or _DEFAULT_PRICING |
| 92 | + |
| 93 | + def cost_per_pattern(self, pattern: str, *, mode: str | None = None) -> float: |
| 94 | + pricing = self.get_pricing() |
| 95 | + mode = mode or self.mode |
| 96 | + entry = pricing.get(pattern) |
| 97 | + if entry is None: |
| 98 | + return 0.0 |
| 99 | + return getattr(entry, f"{mode}_usd", entry.standard_usd) |
| 100 | + |
| 101 | + def cost_per_trace(self) -> float: |
| 102 | + """Cost per single diagnosed trace, accounting for failure upgrade.""" |
| 103 | + base = sum(self.cost_per_pattern(p) for p in self.patterns) |
| 104 | + if self.failure_upgrade: |
| 105 | + forensic_cost = sum(self.cost_per_pattern(p, mode="forensic") for p in self.patterns) |
| 106 | + base += self.failure_rate_assumed * forensic_cost |
| 107 | + return base |
| 108 | + |
| 109 | + def diagnosed_per_day(self) -> int: |
| 110 | + return int(self.traces_per_day * self.sample_rate) |
| 111 | + |
| 112 | + |
| 113 | +@dataclass |
| 114 | +class SimulationResult: |
| 115 | + """Result of simulating a scenario.""" |
| 116 | + |
| 117 | + scenario: Scenario |
| 118 | + daily_diagnosed: int |
| 119 | + daily_cost_usd: float |
| 120 | + monthly_cost_usd: float |
| 121 | + annual_cost_usd: float |
| 122 | + cost_per_pattern: dict[str, float] = field(default_factory=dict) |
| 123 | + cost_per_trace: float = 0.0 |
| 124 | + |
| 125 | + def to_dict(self) -> dict[str, Any]: |
| 126 | + return { |
| 127 | + "scenario_name": self.scenario.name, |
| 128 | + "daily_diagnosed": self.daily_diagnosed, |
| 129 | + "daily_cost_usd": self.daily_cost_usd, |
| 130 | + "monthly_cost_usd": self.monthly_cost_usd, |
| 131 | + "annual_cost_usd": self.annual_cost_usd, |
| 132 | + "cost_per_trace": self.cost_per_trace, |
| 133 | + "cost_per_pattern": dict(self.cost_per_pattern), |
| 134 | + } |
| 135 | + |
| 136 | + |
| 137 | +def simulate(scenario: Scenario) -> SimulationResult: |
| 138 | + """Run the scenario; return the cost projection.""" |
| 139 | + diagnosed = scenario.diagnosed_per_day() |
| 140 | + per_trace = scenario.cost_per_trace() |
| 141 | + |
| 142 | + daily = diagnosed * per_trace |
| 143 | + monthly = daily * 30 |
| 144 | + annual = daily * 365 |
| 145 | + |
| 146 | + per_pattern: dict[str, float] = {} |
| 147 | + for p in scenario.patterns: |
| 148 | + per_pattern[p] = diagnosed * scenario.cost_per_pattern(p) |
| 149 | + if scenario.failure_upgrade: |
| 150 | + per_pattern[p] += ( |
| 151 | + diagnosed |
| 152 | + * scenario.failure_rate_assumed |
| 153 | + * scenario.cost_per_pattern(p, mode="forensic") |
| 154 | + ) |
| 155 | + |
| 156 | + return SimulationResult( |
| 157 | + scenario=scenario, |
| 158 | + daily_diagnosed=diagnosed, |
| 159 | + daily_cost_usd=daily, |
| 160 | + monthly_cost_usd=monthly, |
| 161 | + annual_cost_usd=annual, |
| 162 | + cost_per_pattern=per_pattern, |
| 163 | + cost_per_trace=per_trace, |
| 164 | + ) |
| 165 | + |
| 166 | + |
| 167 | +@dataclass |
| 168 | +class ScenarioComparison: |
| 169 | + """Compare multiple scenarios.""" |
| 170 | + |
| 171 | + results: list[SimulationResult] = field(default_factory=list) |
| 172 | + |
| 173 | + @property |
| 174 | + def cheapest(self) -> SimulationResult | None: |
| 175 | + if not self.results: |
| 176 | + return None |
| 177 | + return min(self.results, key=lambda r: r.daily_cost_usd) |
| 178 | + |
| 179 | + @property |
| 180 | + def most_expensive(self) -> SimulationResult | None: |
| 181 | + if not self.results: |
| 182 | + return None |
| 183 | + return max(self.results, key=lambda r: r.daily_cost_usd) |
| 184 | + |
| 185 | + def to_dict(self) -> dict[str, Any]: |
| 186 | + return {"results": [r.to_dict() for r in self.results]} |
| 187 | + |
| 188 | + def to_markdown(self) -> str: |
| 189 | + lines = ["# Scenario Comparison", ""] |
| 190 | + if not self.results: |
| 191 | + lines.append("_No scenarios._") |
| 192 | + return "\n".join(lines) |
| 193 | + |
| 194 | + lines.append("| Scenario | Diagnosed/day | $/trace | $/day | $/month | $/year |") |
| 195 | + lines.append("|----------|---------------:|--------:|------:|--------:|-------:|") |
| 196 | + for r in self.results: |
| 197 | + lines.append( |
| 198 | + f"| {r.scenario.name} | {r.daily_diagnosed:,} | " |
| 199 | + f"${r.cost_per_trace:.4f} | ${r.daily_cost_usd:,.2f} | " |
| 200 | + f"${r.monthly_cost_usd:,.0f} | ${r.annual_cost_usd:,.0f} |" |
| 201 | + ) |
| 202 | + |
| 203 | + cheapest = self.cheapest |
| 204 | + most_exp = self.most_expensive |
| 205 | + if cheapest and most_exp and cheapest is not most_exp: |
| 206 | + ratio = most_exp.daily_cost_usd / max(cheapest.daily_cost_usd, 0.0001) |
| 207 | + lines.append("") |
| 208 | + lines.append( |
| 209 | + f"_Cheapest: **{cheapest.scenario.name}** " |
| 210 | + f"(${cheapest.daily_cost_usd:.2f}/day). " |
| 211 | + f"Most expensive: **{most_exp.scenario.name}** " |
| 212 | + f"(${most_exp.daily_cost_usd:.2f}/day, {ratio:.1f}× cheapest)._" |
| 213 | + ) |
| 214 | + return "\n".join(lines) |
| 215 | + |
| 216 | + |
| 217 | +def compare_scenarios(scenarios: list[Scenario]) -> ScenarioComparison: |
| 218 | + """Simulate each scenario; return a comparison.""" |
| 219 | + results = [simulate(s) for s in scenarios] |
| 220 | + return ScenarioComparison(results=results) |
0 commit comments