Optimization (#3)

- feat - Optimization (a5cef14ae648e6d3c69fb58408a5ec70d070d773)
- feat - Optimization (47a187a765ee8ae1e0d2e7164e3f629f9b042d4d)


Co-authored-by: Y <q275343119@users.noreply.huggingface.co>

.github/workflows/ci.yml

@@ -0,0 +1,29 @@
+name: CI
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Check out repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+
+      - name: Install package
+        run: python -m pip install --upgrade pip && python -m pip install ".[dev]"
+
+      - name: Check formatting
+        run: ruff format --check .
+
+      - name: Lint
+        run: ruff check .
+
+      - name: Test
+        run: pytest

CHANGELOG.md

@@ -0,0 +1,20 @@
+# Changelog
+
+All notable changes to this project will be documented in this file.
+
+## Unreleased
+
+### Added
+
+- Added full-range overview totals so UV and Sessions are distinct counts across the selected range.
+- Added ordered funnel logic that counts each step only when it occurs after the previous required step.
+- Added benchmark choices, raw data tables, and CSV export support to the dashboard.
+- Added query validation, MongoDB ping checks, and dashboard-friendly error messages.
+- Added pytest coverage for metric totals, query validation, and MongoDB aggregation pipeline shape.
+- Added CI for formatting, linting, and tests.
+
+### Changed
+
+- Updated new vs returning visitor logic to compute first-seen dates from the full available page-view history before applying the selected reporting range.
+- Updated MongoDB aggregation pipelines to prefer an indexed `ts` Date field while retaining fallback support for legacy `timestamp` values.
+- Documented recommended MongoDB indexes for production deployments.

README.md

@@ -23,6 +23,7 @@ The primary purpose of this document is to define **what is measured**, **where
 All analytics are based on the `events` collection and the following stable fields:
 
 - Core dimensions: `event_name`, `timestamp`, `session_id`
+- Preferred event time: `ts` as a MongoDB Date
 - Behavior context: `benchmark`, `filters`
 - Visitor identity (approximate): `properties.visitor_id`
 - Change context: `properties.old_value`, `properties.new_value`, `properties.filter_name`
@@ -51,7 +52,7 @@ Important event names:
 - **Definition**: Number of unique interaction sessions.
 - **Source fields**: `session_id`
 - **Calculation**:
-  - Sessions = count of distinct `session_id` in the selected time range
+  - Sessions = count of distinct non-empty `session_id` values in the selected time range
 
 ### 3) UV (Unique Visitors, Approximate)
 
@@ -59,7 +60,7 @@ Important event names:
 - **Source fields**: `properties.visitor_id`
 - **Calculation**:
   - Remove null/empty `properties.visitor_id`
-  - UV = count of distinct `properties.visitor_id`
+  - UV = count of distinct `properties.visitor_id` values in the selected time range
 
 ### 4) Sessions Per Visitor
 
@@ -106,7 +107,7 @@ Important event names:
 - **Source fields**: `event_name`, `session_id`
 - **Calculation**:
   - For each `filter_change_`* event type:
-    - collect distinct `session_id`
+    - collect distinct non-empty `session_id`
     - coverage = distinct session count
 
 ---
@@ -122,12 +123,12 @@ Recommended session-level funnel:
 
 ### 9) Step Session Count
 
-- **Definition**: Number of sessions that reached each funnel step.
-- **Source fields**: `session_id`, `event_name`
+- **Definition**: Number of sessions that reached each ordered funnel step.
+- **Source fields**: `session_id`, `event_name`, `ts` or `timestamp`
 - **Calculation**:
   - Group events by `session_id`
-  - For each session, mark whether each step exists
-  - Count sessions satisfying each cumulative step condition
+  - Sort events by event time
+  - Count each cumulative step only when it occurs after the previous required step
 
 ### 10) Step Conversion Rate
 
@@ -144,10 +145,10 @@ Recommended session-level funnel:
 ### 11) New Visitors
 
 - **Definition**: Visitors whose current period contains their first observed visit date.
-- **Source fields**: `event_name`, `timestamp`, `properties.visitor_id`
+- **Source fields**: `event_name`, `ts` or `timestamp`, `properties.visitor_id`
 - **Calculation**:
   - Use `page_view` events only
-  - For each `visitor_id`, find earliest timestamp (`first_seen`)
+  - For each `visitor_id`, find earliest timestamp (`first_seen`) from the full available dataset
   - If event date equals `first_seen` date, classify as `new`
   - Count distinct `visitor_id` by period
 
@@ -170,10 +171,11 @@ All trend metrics support these granularities:
 - `week` -> `%G-W%V` (ISO week)
 - `month` -> `%Y-%m`
 
-Time filtering is applied on converted event time:
+Time filtering rules:
 
-- Convert `timestamp` to datetime (`ts`)
-- Keep records where `start_time <= ts <= end_time`
+- Prefer the indexed MongoDB Date field `ts`
+- Fall back to converting legacy `timestamp` values when `ts` is not present
+- Keep records where `start_time <= event time <= end_time`
 
 Optional benchmark filtering:
 
@@ -186,6 +188,26 @@ Optional benchmark filtering:
 1. `visitor_id` is an approximate identifier, not a strict user identity.
 2. For `filter_change_`*, `properties.new_value` may not always represent the actual final filter value; prefer `filters` snapshot for behavioral context.
 3. If `table_download` is not instrumented, funnel step 4 will under-report by design.
+4. Total UV and Sessions are distinct counts across the full selected time range. They are not calculated by summing per-period trend values.
+5. Funnel steps are ordered by event time. A session only reaches a later step when that step happens after the previous required step.
+
+---
+
+## MongoDB Performance Notes
+
+For production deployments, store event time as a MongoDB Date field named `ts`. Keeping only string timestamps forces aggregation pipelines to convert time values at query time and can reduce index usage.
+
+Recommended indexes:
+
+```javascript
+db.events.createIndex({ ts: 1 })
+db.events.createIndex({ ts: 1, benchmark: 1 })
+db.events.createIndex({ event_name: 1, ts: 1 })
+db.events.createIndex({ session_id: 1, ts: 1 })
+db.events.createIndex({ "properties.visitor_id": 1, ts: 1 })
+```
+
+Legacy events with only `timestamp` remain supported, but backfilling `ts` is recommended before running this dashboard against large collections.
 
 ---
 
@@ -203,3 +225,10 @@ uv sync
 uv run leaderboard-analytics
 ```
 
+Run quality checks:
+
+```bash
+uv run ruff format --check .
+uv run ruff check .
+uv run pytest
+```

app.py

@@ -1,5 +1,5 @@
-from pathlib import Path
 import sys
+from pathlib import Path
 
 # Ensure src-layout package is importable in Hugging Face Spaces runtime.
 ROOT_DIR = Path(__file__).resolve().parent
@@ -7,8 +7,7 @@ SRC_DIR = ROOT_DIR / "src"
 if str(SRC_DIR) not in sys.path:
     sys.path.insert(0, str(SRC_DIR))
 
-from leaderboard_analytics.main import run
-
+from leaderboard_analytics.main import run  # noqa: E402
 
 if __name__ == "__main__":
     run()

pyproject.toml

@@ -14,6 +14,12 @@ dependencies = [
   "plotly>=5.24.1",
 ]
 
+[project.optional-dependencies]
+dev = [
+  "pytest>=8.3.0",
+  "ruff>=0.8.0",
+]
+
 [tool.ruff]
 line-length = 100
 target-version = "py311"
@@ -34,3 +40,6 @@ build-backend = "hatchling.build"
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/leaderboard_analytics"]
+
+[tool.pytest.ini_options]
+pythonpath = ["src"]

src/leaderboard_analytics/__init__.py

@@ -1,2 +1 @@
 """Leaderboard analytics package."""
-

src/leaderboard_analytics/config.py

@@ -17,4 +17,3 @@ class Settings(BaseSettings):
 @lru_cache(maxsize=1)
 def get_settings() -> Settings:
     return Settings()
-

src/leaderboard_analytics/db.py

@@ -9,7 +9,9 @@ def get_mongo_client() -> MongoClient:
     settings = get_settings()
     if not settings.mongo_uri:
         raise ValueError("MONGO_URI is not configured. Please set MONGO_URI in .env file.")
-    return MongoClient(settings.mongo_uri)
+    client = MongoClient(settings.mongo_uri, serverSelectionTimeoutMS=5000)
+    client.admin.command("ping")
+    return client
 
 
 def get_database(client: MongoClient) -> Database:
@@ -20,4 +22,3 @@ def get_database(client: MongoClient) -> Database:
 def get_events_collection(db: Database) -> Collection:
     settings = get_settings()
     return db[settings.mongo_collection]
-

src/leaderboard_analytics/main.py

@@ -19,4 +19,3 @@ def run() -> None:
 
 if __name__ == "__main__":
     run()
-

src/leaderboard_analytics/repositories.py

@@ -11,12 +11,34 @@ def _period_expression(granularity: Granularity) -> dict:
         Granularity.WEEK: "%G-W%V",
         Granularity.MONTH: "%Y-%m",
     }
-    return {"$dateToString": {"format": format_map[granularity], "date": "$ts"}}
+    return {"$dateToString": {"format": format_map[granularity], "date": "$event_ts"}}
+
+
+def _with_normalized_time() -> dict:
+    return {
+        "$addFields": {
+            "event_ts": {"$ifNull": ["$ts", {"$toDate": "$timestamp"}]},
+            "visitor_id": "$properties.visitor_id",
+        }
+    }
+
+
+def _indexed_time_prefilter(filters: QueryFilters) -> dict:
+    matcher: dict = {
+        "$or": [
+            {"ts": {"$gte": filters.start_time, "$lte": filters.end_time}},
+            {"ts": None},
+            {"ts": {"$exists": False}},
+        ]
+    }
+    if filters.benchmark:
+        matcher["benchmark"] = filters.benchmark
+    return matcher
 
 
 def _with_time_and_optional_benchmark(filters: QueryFilters) -> dict:
     matcher: dict = {
-        "ts": {
+        "event_ts": {
             "$gte": filters.start_time,
             "$lte": filters.end_time,
         }
@@ -26,6 +48,23 @@ def _with_time_and_optional_benchmark(filters: QueryFilters) -> dict:
     return matcher
 
 
+def _non_empty_set_size(field_name: str, variable_name: str) -> dict:
+    return {
+        "$size": {
+            "$filter": {
+                "input": f"${field_name}",
+                "as": variable_name,
+                "cond": {
+                    "$and": [
+                        {"$ne": [f"$${variable_name}", None]},
+                        {"$ne": [f"$${variable_name}", ""]},
+                    ]
+                },
+            }
+        }
+    }
+
+
 class AnalyticsRepository:
     def __init__(self, events_collection: Collection) -> None:
         self.events_collection = events_collection
@@ -33,7 +72,8 @@ class AnalyticsRepository:
     def overview_timeseries(self, filters: QueryFilters) -> list[dict]:
         period_expr = _period_expression(filters.granularity)
         pipeline: list[dict] = [
-            {"$addFields": {"ts": {"$toDate": "$timestamp"}, "visitor_id": "$properties.visitor_id"}},
+            {"$match": _indexed_time_prefilter(filters)},
+            _with_normalized_time(),
             {"$match": _with_time_and_optional_benchmark(filters)},
             {
                 "$group": {
@@ -50,27 +90,52 @@ class AnalyticsRepository:
                     "period": "$_id.period",
                     "pv": 1,
                     "event_count": 1,
-                    "session_count": {"$size": "$sessions"},
-                    "uv": {
-                        "$size": {
-                            "$filter": {
-                                "input": "$visitors",
-                                "as": "v",
-                                "cond": {"$and": [{"$ne": ["$$v", None]}, {"$ne": ["$$v", ""]}]},
-                            }
-                        }
-                    },
+                    "session_count": _non_empty_set_size("sessions", "s"),
+                    "uv": _non_empty_set_size("visitors", "v"),
                 }
             },
             {"$sort": {"period": 1}},
         ]
         return list(self.events_collection.aggregate(pipeline))
 
+    def overview_totals(self, filters: QueryFilters) -> dict:
+        pipeline: list[dict] = [
+            {"$match": _indexed_time_prefilter(filters)},
+            _with_normalized_time(),
+            {"$match": _with_time_and_optional_benchmark(filters)},
+            {
+                "$group": {
+                    "_id": None,
+                    "pv": {"$sum": {"$cond": [{"$eq": ["$event_name", "page_view"]}, 1, 0]}},
+                    "events": {"$sum": 1},
+                    "sessions": {"$addToSet": "$session_id"},
+                    "visitors": {"$addToSet": "$visitor_id"},
+                }
+            },
+            {
+                "$project": {
+                    "_id": 0,
+                    "pv": 1,
+                    "events": 1,
+                    "sessions": _non_empty_set_size("sessions", "s"),
+                    "uv": _non_empty_set_size("visitors", "v"),
+                }
+            },
+        ]
+        return self.safe_first(self.events_collection.aggregate(pipeline))
+
     def benchmark_top(self, filters: QueryFilters, limit: int = 20) -> list[dict]:
         pipeline: list[dict] = [
-            {"$addFields": {"ts": {"$toDate": "$timestamp"}}},
-            {"$match": {**_with_time_and_optional_benchmark(filters), "event_name": "benchmark_change"}},
+            {"$match": _indexed_time_prefilter(filters)},
+            _with_normalized_time(),
+            {
+                "$match": {
+                    **_with_time_and_optional_benchmark(filters),
+                    "event_name": "benchmark_change",
+                }
+            },
             {"$group": {"_id": "$properties.new_value", "count": {"$sum": 1}}},
+            {"$match": {"_id": {"$nin": [None, ""]}}},
             {"$project": {"_id": 0, "benchmark": "$_id", "count": 1}},
             {"$sort": {"count": -1}},
             {"$limit": limit},
@@ -79,20 +144,27 @@ class AnalyticsRepository:
 
     def filter_distribution(self, filters: QueryFilters) -> list[dict]:
         pipeline: list[dict] = [
-            {"$addFields": {"ts": {"$toDate": "$timestamp"}}},
+            {"$match": _indexed_time_prefilter(filters)},
+            _with_normalized_time(),
             {
                 "$match": {
                     **_with_time_and_optional_benchmark(filters),
                     "event_name": {"$regex": "^filter_change_"},
                 }
             },
-            {"$group": {"_id": "$event_name", "count": {"$sum": 1}, "sessions": {"$addToSet": "$session_id"}}},
+            {
+                "$group": {
+                    "_id": "$event_name",
+                    "count": {"$sum": 1},
+                    "sessions": {"$addToSet": "$session_id"},
+                }
+            },
             {
                 "$project": {
                     "_id": 0,
                     "event_name": "$_id",
                     "count": 1,
-                    "session_coverage": {"$size": "$sessions"},
+                    "session_coverage": _non_empty_set_size("sessions", "s"),
                 }
             },
             {"$sort": {"count": -1}},
@@ -101,41 +173,169 @@ class AnalyticsRepository:
 
     def funnel(self, filters: QueryFilters) -> list[dict]:
         pipeline: list[dict] = [
-            {"$addFields": {"ts": {"$toDate": "$timestamp"}}},
+            {"$match": _indexed_time_prefilter(filters)},
+            _with_normalized_time(),
             {"$match": _with_time_and_optional_benchmark(filters)},
-            {"$group": {"_id": "$session_id", "events": {"$addToSet": "$event_name"}}},
+            {"$sort": {"session_id": 1, "event_ts": 1}},
+            {
+                "$group": {
+                    "_id": "$session_id",
+                    "events": {"$push": {"name": "$event_name", "ts": "$event_ts"}},
+                }
+            },
+            {"$match": {"_id": {"$nin": [None, ""]}}},
+            {
+                "$project": {
+                    "events": 1,
+                    "page_view_at": {
+                        "$arrayElemAt": [
+                            {
+                                "$map": {
+                                    "input": {
+                                        "$filter": {
+                                            "input": "$events",
+                                            "as": "event",
+                                            "cond": {"$eq": ["$$event.name", "page_view"]},
+                                        }
+                                    },
+                                    "as": "event",
+                                    "in": "$$event.ts",
+                                }
+                            },
+                            0,
+                        ]
+                    },
+                }
+            },
+            {
+                "$project": {
+                    "events": 1,
+                    "page_view_at": 1,
+                    "benchmark_change_at": {
+                        "$arrayElemAt": [
+                            {
+                                "$map": {
+                                    "input": {
+                                        "$filter": {
+                                            "input": "$events",
+                                            "as": "event",
+                                            "cond": {
+                                                "$and": [
+                                                    {"$eq": ["$$event.name", "benchmark_change"]},
+                                                    {"$gte": ["$$event.ts", "$page_view_at"]},
+                                                ]
+                                            },
+                                        }
+                                    },
+                                    "as": "event",
+                                    "in": "$$event.ts",
+                                }
+                            },
+                            0,
+                        ]
+                    },
+                }
+            },
+            {
+                "$project": {
+                    "events": 1,
+                    "page_view_at": 1,
+                    "benchmark_change_at": 1,
+                    "filter_change_at": {
+                        "$arrayElemAt": [
+                            {
+                                "$map": {
+                                    "input": {
+                                        "$filter": {
+                                            "input": "$events",
+                                            "as": "event",
+                                            "cond": {
+                                                "$and": [
+                                                    {
+                                                        "$regexMatch": {
+                                                            "input": "$$event.name",
+                                                            "regex": "^filter_change_",
+                                                        }
+                                                    },
+                                                    {
+                                                        "$gte": [
+                                                            "$$event.ts",
+                                                            "$benchmark_change_at",
+                                                        ]
+                                                    },
+                                                ]
+                                            },
+                                        }
+                                    },
+                                    "as": "event",
+                                    "in": "$$event.ts",
+                                }
+                            },
+                            0,
+                        ]
+                    },
+                }
+            },
             {
                 "$project": {
-                    "has_page_view": {"$in": ["page_view", "$events"]},
-                    "has_benchmark_change": {"$in": ["benchmark_change", "$events"]},
-                    "has_filter_change": {
-                        "$gt": [
+                    "page_view_at": 1,
+                    "benchmark_change_at": 1,
+                    "filter_change_at": 1,
+                    "table_download_at": {
+                        "$arrayElemAt": [
                             {
-                                "$size": {
-                                    "$filter": {
-                                        "input": "$events",
-                                        "as": "e",
-                                        "cond": {"$regexMatch": {"input": "$$e", "regex": "^filter_change_"}},
-                                    }
+                                "$map": {
+                                    "input": {
+                                        "$filter": {
+                                            "input": "$events",
+                                            "as": "event",
+                                            "cond": {
+                                                "$and": [
+                                                    {"$eq": ["$$event.name", "table_download"]},
+                                                    {"$gte": ["$$event.ts", "$filter_change_at"]},
+                                                ]
+                                            },
+                                        }
+                                    },
+                                    "as": "event",
+                                    "in": "$$event.ts",
                                 }
                             },
                             0,
                         ]
                     },
-                    "has_table_download": {"$in": ["table_download", "$events"]},
                 }
             },
             {
                 "$group": {
                     "_id": None,
-                    "step1_page_view": {"$sum": {"$cond": ["$has_page_view", 1, 0]}},
+                    "step1_page_view": {
+                        "$sum": {"$cond": [{"$ne": ["$page_view_at", None]}, 1, 0]}
+                    },
                     "step2_benchmark_change": {
-                        "$sum": {"$cond": [{"$and": ["$has_page_view", "$has_benchmark_change"]}, 1, 0]}
+                        "$sum": {
+                            "$cond": [
+                                {
+                                    "$and": [
+                                        {"$ne": ["$page_view_at", None]},
+                                        {"$gte": ["$benchmark_change_at", "$page_view_at"]},
+                                    ]
+                                },
+                                1,
+                                0,
+                            ]
+                        }
                     },
                     "step3_filter_change": {
                         "$sum": {
                             "$cond": [
-                                {"$and": ["$has_page_view", "$has_benchmark_change", "$has_filter_change"]},
+                                {
+                                    "$and": [
+                                        {"$ne": ["$page_view_at", None]},
+                                        {"$gte": ["$benchmark_change_at", "$page_view_at"]},
+                                        {"$gte": ["$filter_change_at", "$benchmark_change_at"]},
+                                    ]
+                                },
                                 1,
                                 0,
                             ]
@@ -146,10 +346,10 @@ class AnalyticsRepository:
                             "$cond": [
                                 {
                                     "$and": [
-                                        "$has_page_view",
-                                        "$has_benchmark_change",
-                                        "$has_filter_change",
-                                        "$has_table_download",
+                                        {"$ne": ["$page_view_at", None]},
+                                        {"$gte": ["$benchmark_change_at", "$page_view_at"]},
+                                        {"$gte": ["$filter_change_at", "$benchmark_change_at"]},
+                                        {"$gte": ["$table_download_at", "$filter_change_at"]},
                                     ]
                                 },
                                 1,
@@ -174,10 +374,9 @@ class AnalyticsRepository:
     def visitors_new_vs_returning(self, filters: QueryFilters) -> list[dict]:
         period_expr = _period_expression(filters.granularity)
         pipeline: list[dict] = [
-            {"$addFields": {"ts": {"$toDate": "$timestamp"}, "visitor_id": "$properties.visitor_id"}},
+            _with_normalized_time(),
             {
                 "$match": {
-                    **_with_time_and_optional_benchmark(filters),
                     "event_name": "page_view",
                     "visitor_id": {"$nin": [None, ""]},
                 }
@@ -185,31 +384,63 @@ class AnalyticsRepository:
             {
                 "$setWindowFields": {
                     "partitionBy": "$visitor_id",
-                    "sortBy": {"ts": 1},
-                    "output": {"first_seen": {"$first": "$ts"}},
+                    "sortBy": {"event_ts": 1},
+                    "output": {"first_seen": {"$first": "$event_ts"}},
                 }
             },
+            {"$match": _with_time_and_optional_benchmark(filters)},
             {
                 "$project": {
                     "period": period_expr,
-                    "is_new": {"$eq": [{"$dateToString": {"format": "%Y-%m-%d", "date": "$ts"}}, {"$dateToString": {"format": "%Y-%m-%d", "date": "$first_seen"}}]},
+                    "is_new": {
+                        "$eq": [
+                            {"$dateToString": {"format": "%Y-%m-%d", "date": "$event_ts"}},
+                            {"$dateToString": {"format": "%Y-%m-%d", "date": "$first_seen"}},
+                        ]
+                    },
                     "visitor_id": 1,
                 }
             },
-            {"$group": {"_id": {"period": "$period", "is_new": "$is_new"}, "visitors": {"$addToSet": "$visitor_id"}}},
+            {
+                "$group": {
+                    "_id": {"period": "$period", "is_new": "$is_new"},
+                    "visitors": {"$addToSet": "$visitor_id"},
+                }
+            },
             {
                 "$project": {
                     "_id": 0,
                     "period": "$_id.period",
                     "is_new": "$_id.is_new",
-                    "visitor_count": {"$size": "$visitors"},
+                    "visitor_count": _non_empty_set_size("visitors", "v"),
                 }
             },
             {"$sort": {"period": 1, "is_new": -1}},
         ]
         return list(self.events_collection.aggregate(pipeline))
 
+    def available_benchmarks(
+        self, filters: QueryFilters | None = None, limit: int = 100
+    ) -> list[str]:
+        pipeline: list[dict] = []
+        if filters is not None:
+            pipeline.extend(
+                [
+                    {"$match": _indexed_time_prefilter(filters)},
+                    _with_normalized_time(),
+                    {"$match": _with_time_and_optional_benchmark(filters)},
+                ]
+            )
+        pipeline.extend(
+            [
+                {"$match": {"benchmark": {"$nin": [None, ""]}}},
+                {"$group": {"_id": "$benchmark"}},
+                {"$sort": {"_id": 1}},
+                {"$limit": limit},
+            ]
+        )
+        return [row["_id"] for row in self.events_collection.aggregate(pipeline)]
+
     @staticmethod
     def safe_first(items: Iterable[dict]) -> dict:
         return next(iter(items), {})
-

src/leaderboard_analytics/schemas.py

@@ -1,7 +1,7 @@
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from enum import StrEnum
 
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
 
 
 class Granularity(StrEnum):
@@ -12,9 +12,16 @@ class Granularity(StrEnum):
 
 class QueryFilters(BaseModel):
     start_time: datetime = Field(
-        default_factory=lambda: datetime.now(tz=timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
+        default_factory=lambda: datetime.now(tz=UTC).replace(
+            hour=0, minute=0, second=0, microsecond=0
+        )
     )
-    end_time: datetime = Field(default_factory=lambda: datetime.now(tz=timezone.utc))
+    end_time: datetime = Field(default_factory=lambda: datetime.now(tz=UTC))
     benchmark: str | None = None
     granularity: Granularity = Granularity.DAY
 
+    @model_validator(mode="after")
+    def validate_time_range(self) -> "QueryFilters":
+        if self.start_time > self.end_time:
+            raise ValueError("start_time must be earlier than or equal to end_time")
+        return self

src/leaderboard_analytics/services.py

@@ -11,25 +11,19 @@ class AnalyticsService:
     def get_overview(self, filters: QueryFilters) -> tuple[pd.DataFrame, dict]:
         rows = self.repository.overview_timeseries(filters)
         frame = pd.DataFrame(rows)
-        if frame.empty:
-            empty = {
-                "pv": 0,
-                "uv": 0,
-                "sessions": 0,
-                "events": 0,
-                "events_per_session": 0.0,
-                "sessions_per_visitor": 0.0,
-            }
-            return frame, empty
-
+        raw_totals = self.repository.overview_totals(filters)
         totals = {
-            "pv": int(frame["pv"].sum()),
-            "uv": int(frame["uv"].sum()),
-            "sessions": int(frame["session_count"].sum()),
-            "events": int(frame["event_count"].sum()),
+            "pv": int(raw_totals.get("pv", 0)),
+            "uv": int(raw_totals.get("uv", 0)),
+            "sessions": int(raw_totals.get("sessions", 0)),
+            "events": int(raw_totals.get("events", 0)),
         }
-        totals["events_per_session"] = round(totals["events"] / totals["sessions"], 2) if totals["sessions"] else 0.0
-        totals["sessions_per_visitor"] = round(totals["sessions"] / totals["uv"], 2) if totals["uv"] else 0.0
+        totals["events_per_session"] = (
+            round(totals["events"] / totals["sessions"], 2) if totals["sessions"] else 0.0
+        )
+        totals["sessions_per_visitor"] = (
+            round(totals["sessions"] / totals["uv"], 2) if totals["uv"] else 0.0
+        )
         return frame, totals
 
     def get_benchmark_top(self, filters: QueryFilters) -> pd.DataFrame:
@@ -60,3 +54,5 @@ class AnalyticsService:
         frame["visitor_type"] = frame["is_new"].map({True: "new", False: "returning"})
         return frame
 
+    def get_available_benchmarks(self, filters: QueryFilters | None = None) -> list[str]:
+        return self.repository.available_benchmarks(filters)

src/leaderboard_analytics/ui.py

@@ -1,8 +1,12 @@
-from datetime import datetime, timedelta, timezone
 import math
+import tempfile
+import zipfile
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
 from typing import Any
 
 import gradio as gr
+import pandas as pd
 import plotly.express as px
 
 from leaderboard_analytics.schemas import Granularity, QueryFilters
@@ -19,7 +23,7 @@ def _to_utc_datetime(value: Any, fallback: datetime) -> datetime:
         if isinstance(value, float) and math.isnan(value):
             return fallback
         # Gradio DateTime may return Unix timestamps as numbers.
-        dt = datetime.fromtimestamp(value, tz=timezone.utc)
+        dt = datetime.fromtimestamp(value, tz=UTC)
     elif isinstance(value, str):
         dt = datetime.fromisoformat(value)
     else:
@@ -27,60 +31,165 @@ def _to_utc_datetime(value: Any, fallback: datetime) -> datetime:
 
     # Gradio DateTime may return naive datetime values in local time.
     if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=timezone.utc)
-    return dt.astimezone(timezone.utc)
+        dt = dt.replace(tzinfo=UTC)
+    return dt.astimezone(UTC)
+
+
+def _empty_plot(title: str):
+    return px.line(title=title)
+
+
+def _query_range_text(filters: QueryFilters) -> str:
+    return f"{filters.start_time.isoformat()} to {filters.end_time.isoformat()}"
+
+
+def _write_csv_archive(tables: dict[str, pd.DataFrame]) -> str | None:
+    if all(table.empty for table in tables.values()):
+        return None
+
+    archive = tempfile.NamedTemporaryFile(
+        prefix="leaderboard-analytics-", suffix=".zip", delete=False
+    )
+    archive.close()
+    with zipfile.ZipFile(archive.name, "w", compression=zipfile.ZIP_DEFLATED) as zip_file:
+        for name, table in tables.items():
+            zip_file.writestr(f"{name}.csv", table.to_csv(index=False))
+    return archive.name
 
 
 def build_dashboard(service: AnalyticsService) -> gr.Blocks:
-    default_end = datetime.now(tz=timezone.utc)
+    default_end = datetime.now(tz=UTC)
     default_start = (default_end - timedelta(days=7)).replace(microsecond=0)
 
+    def load_benchmarks() -> object:
+        try:
+            benchmarks = service.get_available_benchmarks()
+        except Exception:
+            benchmarks = []
+        return gr.update(choices=[""] + benchmarks, value="")
+
     def query(
         start_time: datetime | str | None,
         end_time: datetime | str | None,
         benchmark: str,
         granularity: str,
-    ) -> tuple[str, object, object, object, object, object]:
-        filters = QueryFilters(
-            start_time=_to_utc_datetime(start_time, default_start),
-            end_time=_to_utc_datetime(end_time, default_end),
-            benchmark=benchmark or None,
-            granularity=Granularity(granularity),
-        )
-        overview_df, totals = service.get_overview(filters)
-        benchmark_df = service.get_benchmark_top(filters)
-        filter_df = service.get_filter_distribution(filters)
-        funnel_df = service.get_funnel(filters)
-        visitors_df = service.get_new_vs_returning(filters)
-
-        metrics = (
-            f"PV: {totals['pv']} | UV: {totals['uv']} | Sessions: {totals['sessions']} | "
-            f"Events/Session: {totals['events_per_session']} | Sessions/Visitor: {totals['sessions_per_visitor']}"
-        )
+    ) -> tuple[
+        object,
+        object,
+        object,
+        object,
+        object,
+        object,
+        object,
+        object,
+        object,
+        object,
+        object,
+        object,
+    ]:
+        try:
+            filters = QueryFilters(
+                start_time=_to_utc_datetime(start_time, default_start),
+                end_time=_to_utc_datetime(end_time, default_end),
+                benchmark=benchmark or None,
+                granularity=Granularity(granularity),
+            )
+            overview_df, totals = service.get_overview(filters)
+            benchmark_df = service.get_benchmark_top(filters)
+            filter_df = service.get_filter_distribution(filters)
+            funnel_df = service.get_funnel(filters)
+            visitors_df = service.get_new_vs_returning(filters)
 
-        overview_plot = (
-            px.line(overview_df, x="period", y=["pv", "uv", "session_count"], title="Traffic overview")
-            if not overview_df.empty
-            else px.line(title="Traffic overview (no data)")
-        )
-        benchmark_plot = (
-            px.bar(benchmark_df, x="benchmark", y="count", title="Benchmark Top")
-            if not benchmark_df.empty
-            else px.bar(title="Benchmark Top (no data)")
-        )
-        filter_plot = (
-            px.bar(filter_df, x="event_name", y="count", title="Filter usage")
-            if not filter_df.empty
-            else px.bar(title="Filter usage (no data)")
-        )
-        funnel_plot = px.funnel(funnel_df, x="sessions", y="step", title="Session funnel")
-        visitor_plot = (
-            px.bar(visitors_df, x="period", y="visitor_count", color="visitor_type", barmode="group", title="New vs returning visitors")
-            if not visitors_df.empty
-            else px.bar(title="New vs returning visitors (no data)")
-        )
+            range_text = _query_range_text(filters)
+            if overview_df.empty and benchmark_df.empty and filter_df.empty and visitors_df.empty:
+                metrics = f"No data for {range_text}."
+            else:
+                metrics = (
+                    f"Range: {range_text}  \n"
+                    f"PV: {totals['pv']} | UV: {totals['uv']} | Sessions: {totals['sessions']} | "
+                    f"Events/Session: {totals['events_per_session']} | "
+                    f"Sessions/Visitor: {totals['sessions_per_visitor']}"
+                )
 
-        return metrics, overview_plot, benchmark_plot, filter_plot, funnel_plot, visitor_plot
+            overview_plot = (
+                px.line(
+                    overview_df,
+                    x="period",
+                    y=["pv", "uv", "session_count"],
+                    title="Traffic overview",
+                )
+                if not overview_df.empty
+                else _empty_plot(f"Traffic overview (no data for {range_text})")
+            )
+            benchmark_plot = (
+                px.bar(benchmark_df, x="benchmark", y="count", title="Benchmark Top")
+                if not benchmark_df.empty
+                else px.bar(title=f"Benchmark Top (no data for {range_text})")
+            )
+            filter_plot = (
+                px.bar(filter_df, x="event_name", y="count", title="Filter usage")
+                if not filter_df.empty
+                else px.bar(title=f"Filter usage (no data for {range_text})")
+            )
+            funnel_plot = px.funnel(funnel_df, x="sessions", y="step", title="Session funnel")
+            visitor_plot = (
+                px.bar(
+                    visitors_df,
+                    x="period",
+                    y="visitor_count",
+                    color="visitor_type",
+                    barmode="group",
+                    title="New vs returning visitors",
+                )
+                if not visitors_df.empty
+                else px.bar(title=f"New vs returning visitors (no data for {range_text})")
+            )
+            csv_archive = _write_csv_archive(
+                {
+                    "overview": overview_df,
+                    "benchmarks": benchmark_df,
+                    "filters": filter_df,
+                    "funnel": funnel_df,
+                    "visitors": visitors_df,
+                }
+            )
+
+            return (
+                metrics,
+                overview_plot,
+                benchmark_plot,
+                filter_plot,
+                funnel_plot,
+                visitor_plot,
+                overview_df,
+                benchmark_df,
+                filter_df,
+                funnel_df,
+                visitors_df,
+                csv_archive,
+            )
+        except Exception as exc:
+            message = f"Query failed: {exc}"
+            empty = pd.DataFrame()
+            return (
+                message,
+                _empty_plot(message),
+                px.bar(title=message),
+                px.bar(title=message),
+                px.funnel(
+                    pd.DataFrame({"step": [], "sessions": []}),
+                    x="sessions",
+                    y="step",
+                    title=message,
+                ),
+                px.bar(title=message),
+                empty,
+                empty,
+                empty,
+                empty,
+                empty,
+                None,
+            )
 
     with gr.Blocks() as demo:
         gr.Markdown("# Leaderboard Analytics Dashboard")
@@ -100,7 +209,12 @@ def build_dashboard(service: AnalyticsService) -> gr.Blocks:
                 value=default_end,
                 timezone="UTC",
             )
-            benchmark = gr.Textbox(label="Benchmark (optional)", placeholder="MTEB(eng)")
+            benchmark = gr.Dropdown(
+                label="Benchmark",
+                choices=[""],
+                value="",
+                allow_custom_value=True,
+            )
             granularity = gr.Dropdown(
                 label="Granularity",
                 choices=[Granularity.DAY.value, Granularity.WEEK.value, Granularity.MONTH.value],
@@ -108,7 +222,9 @@ def build_dashboard(service: AnalyticsService) -> gr.Blocks:
             )
             refresh = gr.Button("Refresh", variant="primary")
 
-        metrics_text = gr.Markdown("PV: 0 | UV: 0 | Sessions: 0 | Events/Session: 0 | Sessions/Visitor: 0")
+        metrics_text = gr.Markdown(
+            "PV: 0 | UV: 0 | Sessions: 0 | Events/Session: 0 | Sessions/Visitor: 0"
+        )
 
         with gr.Row():
             overview_plot = gr.Plot(label="Traffic Overview")
@@ -118,31 +234,41 @@ def build_dashboard(service: AnalyticsService) -> gr.Blocks:
             funnel_plot = gr.Plot(label="Funnel")
         visitor_plot = gr.Plot(label="Visitor Segmentation")
 
+        with gr.Accordion("Raw data", open=False):
+            csv_file = gr.File(label="CSV export")
+            overview_table = gr.DataFrame(label="Traffic Overview")
+            benchmark_table = gr.DataFrame(label="Benchmark Analysis")
+            filter_table = gr.DataFrame(label="Filter Behavior")
+            funnel_table = gr.DataFrame(label="Funnel")
+            visitor_table = gr.DataFrame(label="Visitor Segmentation")
+
+        outputs = [
+            metrics_text,
+            overview_plot,
+            benchmark_plot,
+            filter_plot,
+            funnel_plot,
+            visitor_plot,
+            overview_table,
+            benchmark_table,
+            filter_table,
+            funnel_table,
+            visitor_table,
+            csv_file,
+        ]
+
         refresh.click(
             fn=query,
             inputs=[start_time, end_time, benchmark, granularity],
-            outputs=[
-                metrics_text,
-                overview_plot,
-                benchmark_plot,
-                filter_plot,
-                funnel_plot,
-                visitor_plot,
-            ],
+            outputs=outputs,
         )
 
+        demo.load(fn=load_benchmarks, outputs=benchmark)
         demo.load(
             fn=query,
             inputs=[start_time, end_time, benchmark, granularity],
-            outputs=[
-                metrics_text,
-                overview_plot,
-                benchmark_plot,
-                filter_plot,
-                funnel_plot,
-                visitor_plot,
-            ],
+            outputs=outputs,
         )
 
+    Path(tempfile.gettempdir()).mkdir(parents=True, exist_ok=True)
     return demo
-

tests/test_repositories.py

@@ -0,0 +1,74 @@
+from datetime import UTC, datetime
+
+from leaderboard_analytics.repositories import AnalyticsRepository
+from leaderboard_analytics.schemas import QueryFilters
+
+
+class CapturingCollection:
+    def __init__(self, rows: list[dict] | None = None) -> None:
+        self.rows = rows or []
+        self.pipeline: list[dict] | None = None
+
+    def aggregate(self, pipeline: list[dict]):
+        self.pipeline = pipeline
+        return iter(self.rows)
+
+
+def _filters() -> QueryFilters:
+    return QueryFilters(
+        start_time=datetime(2026, 1, 1, tzinfo=UTC),
+        end_time=datetime(2026, 1, 31, tzinfo=UTC),
+    )
+
+
+def test_funnel_pipeline_preserves_ordered_step_logic() -> None:
+    collection = CapturingCollection()
+    repository = AnalyticsRepository(collection)  # type: ignore[arg-type]
+
+    repository.funnel(_filters())
+
+    assert collection.pipeline is not None
+    assert {"$sort": {"session_id": 1, "event_ts": 1}} in collection.pipeline
+    assert any(
+        "$push" in stage.get("$group", {}).get("events", {}) for stage in collection.pipeline
+    )
+    assert not any(
+        "$addToSet" in str(stage) and "events" in str(stage) for stage in collection.pipeline
+    )
+    assert any(
+        "table_download_at" in str(stage) and "$filter_change_at" in str(stage)
+        for stage in collection.pipeline
+    )
+
+
+def test_new_vs_returning_pipeline_computes_first_seen_before_range_match() -> None:
+    collection = CapturingCollection()
+    repository = AnalyticsRepository(collection)  # type: ignore[arg-type]
+
+    repository.visitors_new_vs_returning(_filters())
+
+    assert collection.pipeline is not None
+    window_index = next(
+        i for i, stage in enumerate(collection.pipeline) if "$setWindowFields" in stage
+    )
+    range_match_index = next(
+        i
+        for i, stage in enumerate(collection.pipeline)
+        if stage.get("$match", {}).get("event_ts") is not None
+    )
+    assert window_index < range_match_index
+
+
+def test_overview_totals_filters_empty_identifiers() -> None:
+    collection = CapturingCollection([{"pv": 1, "uv": 1, "sessions": 1, "events": 2}])
+    repository = AnalyticsRepository(collection)  # type: ignore[arg-type]
+
+    totals = repository.overview_totals(_filters())
+
+    assert totals == {"pv": 1, "uv": 1, "sessions": 1, "events": 2}
+    assert collection.pipeline is not None
+    pipeline_text = str(collection.pipeline)
+    assert '"$sessions"' in pipeline_text or "'$sessions'" in pipeline_text
+    assert '"$visitors"' in pipeline_text or "'$visitors'" in pipeline_text
+    assert "$$s" in pipeline_text
+    assert "$$v" in pipeline_text

tests/test_schemas.py

@@ -0,0 +1,16 @@
+from datetime import UTC, datetime
+
+import pytest
+from pydantic import ValidationError
+
+from leaderboard_analytics.schemas import QueryFilters
+
+
+def test_query_filters_rejects_invalid_time_range() -> None:
+    with pytest.raises(
+        ValidationError, match="start_time must be earlier than or equal to end_time"
+    ):
+        QueryFilters(
+            start_time=datetime(2026, 1, 2, tzinfo=UTC),
+            end_time=datetime(2026, 1, 1, tzinfo=UTC),
+        )

tests/test_services.py

@@ -0,0 +1,35 @@
+from datetime import UTC, datetime
+
+from leaderboard_analytics.schemas import QueryFilters
+from leaderboard_analytics.services import AnalyticsService
+
+
+class FakeRepository:
+    def overview_timeseries(self, filters: QueryFilters) -> list[dict]:
+        return [
+            {"period": "2026-01-01", "pv": 2, "uv": 1, "session_count": 1, "event_count": 3},
+            {"period": "2026-01-02", "pv": 1, "uv": 1, "session_count": 1, "event_count": 2},
+        ]
+
+    def overview_totals(self, filters: QueryFilters) -> dict:
+        return {"pv": 3, "uv": 1, "sessions": 1, "events": 5}
+
+
+def test_overview_uses_full_range_distinct_totals() -> None:
+    service = AnalyticsService(FakeRepository())  # type: ignore[arg-type]
+    filters = QueryFilters(
+        start_time=datetime(2026, 1, 1, tzinfo=UTC),
+        end_time=datetime(2026, 1, 2, tzinfo=UTC),
+    )
+
+    frame, totals = service.get_overview(filters)
+
+    assert list(frame["period"]) == ["2026-01-01", "2026-01-02"]
+    assert totals == {
+        "pv": 3,
+        "uv": 1,
+        "sessions": 1,
+        "events": 5,
+        "events_per_session": 5.0,
+        "sessions_per_visitor": 1.0,
+    }