Coverage for backend/django/Economics/results/services/financial_metrics/contracts.py: 95%
160 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Financial metric data contracts."""
3from __future__ import annotations
5from collections.abc import Mapping
6from decimal import Decimal
7from typing import Any, TypeAlias
9from pydantic import BaseModel, ConfigDict, Field, field_validator
11from .metric_catalog import FinancialMetricKey, financial_metric_spec
14ZERO = Decimal("0")
15ONE = Decimal("1")
16FinancialScalar: TypeAlias = str | int | Decimal | bool | None
17FinancialContextValue: TypeAlias = FinancialScalar | tuple[str, ...] | tuple[int, ...] | tuple[dict[str, str], ...] | dict[str, str]
18FinancialContext: TypeAlias = dict[str, FinancialContextValue]
21class EconomicsContract(BaseModel):
22 model_config = ConfigDict(frozen=True, allow_inf_nan=True)
24class AssumptionRecord(EconomicsContract):
25 """One audit assumption exposed through financial metric contracts."""
27 key: str
28 value: FinancialScalar
30class AssumptionSet(EconomicsContract):
31 """Typed assumption collection used internally before JSON persistence.
33 Result lines and future APIs can serialize this model deterministically,
34 while callers that need the legacy JSONField boundary can explicitly ask for
35 a scalar mapping with ``to_mapping``.
36 """
38 records: tuple[AssumptionRecord, ...] = ()
40 @classmethod
41 def from_mapping(cls, values: Mapping[str, object] | None = None) -> "AssumptionSet":
42 if not values: 42 ↛ 43line 42 didn't jump to line 43 because the condition on line 42 was never true
43 return cls()
44 return cls(records=tuple(AssumptionRecord(key=str(key), value=_financial_scalar(value)) for key, value in sorted(values.items())))
46 def merge(self, values: Mapping[str, object] | None = None, **kwargs: object) -> "AssumptionSet":
47 merged: dict[str, object] = self.to_mapping()
48 if values: 48 ↛ 50line 48 didn't jump to line 50 because the condition on line 48 was always true
49 merged.update(values)
50 merged.update(kwargs)
51 return AssumptionSet.from_mapping(merged)
53 def merge_set(self, other: "AssumptionSet") -> "AssumptionSet":
54 return self.merge(other.to_mapping())
56 def to_mapping(self) -> dict[str, FinancialScalar]:
57 """Return the legacy JSON-boundary mapping used by result lines."""
58 return {record.key: record.value for record in self.records}
60 def get(self, key: str, default: FinancialScalar = None) -> FinancialScalar:
61 return self.to_mapping().get(key, default)
63 def __getitem__(self, key: str) -> FinancialScalar:
64 return self.to_mapping()[key]
66class TargetAssumptions(EconomicsContract):
67 """Financial assumptions resolved from the target study boundary."""
69 target_study_id: int
70 assumptions_id: int | None = None
71 project_lifetime_years: int | None = None
72 discount_rate_percent: Decimal | None = None
73 currency: str | None = None
74 basis_date: str | None = None
75 inflation_method: str = ""
76 capital_index_series_id: int | None = None
77 operating_index_series_id: int | None = None
78 annual_operating_hours: Decimal | None = None
79 tax_rate_percent: Decimal | None = None
80 depreciation_enabled: bool = False
81 default_depreciation_life_years: int | None = None
82 default_depreciation_salvage_percent: Decimal | None = None
83 contingency_percent: Decimal | None = None
84 electrical_upgrade_rate_amount: Decimal | None = None
85 electrical_upgrade_rate_unit: str = "NZD/kW"
86 peak_demand_kw: Decimal | None = None
87 default_lang_factor: Decimal | None = None
88 assumptions_source: str = "study"
90 def as_assumption_set(self) -> AssumptionSet:
91 return AssumptionSet.from_mapping(
92 {
93 "target_study_id": self.target_study_id,
94 "assumptions_id": self.assumptions_id,
95 "project_lifetime_years": self.project_lifetime_years,
96 "discount_rate_percent": self.discount_rate_percent,
97 "currency": self.currency,
98 "basis_date": self.basis_date,
99 "inflation_method": self.inflation_method,
100 "capital_index_series_id": self.capital_index_series_id,
101 "operating_index_series_id": self.operating_index_series_id,
102 "annual_operating_hours": self.annual_operating_hours,
103 "tax_rate_percent": self.tax_rate_percent,
104 "depreciation_enabled": self.depreciation_enabled,
105 "default_depreciation_life_years": self.default_depreciation_life_years,
106 "default_depreciation_salvage_percent": self.default_depreciation_salvage_percent,
107 "contingency_percent": self.contingency_percent,
108 "electrical_upgrade_rate_amount": self.electrical_upgrade_rate_amount,
109 "electrical_upgrade_rate_unit": self.electrical_upgrade_rate_unit,
110 "peak_demand_kw": self.peak_demand_kw,
111 "default_lang_factor": self.default_lang_factor,
112 "assumptions_source": self.assumptions_source,
113 }
114 )
116class FinancialWarning(EconomicsContract):
117 code: str
118 severity: str
119 message: str
120 context: FinancialContext = Field(default_factory=dict)
122class FinancialMetric(EconomicsContract):
123 key: FinancialMetricKey
124 value: Decimal | None
125 unit: str
126 assumptions: AssumptionSet = Field(default_factory=AssumptionSet)
127 status: str = "calculated"
128 formula_audit: dict[str, Any] | None = None
129 formula_record_id: int | None = None
131 @classmethod
132 def from_value(
133 cls,
134 *,
135 key: FinancialMetricKey | str,
136 value: Decimal | None,
137 unit: str,
138 assumptions: AssumptionSet | None = None,
139 status: str = "calculated",
140 ) -> "FinancialMetric":
141 metric_key = FinancialMetricKey(key)
142 return cls(
143 key=metric_key,
144 value=value,
145 unit=unit,
146 assumptions=assumptions or AssumptionSet(),
147 status=status if value is not None else "unavailable",
148 )
150 @property
151 def row_key(self) -> str:
152 """Return the catalog-owned persistent result row key."""
153 spec = financial_metric_spec(self.key)
154 return spec.row_key if spec is not None else f"metric.{self.key.value}"
156 @property
157 def label(self) -> str:
158 """Return the catalog-owned user-facing metric label."""
159 spec = financial_metric_spec(self.key)
160 return spec.label if spec is not None else self.key.value.replace("_", " ").title()
162 @property
163 def line_kind(self) -> str:
164 """Return the catalog-owned result-line kind for this metric."""
165 spec = financial_metric_spec(self.key)
166 return spec.line_kind if spec is not None else "financial_metric"
168 @property
169 def sort_order(self) -> int:
170 """Return the catalog-owned result-line sort order for this metric."""
171 spec = financial_metric_spec(self.key)
172 return spec.sort_order if spec is not None else 100
174class DiscountedCashFlowRow(EconomicsContract):
175 year: int
176 cash_flow: Decimal
177 discount_factor: Decimal
178 present_value: Decimal
179 cumulative_cash_flow: Decimal
180 cumulative_present_value: Decimal
182class BaselineResolution(EconomicsContract):
183 source: str
184 is_guided_default: bool
185 capex: Decimal | None
186 annual_opex: Decimal | None
187 annual_heat_basis: Decimal | None
188 annual_heat_basis_unit: str | None
189 residual_value: Decimal
190 project_lifetime_years: int | None
191 discount_rate_percent: Decimal | None
192 assumptions: AssumptionSet = Field(default_factory=AssumptionSet)
194class FinancialCalculationInputs(EconomicsContract):
195 target_capex: Decimal
196 target_annual_opex: Decimal
197 target_annual_revenue: Decimal = ZERO
198 target_annual_depreciation: Decimal = ZERO
199 target_purchase_basis_capex: Decimal = ZERO
200 target_installed_basis_capex: Decimal = ZERO
201 target_contingency_capex: Decimal = ZERO
202 target_electrical_upgrade_capex: Decimal = ZERO
203 target_peak_demand_kw: Decimal | None = None
204 baseline_capex: Decimal | None
205 baseline_annual_opex: Decimal | None
206 project_lifetime_years: int | None
207 discount_rate_percent: Decimal | None
208 tax_rate_percent: Decimal | None = ZERO
209 residual_value: Decimal = ZERO
210 baseline_fully_calculated: bool = False
211 assumptions: AssumptionSet = Field(default_factory=AssumptionSet)
212 warnings: tuple[FinancialWarning, ...] = ()
214 @field_validator("assumptions", mode="before")
215 @classmethod
216 def coerce_assumptions(cls, value):
217 if isinstance(value, AssumptionSet):
218 return value
219 if value is None: 219 ↛ 220line 219 didn't jump to line 220 because the condition on line 219 was never true
220 return AssumptionSet()
221 if isinstance(value, Mapping): 221 ↛ 223line 221 didn't jump to line 223 because the condition on line 221 was always true
222 return AssumptionSet.from_mapping(value)
223 return value
225class FinancialMetricsResult(EconomicsContract):
226 metrics: dict[FinancialMetricKey, FinancialMetric]
227 discounted_cash_flow: tuple[DiscountedCashFlowRow, ...]
228 baseline_resolution: BaselineResolution
229 warnings: tuple[FinancialWarning, ...]
230 input_snapshot_payload: dict[str, Any] | None = None
232class FinancialMetricsError(ValueError):
233 def __init__(self, code: str, message: str, *, context: FinancialContext | None = None):
234 super().__init__(message)
235 self.code = code
236 self.message = message
237 self.context = context or {}
239def _financial_scalar(value: object) -> FinancialScalar:
240 """Normalize flexible assumption inputs into a small JSON-safe scalar set."""
241 if value is None or isinstance(value, (str, int, bool)):
242 return value
243 if isinstance(value, Decimal): 243 ↛ 245line 243 didn't jump to line 245 because the condition on line 243 was always true
244 return str(value)
245 return str(value)
248def _decimal_string(value: Decimal | None) -> str | None:
249 return None if value is None else str(value)