Add quarter hour grouping

Author: azvoleff

gefapi/services/stats_service.py

@@ -578,7 +578,14 @@ def _get_execution_time_series(
             else_=Execution.start_date,
         )
 
-        timestamp_bucket = func.date_trunc(trunc_format, event_timestamp)
+        if group_by == "quarter_hour":
+            interval_seconds = 900  # 15 minutes
+            timestamp_bucket = func.to_timestamp(
+                func.floor(func.extract("epoch", event_timestamp) / interval_seconds)
+                * interval_seconds
+            )
+        else:
+            timestamp_bucket = func.date_trunc(trunc_format, event_timestamp)
 
         query = db.session.query(
             timestamp_bucket.label("timestamp"),

tests/services/test_stats_service.py

@@ -289,6 +289,37 @@ def test_get_cache_info_unavailable(self, mock_get_redis_cache):
         assert result["available"] is False
         assert "error" in result
 
+    @patch("gefapi.services.stats_service.db")
+    def test_execution_time_series_quarter_hour_bucket(self, mock_db):
+        """Quarter-hour grouping should return 15 minute buckets."""
+        query_mock = MagicMock()
+        mock_db.session.query.return_value = query_mock
+
+        query_mock.join.return_value = query_mock
+        query_mock.filter.return_value = query_mock
+        query_mock.group_by.return_value = query_mock
+        query_mock.all.return_value = [
+            SimpleNamespace(
+                timestamp=datetime(2025, 1, 1, 10, 15, 0),
+                total=3,
+                status="FINISHED",
+                slug="productivity-v2.0.0",
+            )
+        ]
+
+        result = StatsService._get_execution_time_series(
+            "last_day", "quarter_hour", None, None
+        )
+
+        assert result == [
+            {
+                "timestamp": "2025-01-01T10:15:00",
+                "total": 3,
+                "by_status": {"FINISHED": 3},
+                "by_task": {"productivity": 3},
+            }
+        ]
+
     @patch("gefapi.services.stats_service.db")
     def test_execution_time_series_aggregates_totals(self, mock_db):
         """Total counts should reflect aggregated values from the query."""