Coverage for backend/django/Economics/results/services/comparison/metrics.py: 98%
74 statements
« prev ^ index » next coverage.py v7.10.7, created at 2026-07-22 05:22 +0000
1"""Metric row and delta calculation for Economics study comparisons."""
3from __future__ import annotations
5from decimal import Decimal, ROUND_HALF_UP
6from typing import Protocol
8from Economics.shared.choices import EconomicsBaselineMode
10from .contracts import (
11 ComparisonMetricValue,
12 ComparisonStatus,
13 ComparisonRowPayload,
14 ComparisonTarget,
15 ComparisonValuePayload,
16 ResultContext,
17)
18from Economics.results.services.financial_metrics.metric_catalog import (
19 MetricComparisonDirection,
20 key_financial_metric_specs,
21)
23PERCENTAGE_QUANTUM = Decimal("0.0001")
26class ComparisonRowSpec(Protocol):
27 row_key: str
28 label: str
29 comparison_direction: MetricComparisonDirection
30 maximum_fraction_digits: int
33def comparison_rows(
34 *,
35 studies: list[ComparisonTarget],
36 baseline_mode: EconomicsBaselineMode,
37 baseline_id: int,
38 baseline_context: ResultContext | None,
39 result_contexts: dict[int | None, ResultContext | None],
40) -> list[ComparisonRowPayload]:
41 """Build comparison rows for the key financial metric set."""
42 return [
43 _comparison_row(
44 row_key=spec.row_key,
45 label=spec.label,
46 comparison_direction=spec.comparison_direction,
47 maximum_fraction_digits=spec.maximum_fraction_digits,
48 baseline_dependent=spec.baseline_dependent,
49 studies=studies,
50 baseline_mode=baseline_mode,
51 baseline_id=baseline_id,
52 baseline_context=baseline_context,
53 result_contexts=result_contexts,
54 )
55 for spec in key_financial_metric_specs()
56 ]
59def comparison_rows_for_specs(
60 *,
61 specs: tuple[ComparisonRowSpec, ...],
62 studies: list[ComparisonTarget],
63 baseline_mode: EconomicsBaselineMode,
64 baseline_id: int,
65 baseline_context: ResultContext | None,
66 result_contexts: dict[int | None, ResultContext | None],
67) -> list[ComparisonRowPayload]:
68 """Build comparable rows for non-financial metric specs."""
69 return [
70 _comparison_row(
71 row_key=spec.row_key,
72 label=spec.label,
73 comparison_direction=spec.comparison_direction,
74 maximum_fraction_digits=spec.maximum_fraction_digits,
75 baseline_dependent=False,
76 studies=studies,
77 baseline_mode=baseline_mode,
78 baseline_id=baseline_id,
79 baseline_context=baseline_context,
80 result_contexts=result_contexts,
81 )
82 for spec in specs
83 ]
86def _comparison_row(
87 *,
88 row_key: str,
89 label: str,
90 comparison_direction: MetricComparisonDirection,
91 maximum_fraction_digits: int,
92 baseline_dependent: bool,
93 studies: list[ComparisonTarget],
94 baseline_mode: EconomicsBaselineMode,
95 baseline_id: int,
96 baseline_context: ResultContext | None,
97 result_contexts: dict[int | None, ResultContext | None],
98) -> ComparisonRowPayload:
99 """Build one metric row and suppress baseline values that need a baseline themselves."""
100 baseline_line = (
101 None
102 if baseline_dependent
103 else _line_for_context(baseline_context, row_key)
104 )
105 values = [
106 _value_payload(
107 target=target,
108 row_key=row_key,
109 baseline_dependent=baseline_dependent,
110 baseline_mode=baseline_mode,
111 baseline_id=baseline_id,
112 baseline_line=baseline_line,
113 context=result_contexts.get(target.study_id),
114 )
115 for target in studies
116 ]
117 _mark_best_values(comparison_direction=comparison_direction, values=values)
118 return ComparisonRowPayload(
119 row_key=row_key,
120 label=label,
121 comparison_direction=comparison_direction,
122 maximum_fraction_digits=maximum_fraction_digits,
123 values=values,
124 )
127def _value_payload(
128 *,
129 target: ComparisonTarget,
130 row_key: str,
131 baseline_dependent: bool,
132 baseline_mode: EconomicsBaselineMode,
133 baseline_id: int,
134 baseline_line: ComparisonMetricValue | None,
135 context: ResultContext | None,
136) -> ComparisonValuePayload:
137 """Render one study's cell value, including comparison status and percentage delta."""
138 is_study_baseline = baseline_mode == EconomicsBaselineMode.STUDY and target.study_id == baseline_id
139 if is_study_baseline and baseline_dependent:
140 return ComparisonValuePayload(
141 study_id=target.study_id,
142 amount=None,
143 unit="",
144 percentage_difference=None,
145 comparison_status=ComparisonStatus.BASELINE_NOT_APPLICABLE,
146 )
147 line = _line_for_context(context, row_key)
148 amount = line.amount if line is not None else None
149 unit = line.unit if line is not None else ""
150 if target.availability_status == "unavailable":
151 status = ComparisonStatus.UNAVAILABLE
152 elif is_study_baseline:
153 status = ComparisonStatus.BASELINE
154 elif context is None:
155 status = ComparisonStatus.MISSING_RESULT
156 elif line is None:
157 status = ComparisonStatus.MISSING_METRIC
158 elif line.comparison_status is not None:
159 status = line.comparison_status
160 else:
161 status = _comparison_status(line=line, baseline_line=baseline_line)
162 return ComparisonValuePayload(
163 study_id=target.study_id,
164 amount=str(amount) if amount is not None else None,
165 unit=unit,
166 percentage_difference=_percentage_difference(
167 line=line,
168 baseline_line=baseline_line,
169 status=status,
170 ),
171 comparison_status=status,
172 )
175def _line_for_context(context: ResultContext | None, row_key: str) -> ComparisonMetricValue | None:
176 if context is None:
177 return None
178 return context.lines.get(row_key)
181def _mark_best_values(
182 *,
183 comparison_direction: MetricComparisonDirection,
184 values: list[ComparisonValuePayload],
185) -> None:
186 """Mark the best available cells for one row using the central metric catalog direction."""
187 direction = comparison_direction
188 if direction == MetricComparisonDirection.NOT_RANKED:
189 return
191 ranked_values = [
192 (value, Decimal(value.amount))
193 for value in values
194 if value.amount is not None
195 ]
196 if len(ranked_values) < 2:
197 return
199 # Mixed units or currencies would make a visual "best" marker misleading, so leave
200 # those rows unranked even when individual cells have numeric values.
201 units = {value.unit for value, _amount in ranked_values}
202 if len(units) != 1:
203 return
205 # Equal values do not give the user a meaningful "best" cell, even when the
206 # metric has a rankable direction.
207 if len({amount for _value, amount in ranked_values}) < 2:
208 return
210 best_amount = (
211 max(amount for _value, amount in ranked_values)
212 if direction == MetricComparisonDirection.HIGHER_IS_BETTER
213 else min(amount for _value, amount in ranked_values)
214 )
215 for value, amount in ranked_values:
216 value.is_best = amount == best_amount
219def _comparison_status(
220 *,
221 line: ComparisonMetricValue,
222 baseline_line: ComparisonMetricValue | None,
223) -> ComparisonStatus:
224 """Classify whether the target value can be compared against the baseline value."""
225 if baseline_line is None or baseline_line.amount is None:
226 return ComparisonStatus.MISSING_BASELINE
227 if baseline_line.amount == 0:
228 return ComparisonStatus.ZERO_BASELINE
229 if line.unit != baseline_line.unit:
230 return ComparisonStatus.UNIT_MISMATCH
231 return ComparisonStatus.COMPARABLE
234def _percentage_difference(
235 *,
236 line: ComparisonMetricValue | None,
237 baseline_line: ComparisonMetricValue | None,
238 status: ComparisonStatus,
239) -> str | None:
240 """Return the target-vs-baseline percentage delta only for comparable values."""
241 if status != ComparisonStatus.COMPARABLE or line is None or baseline_line is None:
242 return None
243 if line.amount is None or baseline_line.amount is None: 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true
244 return None
245 difference = ((line.amount - baseline_line.amount) / baseline_line.amount) * Decimal(
246 "100"
247 )
248 return str(difference.quantize(PERCENTAGE_QUANTUM, rounding=ROUND_HALF_UP))