Skills › Scientific Agent Skills › market-research-reports › calculate_market_sizing.py
calculate_market_sizing.py
389 lines · skills/market-research-reports/scripts/calculate_market_sizing.py at 5f2bc03
1#!/usr/bin/env python3
2"""Calculate bounded TAM/SAM/SOM scenarios and reconcile two sizing methods."""
4from __future__ import annotations
6import argparse
7from pathlib import Path
8from typing import Any
10from _common import (
11 ValidationError,
12 error_exit,
13 parse_currency,
14 parse_fraction,
15 parse_iso_date,
16 parse_number,
17 parse_year,
18 read_json,
19 require_identifier,
20 require_list,
21 require_object,
22 require_text,
23 require_unique_identifiers,
24 write_json_report,
25)
27MAX_COMPONENTS = 1_000
28MAX_SCENARIOS = 20
29MAX_VALUE = 1e18
32def _identifier_array(
33 value: Any, context: str, *, minimum: int = 0
34) -> list[str]:
35 items = require_list(value, context, minimum=minimum, maximum=100)
36 parsed = [
37 require_identifier(item, f"{context}[{index}]")
38 for index, item in enumerate(items)
39 ]
40 require_unique_identifiers(parsed, context)
41 return parsed
44def _metadata(payload: dict[str, Any]) -> dict[str, Any]:
45 raw = require_object(payload.get("metadata"), "metadata")
46 required = (
47 "market_id",
48 "market_definition",
49 "geography",
50 "currency",
51 "base_year",
52 "price_basis",
53 "unit",
54 "taxonomy",
55 "taxonomy_version",
56 "denominator_id",
57 "as_of_date",
58 )
59 missing = [field for field in required if field not in raw]
60 if missing:
61 raise ValidationError(f"metadata is missing: {', '.join(missing)}")
62 price_basis = require_text(raw["price_basis"], "metadata.price_basis", maximum=20)
63 if price_basis not in {"nominal", "real", "current", "constant"}:
64 raise ValidationError(
65 "metadata.price_basis must be nominal, real, current, or constant"
66 )
67 return {
68 "market_id": require_identifier(raw["market_id"], "metadata.market_id"),
69 "market_definition": require_text(
70 raw["market_definition"], "metadata.market_definition"
71 ),
72 "geography": require_text(raw["geography"], "metadata.geography"),
73 "currency": parse_currency(raw["currency"], "metadata.currency"),
74 "base_year": parse_year(raw["base_year"], "metadata.base_year"),
75 "price_basis": price_basis,
76 "unit": require_text(raw["unit"], "metadata.unit", maximum=80),
77 "taxonomy": require_text(raw["taxonomy"], "metadata.taxonomy"),
78 "taxonomy_version": require_text(
79 raw["taxonomy_version"], "metadata.taxonomy_version"
80 ),
81 "denominator_id": require_identifier(
82 raw["denominator_id"], "metadata.denominator_id"
83 ),
84 "as_of_date": parse_iso_date(raw["as_of_date"], "metadata.as_of_date"),
85 }
88def _top_down(payload: dict[str, Any], denominator_id: str) -> tuple[float, list[str]]:
89 section = require_object(payload.get("top_down"), "top_down")
90 components = require_list(
91 section.get("components"),
92 "top_down.components",
93 minimum=1,
94 maximum=MAX_COMPONENTS,
95 )
96 component_ids: list[str] = []
97 coverage_keys: list[str] = []
98 total = 0.0
99 for index, item in enumerate(components):
100 row = require_object(item, f"top_down.components[{index}]")
101 component_id = require_identifier(
102 row.get("component_id"),
103 f"top_down.components[{index}].component_id",
104 )
105 component_ids.append(component_id)
106 coverage_key = require_identifier(
107 row.get("coverage_key"),
108 f"top_down.components[{index}].coverage_key",
109 )
110 coverage_keys.append(coverage_key)
111 row_denominator = require_identifier(
112 row.get("denominator_id"),
113 f"top_down.components[{index}].denominator_id",
114 )
115 if row_denominator != denominator_id:
116 raise ValidationError(
117 f"top_down component {component_id} uses denominator "
118 f"{row_denominator!r}; expected {denominator_id!r}"
119 )
120 value = parse_number(
121 row.get("value"),
122 f"top_down.components[{index}].value",
123 minimum=0.0,
124 maximum=MAX_VALUE,
125 )
126 _identifier_array(
127 row.get("source_ids"),
128 f"top_down.components[{index}].source_ids",
129 minimum=1,
130 )
131 _identifier_array(
132 row.get("assumption_ids", []),
133 f"top_down.components[{index}].assumption_ids",
134 )
135 total += value
136 if total > MAX_VALUE:
137 raise ValidationError("top_down TAM exceeds the supported numeric bound")
138 require_unique_identifiers(component_ids, "top_down component IDs")
139 require_unique_identifiers(
140 coverage_keys,
141 "top_down coverage keys (duplicate coverage can double count the market)",
142 )
143 return total, coverage_keys
146def _bottom_up(
147 payload: dict[str, Any], denominator_id: str
148) -> tuple[float, list[str]]:
149 section = require_object(payload.get("bottom_up"), "bottom_up")
150 components = require_list(
151 section.get("components"),
152 "bottom_up.components",
153 minimum=1,
154 maximum=MAX_COMPONENTS,
155 )
156 component_ids: list[str] = []
157 coverage_keys: list[str] = []
158 total = 0.0
159 for index, item in enumerate(components):
160 row = require_object(item, f"bottom_up.components[{index}]")
161 component_id = require_identifier(
162 row.get("component_id"),
163 f"bottom_up.components[{index}].component_id",
164 )
165 component_ids.append(component_id)
166 coverage_key = require_identifier(
167 row.get("coverage_key"),
168 f"bottom_up.components[{index}].coverage_key",
169 )
170 coverage_keys.append(coverage_key)
171 row_denominator = require_identifier(
172 row.get("denominator_id"),
173 f"bottom_up.components[{index}].denominator_id",
174 )
175 if row_denominator != denominator_id:
176 raise ValidationError(
177 f"bottom_up component {component_id} uses denominator "
178 f"{row_denominator!r}; expected {denominator_id!r}"
179 )
180 customers = parse_number(
181 row.get("customer_count"),
182 f"bottom_up.components[{index}].customer_count",
183 minimum=0.0,
184 maximum=1e15,
185 )
186 annual_quantity = parse_number(
187 row.get("annual_quantity_per_customer"),
188 f"bottom_up.components[{index}].annual_quantity_per_customer",
189 minimum=0.0,
190 maximum=1e12,
191 )
192 price = parse_number(
193 row.get("price_per_unit"),
194 f"bottom_up.components[{index}].price_per_unit",
195 minimum=0.0,
196 maximum=1e15,
197 )
198 addressable_fraction = parse_fraction(
199 row.get("addressable_fraction"),
200 f"bottom_up.components[{index}].addressable_fraction",
201 )
202 _identifier_array(
203 row.get("source_ids"),
204 f"bottom_up.components[{index}].source_ids",
205 minimum=1,
206 )
207 _identifier_array(
208 row.get("assumption_ids", []),
209 f"bottom_up.components[{index}].assumption_ids",
210 )
211 component_value = customers * annual_quantity * price * addressable_fraction
212 if component_value > MAX_VALUE:
213 raise ValidationError(
214 f"bottom_up component {component_id} exceeds the numeric bound"
215 )
216 total += component_value
217 if total > MAX_VALUE:
218 raise ValidationError("bottom_up TAM exceeds the supported numeric bound")
219 require_unique_identifiers(component_ids, "bottom_up component IDs")
220 require_unique_identifiers(
221 coverage_keys,
222 "bottom_up coverage keys (duplicate coverage can double count the market)",
223 )
224 return total, coverage_keys
227def _scenarios(payload: dict[str, Any]) -> list[dict[str, Any]]:
228 raw_scenarios = require_list(
229 payload.get("scenarios"),
230 "scenarios",
231 minimum=2,
232 maximum=MAX_SCENARIOS,
233 )
234 scenarios: list[dict[str, Any]] = []
235 scenario_ids: list[str] = []
236 parameter_pairs: list[tuple[float, float]] = []
237 for index, item in enumerate(raw_scenarios):
238 row = require_object(item, f"scenarios[{index}]")
239 scenario_id = require_identifier(
240 row.get("scenario_id"), f"scenarios[{index}].scenario_id"
241 )
242 scenario_ids.append(scenario_id)
243 serviceable = parse_fraction(
244 row.get("serviceable_fraction"),
245 f"scenarios[{index}].serviceable_fraction",
246 )
247 obtainable = parse_fraction(
248 row.get("obtainable_share"),
249 f"scenarios[{index}].obtainable_share",
250 )
251 parameter_pairs.append((serviceable, obtainable))
252 assumptions = require_list(
253 row.get("assumptions"),
254 f"scenarios[{index}].assumptions",
255 minimum=1,
256 maximum=100,
257 )
258 parsed_assumptions = [
259 require_text(value, f"scenarios[{index}].assumptions[{position}]")
260 for position, value in enumerate(assumptions)
261 ]
262 scenarios.append(
263 {
264 "scenario_id": scenario_id,
265 "label": require_text(
266 row.get("label"), f"scenarios[{index}].label", maximum=120
267 ),
268 "serviceable_fraction": serviceable,
269 "obtainable_share": obtainable,
270 "source_ids": _identifier_array(
271 row.get("source_ids", []),
272 f"scenarios[{index}].source_ids",
273 ),
274 "assumptions": parsed_assumptions,
275 }
276 )
277 require_unique_identifiers(scenario_ids, "scenario IDs")
278 if len(set(parameter_pairs)) < 2:
279 raise ValidationError(
280 "at least two scenarios must use different sizing assumptions"
281 )
282 return scenarios
285def calculate(payload: dict[str, Any]) -> dict[str, Any]:
286 schema_version = require_text(
287 payload.get("schema_version"), "schema_version", maximum=10
288 )
289 if schema_version != "1.0":
290 raise ValidationError("schema_version must be '1.0'")
291 metadata = _metadata(payload)
292 top_down_tam, top_keys = _top_down(payload, metadata["denominator_id"])
293 bottom_up_tam, bottom_keys = _bottom_up(payload, metadata["denominator_id"])
294 scenarios = _scenarios(payload)
295 tolerance = parse_number(
296 payload.get("reconciliation_tolerance_percent", 20.0),
297 "reconciliation_tolerance_percent",
298 minimum=0.0,
299 maximum=100.0,
300 )
302 midpoint = (top_down_tam + bottom_up_tam) / 2.0
303 reconciliation_percent = (
304 0.0
305 if midpoint == 0.0
306 else abs(top_down_tam - bottom_up_tam) / midpoint * 100.0
307 )
308 warnings: list[str] = []
309 if reconciliation_percent > tolerance:
310 warnings.append(
311 "top-down and bottom-up TAM estimates exceed the stated "
312 "reconciliation tolerance; investigate scope, denominators, and assumptions"
313 )
314 if set(top_keys) != set(bottom_keys):
315 warnings.append(
316 "the methods use different coverage-key sets; explain the scope difference "
317 "before treating them as direct reconciliation estimates"
318 )
320 scenario_results: list[dict[str, Any]] = []
321 for scenario in scenarios:
322 methods: dict[str, dict[str, float]] = {}
323 for method, tam in (
324 ("top_down", top_down_tam),
325 ("bottom_up", bottom_up_tam),
326 ):
327 sam = tam * scenario["serviceable_fraction"]
328 som = sam * scenario["obtainable_share"]
329 methods[method] = {"tam": tam, "sam": sam, "som": som}
330 scenario_results.append({**scenario, "methods": methods})
332 all_som = [
333 result["methods"][method]["som"]
334 for result in scenario_results
335 for method in ("top_down", "bottom_up")
336 ]
337 return {
338 "schema_version": "1.0",
339 "metadata": metadata,
340 "method_estimates": {
341 "top_down_tam": top_down_tam,
342 "bottom_up_tam": bottom_up_tam,
343 },
344 "reconciliation": {
345 "absolute_difference": abs(top_down_tam - bottom_up_tam),
346 "difference_percent_of_midpoint": reconciliation_percent,
347 "tolerance_percent": tolerance,
348 "within_tolerance": reconciliation_percent <= tolerance,
349 },
350 "scenario_results": scenario_results,
351 "som_range_across_methods_and_scenarios": {
352 "minimum": min(all_som),
353 "maximum": max(all_som),
354 },
355 "warnings": warnings,
356 "interpretation": (
357 "These are conditional scenario calculations, not a single asserted "
358 "market truth."
359 ),
360 }
363def build_parser() -> argparse.ArgumentParser:
364 parser = argparse.ArgumentParser(
365 description=(
366 "Calculate local top-down and bottom-up TAM/SAM/SOM scenarios, "
367 "detect duplicate coverage keys, and report reconciliation."
368 )
369 )
370 parser.add_argument("input", help="Local .json market-sizing input")
371 parser.add_argument("--output", help="Optional local .json result")
372 parser.add_argument(
373 "--force", action="store_true", help="Replace an existing --output file"
374 )
375 return parser
378def main() -> int:
379 args = build_parser().parse_args()
380 try:
381 payload = require_object(read_json(args.input), "root")
382 write_json_report(calculate(payload), args.output, force=args.force)
383 return 0
384 except ValidationError as exc:
385 return error_exit(exc)
388if __name__ == "__main__":
389 raise SystemExit(main())