from __future__ import annotations from dataclasses import dataclass from typing import Iterable import numpy as np from .dsl import parse_business_model from .models import BusinessModel, ScenarioShock @dataclass(slots=True) class SimulationResult: scenario_name: str seed: int horizon_months: int revenue_paths: np.ndarray cash_paths: np.ndarray runway_months: np.ndarray ending_cash: np.ndarray ltv_cac: np.ndarray gross_margin: np.ndarray monthly_summary: list[dict[str, float]] dilution_table: list[dict[str, float | str]] model_digest: str def summary(self) -> dict[str, float]: return { "runway_p10": float(np.percentile(self.runway_months, 10)), "runway_p50": float(np.percentile(self.runway_months, 50)), "runway_p90": float(np.percentile(self.runway_months, 90)), "ending_cash_p50": float(np.percentile(self.ending_cash, 50)), "ltv_cac_p50": float(np.percentile(self.ltv_cac, 50)), "gross_margin_p50": float(np.percentile(self.gross_margin, 50)), } def _fx(amount: float, currency: str, fx_rates: dict[str, float]) -> float: if currency not in fx_rates: raise KeyError(f"missing fx rate for {currency}") return amount * fx_rates[currency] def _monthly_rate_from_annual(salary: float) -> float: return salary / 12.0 def _dilution_projection(model: BusinessModel) -> list[dict[str, float | str]]: founder = 1.0 pool = 0.0 rows: list[dict[str, float | str]] = [] shares = 100.0 for round_ in sorted(model.financing, key=lambda item: item.month): if round_.instrument == "option_pool": pool += round_.option_pool_percent founder *= max(0.0, 1.0 - round_.option_pool_percent) shares *= max(0.0, 1.0 - round_.option_pool_percent) rows.append({"month": round_.month, "event": "option_pool", "founder_pct": founder, "investor_pct": 0.0, "pool_pct": pool}) continue if round_.instrument == "safe": assumed_valuation = round_.cap or round_.valuation or 1.0 investor = min(0.5, round_.amount / assumed_valuation) else: investor = min(0.9, round_.amount / (round_.valuation or 1.0)) founder *= max(0.0, 1.0 - investor) shares *= max(0.0, 1.0 - investor) rows.append({"month": round_.month, "event": round_.instrument, "founder_pct": founder, "investor_pct": investor, "pool_pct": pool}) return rows def _simulate_single_run(model: BusinessModel, rng: np.random.Generator, scenario: ScenarioShock | None) -> tuple[np.ndarray, np.ndarray, float, float, float]: horizon = model.horizon_months revenue = np.zeros(horizon, dtype=float) cash = np.zeros(horizon + 1, dtype=float) cash[0] = model.starting_cash total_gross_profit = 0.0 total_cac_spend = 0.0 scenario = scenario or ScenarioShock(name="baseline", probability=1.0, demand_multiplier=1.0, cac_multiplier=1.0, churn_delta=0.0, cost_inflation_delta=0.0, funding_environment_delta=0.0, interest_rate_delta=0.0, duration_months=horizon, regime_bias=0.0) regime = 1.0 monthly_customers = [stream.starting_customers for stream in model.revenue_streams] financing_by_month: dict[int, list] = {} for round_ in model.financing: financing_by_month.setdefault(round_.month, []).append(round_) for month in range(horizon): switch_p = model.macro.regime_switch_probability * (1.2 if model.macro.consumer_confidence < 95 else 0.8) if rng.random() < switch_p: regime *= -1.0 active_scenario = 1.0 if month >= scenario.duration_months else scenario.demand_multiplier regime_effect = 1.0 + (scenario.regime_bias if regime < 0 else -scenario.regime_bias / 2.0) demand_shock = 1.0 + (0.02 * model.macro.gdp_growth) + ((model.macro.consumer_confidence - 100.0) / 1000.0) + rng.normal(0.0, 0.02) price_shock = 1.0 + (0.5 * model.macro.inflation) + rng.normal(0.0, 0.01) churn_shift = max(0.0, model.macro.interest_rate * 0.25 + scenario.churn_delta + rng.normal(0.0, 0.005)) cost_inflation = max(0.0, model.macro.inflation + scenario.cost_inflation_delta + rng.normal(0.0, 0.01)) funding_factor = 1.0 + scenario.funding_environment_delta + (model.macro.funding_environment - 0.5) * 0.1 month_revenue = 0.0 month_gross_profit = 0.0 month_cac = 0.0 for i, stream in enumerate(model.revenue_streams): churn = min(0.95, stream.monthly_churn + churn_shift + rng.normal(0.0, 0.003)) acquisition = max(0.0, stream.acquisition_rate * active_scenario * regime_effect * funding_factor * demand_shock) customers = monthly_customers[i] * (1.0 - churn) + acquisition customers = max(0.0, customers * (1.0 + stream.monthly_growth_rate * active_scenario * demand_shock * regime_effect)) monthly_customers[i] = customers price = stream.price_per_customer * price_shock * (1.0 + stream.macro_sensitivity * (demand_shock - 1.0)) stream_revenue = _fx(customers * price, stream.currency, model.fx_rates) stream_gross = stream_revenue * stream.gross_margin month_revenue += stream_revenue month_gross_profit += stream_gross month_cac += max(0.0, acquisition * stream.price_per_customer * scenario.cac_multiplier * 0.15) cost_total = 0.0 for cost in model.costs: inflated = cost.monthly_amount * (1.0 + cost.inflation_sensitivity * cost_inflation) * (1.0 + cost.macro_sensitivity * (demand_shock - 1.0)) cost_total += _fx(inflated, cost.currency, model.fx_rates) for hire in model.hires: if month >= hire.month: ramp = min(1.0, max(0.2, (month - hire.month + 1) / max(1, hire.ramp_months))) cost_total += _fx(_monthly_rate_from_annual(hire.annual_salary) * hire.count * ramp, hire.currency, model.fx_rates) for capex in model.capex: if month == capex.month: cost_total += _fx(capex.amount, capex.currency, model.fx_rates) month_cash = cash[month] + month_gross_profit - cost_total if month in financing_by_month: for round_ in financing_by_month[month]: if round_.instrument in {"equity", "safe"}: month_cash += _fx(round_.amount, round_.currency, model.fx_rates) cash[month + 1] = month_cash revenue[month] = month_revenue total_gross_profit += month_gross_profit total_cac_spend += month_cac runway = int(np.argmax(cash[1:] <= 0)) if np.any(cash[1:] <= 0) else horizon lifetime_value = (np.mean(revenue) / max(1.0, np.mean([s.starting_customers + s.acquisition_rate for s in model.revenue_streams]) or 1.0)) * max(0.0, np.mean([s.gross_margin for s in model.revenue_streams])) / max(0.001, np.mean([s.monthly_churn for s in model.revenue_streams])) cac = max(1.0, total_cac_spend / max(1.0, sum(s.acquisition_rate for s in model.revenue_streams) * horizon)) ltv_cac = lifetime_value / cac gross_margin = total_gross_profit / max(1.0, np.sum(revenue)) return revenue, cash, float(runway), float(cash[-1]), float(ltv_cac), float(gross_margin) def _monthly_summary(revenue_paths: np.ndarray, cash_paths: np.ndarray) -> list[dict[str, float]]: rows: list[dict[str, float]] = [] for month in range(revenue_paths.shape[1]): rows.append( { "month": float(month + 1), "revenue_p10": float(np.percentile(revenue_paths[:, month], 10)), "revenue_p50": float(np.percentile(revenue_paths[:, month], 50)), "revenue_p90": float(np.percentile(revenue_paths[:, month], 90)), "cash_p10": float(np.percentile(cash_paths[:, month + 1], 10)), "cash_p50": float(np.percentile(cash_paths[:, month + 1], 50)), "cash_p90": float(np.percentile(cash_paths[:, month + 1], 90)), } ) return rows def simulate_business_model(model_or_source: BusinessModel | str, n_sims: int = 1000, seed: int = 7, scenario: ScenarioShock | None = None) -> SimulationResult: if isinstance(model_or_source, BusinessModel): model = model_or_source digest = "direct-model" else: model, digest = parse_business_model(model_or_source) rng = np.random.default_rng(seed) revenue_paths = np.zeros((n_sims, model.horizon_months), dtype=float) cash_paths = np.zeros((n_sims, model.horizon_months + 1), dtype=float) runways = np.zeros(n_sims, dtype=float) ending_cash = np.zeros(n_sims, dtype=float) ltv_cac = np.zeros(n_sims, dtype=float) gross_margin = np.zeros(n_sims, dtype=float) for i in range(n_sims): revenue, cash, runway, final_cash, ltv, margin = _simulate_single_run(model, rng, scenario) revenue_paths[i] = revenue cash_paths[i] = cash runways[i] = runway ending_cash[i] = final_cash ltv_cac[i] = ltv gross_margin[i] = margin return SimulationResult( scenario_name=(scenario.name if scenario else "baseline"), seed=seed, horizon_months=model.horizon_months, revenue_paths=revenue_paths, cash_paths=cash_paths, runway_months=runways, ending_cash=ending_cash, ltv_cac=ltv_cac, gross_margin=gross_margin, monthly_summary=_monthly_summary(revenue_paths, cash_paths), dilution_table=_dilution_projection(model), model_digest=digest, ) def compare_scenarios(model_or_source: BusinessModel | str, scenarios: Iterable[ScenarioShock], n_sims: int = 1000, seed: int = 7) -> list[SimulationResult]: model = model_or_source if isinstance(model_or_source, BusinessModel) else parse_business_model(model_or_source)[0] results: list[SimulationResult] = [] for index, scenario in enumerate(scenarios): results.append(simulate_business_model(model, n_sims=n_sims, seed=seed + index, scenario=scenario)) return results def sensitivity_analysis(model_or_source: BusinessModel | str, parameter_path: str, values: Iterable[float], n_sims: int = 500, seed: int = 7) -> list[dict[str, float]]: model = model_or_source if isinstance(model_or_source, BusinessModel) else parse_business_model(model_or_source)[0] if parameter_path != "macro.inflation": raise NotImplementedError("sensitivity analysis currently supports macro.inflation") rows: list[dict[str, float]] = [] for idx, value in enumerate(values): cloned = model.model_copy(deep=True) cloned.macro.inflation = float(value) result = simulate_business_model(cloned, n_sims=n_sims, seed=seed + idx) summary = result.summary() rows.append({"value": float(value), "runway_p50": summary["runway_p50"], "ending_cash_p50": summary["ending_cash_p50"], "ltv_cac_p50": summary["ltv_cac_p50"]}) return rows