Coverage for backend/django/Economics/results/services/financial_metrics/input_snapshots.py: 92%
133 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Financial calculation input snapshots for persisted and comparison flows."""
3from __future__ import annotations
5from collections.abc import Mapping
6from dataclasses import dataclass
7from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
8from typing import Any
10from pydantic import ValidationError
12from Economics.costing.capital.custom_capital_lines import base_capex_for_custom_percentage_lines
13from Economics.costing.capital.electrical_upgrade import PeakDemandBasis, derive_peak_demand_basis
14from Economics.costing.operating.schedule_calculation import calculate_operating_line
15from Economics.formulas.builders.capital import build_electrical_upgrade_formula, build_target_total_capex_formula
16from Economics.formulas.builders.operating import operating_line_is_revenue
17from Economics.results.services.financial_metrics.baselines import (
18 resolve_baseline_for_study,
19 target_assumptions_for_study,
20)
21from Economics.results.services.financial_metrics.contracts import (
22 ZERO,
23 BaselineResolution,
24 FinancialCalculationInputs,
25 FinancialMetricsError,
26 FinancialWarning,
27 TargetAssumptions,
28)
29from Economics.shared.payloads import json_ready
30from Economics.settings_profiles.services.depreciation import DepreciationSchedule, build_straight_line_depreciation_schedule
31from Economics.studies.models import EconomicsStudy
34FINANCIAL_INPUT_SNAPSHOT_PAYLOAD_KEY = "financial_input_snapshot"
37@dataclass(frozen=True)
38class FinancialInputSnapshot:
39 """Target financial inputs before baseline-dependent assumptions are applied."""
41 study: EconomicsStudy
42 target_capex: Decimal
43 target_annual_opex: Decimal
44 target_annual_revenue: Decimal
45 target_annual_depreciation: Decimal
46 target_purchase_basis_capex: Decimal
47 target_installed_basis_capex: Decimal
48 target_contingency_capex: Decimal
49 target_electrical_upgrade_capex: Decimal
50 target_peak_demand_kw: Decimal | None
51 target_assumptions: TargetAssumptions
53 def to_payload(self) -> dict[str, Any]:
54 """Serialize the snapshot for result-run JSON metadata."""
56 return json_ready(
57 {
58 "version": 1,
59 "target_capex": self.target_capex,
60 "target_annual_opex": self.target_annual_opex,
61 "target_annual_revenue": self.target_annual_revenue,
62 "target_annual_depreciation": self.target_annual_depreciation,
63 "target_purchase_basis_capex": self.target_purchase_basis_capex,
64 "target_installed_basis_capex": self.target_installed_basis_capex,
65 "target_contingency_capex": self.target_contingency_capex,
66 "target_electrical_upgrade_capex": self.target_electrical_upgrade_capex,
67 "target_peak_demand_kw": self.target_peak_demand_kw,
68 "target_assumptions": self.target_assumptions.model_dump(mode="json"),
69 }
70 )
72 def calculation_inputs(
73 self,
74 *,
75 baseline_resolution: BaselineResolution,
76 warnings: tuple[FinancialWarning, ...] | list[FinancialWarning] = (),
77 ) -> FinancialCalculationInputs:
78 """Return pure calculation inputs using a resolved baseline contract."""
80 return FinancialCalculationInputs(
81 target_capex=self.target_capex,
82 target_annual_opex=self.target_annual_opex,
83 target_annual_revenue=self.target_annual_revenue,
84 target_annual_depreciation=self.target_annual_depreciation,
85 target_purchase_basis_capex=self.target_purchase_basis_capex,
86 target_installed_basis_capex=self.target_installed_basis_capex,
87 target_contingency_capex=self.target_contingency_capex,
88 target_electrical_upgrade_capex=self.target_electrical_upgrade_capex,
89 target_peak_demand_kw=self.target_peak_demand_kw,
90 baseline_capex=baseline_resolution.capex,
91 baseline_annual_opex=baseline_resolution.annual_opex,
92 project_lifetime_years=baseline_resolution.project_lifetime_years,
93 discount_rate_percent=baseline_resolution.discount_rate_percent,
94 tax_rate_percent=self.target_assumptions.tax_rate_percent,
95 residual_value=baseline_resolution.residual_value,
96 baseline_fully_calculated=(
97 not baseline_resolution.is_guided_default
98 and baseline_resolution.capex is not None
99 and baseline_resolution.annual_opex is not None
100 ),
101 assumptions=self.target_assumptions.as_assumption_set().merge_set(baseline_resolution.assumptions),
102 warnings=tuple(warnings),
103 )
106def build_financial_input_snapshot(
107 *,
108 study: EconomicsStudy,
109 target_capex: Decimal,
110 capital_breakdown: dict[str, Decimal],
111 depreciation_schedule: DepreciationSchedule,
112 target_annual_opex: Decimal,
113 target_annual_revenue: Decimal,
114 peak_demand_basis: PeakDemandBasis,
115 target_assumptions: TargetAssumptions,
116) -> FinancialInputSnapshot:
117 """Create a snapshot from lifecycle-owned target calculations."""
119 return FinancialInputSnapshot(
120 study=study,
121 target_capex=target_capex,
122 target_annual_opex=target_annual_opex,
123 target_annual_revenue=target_annual_revenue,
124 target_annual_depreciation=depreciation_schedule.annual_depreciation,
125 target_purchase_basis_capex=capital_breakdown["purchase_basis"],
126 target_installed_basis_capex=capital_breakdown["installed_basis"],
127 target_contingency_capex=capital_breakdown["contingency"],
128 target_electrical_upgrade_capex=capital_breakdown["electrical_upgrade"],
129 target_peak_demand_kw=peak_demand_basis.quantity_kw,
130 target_assumptions=target_assumptions,
131 )
134def snapshot_from_study_target_state(study: EconomicsStudy) -> FinancialInputSnapshot:
135 """Build a financial input snapshot from the study's current persisted target state."""
137 target_capex = target_total_capex(study)
138 capital_breakdown = target_capital_breakdown(study)
139 depreciation_schedule = build_straight_line_depreciation_schedule(study)
140 target_annual_opex, target_annual_revenue = target_operating_totals(study)
141 peak_demand_basis = derive_peak_demand_basis(study)
142 target_assumptions = target_assumptions_for_study(study)
143 return build_financial_input_snapshot(
144 study=study,
145 target_capex=target_capex,
146 capital_breakdown=capital_breakdown,
147 depreciation_schedule=depreciation_schedule,
148 target_annual_opex=target_annual_opex,
149 target_annual_revenue=target_annual_revenue,
150 peak_demand_basis=peak_demand_basis,
151 target_assumptions=target_assumptions,
152 )
155def snapshot_from_result_run(result_run) -> FinancialInputSnapshot:
156 """Load the structured financial input snapshot persisted with a result run."""
158 snapshot_payload = financial_input_snapshot_payload_for_result_run(result_run)
159 if not isinstance(snapshot_payload, Mapping):
160 raise FinancialMetricsError(
161 "financial_input_snapshot_missing",
162 "The saved result run does not include structured financial calculation inputs.",
163 context={"result_run_id": result_run.pk, "target_study_id": result_run.study_id},
164 )
165 return financial_input_snapshot_from_payload(study=result_run.study, payload=snapshot_payload)
168def financial_input_snapshot_from_payload(
169 *,
170 study: EconomicsStudy,
171 payload: Mapping[str, Any],
172) -> FinancialInputSnapshot:
173 """Deserialize a result-run financial input snapshot payload."""
175 try:
176 if payload.get("version") != 1:
177 raise ValueError("version")
178 target_assumptions = TargetAssumptions.model_validate(payload["target_assumptions"])
179 if target_assumptions.target_study_id != study.pk:
180 raise ValueError("target_assumptions.target_study_id")
181 return FinancialInputSnapshot(
182 study=study,
183 target_capex=_required_decimal(payload, "target_capex"),
184 target_annual_opex=_required_decimal(payload, "target_annual_opex"),
185 target_annual_revenue=_required_decimal(payload, "target_annual_revenue"),
186 target_annual_depreciation=_required_decimal(payload, "target_annual_depreciation"),
187 target_purchase_basis_capex=_required_decimal(payload, "target_purchase_basis_capex"),
188 target_installed_basis_capex=_required_decimal(payload, "target_installed_basis_capex"),
189 target_contingency_capex=_required_decimal(payload, "target_contingency_capex"),
190 target_electrical_upgrade_capex=_required_decimal(payload, "target_electrical_upgrade_capex"),
191 target_peak_demand_kw=_optional_decimal(payload.get("target_peak_demand_kw")),
192 target_assumptions=target_assumptions,
193 )
194 except (KeyError, TypeError, ValueError, ValidationError) as exc:
195 raise FinancialMetricsError(
196 "financial_input_snapshot_invalid",
197 "The saved result run has invalid structured financial calculation inputs.",
198 context={"target_study_id": study.pk, "field": str(exc)},
199 ) from exc
202def financial_input_snapshot_payload_for_result_run(result_run) -> object:
203 """Return the raw snapshot payload from a persisted result run.
205 Result runs currently store calculation metadata in their JSON warning
206 payload. Keeping that access behind this helper prevents comparison and
207 lifecycle code from treating the diagnostics field as their own contract.
208 """
210 payload = result_run.warning_payload if isinstance(result_run.warning_payload, Mapping) else {}
211 return payload.get(FINANCIAL_INPUT_SNAPSHOT_PAYLOAD_KEY)
214def warning_payload_with_financial_input_snapshot(
215 *,
216 warning_payload: Mapping[str, Any],
217 snapshot_payload: Mapping[str, Any] | None,
218) -> dict[str, Any]:
219 """Attach the structured financial input snapshot to a result-run payload."""
221 payload = dict(warning_payload)
222 if snapshot_payload is not None:
223 payload[FINANCIAL_INPUT_SNAPSHOT_PAYLOAD_KEY] = dict(snapshot_payload)
224 return payload
227def calculation_inputs_for_study(
228 *,
229 study: EconomicsStudy,
230 snapshot: FinancialInputSnapshot,
231) -> tuple[FinancialCalculationInputs, BaselineResolution, list[FinancialWarning]]:
232 """Resolve the study's selected baseline and return calculation inputs."""
234 baseline_resolution, warnings = resolve_baseline_for_study(
235 study=study,
236 target_capex=snapshot.target_capex,
237 target_annual_opex=snapshot.target_annual_opex,
238 target_assumptions=snapshot.target_assumptions,
239 )
240 return (
241 snapshot.calculation_inputs(
242 baseline_resolution=baseline_resolution,
243 warnings=tuple(warnings),
244 ),
245 baseline_resolution,
246 warnings,
247 )
250def target_base_capital_cost(study: EconomicsStudy) -> Decimal:
251 """Return the generated unit-operation CAPEX subtotal for native properties."""
253 return base_capex_for_custom_percentage_lines(study)
256def target_annual_operating_expense(study: EconomicsStudy) -> Decimal:
257 """Return annual operating expenses before output revenue offsets."""
259 expense_total, _revenue_total = target_operating_totals(study)
260 return expense_total
263def target_total_capex(study: EconomicsStudy) -> Decimal:
264 """Return the target study's total capital cost."""
266 total_formula = build_target_total_capex_formula(study)
267 total = total_formula.evaluate()
268 if total is None:
269 raise FinancialMetricsError(
270 "target_capex_formula_blocked",
271 total_formula.formula.blocked_reason,
272 context={"blocked_children": total_formula.formula.blocked_children},
273 )
274 return total.quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
277def target_capital_breakdown(study: EconomicsStudy) -> dict[str, Decimal]:
278 """Return target capital subtotals used by financial metrics."""
280 totals = {
281 "purchase_basis": ZERO,
282 "installed_basis": ZERO,
283 "contingency": ZERO,
284 "electrical_upgrade": ZERO,
285 }
286 for payload in study.capital_lines.filter(included=True).values_list("warning_payload", flat=True):
287 if not isinstance(payload, Mapping): 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true
288 continue
289 totals["purchase_basis"] += _decimal_from_payload(payload.get("purchase_basis_amount"))
290 totals["installed_basis"] += _decimal_from_payload(payload.get("installed_basis_amount"))
291 totals["contingency"] += _decimal_from_payload(payload.get("contingency_amount"))
292 totals["electrical_upgrade"] = _electrical_upgrade_capex(study)
293 return totals
296def target_operating_totals(study: EconomicsStudy) -> tuple[Decimal, Decimal]:
297 """Return target annual expense and revenue totals."""
299 expense_total = ZERO
300 revenue_total = ZERO
301 blocked_children = []
302 lines = study.operating_lines.filter(included=True).select_related(
303 "source_default_rate",
304 "source_property_info",
305 )
306 for line in lines.order_by("pk"):
307 calculation = calculate_operating_line(line, study=study)
308 if calculation.annual_amount is None:
309 blocked_children.append(
310 {
311 "key": f"operating_line:{line.pk}",
312 "reason": calculation.blocked_reason or "Operating line cannot be calculated.",
313 }
314 )
315 continue
316 if operating_line_is_revenue(line):
317 revenue_total += calculation.annual_amount
318 else:
319 expense_total += calculation.annual_amount
320 if blocked_children:
321 raise FinancialMetricsError(
322 "target_annual_opex_formula_blocked",
323 "Annual operating costs cannot be calculated until all included operating lines are complete.",
324 context={"blocked_children": blocked_children},
325 )
326 return expense_total, revenue_total
329def _electrical_upgrade_capex(study: EconomicsStudy) -> Decimal:
330 """Calculate project-level electrical-upgrade capex without creating a capital line."""
332 formula = build_electrical_upgrade_formula(study)
333 amount = formula.evaluate()
334 if amount is None: 334 ↛ 335line 334 didn't jump to line 335 because the condition on line 334 was never true
335 raise FinancialMetricsError(
336 "electrical_upgrade_formula_blocked",
337 formula.formula.blocked_reason,
338 context={"blocked_children": formula.formula.blocked_children},
339 )
340 return amount
343def _decimal_from_payload(value: object) -> Decimal:
344 if value in (None, ""):
345 return ZERO
346 try:
347 return Decimal(str(value))
348 except (InvalidOperation, ValueError):
349 return ZERO
352def _required_decimal(payload: Mapping[str, Any], key: str) -> Decimal:
353 value = payload[key]
354 try:
355 return Decimal(str(value))
356 except (InvalidOperation, ValueError, TypeError) as exc:
357 raise ValueError(key) from exc
360def _optional_decimal(value: object) -> Decimal | None:
361 if value in (None, ""):
362 return None
363 try:
364 return Decimal(str(value))
365 except (InvalidOperation, ValueError, TypeError) as exc:
366 raise ValueError("target_peak_demand_kw") from exc