Skills › Scientific Agent Skills › market-research-reports › forecast_sensitivity.py
forecast_sensitivity.py
326 lines · skills/market-research-reports/scripts/forecast_sensitivity.py at 5f2bc03
1#!/usr/bin/env python3
2"""Generate deterministic scenario forecasts and one-way growth sensitivity."""
4from __future__ import annotations
6import argparse
7from typing import Any
9from _common import (
10 ValidationError,
11 error_exit,
12 parse_currency,
13 parse_iso_date,
14 parse_number,
15 parse_year,
16 read_json,
17 require_identifier,
18 require_list,
19 require_object,
20 require_text,
21 require_unique_identifiers,
22 write_json_report,
23)
25MAX_VALUE = 1e18
26MAX_HORIZON = 50
27MAX_SCENARIOS = 20
30def _identifiers(value: Any, context: str, *, minimum: int = 0) -> list[str]:
31 items = require_list(value, context, minimum=minimum, maximum=100)
32 parsed = [
33 require_identifier(item, f"{context}[{index}]")
34 for index, item in enumerate(items)
35 ]
36 require_unique_identifiers(parsed, context)
37 return parsed
40def _integer(value: Any, context: str, minimum: int, maximum: int) -> int:
41 if type(value) is not int:
42 raise ValidationError(f"{context} must be an integer")
43 if not minimum <= value <= maximum:
44 raise ValidationError(f"{context} must be between {minimum} and {maximum}")
45 return value
48def _metadata(payload: dict[str, Any]) -> dict[str, Any]:
49 raw = require_object(payload.get("metadata"), "metadata")
50 required = (
51 "series_id",
52 "description",
53 "geography",
54 "currency",
55 "base_year",
56 "price_basis",
57 "unit",
58 "measure_type",
59 "as_of_date",
60 "source_ids",
61 )
62 missing = [field for field in required if field not in raw]
63 if missing:
64 raise ValidationError(f"metadata is missing: {', '.join(missing)}")
65 currency = parse_currency(raw["currency"], "metadata.currency", allow_empty=True)
66 base_year_text = require_text(
67 raw["base_year"], "metadata.base_year", allow_empty=True, maximum=4
68 )
69 base_year: int | None = None
70 if base_year_text:
71 base_year = parse_year(base_year_text, "metadata.base_year")
72 price_basis = require_text(raw["price_basis"], "metadata.price_basis", maximum=20)
73 if price_basis not in {
74 "nominal",
75 "real",
76 "current",
77 "constant",
78 "not-applicable",
79 }:
80 raise ValidationError("metadata.price_basis has an unsupported value")
81 if currency and (base_year is None or price_basis == "not-applicable"):
82 raise ValidationError(
83 "monetary forecasts require currency, base_year, and price_basis"
84 )
85 measure_type = require_text(
86 raw["measure_type"], "metadata.measure_type", maximum=20
87 )
88 if measure_type not in {"stock", "flow", "index", "share", "count", "rate", "price"}:
89 raise ValidationError("metadata.measure_type has an unsupported value")
90 return {
91 "series_id": require_identifier(raw["series_id"], "metadata.series_id"),
92 "description": require_text(raw["description"], "metadata.description"),
93 "geography": require_text(raw["geography"], "metadata.geography"),
94 "currency": currency,
95 "base_year": base_year,
96 "price_basis": price_basis,
97 "unit": require_text(raw["unit"], "metadata.unit", maximum=80),
98 "measure_type": measure_type,
99 "as_of_date": parse_iso_date(raw["as_of_date"], "metadata.as_of_date"),
100 "source_ids": _identifiers(
101 raw["source_ids"], "metadata.source_ids", minimum=1
102 ),
103 }
106def _rate_path(value: Any, context: str, horizon: int) -> list[float]:
107 items = require_list(value, context, minimum=horizon, maximum=horizon)
108 return [
109 parse_number(
110 item,
111 f"{context}[{index}]",
112 minimum=-0.99,
113 maximum=10.0,
114 )
115 for index, item in enumerate(items)
116 ]
119def _project(
120 start_year: int, start_value: float, rates: list[float]
121) -> list[dict[str, float | int]]:
122 observations: list[dict[str, float | int]] = [
123 {"year": start_year, "value": start_value}
124 ]
125 value = start_value
126 for offset, rate in enumerate(rates, start=1):
127 value *= 1.0 + rate
128 if value > MAX_VALUE:
129 raise ValidationError("forecast exceeds the supported numeric bound")
130 observations.append(
131 {"year": start_year + offset, "value": value, "growth_rate": rate}
132 )
133 return observations
136def _cagr(start: float, end: float, periods: int) -> float | None:
137 if periods <= 0 or start <= 0.0 or end < 0.0:
138 return None
139 return (end / start) ** (1.0 / periods) - 1.0
142def forecast(payload: dict[str, Any]) -> dict[str, Any]:
143 schema_version = require_text(
144 payload.get("schema_version"), "schema_version", maximum=10
145 )
146 if schema_version != "1.0":
147 raise ValidationError("schema_version must be '1.0'")
148 metadata = _metadata(payload)
149 start_year = parse_year(payload.get("start_year"), "start_year")
150 start_value = parse_number(
151 payload.get("start_value"),
152 "start_value",
153 minimum=0.0,
154 maximum=MAX_VALUE,
155 )
156 horizon = _integer(
157 payload.get("horizon_years"),
158 "horizon_years",
159 minimum=1,
160 maximum=MAX_HORIZON,
161 )
163 raw_scenarios = require_list(
164 payload.get("scenarios"),
165 "scenarios",
166 minimum=2,
167 maximum=MAX_SCENARIOS,
168 )
169 scenario_ids: list[str] = []
170 unique_paths: set[tuple[float, ...]] = set()
171 scenario_results: list[dict[str, Any]] = []
172 scenario_rates: dict[str, list[float]] = {}
173 for index, item in enumerate(raw_scenarios):
174 row = require_object(item, f"scenarios[{index}]")
175 if "probability" in row:
176 raise ValidationError(
177 f"scenarios[{index}].probability is not accepted; do not assign "
178 "probabilities without a separately validated probabilistic model"
179 )
180 scenario_id = require_identifier(
181 row.get("scenario_id"), f"scenarios[{index}].scenario_id"
182 )
183 scenario_ids.append(scenario_id)
184 rates = _rate_path(
185 row.get("annual_growth_rates"),
186 f"scenarios[{index}].annual_growth_rates",
187 horizon,
188 )
189 unique_paths.add(tuple(rates))
190 scenario_rates[scenario_id] = rates
191 assumptions = require_list(
192 row.get("assumptions"),
193 f"scenarios[{index}].assumptions",
194 minimum=1,
195 maximum=100,
196 )
197 observations = _project(start_year, start_value, rates)
198 scenario_results.append(
199 {
200 "scenario_id": scenario_id,
201 "label": require_text(
202 row.get("label"), f"scenarios[{index}].label", maximum=120
203 ),
204 "source_ids": _identifiers(
205 row.get("source_ids", []),
206 f"scenarios[{index}].source_ids",
207 ),
208 "assumptions": [
209 require_text(
210 assumption,
211 f"scenarios[{index}].assumptions[{position}]",
212 )
213 for position, assumption in enumerate(assumptions)
214 ],
215 "annual_growth_rates": rates,
216 "observations": observations,
217 "endpoint": observations[-1]["value"],
218 "cagr": _cagr(start_value, float(observations[-1]["value"]), horizon),
219 }
220 )
221 require_unique_identifiers(scenario_ids, "scenario IDs")
222 if len(unique_paths) < 2:
223 raise ValidationError("at least two scenarios must use different rate paths")
225 sensitivity = require_object(payload.get("sensitivity"), "sensitivity")
226 base_scenario_id = require_identifier(
227 sensitivity.get("base_scenario_id"), "sensitivity.base_scenario_id"
228 )
229 if base_scenario_id not in scenario_rates:
230 raise ValidationError(
231 "sensitivity.base_scenario_id must reference one of the scenarios"
232 )
233 shifts_raw = require_list(
234 sensitivity.get("growth_rate_shifts"),
235 "sensitivity.growth_rate_shifts",
236 minimum=2,
237 maximum=21,
238 )
239 shifts = [
240 parse_number(
241 value,
242 f"sensitivity.growth_rate_shifts[{index}]",
243 minimum=-0.5,
244 maximum=0.5,
245 )
246 for index, value in enumerate(shifts_raw)
247 ]
248 if len(shifts) != len(set(shifts)):
249 raise ValidationError("sensitivity.growth_rate_shifts must be unique")
250 sensitivity_results: list[dict[str, Any]] = []
251 for shift in sorted(shifts):
252 shifted_rates = [rate + shift for rate in scenario_rates[base_scenario_id]]
253 if any(rate < -0.99 or rate > 10.0 for rate in shifted_rates):
254 raise ValidationError(
255 f"sensitivity shift {shift} produces an out-of-bounds growth rate"
256 )
257 observations = _project(start_year, start_value, shifted_rates)
258 sensitivity_results.append(
259 {
260 "growth_rate_shift": shift,
261 "annual_growth_rates": shifted_rates,
262 "endpoint": observations[-1]["value"],
263 "observations": observations,
264 }
265 )
267 ranges: list[dict[str, float | int]] = []
268 for offset in range(horizon + 1):
269 values = [
270 float(result["observations"][offset]["value"])
271 for result in scenario_results
272 ]
273 ranges.append(
274 {
275 "year": start_year + offset,
276 "minimum": min(values),
277 "maximum": max(values),
278 }
279 )
281 return {
282 "schema_version": "1.0",
283 "metadata": metadata,
284 "start_year": start_year,
285 "start_value": start_value,
286 "horizon_years": horizon,
287 "scenarios": scenario_results,
288 "scenario_range_by_year": ranges,
289 "sensitivity": {
290 "base_scenario_id": base_scenario_id,
291 "results": sensitivity_results,
292 },
293 "interpretation": (
294 "Ranges are conditional on documented scenarios and are not confidence "
295 "intervals or assigned probabilities."
296 ),
297 }
300def build_parser() -> argparse.ArgumentParser:
301 parser = argparse.ArgumentParser(
302 description=(
303 "Generate bounded local scenario forecasts and one-way growth-rate "
304 "sensitivity. No network or random sampling is used."
305 )
306 )
307 parser.add_argument("input", help="Local .json forecast input")
308 parser.add_argument("--output", help="Optional local .json result")
309 parser.add_argument(
310 "--force", action="store_true", help="Replace an existing --output file"
311 )
312 return parser
315def main() -> int:
316 args = build_parser().parse_args()
317 try:
318 payload = require_object(read_json(args.input), "root")
319 write_json_report(forecast(payload), args.output, force=args.force)
320 return 0
321 except ValidationError as exc:
322 return error_exit(exc)
325if __name__ == "__main__":
326 raise SystemExit(main())