Coverage for backend/django/Economics/results/services/comparison/recalculation.py: 94%
87 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Ephemeral financial metric recalculation for study comparisons."""
3from __future__ import annotations
5from pydantic import Field
7from Economics.results.services.financial_metrics.baselines import (
8 manual_baseline_profile_for_study,
9 resolve_manual_baseline_for_target,
10 resolve_study_baseline,
11 target_residual_value,
12)
13from Economics.results.services.financial_metrics.calculation import calculate_financial_metrics
14from Economics.results.services.financial_metrics.contracts import (
15 AssumptionSet,
16 BaselineResolution,
17 FinancialCalculationInputs,
18 FinancialMetricsError,
19 FinancialWarning,
20)
21from Economics.results.services.financial_metrics.input_snapshots import (
22 FinancialInputSnapshot,
23 snapshot_from_result_run,
24)
25from Economics.shared.payloads import json_ready
26from Economics.shared.choices import EconomicsBaselineMode
27from Economics.results.services.lifecycle.runs import ResultRunClassification
28from Economics.settings_profiles.models import EconomicsSettingsProfile
29from Economics.studies.models import EconomicsStudy
31from .contracts import (
32 ComparisonMetricValue,
33 ComparisonStatus,
34 ComparisonTarget,
35 EconomicsComparisonContract,
36 ResultContext,
37 WarningSummaryPayload,
38)
39from Economics.results.services.financial_metrics.metric_catalog import financial_metric_spec, key_financial_metric_specs
42class ComparisonRecalculationResult(EconomicsComparisonContract):
43 """Temporary comparison contexts and non-persistent warnings."""
45 contexts: dict[int | None, ResultContext | None]
46 warnings: list[WarningSummaryPayload] = Field(default_factory=list)
49def comparison_contexts_with_ephemeral_baseline(
50 *,
51 active_study: EconomicsStudy,
52 studies: list[ComparisonTarget],
53 baseline_mode: EconomicsBaselineMode,
54 baseline_id: int,
55 result_contexts: dict[int | None, ResultContext | None],
56) -> ComparisonRecalculationResult:
57 """Recalculate baseline-dependent metrics for this comparison only."""
58 recalculated_contexts = dict(result_contexts)
59 warnings: list[WarningSummaryPayload] = []
60 baseline_study = next(
61 (target.study for target in studies if target.study_id == baseline_id),
62 None,
63 )
64 manual_baseline = (
65 manual_baseline_profile_for_study(active_study)
66 if baseline_mode == EconomicsBaselineMode.MANUAL
67 else None
68 )
69 for target in studies:
70 if target.study is None:
71 continue
72 if baseline_mode == EconomicsBaselineMode.STUDY and target.study_id == baseline_id:
73 continue
74 context = result_contexts.get(target.study_id)
75 if context is None:
76 continue
77 recalculation = _recalculated_context(
78 target_study=target.study,
79 context=context,
80 baseline_mode=baseline_mode,
81 baseline_study=baseline_study,
82 manual_baseline=manual_baseline,
83 )
84 recalculated_contexts[target.study_id] = recalculation.context
85 warnings.extend(recalculation.warnings)
86 return ComparisonRecalculationResult(contexts=recalculated_contexts, warnings=warnings)
89class _TargetRecalculation(EconomicsComparisonContract):
90 """One study's comparison-only context plus warnings produced while rebuilding it."""
92 context: ResultContext
93 warnings: list[WarningSummaryPayload] = Field(default_factory=list)
96def _recalculated_context(
97 *,
98 target_study: EconomicsStudy,
99 context: ResultContext,
100 baseline_mode: EconomicsBaselineMode,
101 baseline_study: EconomicsStudy | None,
102 manual_baseline: EconomicsSettingsProfile | None,
103) -> _TargetRecalculation:
104 """Replace only baseline-dependent rows with freshly calculated comparison values.
106 The incoming ``context`` represents the persisted current result. This helper
107 keeps all non-baseline rows from that stored result and overlays the metrics
108 that depend on the selected comparison baseline, so comparison rendering does
109 not mutate or imply a recalculation of the study's saved result run.
110 """
111 lines = {
112 row_key: value
113 for row_key, value in context.lines.items()
114 if not _row_requires_baseline(row_key)
115 }
116 if context.classification != ResultRunClassification.CURRENT:
117 warning = FinancialWarning(
118 code="comparison_target_result_stale",
119 severity="warning",
120 message="Comparison metrics could not be recalculated because the target study result is stale.",
121 context={
122 "target_study_id": target_study.pk,
123 "result_run_id": context.run.pk,
124 "classification": context.classification,
125 },
126 )
127 return _TargetRecalculation(
128 context=context.model_copy(update={"lines": _with_recalculation_unavailable_rows(lines)}),
129 warnings=[_warning_payload(warning)],
130 )
131 snapshot, warnings = _target_snapshot(target_study=target_study, context=context)
132 if snapshot is None:
133 return _TargetRecalculation(
134 context=context.model_copy(update={"lines": _with_recalculation_unavailable_rows(lines)}),
135 warnings=[_warning_payload(warning) for warning in warnings],
136 )
137 inputs, warnings = _calculation_inputs(
138 target_study=target_study,
139 snapshot=snapshot,
140 baseline_mode=baseline_mode,
141 baseline_study=baseline_study,
142 manual_baseline=manual_baseline,
143 )
144 try:
145 calculation = calculate_financial_metrics(inputs, formula_store=None)
146 except FinancialMetricsError as exc:
147 return _TargetRecalculation(
148 context=context.model_copy(update={"lines": _with_recalculation_unavailable_rows(lines)}),
149 warnings=[
150 WarningSummaryPayload(
151 code=exc.code,
152 severity="error",
153 message=exc.message,
154 context=json_ready(exc.context),
155 )
156 ],
157 )
158 for metric in calculation.metrics.values():
159 spec = financial_metric_spec(metric.key)
160 if spec is None or not spec.baseline_dependent:
161 continue
162 lines[spec.row_key] = ComparisonMetricValue(amount=metric.value, unit=metric.unit)
163 lines = _with_recalculation_unavailable_rows(lines)
164 return _TargetRecalculation(
165 context=context.model_copy(
166 update={
167 "lines": lines,
168 "discounted_cash_flow": calculation.discounted_cash_flow,
169 }
170 ),
171 warnings=[_warning_payload(warning) for warning in calculation.warnings],
172 )
175def _calculation_inputs(
176 *,
177 target_study: EconomicsStudy,
178 snapshot: FinancialInputSnapshot,
179 baseline_mode: EconomicsBaselineMode,
180 baseline_study: EconomicsStudy | None,
181 manual_baseline: EconomicsSettingsProfile | None,
182) -> tuple[FinancialCalculationInputs, list[FinancialWarning]]:
183 """Build financial-calculation inputs from a target snapshot and transient baseline data."""
184 target_assumptions = snapshot.target_assumptions
185 baseline_resolution, warnings = _baseline_resolution(
186 target_study=target_study,
187 target_assumptions=target_assumptions,
188 baseline_mode=baseline_mode,
189 baseline_study=baseline_study,
190 manual_baseline=manual_baseline,
191 )
192 return snapshot.calculation_inputs(
193 baseline_resolution=baseline_resolution,
194 warnings=tuple(warnings),
195 ), warnings
198def _target_snapshot(
199 *,
200 target_study: EconomicsStudy,
201 context: ResultContext,
202) -> tuple[FinancialInputSnapshot | None, list[FinancialWarning]]:
203 """Load the target financial input snapshot used for comparison-only recalculation."""
205 try:
206 return snapshot_from_result_run(context.run), []
207 except FinancialMetricsError as exc:
208 return None, [
209 FinancialWarning(
210 code=exc.code,
211 severity="warning",
212 message=(
213 "Comparison metrics could not be recalculated because the target study's saved financial "
214 "inputs are unavailable."
215 ),
216 context={
217 "target_study_id": target_study.pk,
218 "result_run_id": context.run.pk,
219 **exc.context,
220 },
221 )
222 ]
225def _baseline_resolution(
226 *,
227 target_study: EconomicsStudy,
228 target_assumptions,
229 baseline_mode: EconomicsBaselineMode,
230 baseline_study: EconomicsStudy | None,
231 manual_baseline: EconomicsSettingsProfile | None,
232) -> tuple[BaselineResolution, list[FinancialWarning]]:
233 """Resolve the baseline contract selected for this comparison target.
235 Manual comparison mode reuses the active study's configured manual baseline
236 but applies the target study's assumptions where the shared financial
237 calculation pathway expects target-local values. Study mode delegates to the
238 normal study-baseline resolver so the comparison path cannot drift from the
239 persisted financial metric calculation.
240 """
241 if baseline_mode == EconomicsBaselineMode.MANUAL:
242 if manual_baseline is None: 242 ↛ 243line 242 didn't jump to line 243 because the condition on line 242 was never true
243 return _unavailable_baseline_resolution(
244 target_study=target_study,
245 target_assumptions=target_assumptions,
246 code="comparison_manual_baseline_missing",
247 message="Manual baseline values are not configured for this comparison.",
248 )
249 if manual_baseline.currency and target_assumptions.currency and manual_baseline.currency != target_assumptions.currency: 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true
250 return _unavailable_baseline_resolution(
251 target_study=target_study,
252 target_assumptions=target_assumptions,
253 code="comparison_manual_baseline_currency_mismatch",
254 message="Manual comparison baseline currency does not match the target study currency.",
255 )
256 resolution, warnings = resolve_manual_baseline_for_target(
257 target_study=target_study,
258 baseline_profile=manual_baseline,
259 target_assumptions=target_assumptions,
260 )
261 return resolution.model_copy(update={"residual_value": target_residual_value(target_study)}), warnings
262 return resolve_study_baseline(
263 target_study=target_study,
264 target_assumptions=target_assumptions,
265 baseline_study=baseline_study,
266 )
269def _unavailable_baseline_resolution(
270 *,
271 target_study: EconomicsStudy,
272 target_assumptions,
273 code: str,
274 message: str,
275) -> tuple[BaselineResolution, list[FinancialWarning]]:
276 """Return a target-only baseline resolution that marks recalculation unavailable."""
277 return BaselineResolution(
278 source="comparison",
279 is_guided_default=True,
280 capex=None,
281 annual_opex=None,
282 annual_heat_basis=None,
283 annual_heat_basis_unit=None,
284 residual_value=target_residual_value(target_study),
285 project_lifetime_years=target_assumptions.project_lifetime_years,
286 discount_rate_percent=target_assumptions.discount_rate_percent,
287 assumptions=AssumptionSet.from_mapping(
288 {
289 "baseline_source": "comparison",
290 "target_study_id": target_study.pk,
291 "residual_value": target_residual_value(target_study),
292 }
293 ),
294 ), [
295 FinancialWarning(
296 code=code,
297 severity="warning",
298 message=message,
299 context={"target_study_id": target_study.pk},
300 )
301 ]
304def _row_requires_baseline(row_key: str) -> bool:
305 """Ask the central metric catalog whether a row must be recalculated for comparisons."""
306 spec = next((spec for spec in key_financial_metric_specs() if spec.row_key == row_key), None)
307 return spec.baseline_dependent if spec is not None else False
310def _with_recalculation_unavailable_rows(
311 lines: dict[str, ComparisonMetricValue],
312) -> dict[str, ComparisonMetricValue]:
313 """Insert placeholder values for baseline-dependent rows that could not be rebuilt."""
314 next_lines = dict(lines)
315 for spec in key_financial_metric_specs():
316 if spec.baseline_dependent and spec.row_key not in next_lines:
317 next_lines[spec.row_key] = ComparisonMetricValue(
318 amount=None,
319 unit="",
320 comparison_status=ComparisonStatus.RECALCULATION_UNAVAILABLE,
321 )
322 return next_lines
325def _warning_payload(warning: FinancialWarning) -> WarningSummaryPayload:
326 """Convert shared financial warnings into the comparison response contract."""
327 return WarningSummaryPayload(
328 code=warning.code,
329 severity=warning.severity,
330 message=warning.message,
331 context=json_ready(warning.context),
332 )