Add first_value and last_value aggregations and allow winsorization.

Author: chasdevs

eppo_metrics_sync/schema/eppo_metric_schema.json

@@ -156,7 +156,7 @@
                             },
                             "operation": {
                                 "description": "Which aggregation to apply to the fact",
-                                "enum": ["sum", "count", "distinct_entity", "threshold", "conversion", "retention", "count_distinct"]
+                                "enum": ["sum", "count", "distinct_entity", "threshold", "conversion", "retention", "count_distinct", "last_value", "first_value"]
                             },
                             "filters": {
                                 "description": "Optional fact property filters to apply",
@@ -254,7 +254,7 @@
                             },
                             "operation": {
                                 "description": "How to aggregate fact",
-                                "enum": ["sum", "count", "distinct_entity", "count_distinct"]
+                                "enum": ["sum", "count", "distinct_entity", "count_distinct", "last_value", "first_value"]
                             },
                             "filters": {
                                 "description": "Optional fact property filters to apply",

eppo_metrics_sync/validation.py

@@ -112,13 +112,13 @@ def aggregation_is_valid(aggregation):
     error_message = []
 
     if aggregation['operation'] not in ['sum', 'count', 'count_distinct', 'distinct_entity', 'threshold', 'retention',
-                                        'conversion']:
+                                        'conversion', 'last_value', 'first_value']:
         error_message.append(
             'Invalid aggregation operation: ' + aggregation['operation']
         )
 
     # can only winsorize sum or count metrics
-    if aggregation['operation'] not in ['sum', 'count']:
+    if aggregation['operation'] not in ['sum', 'count', 'last_value', 'first_value']:
         if [name for name in winsorization_parameters if name in aggregation]:
             error_message.append(
                 'Cannot winsorize a metric with operation ' + aggregation['operation']

tests/test_validation.py

@@ -129,3 +129,16 @@ def test_count_distinct():
 
     res = aggregation_is_valid(test_agg)
     assert res == None
+
+def test_last_value():
+    res = aggregation_is_valid({'operation': 'last_value'})
+    assert res == None
+
+def test_first_value():
+    res = aggregation_is_valid({'operation': 'first_value'})
+    assert res == None
+
+def test_valid_yaml():
+    eppo_metrics_sync = EppoMetricsSync(directory=None)
+    eppo_metrics_sync.load_eppo_yaml(path='tests/yaml/valid/purchases.yaml')
+    eppo_metrics_sync.validate()

tests/yaml/valid/purchases.yaml

@@ -52,4 +52,11 @@ metrics:
         operation: equals
         values:
           - "UNITED STATES"
-
+- name: Last Purchase Amount
+  entity: User
+  description: The last purchase amount with winsorization
+  numerator:
+    fact_name: Purchase
+    operation: last_value
+    winsorization_lower_percentile: 0.05
+    winsorization_upper_percentile: 0.95