Coverage for backend/django/Economics/results/services/financial_metrics/calculation.py: 93%
172 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Pure financial metric calculation and metric construction."""
3from __future__ import annotations
5from decimal import Decimal, InvalidOperation, ROUND_HALF_UP, localcontext
7from Economics.formulas.builders.metrics import (
8 BoundMetricFormula,
9 annual_net_benefit_formula,
10 annual_profit_formula,
11 annual_savings_formula,
12 after_tax_annual_cash_flow_formula,
13 cash_flow_formula,
14 cumulative_cash_flow_formula,
15 cumulative_present_value_formula,
16 discounted_cash_flow_formula,
17 discount_factor_formula,
18 depreciation_tax_shield_formula,
19 incremental_capex_formula,
20 metric_value_formula,
21 npv_formula,
22 roi_percent_formula,
23)
24from Economics.formulas.builders.metric_formulas import MetricFormulaStore
25from Economics.formulas.engine.core import FormulaEvaluation
26from .contracts import (
27 ONE,
28 ZERO,
29 AssumptionSet,
30 BaselineResolution,
31 DiscountedCashFlowRow,
32 FinancialCalculationInputs,
33 FinancialMetric,
34 FinancialMetricsError,
35 FinancialMetricsResult,
36 FinancialWarning,
37)
38from .warnings import add_incomplete_cashflow_warning
39from .metric_catalog import FinancialMetricKey
42def calculate_financial_metrics(
43 inputs: FinancialCalculationInputs,
44 *,
45 formula_store: MetricFormulaStore | None = None,
46) -> FinancialMetricsResult:
47 """Calculate deterministic v1 metrics from already-resolved numeric inputs.
49 The formulas are fixed for v1: capex is a year-0 outflow, annual savings
50 are avoided operating cost, annual net benefit adds target revenue,
51 discounted cash flow uses the supplied lifetime/rate, and ROI requires
52 positive incremental capital outlay.
53 """
54 warnings = list(inputs.warnings)
55 assumptions = inputs.assumptions
56 metrics = _base_metrics(inputs=inputs, assumptions=assumptions, formula_store=formula_store)
57 annual_savings = _add_annual_savings_metric(
58 inputs=inputs,
59 assumptions=assumptions,
60 metrics=metrics,
61 warnings=warnings,
62 formula_store=formula_store,
63 )
64 annual_net_benefit = _add_annual_net_benefit_metric(
65 inputs=inputs,
66 assumptions=assumptions,
67 annual_savings=annual_savings,
68 metrics=metrics,
69 formula_store=formula_store,
70 )
71 annual_cash_flow = _add_tax_metrics(
72 inputs=inputs,
73 assumptions=assumptions,
74 annual_net_benefit=annual_net_benefit,
75 metrics=metrics,
76 formula_store=formula_store,
77 )
78 _discount_rate_from_percent(inputs.discount_rate_percent)
79 incremental_capex = (
80 (incremental_capex_formula(
81 target_capex=inputs.target_capex,
82 baseline_capex=inputs.baseline_capex,
83 unit=_currency_unit(assumptions),
84 ).evaluate() or ZERO).quantize(inputs.target_capex, rounding=ROUND_HALF_UP)
85 if inputs.baseline_capex is not None
86 else None
87 )
88 _add_incremental_capex_metric(
89 inputs=inputs,
90 assumptions=assumptions,
91 incremental_capex=incremental_capex,
92 metrics=metrics,
93 warnings=warnings,
94 formula_store=formula_store,
95 )
96 discounted_cash_flow = _add_cashflow_metrics(
97 inputs=inputs,
98 assumptions=assumptions,
99 incremental_capex=incremental_capex,
100 annual_savings=annual_savings,
101 annual_net_benefit=annual_net_benefit,
102 annual_cash_flow=annual_cash_flow,
103 metrics=metrics,
104 warnings=warnings,
105 formula_store=formula_store,
106 )
107 return FinancialMetricsResult(
108 metrics=metrics,
109 discounted_cash_flow=discounted_cash_flow,
110 baseline_resolution=_pure_calculation_baseline_resolution(inputs=inputs),
111 warnings=tuple(warnings),
112 )
114def calculate_discounted_cash_flow(
115 *,
116 incremental_capex: Decimal,
117 annual_cash_flow: Decimal,
118 project_lifetime_years: int,
119 discount_rate_percent: Decimal,
120 residual_value: Decimal = ZERO,
121) -> list[DiscountedCashFlowRow]:
122 discount_rate = _discount_rate_from_percent(discount_rate_percent)
123 rows: list[DiscountedCashFlowRow] = []
124 for year in range(0, project_lifetime_years + 1):
125 cash_flow_formula_result = cash_flow_formula(
126 year=year,
127 project_lifetime_years=project_lifetime_years,
128 incremental_capex=incremental_capex,
129 annual_cash_flow=annual_cash_flow,
130 residual_value=residual_value,
131 unit="currency",
132 )
133 cash_flow = cash_flow_formula_result.evaluate()
134 if year == 0:
135 cash_flow = _quantize_like(cash_flow, incremental_capex)
136 elif year == project_lifetime_years:
137 cash_flow = _quantize_like(cash_flow, annual_cash_flow, residual_value)
138 else:
139 cash_flow = _quantize_like(cash_flow, annual_cash_flow)
140 discount_factor = _discount_factor(discount_rate=discount_rate, year=year)
141 present_value = discounted_cash_flow_formula(
142 year=year,
143 cash_flow=cash_flow,
144 discount_rate=discount_rate,
145 unit="currency",
146 ).evaluate()
147 cumulative_cash_flow = cumulative_cash_flow_formula(
148 year=year,
149 project_lifetime_years=project_lifetime_years,
150 incremental_capex=incremental_capex,
151 annual_cash_flow=annual_cash_flow,
152 residual_value=residual_value,
153 unit="currency",
154 ).evaluate()
155 cumulative_present_value = cumulative_present_value_formula(
156 key=f"cumulative_present_value_year_{year}",
157 year=year,
158 project_lifetime_years=project_lifetime_years,
159 incremental_capex=incremental_capex,
160 annual_cash_flow=annual_cash_flow,
161 discount_rate=discount_rate,
162 residual_value=residual_value,
163 unit="currency",
164 ).evaluate()
165 rows.append(
166 DiscountedCashFlowRow(
167 year=year,
168 cash_flow=cash_flow,
169 discount_factor=discount_factor,
170 present_value=present_value,
171 cumulative_cash_flow=cumulative_cash_flow,
172 cumulative_present_value=cumulative_present_value,
173 )
174 )
175 return rows
177def calculate_simple_payback_years(
178 *,
179 incremental_capex: Decimal,
180 annual_cash_flow: Decimal,
181 project_lifetime_years: int,
182 residual_value: Decimal = ZERO,
183) -> Decimal | None:
184 if incremental_capex <= 0:
185 return ZERO
186 cumulative = -incremental_capex
187 for year in range(1, project_lifetime_years + 1):
188 cash_flow = annual_cash_flow + (residual_value if year == project_lifetime_years else ZERO)
189 next_cumulative = cumulative + cash_flow
190 if cash_flow > 0 and next_cumulative >= 0:
191 return Decimal(year - 1) + (-cumulative / cash_flow)
192 cumulative = next_cumulative
193 return None
195def calculate_roi_percent(
196 *,
197 incremental_capex: Decimal,
198 annual_cash_flow: Decimal,
199 project_lifetime_years: int,
200 residual_value: Decimal = ZERO,
201) -> Decimal | None:
202 if incremental_capex <= 0:
203 return None
204 return roi_percent_formula(
205 incremental_capex=incremental_capex,
206 annual_cash_flow=annual_cash_flow,
207 project_lifetime_years=project_lifetime_years,
208 residual_value=residual_value,
209 ).evaluate()
211def _base_metrics(
212 *,
213 inputs: FinancialCalculationInputs,
214 assumptions: AssumptionSet,
215 formula_store: MetricFormulaStore | None,
216) -> dict[FinancialMetricKey, FinancialMetric]:
217 """Create always-visible target-cost metrics before baseline comparisons."""
218 currency_unit = _currency_unit(assumptions)
219 annual_currency_unit = _currency_unit(assumptions, per_year=True)
220 target_annual_profit_formula = annual_profit_formula(
221 target_annual_revenue=inputs.target_annual_revenue,
222 target_annual_opex=inputs.target_annual_opex,
223 unit=annual_currency_unit,
224 )
225 return {
226 FinancialMetricKey.CAPEX: _metric_from_scalar_formula(
227 key=FinancialMetricKey.CAPEX,
228 value=inputs.target_capex,
229 unit=currency_unit,
230 assumptions=assumptions,
231 formula_store=formula_store,
232 ),
233 FinancialMetricKey.PURCHASE_BASIS_EQUIPMENT: _metric_from_scalar_formula(
234 key=FinancialMetricKey.PURCHASE_BASIS_EQUIPMENT,
235 value=inputs.target_purchase_basis_capex,
236 unit=currency_unit,
237 assumptions=assumptions,
238 formula_store=formula_store,
239 ),
240 FinancialMetricKey.INSTALLED_BASIS_EQUIPMENT: _metric_from_scalar_formula(
241 key=FinancialMetricKey.INSTALLED_BASIS_EQUIPMENT,
242 value=inputs.target_installed_basis_capex,
243 unit=currency_unit,
244 assumptions=assumptions,
245 formula_store=formula_store,
246 ),
247 FinancialMetricKey.CONTINGENCY: _metric_from_scalar_formula(
248 key=FinancialMetricKey.CONTINGENCY,
249 value=inputs.target_contingency_capex,
250 unit=currency_unit,
251 assumptions=assumptions,
252 formula_store=formula_store,
253 ),
254 FinancialMetricKey.ELECTRICAL_UPGRADE: _metric_from_scalar_formula(
255 key=FinancialMetricKey.ELECTRICAL_UPGRADE,
256 value=inputs.target_electrical_upgrade_capex,
257 unit=currency_unit,
258 assumptions=assumptions,
259 formula_store=formula_store,
260 ),
261 FinancialMetricKey.PEAK_DEMAND: _metric_from_scalar_formula(
262 key=FinancialMetricKey.PEAK_DEMAND,
263 value=inputs.target_peak_demand_kw,
264 unit="kW",
265 assumptions=assumptions,
266 formula_store=formula_store,
267 ),
268 FinancialMetricKey.ANNUAL_OPEX: _metric_from_scalar_formula(
269 key=FinancialMetricKey.ANNUAL_OPEX,
270 value=inputs.target_annual_opex,
271 unit=annual_currency_unit,
272 assumptions=assumptions,
273 formula_store=formula_store,
274 ),
275 FinancialMetricKey.ANNUAL_REVENUE: _metric_from_scalar_formula(
276 key=FinancialMetricKey.ANNUAL_REVENUE,
277 value=inputs.target_annual_revenue,
278 unit=annual_currency_unit,
279 assumptions=assumptions,
280 formula_store=formula_store,
281 ),
282 FinancialMetricKey.ANNUAL_DEPRECIATION: _metric_from_scalar_formula(
283 key=FinancialMetricKey.ANNUAL_DEPRECIATION,
284 value=inputs.target_annual_depreciation,
285 unit=annual_currency_unit,
286 assumptions=assumptions,
287 formula_store=formula_store,
288 ),
289 FinancialMetricKey.ANNUAL_PROFIT: _metric_from_formula(
290 key=FinancialMetricKey.ANNUAL_PROFIT,
291 formula=target_annual_profit_formula,
292 assumptions=assumptions.merge(
293 {
294 "target_annual_opex": inputs.target_annual_opex,
295 "target_annual_revenue": inputs.target_annual_revenue,
296 }
297 ),
298 formula_store=formula_store,
299 ),
300 }
302def _metric_from_scalar_formula(
303 *,
304 key: FinancialMetricKey,
305 value: Decimal | None,
306 unit: str,
307 assumptions: AssumptionSet,
308 status: str = "calculated",
309 formula_store: MetricFormulaStore | None = None,
310) -> FinancialMetric:
311 if value is None:
312 return FinancialMetric.from_value(
313 key=key,
314 value=None,
315 unit=unit,
316 assumptions=assumptions,
317 status=status,
318 )
319 return _metric_from_formula(
320 key=key,
321 formula=metric_value_formula(key=key.value, value=value, unit=unit),
322 assumptions=assumptions,
323 status=status,
324 formula_store=formula_store,
325 )
328def _metric_from_formula(
329 *,
330 key: FinancialMetricKey,
331 formula: BoundMetricFormula,
332 assumptions: AssumptionSet | None = None,
333 status: str = "calculated",
334 value: Decimal | None = None,
335 formula_store: MetricFormulaStore | None = None,
336) -> FinancialMetric:
337 """Evaluate a metric formula and persist its audit record when a store is available."""
339 evaluated_value = formula.evaluate()
340 metric_value = evaluated_value if value is None else value
341 metric_status = status if metric_value is not None else "unavailable"
342 resolved_assumptions = assumptions or AssumptionSet()
343 formula_audit = formula.formula.audit_payload(
344 FormulaEvaluation(value=metric_value, bindings=formula.bindings)
345 )
346 formula_record_id = None
347 if formula_store is not None:
348 persisted_formula = formula_store.persist_metric_formula(
349 metric_key=key.value,
350 formula=formula,
351 value=metric_value,
352 status=metric_status,
353 assumptions=resolved_assumptions,
354 formula_audit=formula_audit,
355 )
356 formula_record_id = persisted_formula.id
357 formula_audit = persisted_formula.formula_audit
358 return FinancialMetric(
359 key=key,
360 value=metric_value,
361 unit=formula.formula.unit,
362 assumptions=resolved_assumptions,
363 status=metric_status,
364 formula_audit=formula_audit,
365 formula_record_id=formula_record_id,
366 )
369def _add_annual_savings_metric(
370 *,
371 inputs: FinancialCalculationInputs,
372 assumptions: AssumptionSet,
373 metrics: dict[FinancialMetricKey, FinancialMetric],
374 warnings: list[FinancialWarning],
375 formula_store: MetricFormulaStore | None,
376) -> Decimal | None:
377 if inputs.baseline_annual_opex is None:
378 warnings.append(
379 FinancialWarning(
380 code="baseline_annual_opex_missing",
381 severity="warning",
382 message="Annual savings cannot be calculated until a baseline annual opex is available.",
383 context={"field": "manual_annual_opex"},
384 )
385 )
386 return None
387 formula = annual_savings_formula(
388 baseline_annual_opex=inputs.baseline_annual_opex,
389 target_annual_opex=inputs.target_annual_opex,
390 unit=_currency_unit(assumptions, per_year=True),
391 )
392 annual_savings = formula.evaluate()
393 metrics[FinancialMetricKey.ANNUAL_SAVINGS] = _metric_from_formula(
394 key=FinancialMetricKey.ANNUAL_SAVINGS,
395 formula=formula,
396 assumptions=assumptions.merge(
397 {
398 "baseline_annual_opex": inputs.baseline_annual_opex,
399 "target_annual_opex": inputs.target_annual_opex,
400 }
401 ),
402 status="calculated" if inputs.baseline_fully_calculated else "incomplete_baseline",
403 formula_store=formula_store,
404 )
405 if not inputs.baseline_fully_calculated:
406 warnings.append(
407 FinancialWarning(
408 code="baseline_incomplete",
409 severity="warning",
410 message="Manual baseline values are incomplete, so dependent metrics are blocked.",
411 context={},
412 )
413 )
414 return annual_savings
416def _add_annual_net_benefit_metric(
417 *,
418 inputs: FinancialCalculationInputs,
419 assumptions: AssumptionSet,
420 annual_savings: Decimal | None,
421 metrics: dict[FinancialMetricKey, FinancialMetric],
422 formula_store: MetricFormulaStore | None,
423) -> Decimal | None:
424 if annual_savings is None:
425 return None
426 formula = annual_net_benefit_formula(
427 annual_savings=annual_savings,
428 target_annual_revenue=inputs.target_annual_revenue,
429 unit=_currency_unit(assumptions, per_year=True),
430 )
431 annual_net_benefit = formula.evaluate()
432 metrics[FinancialMetricKey.ANNUAL_NET_BENEFIT] = _metric_from_formula(
433 key=FinancialMetricKey.ANNUAL_NET_BENEFIT,
434 formula=formula,
435 assumptions=assumptions.merge(
436 {
437 "annual_savings": annual_savings,
438 "target_annual_revenue": inputs.target_annual_revenue,
439 }
440 ),
441 status="calculated" if inputs.baseline_fully_calculated else "incomplete_baseline",
442 formula_store=formula_store,
443 )
444 return annual_net_benefit
446def _add_tax_metrics(
447 *,
448 inputs: FinancialCalculationInputs,
449 assumptions: AssumptionSet,
450 annual_net_benefit: Decimal | None,
451 metrics: dict[FinancialMetricKey, FinancialMetric],
452 formula_store: MetricFormulaStore | None,
453) -> Decimal | None:
454 tax_rate = _tax_rate_from_percent(inputs.tax_rate_percent)
455 annual_currency_unit = _currency_unit(assumptions, per_year=True)
456 # Screening economics assumes the straight-line depreciation tax shield is
457 # usable in the same project year; tax-loss carry-forward is out of scope.
458 tax_assumptions = assumptions.merge(
459 {
460 "annual_depreciation": inputs.target_annual_depreciation,
461 "tax_rate_percent": inputs.tax_rate_percent or ZERO,
462 "tax_rate": tax_rate,
463 }
464 )
465 shield_formula = depreciation_tax_shield_formula(
466 annual_depreciation=inputs.target_annual_depreciation,
467 tax_rate=tax_rate,
468 unit=annual_currency_unit,
469 )
470 metrics[FinancialMetricKey.DEPRECIATION_TAX_SHIELD] = _metric_from_formula(
471 key=FinancialMetricKey.DEPRECIATION_TAX_SHIELD,
472 formula=shield_formula,
473 assumptions=tax_assumptions,
474 formula_store=formula_store,
475 )
476 if annual_net_benefit is None:
477 return None
478 cash_flow_formula_result = after_tax_annual_cash_flow_formula(
479 annual_net_benefit=annual_net_benefit,
480 annual_depreciation=inputs.target_annual_depreciation,
481 tax_rate=tax_rate,
482 unit=annual_currency_unit,
483 )
484 after_tax_cash_flow = cash_flow_formula_result.evaluate()
485 metrics[FinancialMetricKey.AFTER_TAX_ANNUAL_CASH_FLOW] = _metric_from_formula(
486 key=FinancialMetricKey.AFTER_TAX_ANNUAL_CASH_FLOW,
487 formula=cash_flow_formula_result,
488 assumptions=tax_assumptions.merge({"annual_net_benefit": annual_net_benefit}),
489 status="calculated" if inputs.baseline_fully_calculated else "incomplete_baseline",
490 formula_store=formula_store,
491 )
492 return after_tax_cash_flow
494def _add_incremental_capex_metric(
495 *,
496 inputs: FinancialCalculationInputs,
497 assumptions: AssumptionSet,
498 incremental_capex: Decimal | None,
499 metrics: dict[FinancialMetricKey, FinancialMetric],
500 warnings: list[FinancialWarning],
501 formula_store: MetricFormulaStore | None,
502) -> None:
503 if incremental_capex is None:
504 warnings.append(
505 FinancialWarning(
506 code="baseline_capex_missing",
507 severity="warning",
508 message="Incremental capex cannot be calculated until a baseline capex is available.",
509 context={"field": "manual_capex"},
510 )
511 )
512 return
513 status = "calculated" if inputs.baseline_fully_calculated else "incomplete_baseline"
514 formula = incremental_capex_formula(
515 target_capex=inputs.target_capex,
516 baseline_capex=inputs.baseline_capex,
517 unit=_currency_unit(assumptions),
518 )
519 metrics[FinancialMetricKey.INCREMENTAL_CAPEX] = _metric_from_formula(
520 key=FinancialMetricKey.INCREMENTAL_CAPEX,
521 formula=formula,
522 assumptions=assumptions.merge(
523 {
524 "baseline_capex": inputs.baseline_capex,
525 "target_capex": inputs.target_capex,
526 }
527 ),
528 status=status,
529 value=incremental_capex,
530 formula_store=formula_store,
531 )
533def _currency_unit(assumptions: AssumptionSet, *, per_year: bool = False) -> str:
534 """Return a concrete study currency unit when the study supplied one."""
535 currency = assumptions.get("currency")
536 base_unit = currency if isinstance(currency, str) and currency else "currency"
537 return f"{base_unit}/year" if per_year else base_unit
540def currency_unit(assumptions: AssumptionSet, *, per_year: bool = False) -> str:
541 """Return the currency unit label used by financial metric result lines."""
543 return _currency_unit(assumptions, per_year=per_year)
545def _add_cashflow_metrics(
546 *,
547 inputs: FinancialCalculationInputs,
548 assumptions: AssumptionSet,
549 incremental_capex: Decimal | None,
550 annual_savings: Decimal | None,
551 annual_net_benefit: Decimal | None,
552 annual_cash_flow: Decimal | None,
553 metrics: dict[FinancialMetricKey, FinancialMetric],
554 warnings: list[FinancialWarning],
555 formula_store: MetricFormulaStore | None,
556) -> tuple[DiscountedCashFlowRow, ...]:
557 if not inputs.baseline_fully_calculated:
558 return ()
560 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:
561 add_incomplete_cashflow_warning(inputs=inputs, warnings=warnings)
562 return ()
564 metric_assumptions = _cashflow_metric_assumptions(
565 assumptions=assumptions,
566 incremental_capex=incremental_capex,
567 annual_savings=annual_savings,
568 annual_net_benefit=annual_net_benefit,
569 annual_cash_flow=annual_cash_flow,
570 project_lifetime_years=inputs.project_lifetime_years,
571 discount_rate_percent=inputs.discount_rate_percent,
572 tax_rate_percent=inputs.tax_rate_percent or ZERO,
573 annual_depreciation=inputs.target_annual_depreciation,
574 residual_value=inputs.residual_value,
575 )
576 discounted_cash_flow = tuple(
577 calculate_discounted_cash_flow(
578 incremental_capex=incremental_capex,
579 annual_cash_flow=annual_cash_flow,
580 project_lifetime_years=inputs.project_lifetime_years,
581 discount_rate_percent=inputs.discount_rate_percent,
582 residual_value=inputs.residual_value,
583 )
584 )
585 discount_rate = _discount_rate_from_percent(inputs.discount_rate_percent)
586 metrics[FinancialMetricKey.NPV] = _metric_from_formula(
587 key=FinancialMetricKey.NPV,
588 formula=npv_formula(
589 incremental_capex=incremental_capex,
590 annual_cash_flow=annual_cash_flow,
591 project_lifetime_years=inputs.project_lifetime_years,
592 discount_rate=discount_rate,
593 residual_value=inputs.residual_value,
594 unit=_currency_unit(metric_assumptions),
595 ),
596 assumptions=metric_assumptions,
597 formula_store=formula_store,
598 )
599 metrics[FinancialMetricKey.SIMPLE_PAYBACK_YEARS] = FinancialMetric.from_value(
600 key=FinancialMetricKey.SIMPLE_PAYBACK_YEARS,
601 value=calculate_simple_payback_years(
602 incremental_capex=incremental_capex,
603 annual_cash_flow=annual_cash_flow,
604 project_lifetime_years=inputs.project_lifetime_years,
605 residual_value=inputs.residual_value,
606 ),
607 unit="years",
608 assumptions=metric_assumptions,
609 )
610 if incremental_capex <= 0:
611 warnings.append(
612 FinancialWarning(
613 code="roi_capital_outlay_non_positive",
614 severity="warning",
615 message="ROI is unavailable because incremental capital outlay is not positive.",
616 context={"incremental_capex": str(incremental_capex)},
617 )
618 )
619 metrics[FinancialMetricKey.ROI_PERCENT] = FinancialMetric.from_value(
620 key=FinancialMetricKey.ROI_PERCENT,
621 value=None,
622 unit="percent",
623 assumptions=metric_assumptions,
624 )
625 else:
626 metrics[FinancialMetricKey.ROI_PERCENT] = _metric_from_formula(
627 key=FinancialMetricKey.ROI_PERCENT,
628 formula=roi_percent_formula(
629 incremental_capex=incremental_capex,
630 annual_cash_flow=annual_cash_flow,
631 project_lifetime_years=inputs.project_lifetime_years,
632 residual_value=inputs.residual_value,
633 ),
634 assumptions=metric_assumptions,
635 formula_store=formula_store,
636 )
637 return discounted_cash_flow
639def _pure_calculation_baseline_resolution(*, inputs: FinancialCalculationInputs) -> BaselineResolution:
640 return BaselineResolution(
641 source="pure_calculation",
642 is_guided_default=not inputs.baseline_fully_calculated,
643 capex=inputs.baseline_capex,
644 annual_opex=inputs.baseline_annual_opex,
645 annual_heat_basis=None,
646 annual_heat_basis_unit=None,
647 residual_value=inputs.residual_value,
648 project_lifetime_years=inputs.project_lifetime_years,
649 discount_rate_percent=inputs.discount_rate_percent,
650 assumptions=inputs.assumptions,
651 )
653def _discount_factor(*, discount_rate: Decimal, year: int) -> Decimal:
654 with localcontext() as context:
655 context.prec = 34
656 return discount_factor_formula(year=year, discount_rate=discount_rate).evaluate()
658def _quantize_like(value: Decimal, *references: Decimal) -> Decimal:
659 exponent = min(reference.as_tuple().exponent for reference in references)
660 quantum = Decimal("1").scaleb(exponent)
661 with localcontext() as context:
662 context.prec = max(
663 34,
664 len(value.as_tuple().digits),
665 *(len(reference.as_tuple().digits) for reference in references),
666 )
667 return value.quantize(quantum, rounding=ROUND_HALF_UP)
669def _discount_rate_from_percent(discount_rate_percent: Decimal | None) -> Decimal | None:
670 if discount_rate_percent is None:
671 return None
672 if not discount_rate_percent.is_finite():
673 raise FinancialMetricsError(
674 "invalid_discount_rate",
675 "Discount rate must be finite.",
676 context={"discount_rate_percent": str(discount_rate_percent)},
677 )
678 discount_rate = discount_rate_percent / Decimal("100")
679 if ONE + discount_rate <= 0:
680 raise FinancialMetricsError(
681 "invalid_discount_rate",
682 "Discount rate must keep 1 + rate greater than zero.",
683 context={"discount_rate_percent": str(discount_rate_percent), "discount_rate": str(discount_rate)},
684 )
685 return discount_rate
687def _tax_rate_from_percent(tax_rate_percent: Decimal | None) -> Decimal:
688 if tax_rate_percent is None:
689 return ZERO
690 if not tax_rate_percent.is_finite():
691 raise FinancialMetricsError(
692 "invalid_tax_rate",
693 "Tax rate must be finite.",
694 context={"tax_rate_percent": str(tax_rate_percent)},
695 )
696 if tax_rate_percent < ZERO or tax_rate_percent > Decimal("100"):
697 raise FinancialMetricsError(
698 "invalid_tax_rate",
699 "Tax rate must be between 0 and 100 percent.",
700 context={"tax_rate_percent": str(tax_rate_percent)},
701 )
702 return tax_rate_percent / Decimal("100")
704def _cashflow_metric_assumptions(
705 *,
706 assumptions: AssumptionSet,
707 incremental_capex: Decimal,
708 annual_savings: Decimal | None,
709 annual_net_benefit: Decimal | None,
710 annual_cash_flow: Decimal,
711 project_lifetime_years: int,
712 discount_rate_percent: Decimal,
713 tax_rate_percent: Decimal,
714 annual_depreciation: Decimal,
715 residual_value: Decimal,
716) -> AssumptionSet:
717 return assumptions.merge(
718 {
719 "incremental_capex": incremental_capex,
720 "annual_savings": annual_savings,
721 "annual_net_benefit": annual_net_benefit,
722 "annual_cash_flow": annual_cash_flow,
723 "annual_depreciation": annual_depreciation,
724 "tax_rate_percent": tax_rate_percent,
725 "project_lifetime_years": project_lifetime_years,
726 "discount_rate_percent": discount_rate_percent,
727 "residual_value": residual_value,
728 }
729 )
731def _decimal_string(value: Decimal | None) -> str | None:
732 return None if value is None else str(value)
734def parse_decimal(value: Decimal | int | str, *, field_name: str) -> Decimal:
735 try:
736 decimal_value = Decimal(str(value))
737 except (InvalidOperation, ValueError) as exc:
738 raise FinancialMetricsError(
739 "invalid_decimal",
740 "Financial metric input must be numeric.",
741 context={"field": field_name, "value": str(value)},
742 ) from exc
743 if not decimal_value.is_finite():
744 raise FinancialMetricsError(
745 "invalid_decimal",
746 "Financial metric input must be finite.",
747 context={"field": field_name, "value": str(value)},
748 )
749 return decimal_value