Bases: KPIDefinition
KPI definition for comparing values between two datasets.
Creates KPIs by
- Generating base KPIs from base_definition for reference and variation datasets
- Applying comparison operation to corresponding KPI values
- Creating comparison KPI instances with proper metadata
This is typically used to compare scenarios (e.g., "Price increase in
high_res vs base scenario").
Attributes:
| Name |
Type |
Description |
base_definition |
KPIDefinition
|
Definition to generate base KPIs from
|
comparison |
ValueComparison
|
Comparison operation to apply (e.g., ValueComparisons.Increase)
|
name_prefix |
str
|
Optional prefix for comparison KPI names
|
name_suffix |
str
|
Optional suffix for comparison KPI names
|
Source code in submodules/mesqual/mesqual/kpis/definitions/comparison.py
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136 | @dataclass
class ComparisonKPIDefinition(KPIDefinition):
"""
KPI definition for comparing values between two datasets.
Creates KPIs by:
1. Generating base KPIs from base_definition for reference and variation datasets
2. Applying comparison operation to corresponding KPI values
3. Creating comparison KPI instances with proper metadata
This is typically used to compare scenarios (e.g., "Price increase in
high_res vs base scenario").
Attributes:
base_definition: Definition to generate base KPIs from
comparison: Comparison operation to apply (e.g., ValueComparisons.Increase)
name_prefix: Optional prefix for comparison KPI names
name_suffix: Optional suffix for comparison KPI names
"""
base_definition: KPIDefinition
comparison: ValueComparison
name_prefix: str = ''
name_suffix: str = ''
extra_attributes: dict = None
custom_name: str | None = None
def generate_kpis(self, dataset: DatasetComparison) -> list[KPI]:
"""
Generate comparison KPIs from a DatasetComparison.
Process:
1. Get reference and variation datasets from comparison dataset
2. Generate base KPIs for both datasets
3. Match KPIs by object_name
4. Apply comparison operation
5. Create comparison KPI instances
Args:
dataset: DatasetComparison containing reference and variation datasets
Returns:
List of comparison KPI instances
Raises:
AttributeError: If dataset is not a DatasetComparison
"""
if not isinstance(dataset, DatasetComparison):
raise TypeError(
f"ComparisonKPIDefinition requires a {DatasetComparison.__name__}, got {type(dataset).__name__}"
)
reference_dataset = dataset.reference_dataset
variation_dataset = dataset.variation_dataset
reference_kpis = self.base_definition.generate_kpis(reference_dataset)
variation_kpis = self.base_definition.generate_kpis(variation_dataset)
variation_kpis_by_object = {
kpi.attributes.object_name: kpi
for kpi in variation_kpis
}
comparison_kpis = []
for ref_kpi in reference_kpis:
obj_name = ref_kpi.attributes.object_name
if obj_name not in variation_kpis_by_object:
continue
var_kpi = variation_kpis_by_object[obj_name]
comparison_value = self.comparison(var_kpi.value, ref_kpi.value)
# If custom_name is set and there are multiple objects, append object name
kpi_custom_name = self.custom_name
if self.custom_name and len(reference_kpis) > 1:
kpi_custom_name = f"{self.custom_name} {obj_name}"
unit = self.comparison.unit or ref_kpi.attributes.unit
attributes = KPIAttributes(
flag=ref_kpi.attributes.flag,
model_flag=ref_kpi.attributes.model_flag,
object_name=obj_name,
aggregation=ref_kpi.attributes.aggregation,
dataset_name=dataset.name,
dataset_type=type(dataset),
value_comparison=self.comparison,
reference_dataset_name=reference_dataset.name,
variation_dataset_name=variation_dataset.name,
name_prefix=self.name_prefix,
name_suffix=self.name_suffix,
custom_name=kpi_custom_name,
unit=unit,
dataset_attributes=dataset.attributes,
extra_attributes=self.extra_attributes or dict()
)
comparison_kpi = KPI(
value=comparison_value,
attributes=attributes,
dataset=dataset
)
for k in [var_kpi, ref_kpi]:
obj_info = k.get_object_info_from_model()
if isinstance(obj_info, pd.Series) and not obj_info.empty:
comparison_kpi._object_info = obj_info
break
comparison_kpis.append(comparison_kpi)
return comparison_kpis
def required_flags(self) -> set[FlagTypeProtocol]:
"""
Return required flags from base definition.
Returns:
Set of flags required by base definition
"""
return self.base_definition.required_flags()
|
Generate comparison KPIs from a DatasetComparison.
Process
- Get reference and variation datasets from comparison dataset
- Generate base KPIs for both datasets
- Match KPIs by object_name
- Apply comparison operation
- Create comparison KPI instances
Parameters:
| Name |
Type |
Description |
Default |
dataset
|
DatasetComparison
|
DatasetComparison containing reference and variation datasets
|
required
|
Returns:
| Type |
Description |
list[KPI]
|
List of comparison KPI instances
|
Raises:
| Type |
Description |
AttributeError
|
If dataset is not a DatasetComparison
|
Source code in submodules/mesqual/mesqual/kpis/definitions/comparison.py
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127 | def generate_kpis(self, dataset: DatasetComparison) -> list[KPI]:
"""
Generate comparison KPIs from a DatasetComparison.
Process:
1. Get reference and variation datasets from comparison dataset
2. Generate base KPIs for both datasets
3. Match KPIs by object_name
4. Apply comparison operation
5. Create comparison KPI instances
Args:
dataset: DatasetComparison containing reference and variation datasets
Returns:
List of comparison KPI instances
Raises:
AttributeError: If dataset is not a DatasetComparison
"""
if not isinstance(dataset, DatasetComparison):
raise TypeError(
f"ComparisonKPIDefinition requires a {DatasetComparison.__name__}, got {type(dataset).__name__}"
)
reference_dataset = dataset.reference_dataset
variation_dataset = dataset.variation_dataset
reference_kpis = self.base_definition.generate_kpis(reference_dataset)
variation_kpis = self.base_definition.generate_kpis(variation_dataset)
variation_kpis_by_object = {
kpi.attributes.object_name: kpi
for kpi in variation_kpis
}
comparison_kpis = []
for ref_kpi in reference_kpis:
obj_name = ref_kpi.attributes.object_name
if obj_name not in variation_kpis_by_object:
continue
var_kpi = variation_kpis_by_object[obj_name]
comparison_value = self.comparison(var_kpi.value, ref_kpi.value)
# If custom_name is set and there are multiple objects, append object name
kpi_custom_name = self.custom_name
if self.custom_name and len(reference_kpis) > 1:
kpi_custom_name = f"{self.custom_name} {obj_name}"
unit = self.comparison.unit or ref_kpi.attributes.unit
attributes = KPIAttributes(
flag=ref_kpi.attributes.flag,
model_flag=ref_kpi.attributes.model_flag,
object_name=obj_name,
aggregation=ref_kpi.attributes.aggregation,
dataset_name=dataset.name,
dataset_type=type(dataset),
value_comparison=self.comparison,
reference_dataset_name=reference_dataset.name,
variation_dataset_name=variation_dataset.name,
name_prefix=self.name_prefix,
name_suffix=self.name_suffix,
custom_name=kpi_custom_name,
unit=unit,
dataset_attributes=dataset.attributes,
extra_attributes=self.extra_attributes or dict()
)
comparison_kpi = KPI(
value=comparison_value,
attributes=attributes,
dataset=dataset
)
for k in [var_kpi, ref_kpi]:
obj_info = k.get_object_info_from_model()
if isinstance(obj_info, pd.Series) and not obj_info.empty:
comparison_kpi._object_info = obj_info
break
comparison_kpis.append(comparison_kpi)
return comparison_kpis
|
required_flags() -> set[FlagTypeProtocol]
Return required flags from base definition.
Returns:
| Type |
Description |
set[FlagTypeProtocol]
|
Set of flags required by base definition
|
Source code in submodules/mesqual/mesqual/kpis/definitions/comparison.py
129
130
131
132
133
134
135
136 | def required_flags(self) -> set[FlagTypeProtocol]:
"""
Return required flags from base definition.
Returns:
Set of flags required by base definition
"""
return self.base_definition.required_flags()
|