Coverage for backend/django/Economics/results/services/financial_metrics.py: 85%
495 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-06-23 21:51 +0000
1"""Pure financial metric contracts and v1 calculation formulas.
3The public functions in this module either resolve persisted Economics study
4state at the boundary or calculate from already-resolved numeric inputs. Return
5values are frozen Pydantic models so metrics, warnings, baseline resolution, and
6cash-flow rows have explicit serialization contracts before they are copied into
7result-line JSON payloads or future API responses.
8"""
10from __future__ import annotations
12from collections.abc import Mapping
13from decimal import Decimal, InvalidOperation, ROUND_HALF_UP, localcontext
14from typing import Any, TypeAlias
16from pydantic import BaseModel, ConfigDict, Field, field_validator
18from Economics.settings_profiles.models import EconomicsSettingsProfile
20from Economics.studies.models import EconomicsStudy
22from Economics.costing.models import OperatingCostLine
24from Economics.shared.choices import OperatingLineCategory
25from Economics.costing.capital.custom_capital_lines import base_capex_for_custom_percentage_lines
26from Economics.costing.capital.electrical_upgrade import derive_peak_demand_basis
27from Economics.settings_profiles.services.depreciation import build_straight_line_depreciation_schedule
28from Economics.formulas.builders.capital import (
29 build_electrical_upgrade_formula,
30 build_target_total_capex_formula,
31)
32from Economics.formulas.engine.core import FormulaError, FormulaEvaluation
33from Economics.formulas.builders.metrics import (
34 BoundMetricFormula,
35 annual_profit_formula,
36 annual_savings_formula,
37 after_tax_annual_cash_flow_formula,
38 cash_flow_formula,
39 cumulative_cash_flow_formula,
40 cumulative_present_value_formula,
41 discounted_cash_flow_formula,
42 discount_factor_formula,
43 depreciation_tax_shield_formula,
44 incremental_capex_formula,
45 lcoh_formula,
46 metric_value_formula,
47 npv_formula,
48 roi_percent_formula,
49)
50from Economics.formulas.builders.operating import (
51 build_annual_operating_expense_formula,
52 build_annual_operating_revenue_formula,
53 build_operating_line_formula,
54)
55from Economics.formulas.builders.metric_formulas import MetricFormulaStore
56from Economics.costing.operating.line_calculation import annualized_basis_quantity
57from Economics.costing.operating.resource_basis import (
58 TARGET_PROCESS_ENERGY_BASIS_UNIT,
59 TargetProcessEnergyBasis,
60 derive_target_process_energy_basis,
61)
62from Economics.settings_profiles.services.settings_profiles import get_settings_profile
65ZERO = Decimal("0")
66ONE = Decimal("1")
67FinancialScalar: TypeAlias = str | int | Decimal | bool | None
68FinancialContextValue: TypeAlias = FinancialScalar | tuple[str, ...] | tuple[int, ...] | tuple[dict[str, str], ...] | dict[str, str]
69FinancialContext: TypeAlias = dict[str, FinancialContextValue]
72class EconomicsContract(BaseModel):
73 model_config = ConfigDict(frozen=True, allow_inf_nan=True)
76class AssumptionRecord(EconomicsContract):
77 """One audit assumption exposed through financial metric contracts."""
79 key: str
80 value: FinancialScalar
83class AssumptionSet(EconomicsContract):
84 """Typed assumption collection used internally before JSON persistence.
86 Result lines and future APIs can serialize this model deterministically,
87 while callers that need the legacy JSONField boundary can explicitly ask for
88 a scalar mapping with ``to_mapping``.
89 """
91 records: tuple[AssumptionRecord, ...] = ()
93 @classmethod
94 def from_mapping(cls, values: Mapping[str, object] | None = None) -> "AssumptionSet":
95 if not values: 95 ↛ 96line 95 didn't jump to line 96 because the condition on line 95 was never true
96 return cls()
97 return cls(records=tuple(AssumptionRecord(key=str(key), value=_financial_scalar(value)) for key, value in sorted(values.items())))
99 def merge(self, values: Mapping[str, object] | None = None, **kwargs: object) -> "AssumptionSet":
100 merged: dict[str, object] = self.to_mapping()
101 if values: 101 ↛ 103line 101 didn't jump to line 103 because the condition on line 101 was always true
102 merged.update(values)
103 merged.update(kwargs)
104 return AssumptionSet.from_mapping(merged)
106 def merge_set(self, other: "AssumptionSet") -> "AssumptionSet":
107 return self.merge(other.to_mapping())
109 def to_mapping(self) -> dict[str, FinancialScalar]:
110 """Return the legacy JSON-boundary mapping used by result lines."""
111 return {record.key: record.value for record in self.records}
113 def get(self, key: str, default: FinancialScalar = None) -> FinancialScalar:
114 return self.to_mapping().get(key, default)
116 def __getitem__(self, key: str) -> FinancialScalar:
117 return self.to_mapping()[key]
120class TargetAssumptions(EconomicsContract):
121 """Financial assumptions resolved from the target study boundary."""
123 target_study_id: int
124 assumptions_id: int | None = None
125 project_lifetime_years: int | None = None
126 discount_rate_percent: Decimal | None = None
127 currency: str | None = None
128 basis_date: str | None = None
129 inflation_method: str = ""
130 capital_index_series_id: int | None = None
131 operating_index_series_id: int | None = None
132 annual_operating_hours: Decimal | None = None
133 tax_rate_percent: Decimal | None = None
134 depreciation_enabled: bool = False
135 default_depreciation_life_years: int | None = None
136 default_depreciation_salvage_percent: Decimal | None = None
137 contingency_percent: Decimal | None = None
138 electrical_upgrade_rate_amount: Decimal | None = None
139 electrical_upgrade_rate_unit: str = "NZD/kW"
140 peak_demand_kw: Decimal | None = None
141 default_lang_factor: Decimal | None = None
142 assumptions_source: str = "study"
144 def as_assumption_set(self) -> AssumptionSet:
145 return AssumptionSet.from_mapping(
146 {
147 "target_study_id": self.target_study_id,
148 "assumptions_id": self.assumptions_id,
149 "project_lifetime_years": self.project_lifetime_years,
150 "discount_rate_percent": self.discount_rate_percent,
151 "currency": self.currency,
152 "basis_date": self.basis_date,
153 "inflation_method": self.inflation_method,
154 "capital_index_series_id": self.capital_index_series_id,
155 "operating_index_series_id": self.operating_index_series_id,
156 "annual_operating_hours": self.annual_operating_hours,
157 "tax_rate_percent": self.tax_rate_percent,
158 "depreciation_enabled": self.depreciation_enabled,
159 "default_depreciation_life_years": self.default_depreciation_life_years,
160 "default_depreciation_salvage_percent": self.default_depreciation_salvage_percent,
161 "contingency_percent": self.contingency_percent,
162 "electrical_upgrade_rate_amount": self.electrical_upgrade_rate_amount,
163 "electrical_upgrade_rate_unit": self.electrical_upgrade_rate_unit,
164 "peak_demand_kw": self.peak_demand_kw,
165 "default_lang_factor": self.default_lang_factor,
166 "assumptions_source": self.assumptions_source,
167 }
168 )
171class FinancialWarning(EconomicsContract):
172 code: str
173 severity: str
174 message: str
175 context: FinancialContext = Field(default_factory=dict)
178class FinancialMetric(EconomicsContract):
179 key: str
180 value: Decimal | None
181 unit: str
182 assumptions: AssumptionSet = Field(default_factory=AssumptionSet)
183 status: str = "calculated"
184 formula_audit: dict[str, Any] | None = None
185 formula_record_id: int | None = None
187 @classmethod
188 def from_value(
189 cls,
190 *,
191 key: str,
192 value: Decimal | None,
193 unit: str,
194 assumptions: AssumptionSet | None = None,
195 status: str = "calculated",
196 ) -> "FinancialMetric":
197 return cls(
198 key=key,
199 value=value,
200 unit=unit,
201 assumptions=assumptions or AssumptionSet(),
202 status=status if value is not None else "unavailable",
203 )
205 @classmethod
206 def from_formula(
207 cls,
208 *,
209 key: str,
210 formula: BoundMetricFormula,
211 assumptions: AssumptionSet | None = None,
212 status: str = "calculated",
213 value: Decimal | None = None,
214 formula_store: MetricFormulaStore | None = None,
215 ) -> "FinancialMetric":
216 evaluated_value = formula.evaluate()
217 metric_value = evaluated_value if value is None else value
218 metric_status = status if metric_value is not None else "unavailable"
219 formula_audit = formula.formula.audit_payload(
220 FormulaEvaluation(value=metric_value, bindings=formula.bindings)
221 )
222 formula_record_id = None
223 if formula_store is not None:
224 persisted_formula = formula_store.persist_metric_formula(
225 metric_key=key,
226 formula=formula,
227 value=metric_value,
228 status=metric_status,
229 assumptions=assumptions or AssumptionSet(),
230 formula_audit=formula_audit,
231 )
232 formula_record_id = persisted_formula.id
233 formula_audit = persisted_formula.formula_audit
234 return cls(
235 key=key,
236 value=metric_value,
237 unit=formula.formula.unit,
238 assumptions=assumptions or AssumptionSet(),
239 status=metric_status,
240 formula_audit=formula_audit,
241 formula_record_id=formula_record_id,
242 )
245class DiscountedCashFlowRow(EconomicsContract):
246 year: int
247 cash_flow: Decimal
248 discount_factor: Decimal
249 present_value: Decimal
250 cumulative_cash_flow: Decimal
251 cumulative_present_value: Decimal
254class BaselineResolution(EconomicsContract):
255 source: str
256 is_guided_default: bool
257 capex: Decimal | None
258 annual_opex: Decimal | None
259 annual_heat_basis: Decimal | None
260 annual_heat_basis_unit: str | None
261 residual_value: Decimal
262 project_lifetime_years: int | None
263 discount_rate_percent: Decimal | None
264 assumptions: AssumptionSet = Field(default_factory=AssumptionSet)
267class FinancialCalculationInputs(EconomicsContract):
268 target_capex: Decimal
269 target_annual_opex: Decimal
270 target_annual_revenue: Decimal = ZERO
271 target_annual_depreciation: Decimal = ZERO
272 target_purchase_basis_capex: Decimal = ZERO
273 target_installed_basis_capex: Decimal = ZERO
274 target_contingency_capex: Decimal = ZERO
275 target_electrical_upgrade_capex: Decimal = ZERO
276 target_peak_demand_kw: Decimal | None = None
277 baseline_capex: Decimal | None
278 baseline_annual_opex: Decimal | None
279 target_annual_process_energy_basis: Decimal | None
280 target_annual_process_energy_basis_unit: str | None = TARGET_PROCESS_ENERGY_BASIS_UNIT
281 target_process_energy_basis_source_row_keys: str = ""
282 target_process_energy_basis_contributor_count: int = 0
283 project_lifetime_years: int | None
284 discount_rate_percent: Decimal | None
285 tax_rate_percent: Decimal | None = ZERO
286 residual_value: Decimal = ZERO
287 baseline_fully_calculated: bool = False
288 assumptions: AssumptionSet = Field(default_factory=AssumptionSet)
289 warnings: tuple[FinancialWarning, ...] = ()
291 @field_validator("assumptions", mode="before")
292 @classmethod
293 def coerce_assumptions(cls, value):
294 if isinstance(value, AssumptionSet):
295 return value
296 if value is None: 296 ↛ 297line 296 didn't jump to line 297 because the condition on line 296 was never true
297 return AssumptionSet()
298 if isinstance(value, Mapping): 298 ↛ 300line 298 didn't jump to line 300 because the condition on line 298 was always true
299 return AssumptionSet.from_mapping(value)
300 return value
303class FinancialMetricsResult(EconomicsContract):
304 metrics: dict[str, FinancialMetric]
305 discounted_cash_flow: tuple[DiscountedCashFlowRow, ...]
306 baseline_resolution: BaselineResolution
307 warnings: tuple[FinancialWarning, ...]
310class FinancialMetricsError(ValueError):
311 def __init__(self, code: str, message: str, *, context: FinancialContext | None = None):
312 super().__init__(message)
313 self.code = code
314 self.message = message
315 self.context = context or {}
318def calculate_study_financial_metrics(study: EconomicsStudy) -> FinancialMetricsResult:
319 """Resolve persisted study inputs and calculate fixed v1 financial metrics.
321 This is the database boundary for the service. It resolves target study
322 assumptions, persisted capital/operating lines, and baseline state before
323 delegating to ``calculate_financial_metrics``, which remains pure.
324 """
325 formula_store = MetricFormulaStore(study)
326 try:
327 target_capex = _sum_capital_lines(study)
328 capital_breakdown = _sum_capital_breakdown(study)
329 depreciation_schedule = build_straight_line_depreciation_schedule(study)
330 target_annual_opex, target_annual_revenue = _sum_operating_lines(study)
331 target_process_energy_basis = derive_target_process_energy_basis(study)
332 peak_demand_basis = derive_peak_demand_basis(study)
333 target_assumptions = _target_assumptions(study)
334 baseline_resolution, warnings = resolve_baseline_for_study(
335 study=study,
336 target_capex=target_capex,
337 target_annual_opex=target_annual_opex,
338 target_assumptions=target_assumptions,
339 )
340 calculation = calculate_financial_metrics(
341 FinancialCalculationInputs(
342 target_capex=target_capex,
343 target_annual_opex=target_annual_opex,
344 target_annual_revenue=target_annual_revenue,
345 target_annual_depreciation=depreciation_schedule.annual_depreciation,
346 target_purchase_basis_capex=capital_breakdown["purchase_basis"],
347 target_installed_basis_capex=capital_breakdown["installed_basis"],
348 target_contingency_capex=capital_breakdown["contingency"],
349 target_electrical_upgrade_capex=capital_breakdown["electrical_upgrade"],
350 target_peak_demand_kw=peak_demand_basis.quantity_kw,
351 baseline_capex=baseline_resolution.capex,
352 baseline_annual_opex=baseline_resolution.annual_opex,
353 target_annual_process_energy_basis=target_process_energy_basis.quantity,
354 target_annual_process_energy_basis_unit=target_process_energy_basis.unit,
355 target_process_energy_basis_source_row_keys=_process_energy_basis_source_row_keys(
356 target_process_energy_basis
357 ),
358 target_process_energy_basis_contributor_count=len(target_process_energy_basis.contributions),
359 project_lifetime_years=baseline_resolution.project_lifetime_years,
360 discount_rate_percent=baseline_resolution.discount_rate_percent,
361 tax_rate_percent=target_assumptions.tax_rate_percent,
362 residual_value=baseline_resolution.residual_value,
363 baseline_fully_calculated=(
364 not baseline_resolution.is_guided_default
365 and baseline_resolution.capex is not None
366 and baseline_resolution.annual_opex is not None
367 ),
368 assumptions=target_assumptions.as_assumption_set().merge_set(baseline_resolution.assumptions),
369 warnings=tuple(warnings),
370 ),
371 formula_store=formula_store,
372 )
373 formula_store.delete_stale_metric_formulas()
374 return FinancialMetricsResult(
375 metrics=calculation.metrics,
376 discounted_cash_flow=calculation.discounted_cash_flow,
377 baseline_resolution=baseline_resolution,
378 warnings=calculation.warnings,
379 )
380 except FinancialMetricsError as exc:
381 formula_store.delete_all_metric_formulas()
382 target_assumptions = _target_assumptions(study)
383 return FinancialMetricsResult(
384 metrics={},
385 discounted_cash_flow=(),
386 baseline_resolution=_target_only_baseline_resolution(
387 study=study,
388 target_assumptions=target_assumptions,
389 ),
390 warnings=(
391 FinancialWarning(
392 code=exc.code,
393 severity="error",
394 message=exc.message,
395 context=exc.context,
396 ),
397 ),
398 )
399 except Exception:
400 formula_store.delete_all_metric_formulas()
401 raise
404def target_base_capital_cost(study: EconomicsStudy) -> Decimal:
405 """Return the generated unit-operation CAPEX subtotal for native properties."""
407 return base_capex_for_custom_percentage_lines(study)
410def target_annual_operating_expense(study: EconomicsStudy) -> Decimal:
411 """Return annual operating expenses before output revenue offsets."""
413 expense_total, _revenue_total = _sum_operating_lines(study)
414 return expense_total
417def calculate_financial_metrics(
418 inputs: FinancialCalculationInputs,
419 *,
420 formula_store: MetricFormulaStore | None = None,
421) -> FinancialMetricsResult:
422 """Calculate deterministic v1 metrics from already-resolved numeric inputs.
424 The formulas are fixed for v1: capex is a year-0 outflow, annual savings
425 are baseline opex minus target opex plus target revenue, discounted cash
426 flow uses the supplied lifetime/rate, ROI requires positive incremental
427 capital outlay, and LCOH is hidden until a positive heat basis exists.
428 """
429 warnings = list(inputs.warnings)
430 assumptions = inputs.assumptions
431 metrics = _base_metrics(inputs=inputs, assumptions=assumptions, formula_store=formula_store)
432 annual_savings = _add_annual_savings_metric(
433 inputs=inputs,
434 assumptions=assumptions,
435 metrics=metrics,
436 warnings=warnings,
437 formula_store=formula_store,
438 )
439 annual_cash_flow = _add_tax_metrics(
440 inputs=inputs,
441 assumptions=assumptions,
442 annual_savings=annual_savings,
443 metrics=metrics,
444 formula_store=formula_store,
445 )
446 _discount_rate_from_percent(inputs.discount_rate_percent)
447 incremental_capex = (
448 (incremental_capex_formula(
449 target_capex=inputs.target_capex,
450 baseline_capex=inputs.baseline_capex,
451 unit=_currency_unit(assumptions),
452 ).evaluate() or ZERO).quantize(inputs.target_capex, rounding=ROUND_HALF_UP)
453 if inputs.baseline_capex is not None
454 else None
455 )
456 _add_incremental_capex_metric(
457 inputs=inputs,
458 assumptions=assumptions,
459 incremental_capex=incremental_capex,
460 metrics=metrics,
461 warnings=warnings,
462 formula_store=formula_store,
463 )
464 discounted_cash_flow = _add_cashflow_metrics(
465 inputs=inputs,
466 assumptions=assumptions,
467 incremental_capex=incremental_capex,
468 annual_savings=annual_savings,
469 annual_cash_flow=annual_cash_flow,
470 metrics=metrics,
471 warnings=warnings,
472 formula_store=formula_store,
473 )
474 _add_lcoh_metric(
475 inputs=inputs,
476 assumptions=assumptions,
477 metrics=metrics,
478 warnings=warnings,
479 formula_store=formula_store,
480 )
482 return FinancialMetricsResult(
483 metrics=metrics,
484 discounted_cash_flow=discounted_cash_flow,
485 baseline_resolution=_pure_calculation_baseline_resolution(inputs=inputs),
486 warnings=tuple(warnings),
487 )
490def validate_manual_baseline(baseline: EconomicsSettingsProfile, *, target_assumptions: TargetAssumptions) -> None:
491 """Validate the full manual baseline before presenting calculated metrics."""
492 errors = _manual_baseline_errors(baseline, target_assumptions=target_assumptions)
493 if errors:
494 raise FinancialMetricsError(
495 "manual_baseline_invalid",
496 "Manual economics baseline is incomplete or invalid.",
497 context={"errors": errors, "baseline_id": baseline.pk},
498 )
501def _manual_baseline_errors(baseline: EconomicsSettingsProfile, *, target_assumptions: TargetAssumptions) -> dict[str, str]:
502 """Return field-addressable blockers for calculations that depend on the manual baseline."""
503 errors: dict[str, str] = {}
504 if baseline.manual_capex is None: 504 ↛ 505line 504 didn't jump to line 505 because the condition on line 504 was never true
505 errors["manual_capex"] = "Manual baseline capex is required."
506 elif baseline.manual_capex < 0: 506 ↛ 507line 506 didn't jump to line 507 because the condition on line 506 was never true
507 errors["manual_capex"] = "Manual baseline capex cannot be negative."
509 if baseline.manual_annual_opex is None:
510 errors["manual_annual_opex"] = "Manual baseline annual opex is required."
511 elif baseline.manual_annual_opex < 0: 511 ↛ 512line 511 didn't jump to line 512 because the condition on line 511 was never true
512 errors["manual_annual_opex"] = "Manual baseline annual opex cannot be negative."
514 if baseline.annual_heat_basis_mode == "average_power":
515 if baseline.average_power_input is None: 515 ↛ 516line 515 didn't jump to line 516 because the condition on line 515 was never true
516 errors["average_power_input"] = "Hourly heat quantity is required when the baseline heat basis uses hourly heat."
517 elif baseline.average_power_input <= 0: 517 ↛ 518line 517 didn't jump to line 518 because the condition on line 517 was never true
518 errors["average_power_input"] = "Hourly heat quantity must be positive when supplied."
519 if target_assumptions.annual_operating_hours is None: 519 ↛ 520line 519 didn't jump to line 520 because the condition on line 519 was never true
520 errors["annual_operating_hours"] = "Annual operating hours are required for an hourly-heat baseline heat basis."
521 elif target_assumptions.annual_operating_hours <= 0: 521 ↛ 522line 521 didn't jump to line 522 because the condition on line 521 was never true
522 errors["annual_operating_hours"] = "Annual operating hours must be positive for an hourly-heat baseline heat basis."
523 elif baseline.manual_annual_heat_basis is not None and baseline.manual_annual_heat_basis <= 0: 523 ↛ 524line 523 didn't jump to line 524 because the condition on line 523 was never true
524 errors["manual_annual_heat_basis"] = "Manual annual heat basis must be positive when supplied."
526 if target_assumptions.project_lifetime_years is None: 526 ↛ 527line 526 didn't jump to line 527 because the condition on line 526 was never true
527 errors["project_lifetime_years"] = "Study project lifetime is required for manual baseline metrics."
528 elif target_assumptions.project_lifetime_years <= 0: 528 ↛ 529line 528 didn't jump to line 529 because the condition on line 528 was never true
529 errors["project_lifetime_years"] = "Study project lifetime must be positive for manual baseline metrics."
531 if target_assumptions.discount_rate_percent is None: 531 ↛ 532line 531 didn't jump to line 532 because the condition on line 531 was never true
532 errors["discount_rate_percent"] = "Study discount rate is required for manual baseline metrics."
533 elif target_assumptions.discount_rate_percent < 0: 533 ↛ 534line 533 didn't jump to line 534 because the condition on line 533 was never true
534 errors["discount_rate_percent"] = "Study discount rate cannot be negative for manual baseline metrics."
536 if baseline.residual_value is not None and baseline.residual_value < 0: 536 ↛ 537line 536 didn't jump to line 537 because the condition on line 536 was never true
537 errors["residual_value"] = "Residual value cannot be negative."
539 return errors
542def resolve_baseline_for_study(
543 *,
544 study: EconomicsStudy,
545 target_capex: Decimal,
546 target_annual_opex: Decimal,
547 target_assumptions: TargetAssumptions,
548) -> tuple[BaselineResolution, list[FinancialWarning]]:
549 """Resolve baseline inputs using explicit v1 precedence rules.
551 V1 resolves manual baselines only. Incomplete manual drafts are preserved
552 as source state, but field-addressable warnings block dependent metrics
553 until required values are present.
554 """
555 warnings = _target_assumption_warnings(study=study, target_assumptions=target_assumptions)
556 baseline = _get_baseline(study)
557 if baseline is None:
558 return _resolve_missing_baseline(study=study, target_assumptions=target_assumptions, warnings=warnings)
560 resolution, baseline_warnings = _resolve_manual_baseline(
561 study=study,
562 baseline=baseline,
563 target_assumptions=target_assumptions,
564 )
565 warnings.extend(baseline_warnings)
566 return resolution, warnings
569def _resolve_missing_baseline(
570 *,
571 study: EconomicsStudy,
572 target_assumptions: TargetAssumptions,
573 warnings: list[FinancialWarning],
574) -> tuple[BaselineResolution, list[FinancialWarning]]:
575 warnings.append(
576 FinancialWarning(
577 code="baseline_missing",
578 severity="warning",
579 message="No baseline has been configured; savings and comparison metrics are not fully calculated.",
580 context={"study_id": study.pk},
581 )
582 )
583 return _target_only_baseline_resolution(study=study, target_assumptions=target_assumptions), warnings
586def _resolve_manual_baseline(
587 *,
588 study: EconomicsStudy,
589 baseline: EconomicsSettingsProfile,
590 target_assumptions: TargetAssumptions,
591) -> tuple[BaselineResolution, list[FinancialWarning]]:
592 errors = _manual_baseline_errors(baseline, target_assumptions=target_assumptions)
593 is_incomplete = bool(errors)
594 warnings = []
595 if errors:
596 warnings.append(
597 FinancialWarning(
598 code="manual_baseline_invalid",
599 severity="error",
600 message="Manual economics baseline is incomplete or invalid.",
601 context={"errors": errors, "baseline_id": baseline.pk},
602 )
603 )
604 return BaselineResolution(
605 source="manual",
606 is_guided_default=is_incomplete,
607 capex=baseline.manual_capex,
608 annual_opex=baseline.manual_annual_opex,
609 annual_heat_basis=_resolve_baseline_annual_heat_basis(
610 study=study,
611 baseline=baseline,
612 target_assumptions=target_assumptions,
613 ),
614 annual_heat_basis_unit=_resolve_baseline_annual_heat_basis_unit(baseline),
615 residual_value=baseline.residual_value or ZERO,
616 project_lifetime_years=target_assumptions.project_lifetime_years,
617 discount_rate_percent=target_assumptions.discount_rate_percent,
618 assumptions=AssumptionSet.from_mapping(
619 {
620 "baseline_id": baseline.pk,
621 "baseline_source": "manual",
622 "baseline_capex_source": "manual_capex",
623 "baseline_annual_opex_source": "manual_annual_opex",
624 "annual_heat_basis_mode": baseline.annual_heat_basis_mode,
625 "annual_heat_basis": _resolve_baseline_annual_heat_basis(
626 study=study,
627 baseline=baseline,
628 target_assumptions=target_assumptions,
629 ),
630 "annual_heat_basis_unit": _resolve_baseline_annual_heat_basis_unit(baseline),
631 "average_power_input": baseline.average_power_input,
632 "average_power_unit": baseline.average_power_unit,
633 "currency_source": "study_assumptions",
634 "basis_date_source": "study_assumptions",
635 "project_lifetime_source": "study_assumptions",
636 "discount_rate_source": "study_assumptions",
637 "index_method_source": "study_assumptions",
638 "residual_value": baseline.residual_value or ZERO,
639 }
640 ),
641 ), warnings
644def _resolve_baseline_annual_heat_basis(
645 *,
646 study: EconomicsStudy,
647 baseline: EconomicsSettingsProfile,
648 target_assumptions: TargetAssumptions,
649) -> Decimal | None:
650 """Resolve the baseline heat basis from explicit annual heat or hourly heat quantity."""
651 if baseline.annual_heat_basis_mode != "average_power":
652 return baseline.manual_annual_heat_basis
653 if baseline.average_power_input is None or target_assumptions.annual_operating_hours is None: 653 ↛ 654line 653 didn't jump to line 654 because the condition on line 653 was never true
654 return None
655 if baseline.average_power_input <= 0 or target_assumptions.annual_operating_hours <= 0: 655 ↛ 656line 655 didn't jump to line 656 because the condition on line 655 was never true
656 return None
657 return annualized_basis_quantity(
658 baseline.average_power_input,
659 source_unit=baseline.average_power_unit,
660 target_unit="GJ",
661 study=study,
662 )
665def _resolve_baseline_annual_heat_basis_unit(baseline: EconomicsSettingsProfile) -> str:
666 if baseline.annual_heat_basis_mode == "average_power":
667 return "GJ/year"
668 return baseline.manual_annual_heat_basis_unit
671def calculate_discounted_cash_flow(
672 *,
673 incremental_capex: Decimal,
674 annual_cash_flow: Decimal,
675 project_lifetime_years: int,
676 discount_rate_percent: Decimal,
677 residual_value: Decimal = ZERO,
678) -> list[DiscountedCashFlowRow]:
679 discount_rate = _discount_rate_from_percent(discount_rate_percent)
680 rows: list[DiscountedCashFlowRow] = []
681 for year in range(0, project_lifetime_years + 1):
682 cash_flow_formula_result = cash_flow_formula(
683 year=year,
684 project_lifetime_years=project_lifetime_years,
685 incremental_capex=incremental_capex,
686 annual_cash_flow=annual_cash_flow,
687 residual_value=residual_value,
688 unit="currency",
689 )
690 cash_flow = cash_flow_formula_result.evaluate()
691 if year == 0:
692 cash_flow = _quantize_like(cash_flow, incremental_capex)
693 elif year == project_lifetime_years:
694 cash_flow = _quantize_like(cash_flow, annual_cash_flow, residual_value)
695 else:
696 cash_flow = _quantize_like(cash_flow, annual_cash_flow)
697 discount_factor = _discount_factor(discount_rate=discount_rate, year=year)
698 present_value = discounted_cash_flow_formula(
699 year=year,
700 cash_flow=cash_flow,
701 discount_rate=discount_rate,
702 unit="currency",
703 ).evaluate()
704 cumulative_cash_flow = cumulative_cash_flow_formula(
705 year=year,
706 project_lifetime_years=project_lifetime_years,
707 incremental_capex=incremental_capex,
708 annual_cash_flow=annual_cash_flow,
709 residual_value=residual_value,
710 unit="currency",
711 ).evaluate()
712 cumulative_present_value = cumulative_present_value_formula(
713 key=f"cumulative_present_value_year_{year}",
714 year=year,
715 project_lifetime_years=project_lifetime_years,
716 incremental_capex=incremental_capex,
717 annual_cash_flow=annual_cash_flow,
718 discount_rate=discount_rate,
719 residual_value=residual_value,
720 unit="currency",
721 ).evaluate()
722 rows.append(
723 DiscountedCashFlowRow(
724 year=year,
725 cash_flow=cash_flow,
726 discount_factor=discount_factor,
727 present_value=present_value,
728 cumulative_cash_flow=cumulative_cash_flow,
729 cumulative_present_value=cumulative_present_value,
730 )
731 )
732 return rows
735def calculate_simple_payback_years(
736 *,
737 incremental_capex: Decimal,
738 annual_cash_flow: Decimal,
739 project_lifetime_years: int,
740 residual_value: Decimal = ZERO,
741) -> Decimal | None:
742 if incremental_capex <= 0:
743 return ZERO
744 cumulative = -incremental_capex
745 for year in range(1, project_lifetime_years + 1):
746 cash_flow = annual_cash_flow + (residual_value if year == project_lifetime_years else ZERO)
747 next_cumulative = cumulative + cash_flow
748 if cash_flow > 0 and next_cumulative >= 0:
749 return Decimal(year - 1) + (-cumulative / cash_flow)
750 cumulative = next_cumulative
751 return None
754def calculate_roi_percent(
755 *,
756 incremental_capex: Decimal,
757 annual_cash_flow: Decimal,
758 project_lifetime_years: int,
759 residual_value: Decimal = ZERO,
760) -> Decimal | None:
761 if incremental_capex <= 0:
762 return None
763 return roi_percent_formula(
764 incremental_capex=incremental_capex,
765 annual_cash_flow=annual_cash_flow,
766 project_lifetime_years=project_lifetime_years,
767 residual_value=residual_value,
768 ).evaluate()
771def calculate_lcoh(
772 *,
773 target_capex: Decimal,
774 target_annual_opex: Decimal,
775 target_annual_process_energy_basis: Decimal | None,
776 project_lifetime_years: int | None,
777 discount_rate_percent: Decimal | None,
778 residual_value: Decimal = ZERO,
779) -> Decimal | None:
780 if discount_rate_percent is None:
781 return None
782 discount_rate = _discount_rate_from_percent(discount_rate_percent)
783 if (
784 target_annual_process_energy_basis is None
785 or target_annual_process_energy_basis <= 0
786 or project_lifetime_years is None
787 or project_lifetime_years <= 0
788 ):
789 return None
790 return lcoh_formula(
791 target_capex=target_capex,
792 target_annual_opex=target_annual_opex,
793 target_annual_process_energy_basis=target_annual_process_energy_basis,
794 project_lifetime_years=project_lifetime_years,
795 discount_rate=discount_rate,
796 residual_value=residual_value,
797 unit="currency/energy",
798 ).evaluate()
801def _base_metrics(
802 *,
803 inputs: FinancialCalculationInputs,
804 assumptions: AssumptionSet,
805 formula_store: MetricFormulaStore | None,
806) -> dict[str, FinancialMetric]:
807 """Create always-visible target-cost metrics before baseline comparisons."""
808 currency_unit = _currency_unit(assumptions)
809 annual_currency_unit = _currency_unit(assumptions, per_year=True)
810 target_annual_profit_formula = annual_profit_formula(
811 target_annual_revenue=inputs.target_annual_revenue,
812 target_annual_opex=inputs.target_annual_opex,
813 unit=annual_currency_unit,
814 )
815 return {
816 "capex": _metric_from_scalar_formula(
817 key="capex",
818 value=inputs.target_capex,
819 unit=currency_unit,
820 assumptions=assumptions,
821 formula_store=formula_store,
822 ),
823 "purchase_basis_equipment": _metric_from_scalar_formula(
824 key="purchase_basis_equipment",
825 value=inputs.target_purchase_basis_capex,
826 unit=currency_unit,
827 assumptions=assumptions,
828 formula_store=formula_store,
829 ),
830 "installed_basis_equipment": _metric_from_scalar_formula(
831 key="installed_basis_equipment",
832 value=inputs.target_installed_basis_capex,
833 unit=currency_unit,
834 assumptions=assumptions,
835 formula_store=formula_store,
836 ),
837 "contingency": _metric_from_scalar_formula(
838 key="contingency",
839 value=inputs.target_contingency_capex,
840 unit=currency_unit,
841 assumptions=assumptions,
842 formula_store=formula_store,
843 ),
844 "electrical_upgrade": _metric_from_scalar_formula(
845 key="electrical_upgrade",
846 value=inputs.target_electrical_upgrade_capex,
847 unit=currency_unit,
848 assumptions=assumptions,
849 formula_store=formula_store,
850 ),
851 "peak_demand": _metric_from_scalar_formula(
852 key="peak_demand",
853 value=inputs.target_peak_demand_kw,
854 unit="kW",
855 assumptions=assumptions,
856 formula_store=formula_store,
857 ),
858 "annual_opex": _metric_from_scalar_formula(
859 key="annual_opex",
860 value=inputs.target_annual_opex,
861 unit=annual_currency_unit,
862 assumptions=assumptions,
863 formula_store=formula_store,
864 ),
865 "annual_revenue": _metric_from_scalar_formula(
866 key="annual_revenue",
867 value=inputs.target_annual_revenue,
868 unit=annual_currency_unit,
869 assumptions=assumptions,
870 formula_store=formula_store,
871 ),
872 "annual_depreciation": _metric_from_scalar_formula(
873 key="annual_depreciation",
874 value=inputs.target_annual_depreciation,
875 unit=annual_currency_unit,
876 assumptions=assumptions,
877 formula_store=formula_store,
878 ),
879 "annual_profit": FinancialMetric.from_formula(
880 key="annual_profit",
881 formula=target_annual_profit_formula,
882 assumptions=assumptions.merge(
883 {
884 "target_annual_opex": inputs.target_annual_opex,
885 "target_annual_revenue": inputs.target_annual_revenue,
886 }
887 ),
888 formula_store=formula_store,
889 ),
890 }
893def _metric_from_scalar_formula(
894 *,
895 key: str,
896 value: Decimal | None,
897 unit: str,
898 assumptions: AssumptionSet,
899 status: str = "calculated",
900 formula_store: MetricFormulaStore | None = None,
901) -> FinancialMetric:
902 if value is None:
903 return FinancialMetric.from_value(
904 key=key,
905 value=None,
906 unit=unit,
907 assumptions=assumptions,
908 status=status,
909 )
910 return FinancialMetric.from_formula(
911 key=key,
912 formula=metric_value_formula(key=key, value=value, unit=unit),
913 assumptions=assumptions,
914 status=status,
915 formula_store=formula_store,
916 )
919def _add_annual_savings_metric(
920 *,
921 inputs: FinancialCalculationInputs,
922 assumptions: AssumptionSet,
923 metrics: dict[str, FinancialMetric],
924 warnings: list[FinancialWarning],
925 formula_store: MetricFormulaStore | None,
926) -> Decimal | None:
927 if inputs.baseline_annual_opex is None:
928 warnings.append(
929 FinancialWarning(
930 code="baseline_annual_opex_missing",
931 severity="warning",
932 message="Annual savings cannot be calculated until a baseline annual opex is available.",
933 context={"field": "manual_annual_opex"},
934 )
935 )
936 return None
937 formula = annual_savings_formula(
938 baseline_annual_opex=inputs.baseline_annual_opex,
939 target_annual_opex=inputs.target_annual_opex,
940 target_annual_revenue=inputs.target_annual_revenue,
941 unit=_currency_unit(assumptions, per_year=True),
942 )
943 annual_savings = formula.evaluate()
944 metrics["annual_savings"] = FinancialMetric.from_formula(
945 key="annual_savings",
946 formula=formula,
947 assumptions=assumptions.merge(
948 {
949 "baseline_annual_opex": inputs.baseline_annual_opex,
950 "target_annual_opex": inputs.target_annual_opex,
951 "target_annual_revenue": inputs.target_annual_revenue,
952 }
953 ),
954 status="calculated" if inputs.baseline_fully_calculated else "incomplete_baseline",
955 formula_store=formula_store,
956 )
957 if not inputs.baseline_fully_calculated:
958 warnings.append(
959 FinancialWarning(
960 code="baseline_incomplete",
961 severity="warning",
962 message="Manual baseline values are incomplete, so dependent metrics are blocked.",
963 context={},
964 )
965 )
966 return annual_savings
969def _add_tax_metrics(
970 *,
971 inputs: FinancialCalculationInputs,
972 assumptions: AssumptionSet,
973 annual_savings: Decimal | None,
974 metrics: dict[str, FinancialMetric],
975 formula_store: MetricFormulaStore | None,
976) -> Decimal | None:
977 tax_rate = _tax_rate_from_percent(inputs.tax_rate_percent)
978 annual_currency_unit = _currency_unit(assumptions, per_year=True)
979 # Screening economics assumes the straight-line depreciation tax shield is
980 # usable in the same project year; tax-loss carry-forward is out of scope.
981 tax_assumptions = assumptions.merge(
982 {
983 "annual_depreciation": inputs.target_annual_depreciation,
984 "tax_rate_percent": inputs.tax_rate_percent or ZERO,
985 "tax_rate": tax_rate,
986 }
987 )
988 shield_formula = depreciation_tax_shield_formula(
989 annual_depreciation=inputs.target_annual_depreciation,
990 tax_rate=tax_rate,
991 unit=annual_currency_unit,
992 )
993 metrics["depreciation_tax_shield"] = FinancialMetric.from_formula(
994 key="depreciation_tax_shield",
995 formula=shield_formula,
996 assumptions=tax_assumptions,
997 formula_store=formula_store,
998 )
999 if annual_savings is None:
1000 return None
1001 cash_flow_formula_result = after_tax_annual_cash_flow_formula(
1002 annual_savings=annual_savings,
1003 annual_depreciation=inputs.target_annual_depreciation,
1004 tax_rate=tax_rate,
1005 unit=annual_currency_unit,
1006 )
1007 after_tax_cash_flow = cash_flow_formula_result.evaluate()
1008 metrics["after_tax_annual_cash_flow"] = FinancialMetric.from_formula(
1009 key="after_tax_annual_cash_flow",
1010 formula=cash_flow_formula_result,
1011 assumptions=tax_assumptions.merge({"annual_savings": annual_savings}),
1012 status="calculated" if inputs.baseline_fully_calculated else "incomplete_baseline",
1013 formula_store=formula_store,
1014 )
1015 return after_tax_cash_flow
1018def _add_incremental_capex_metric(
1019 *,
1020 inputs: FinancialCalculationInputs,
1021 assumptions: AssumptionSet,
1022 incremental_capex: Decimal | None,
1023 metrics: dict[str, FinancialMetric],
1024 warnings: list[FinancialWarning],
1025 formula_store: MetricFormulaStore | None,
1026) -> None:
1027 if incremental_capex is None:
1028 warnings.append(
1029 FinancialWarning(
1030 code="baseline_capex_missing",
1031 severity="warning",
1032 message="Incremental capex cannot be calculated until a baseline capex is available.",
1033 context={"field": "manual_capex"},
1034 )
1035 )
1036 return
1037 status = "calculated" if inputs.baseline_fully_calculated else "incomplete_baseline"
1038 formula = incremental_capex_formula(
1039 target_capex=inputs.target_capex,
1040 baseline_capex=inputs.baseline_capex,
1041 unit=_currency_unit(assumptions),
1042 )
1043 metrics["incremental_capex"] = FinancialMetric.from_formula(
1044 key="incremental_capex",
1045 formula=formula,
1046 assumptions=assumptions.merge(
1047 {
1048 "baseline_capex": inputs.baseline_capex,
1049 "target_capex": inputs.target_capex,
1050 }
1051 ),
1052 status=status,
1053 value=incremental_capex,
1054 formula_store=formula_store,
1055 )
1058def _currency_unit(assumptions: AssumptionSet, *, per_year: bool = False) -> str:
1059 """Return a concrete study currency unit when the study supplied one."""
1060 currency = assumptions.get("currency")
1061 base_unit = currency if isinstance(currency, str) and currency else "currency"
1062 return f"{base_unit}/year" if per_year else base_unit
1065def _add_cashflow_metrics(
1066 *,
1067 inputs: FinancialCalculationInputs,
1068 assumptions: AssumptionSet,
1069 incremental_capex: Decimal | None,
1070 annual_savings: Decimal | None,
1071 annual_cash_flow: Decimal | None,
1072 metrics: dict[str, FinancialMetric],
1073 warnings: list[FinancialWarning],
1074 formula_store: MetricFormulaStore | None,
1075) -> tuple[DiscountedCashFlowRow, ...]:
1076 if not inputs.baseline_fully_calculated:
1077 return ()
1079 if incremental_capex is None or annual_cash_flow is None or inputs.project_lifetime_years is None or inputs.discount_rate_percent is None:
1080 _add_incomplete_cashflow_warning(inputs=inputs, warnings=warnings)
1081 return ()
1083 metric_assumptions = _cashflow_metric_assumptions(
1084 assumptions=assumptions,
1085 incremental_capex=incremental_capex,
1086 annual_savings=annual_savings,
1087 annual_cash_flow=annual_cash_flow,
1088 project_lifetime_years=inputs.project_lifetime_years,
1089 discount_rate_percent=inputs.discount_rate_percent,
1090 tax_rate_percent=inputs.tax_rate_percent or ZERO,
1091 annual_depreciation=inputs.target_annual_depreciation,
1092 residual_value=inputs.residual_value,
1093 )
1094 discounted_cash_flow = tuple(
1095 calculate_discounted_cash_flow(
1096 incremental_capex=incremental_capex,
1097 annual_cash_flow=annual_cash_flow,
1098 project_lifetime_years=inputs.project_lifetime_years,
1099 discount_rate_percent=inputs.discount_rate_percent,
1100 residual_value=inputs.residual_value,
1101 )
1102 )
1103 discount_rate = _discount_rate_from_percent(inputs.discount_rate_percent)
1104 metrics["npv"] = FinancialMetric.from_formula(
1105 key="npv",
1106 formula=npv_formula(
1107 incremental_capex=incremental_capex,
1108 annual_cash_flow=annual_cash_flow,
1109 project_lifetime_years=inputs.project_lifetime_years,
1110 discount_rate=discount_rate,
1111 residual_value=inputs.residual_value,
1112 unit=_currency_unit(metric_assumptions),
1113 ),
1114 assumptions=metric_assumptions,
1115 formula_store=formula_store,
1116 )
1117 metrics["simple_payback_years"] = FinancialMetric.from_value(
1118 key="simple_payback_years",
1119 value=calculate_simple_payback_years(
1120 incremental_capex=incremental_capex,
1121 annual_cash_flow=annual_cash_flow,
1122 project_lifetime_years=inputs.project_lifetime_years,
1123 residual_value=inputs.residual_value,
1124 ),
1125 unit="years",
1126 assumptions=metric_assumptions,
1127 )
1128 if incremental_capex <= 0:
1129 warnings.append(
1130 FinancialWarning(
1131 code="roi_capital_outlay_non_positive",
1132 severity="warning",
1133 message="ROI is unavailable because incremental capital outlay is not positive.",
1134 context={"incremental_capex": str(incremental_capex)},
1135 )
1136 )
1137 metrics["roi_percent"] = FinancialMetric.from_value(
1138 key="roi_percent",
1139 value=None,
1140 unit="percent",
1141 assumptions=metric_assumptions,
1142 )
1143 else:
1144 metrics["roi_percent"] = FinancialMetric.from_formula(
1145 key="roi_percent",
1146 formula=roi_percent_formula(
1147 incremental_capex=incremental_capex,
1148 annual_cash_flow=annual_cash_flow,
1149 project_lifetime_years=inputs.project_lifetime_years,
1150 residual_value=inputs.residual_value,
1151 ),
1152 assumptions=metric_assumptions,
1153 formula_store=formula_store,
1154 )
1155 return discounted_cash_flow
1158def _add_incomplete_cashflow_warning(
1159 *,
1160 inputs: FinancialCalculationInputs,
1161 warnings: list[FinancialWarning],
1162) -> None:
1163 warnings.append(
1164 FinancialWarning(
1165 code="financial_assumptions_incomplete",
1166 severity="warning",
1167 message="Project lifetime and discount rate are required for cash-flow, payback, NPV, and ROI.",
1168 context={
1169 "project_lifetime_years": inputs.project_lifetime_years,
1170 "discount_rate_percent": _decimal_string(inputs.discount_rate_percent),
1171 },
1172 )
1173 )
1176def _add_lcoh_metric(
1177 *,
1178 inputs: FinancialCalculationInputs,
1179 assumptions: AssumptionSet,
1180 metrics: dict[str, FinancialMetric],
1181 warnings: list[FinancialWarning],
1182 formula_store: MetricFormulaStore | None,
1183) -> None:
1184 lcoh_assumptions = _lcoh_metric_assumptions(
1185 assumptions=assumptions,
1186 target_capex=inputs.target_capex,
1187 target_annual_opex=inputs.target_annual_opex,
1188 target_annual_process_energy_basis=inputs.target_annual_process_energy_basis,
1189 target_annual_process_energy_basis_unit=inputs.target_annual_process_energy_basis_unit,
1190 target_process_energy_basis_source_row_keys=inputs.target_process_energy_basis_source_row_keys,
1191 target_process_energy_basis_contributor_count=inputs.target_process_energy_basis_contributor_count,
1192 project_lifetime_years=inputs.project_lifetime_years,
1193 discount_rate_percent=inputs.discount_rate_percent,
1194 residual_value=inputs.residual_value,
1195 )
1196 if (
1197 inputs.discount_rate_percent is None
1198 or inputs.target_annual_process_energy_basis is None
1199 or inputs.target_annual_process_energy_basis <= 0
1200 or inputs.project_lifetime_years is None
1201 or inputs.project_lifetime_years <= 0
1202 ):
1203 warnings.append(
1204 FinancialWarning(
1205 code="lcoh_process_energy_basis_missing",
1206 severity="warning",
1207 message="LCOH is hidden until a positive target process energy basis and financial assumptions are available.",
1208 context={
1209 "target_annual_process_energy_basis": _decimal_string(
1210 inputs.target_annual_process_energy_basis
1211 ),
1212 "target_annual_process_energy_basis_unit": inputs.target_annual_process_energy_basis_unit,
1213 "target_process_energy_basis_contributor_count": inputs.target_process_energy_basis_contributor_count,
1214 "project_lifetime_years": inputs.project_lifetime_years,
1215 "discount_rate_percent": _decimal_string(inputs.discount_rate_percent),
1216 "field": "target_annual_process_energy_basis",
1217 },
1218 )
1219 )
1220 return
1221 discount_rate = _discount_rate_from_percent(inputs.discount_rate_percent)
1222 metrics["lcoh"] = FinancialMetric.from_formula(
1223 key="lcoh",
1224 formula=lcoh_formula(
1225 target_capex=inputs.target_capex,
1226 target_annual_opex=inputs.target_annual_opex,
1227 target_annual_process_energy_basis=inputs.target_annual_process_energy_basis,
1228 project_lifetime_years=inputs.project_lifetime_years,
1229 discount_rate=discount_rate,
1230 residual_value=inputs.residual_value,
1231 unit=f"{_currency_unit(lcoh_assumptions)}/{_energy_basis_denominator(inputs.target_annual_process_energy_basis_unit)}",
1232 ),
1233 assumptions=lcoh_assumptions,
1234 formula_store=formula_store,
1235 )
1238def _pure_calculation_baseline_resolution(*, inputs: FinancialCalculationInputs) -> BaselineResolution:
1239 return BaselineResolution(
1240 source="pure_calculation",
1241 is_guided_default=not inputs.baseline_fully_calculated,
1242 capex=inputs.baseline_capex,
1243 annual_opex=inputs.baseline_annual_opex,
1244 annual_heat_basis=None,
1245 annual_heat_basis_unit=None,
1246 residual_value=inputs.residual_value,
1247 project_lifetime_years=inputs.project_lifetime_years,
1248 discount_rate_percent=inputs.discount_rate_percent,
1249 assumptions=inputs.assumptions,
1250 )
1253def _target_only_baseline_resolution(
1254 *,
1255 study: EconomicsStudy,
1256 target_assumptions: TargetAssumptions,
1257) -> BaselineResolution:
1258 return BaselineResolution(
1259 source="target_only",
1260 is_guided_default=True,
1261 capex=None,
1262 annual_opex=None,
1263 annual_heat_basis=None,
1264 annual_heat_basis_unit=None,
1265 residual_value=ZERO,
1266 project_lifetime_years=target_assumptions.project_lifetime_years,
1267 discount_rate_percent=target_assumptions.discount_rate_percent,
1268 assumptions=AssumptionSet.from_mapping(
1269 {
1270 "target_study_id": study.pk,
1271 "baseline_source": "target_only",
1272 "project_lifetime_source": "target_study",
1273 "discount_rate_source": "target_study",
1274 }
1275 ),
1276 )
1279def _target_assumption_warnings(
1280 *,
1281 study: EconomicsStudy,
1282 target_assumptions: TargetAssumptions,
1283) -> list[FinancialWarning]:
1284 if target_assumptions.assumptions_source == "missing": 1284 ↛ 1285line 1284 didn't jump to line 1285 because the condition on line 1284 was never true
1285 return [
1286 FinancialWarning(
1287 code="study_assumptions_missing",
1288 severity="warning",
1289 message="Study assumptions are required for project lifetime, discount rate, and basis metadata.",
1290 context={"study_id": study.pk},
1291 )
1292 ]
1293 missing_fields = [
1294 field_name
1295 for field_name in ("project_lifetime_years", "discount_rate_percent")
1296 if getattr(target_assumptions, field_name) is None
1297 ]
1298 if not missing_fields: 1298 ↛ 1300line 1298 didn't jump to line 1300 because the condition on line 1298 was always true
1299 return []
1300 return [
1301 FinancialWarning(
1302 code="target_financial_assumptions_incomplete",
1303 severity="warning",
1304 message="Target study financial assumptions are incomplete.",
1305 context={"study_id": study.pk, "missing_fields": tuple(missing_fields)},
1306 )
1307 ]
1310def _target_assumptions(study: EconomicsStudy) -> TargetAssumptions:
1311 assumptions = get_settings_profile(study)
1312 if assumptions is None:
1313 return TargetAssumptions(target_study_id=study.pk, assumptions_source="missing")
1314 peak_demand_basis = derive_peak_demand_basis(study)
1315 return TargetAssumptions(
1316 target_study_id=study.pk,
1317 assumptions_id=assumptions.pk,
1318 project_lifetime_years=assumptions.project_lifetime_years,
1319 discount_rate_percent=assumptions.discount_rate_percent,
1320 currency=assumptions.currency,
1321 basis_date=assumptions.basis_date.isoformat() if assumptions.basis_date else None,
1322 inflation_method=assumptions.inflation_method,
1323 capital_index_series_id=assumptions.capital_index_series_id,
1324 operating_index_series_id=assumptions.operating_index_series_id,
1325 annual_operating_hours=assumptions.annual_operating_hours,
1326 tax_rate_percent=assumptions.tax_rate_percent,
1327 depreciation_enabled=assumptions.depreciation_enabled,
1328 default_depreciation_life_years=assumptions.default_depreciation_life_years,
1329 default_depreciation_salvage_percent=assumptions.default_depreciation_salvage_percent,
1330 contingency_percent=assumptions.contingency_percent,
1331 electrical_upgrade_rate_amount=assumptions.electrical_upgrade_rate_amount,
1332 electrical_upgrade_rate_unit=assumptions.electrical_upgrade_rate_unit,
1333 peak_demand_kw=peak_demand_basis.quantity_kw,
1334 default_lang_factor=assumptions.default_lang_factor,
1335 assumptions_source="study",
1336 )
1339def _get_baseline(study: EconomicsStudy) -> EconomicsSettingsProfile | None:
1340 baseline = get_settings_profile(study)
1341 if baseline is None or not _has_manual_baseline_values(baseline):
1342 return None
1343 return baseline
1346def _has_manual_baseline_values(profile: EconomicsSettingsProfile) -> bool:
1347 return any(
1348 value is not None
1349 for value in (
1350 profile.manual_capex,
1351 profile.manual_annual_opex,
1352 profile.manual_annual_heat_basis,
1353 profile.average_power_input,
1354 profile.residual_value,
1355 )
1356 )
1359def _sum_capital_lines(study: EconomicsStudy) -> Decimal:
1360 total_formula = build_target_total_capex_formula(study)
1361 total = total_formula.evaluate()
1362 if total is None:
1363 raise FinancialMetricsError(
1364 "target_capex_formula_blocked",
1365 total_formula.formula.blocked_reason,
1366 context={"blocked_children": total_formula.formula.blocked_children},
1367 )
1368 return total.quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP)
1371def _sum_capital_breakdown(study: EconomicsStudy) -> dict[str, Decimal]:
1372 totals = {
1373 "purchase_basis": ZERO,
1374 "installed_basis": ZERO,
1375 "contingency": ZERO,
1376 "electrical_upgrade": ZERO,
1377 }
1378 for payload in study.capital_lines.filter(included=True).values_list("warning_payload", flat=True):
1379 if not isinstance(payload, Mapping): 1379 ↛ 1380line 1379 didn't jump to line 1380 because the condition on line 1379 was never true
1380 continue
1381 totals["purchase_basis"] += _decimal_from_payload(payload.get("purchase_basis_amount"))
1382 totals["installed_basis"] += _decimal_from_payload(payload.get("installed_basis_amount"))
1383 totals["contingency"] += _decimal_from_payload(payload.get("contingency_amount"))
1384 totals["electrical_upgrade"] = _electrical_upgrade_capex(study)
1385 return totals
1388def _electrical_upgrade_capex(study: EconomicsStudy) -> Decimal:
1389 """Calculate project-level electrical-upgrade capex without creating a capital line."""
1390 formula = build_electrical_upgrade_formula(study)
1391 amount = formula.evaluate()
1392 if amount is None: 1392 ↛ 1393line 1392 didn't jump to line 1393 because the condition on line 1392 was never true
1393 raise FinancialMetricsError(
1394 "electrical_upgrade_formula_blocked",
1395 formula.formula.blocked_reason,
1396 context={"blocked_children": formula.formula.blocked_children},
1397 )
1398 return amount
1401def _decimal_from_payload(value: object) -> Decimal:
1402 if value in (None, ""):
1403 return ZERO
1404 try:
1405 return Decimal(str(value))
1406 except (InvalidOperation, ValueError):
1407 return ZERO
1410def _sum_operating_lines(study: EconomicsStudy) -> tuple[Decimal, Decimal]:
1411 expense_formula = build_annual_operating_expense_formula(study)
1412 revenue_formula = build_annual_operating_revenue_formula(study)
1413 expense_total = expense_formula.evaluate()
1414 revenue_total = revenue_formula.evaluate()
1415 if expense_total is None:
1416 raise FinancialMetricsError(
1417 "target_annual_opex_formula_blocked",
1418 expense_formula.formula.blocked_reason,
1419 context={"blocked_children": expense_formula.formula.blocked_children},
1420 )
1421 if revenue_total is None: 1421 ↛ 1422line 1421 didn't jump to line 1422 because the condition on line 1421 was never true
1422 raise FinancialMetricsError(
1423 "target_annual_revenue_formula_blocked",
1424 revenue_formula.formula.blocked_reason,
1425 context={"blocked_children": revenue_formula.formula.blocked_children},
1426 )
1427 return expense_total, revenue_total
1430def _operating_line_annual_amount(line: OperatingCostLine, *, study: EconomicsStudy) -> Decimal | None:
1431 try:
1432 return build_operating_line_formula(line, study=study).evaluate()
1433 except FormulaError:
1434 return None
1437def _discount_factor(*, discount_rate: Decimal, year: int) -> Decimal:
1438 with localcontext() as context:
1439 context.prec = 34
1440 return discount_factor_formula(year=year, discount_rate=discount_rate).evaluate()
1443def _quantize_like(value: Decimal, *references: Decimal) -> Decimal:
1444 exponent = min(reference.as_tuple().exponent for reference in references)
1445 quantum = Decimal("1").scaleb(exponent)
1446 with localcontext() as context:
1447 context.prec = max(
1448 34,
1449 len(value.as_tuple().digits),
1450 *(len(reference.as_tuple().digits) for reference in references),
1451 )
1452 return value.quantize(quantum, rounding=ROUND_HALF_UP)
1455def _discount_rate_from_percent(discount_rate_percent: Decimal | None) -> Decimal | None:
1456 if discount_rate_percent is None:
1457 return None
1458 if not discount_rate_percent.is_finite():
1459 raise FinancialMetricsError(
1460 "invalid_discount_rate",
1461 "Discount rate must be finite.",
1462 context={"discount_rate_percent": str(discount_rate_percent)},
1463 )
1464 discount_rate = discount_rate_percent / Decimal("100")
1465 if ONE + discount_rate <= 0:
1466 raise FinancialMetricsError(
1467 "invalid_discount_rate",
1468 "Discount rate must keep 1 + rate greater than zero.",
1469 context={"discount_rate_percent": str(discount_rate_percent), "discount_rate": str(discount_rate)},
1470 )
1471 return discount_rate
1474def _tax_rate_from_percent(tax_rate_percent: Decimal | None) -> Decimal:
1475 if tax_rate_percent is None: 1475 ↛ 1476line 1475 didn't jump to line 1476 because the condition on line 1475 was never true
1476 return ZERO
1477 if not tax_rate_percent.is_finite():
1478 raise FinancialMetricsError(
1479 "invalid_tax_rate",
1480 "Tax rate must be finite.",
1481 context={"tax_rate_percent": str(tax_rate_percent)},
1482 )
1483 if tax_rate_percent < ZERO or tax_rate_percent > Decimal("100"):
1484 raise FinancialMetricsError(
1485 "invalid_tax_rate",
1486 "Tax rate must be between 0 and 100 percent.",
1487 context={"tax_rate_percent": str(tax_rate_percent)},
1488 )
1489 return tax_rate_percent / Decimal("100")
1492def _cashflow_metric_assumptions(
1493 *,
1494 assumptions: AssumptionSet,
1495 incremental_capex: Decimal,
1496 annual_savings: Decimal | None,
1497 annual_cash_flow: Decimal,
1498 project_lifetime_years: int,
1499 discount_rate_percent: Decimal,
1500 tax_rate_percent: Decimal,
1501 annual_depreciation: Decimal,
1502 residual_value: Decimal,
1503) -> AssumptionSet:
1504 return assumptions.merge(
1505 {
1506 "incremental_capex": incremental_capex,
1507 "annual_savings": annual_savings,
1508 "annual_cash_flow": annual_cash_flow,
1509 "annual_depreciation": annual_depreciation,
1510 "tax_rate_percent": tax_rate_percent,
1511 "project_lifetime_years": project_lifetime_years,
1512 "discount_rate_percent": discount_rate_percent,
1513 "residual_value": residual_value,
1514 }
1515 )
1518def _lcoh_metric_assumptions(
1519 *,
1520 assumptions: AssumptionSet,
1521 target_capex: Decimal,
1522 target_annual_opex: Decimal,
1523 target_annual_process_energy_basis: Decimal | None,
1524 target_annual_process_energy_basis_unit: str | None,
1525 target_process_energy_basis_source_row_keys: str,
1526 target_process_energy_basis_contributor_count: int,
1527 project_lifetime_years: int | None,
1528 discount_rate_percent: Decimal | None,
1529 residual_value: Decimal,
1530) -> AssumptionSet:
1531 return assumptions.merge(
1532 {
1533 "target_capex": target_capex,
1534 "target_annual_opex": target_annual_opex,
1535 "target_annual_process_energy_basis": target_annual_process_energy_basis,
1536 "target_annual_process_energy_basis_unit": target_annual_process_energy_basis_unit,
1537 "target_process_energy_basis_source_row_keys": target_process_energy_basis_source_row_keys,
1538 "target_process_energy_basis_contributor_count": target_process_energy_basis_contributor_count,
1539 "project_lifetime_years": project_lifetime_years,
1540 "discount_rate_percent": discount_rate_percent,
1541 "residual_value": residual_value,
1542 }
1543 )
1546def _energy_basis_denominator(unit: str | None) -> str:
1547 if unit and "/" in unit: 1547 ↛ 1549line 1547 didn't jump to line 1549 because the condition on line 1547 was always true
1548 return unit.split("/", maxsplit=1)[0]
1549 return unit or "energy"
1552def _process_energy_basis_source_row_keys(basis: TargetProcessEnergyBasis) -> str:
1553 return ";".join(
1554 f"operating_line.{contribution.operating_line_id}" for contribution in basis.contributions
1555 )
1558def _financial_scalar(value: object) -> FinancialScalar:
1559 """Normalize flexible assumption inputs into a small JSON-safe scalar set."""
1560 if value is None or isinstance(value, (str, int, bool)):
1561 return value
1562 if isinstance(value, Decimal): 1562 ↛ 1564line 1562 didn't jump to line 1564 because the condition on line 1562 was always true
1563 return str(value)
1564 return str(value)
1567def _decimal_string(value: Decimal | None) -> str | None:
1568 return None if value is None else str(value)
1571def parse_decimal(value: Decimal | int | str, *, field_name: str) -> Decimal:
1572 try:
1573 decimal_value = Decimal(str(value))
1574 except (InvalidOperation, ValueError) as exc:
1575 raise FinancialMetricsError(
1576 "invalid_decimal",
1577 "Financial metric input must be numeric.",
1578 context={"field": field_name, "value": str(value)},
1579 ) from exc
1580 if not decimal_value.is_finite():
1581 raise FinancialMetricsError(
1582 "invalid_decimal",
1583 "Financial metric input must be finite.",
1584 context={"field": field_name, "value": str(value)},
1585 )
1586 return decimal_value