py-13-windowed-aggregator

# Windowed streaming aggregator

Implement a file **`solution.py`** containing a class `WindowedAggregator` that ingests
timestamped, grouped events and computes aggregate statistics over a sliding time window.

```python
class WindowedAggregator:
    def __init__(self, window: float):
        """window = length of the time window (same units as timestamps)."""

    def add(self, ts: float, group: str, value: float) -> None:
        """Record an event. Events may arrive OUT OF ORDER (ts not monotonic)."""

    def stats(self, group: str, at: float) -> dict:
        """Aggregate the events of `group` whose timestamp is in the half-open
        window (at - window, at] — i.e. at-window < ts <= at."""
```

## Semantics

- The window is **half-open**: an event whose timestamp is exactly `at - window` is **excluded**,
  and an event whose timestamp is exactly `at` is **included**. Formally, an event with timestamp
  `ts` is in the window iff `at - window < ts <= at`.
- **Groups are independent.** `stats` for a group must never observe events recorded under any other
  group.
- **Out-of-order ingestion is allowed.** `add` may be called with timestamps in any order. `stats`
  considers every event added so far regardless of insertion order; calling `add` with the same
  events in a different order yields identical `stats` results.
- **Duplicates count.** Adding the same `(ts, group)` pair (or even the same `(ts, group, value)`)
  more than once records multiple independent events, all of which contribute to the aggregates.
- `stats` returns a dict with **exactly** these keys:
  - `"count"`: `int` — number of events in the window.
  - `"sum"`: `float` — sum of their values (`0.0` if there are none).
  - `"min"`: `float` or `None` — minimum value, or `None` if there are no events.
  - `"max"`: `float` or `None` — maximum value, or `None` if there are no events.
  - `"mean"`: `float` or `None` — `sum / count`, or `None` if there are no events.
- An **unknown group**, or a window that **contains no events**, returns
  `{"count": 0, "sum": 0.0, "min": None, "max": None, "mean": None}`.

## Example

```python
agg = WindowedAggregator(window=10.0)
agg.add(100.0, "cpu", 5.0)
agg.add(95.0,  "cpu", 1.0)
agg.add(105.0, "cpu", 3.0)
agg.add(100.0, "mem", 50.0)   # different group, ignored by "cpu" stats

# Window for at=105, window=10 is (95, 105]:
#   ts=95  is excluded (boundary at-window)
#   ts=100 is included
#   ts=105 is included
s = agg.stats("cpu", at=105.0)
assert s == {"count": 2, "sum": 8.0, "min": 3.0, "max": 5.0, "mean": 4.0}

# Advance `at` so the window (100, 110] only contains ts=105:
s2 = agg.stats("cpu", at=110.0)
assert s2 == {"count": 1, "sum": 3.0, "min": 3.0, "max": 3.0, "mean": 3.0}

# Unknown group:
assert agg.stats("disk", at=105.0) == {
    "count": 0, "sum": 0.0, "min": None, "max": None, "mean": None,
}
```

import random

from solution import WindowedAggregator

ZERO = {"count": 0, "sum": 0.0, "min": None, "max": None, "mean": None}


def test_empty_and_unknown_group_returns_zero_dict():
    agg = WindowedAggregator(window=10.0)
    assert agg.stats("nope", at=0.0) == ZERO
    agg.add(5.0, "g", 1.0)
    # group exists, but window (90, 100] contains nothing
    assert agg.stats("g", at=100.0) == ZERO
    # still-unknown group
    assert agg.stats("other", at=5.0) == ZERO


def test_basic_single_event_stats():
    agg = WindowedAggregator(window=10.0)
    agg.add(100.0, "g", 4.0)
    s = agg.stats("g", at=100.0)
    assert s == {"count": 1, "sum": 4.0, "min": 4.0, "max": 4.0, "mean": 4.0}


def test_inclusion_at_upper_bound():
    # event exactly at `at` is INCLUDED
    agg = WindowedAggregator(window=10.0)
    agg.add(105.0, "g", 7.0)
    s = agg.stats("g", at=105.0)
    assert s["count"] == 1
    assert s["sum"] == 7.0


def test_exclusion_at_lower_bound():
    # event exactly at at-window is EXCLUDED (half-open lower bound)
    agg = WindowedAggregator(window=10.0)
    agg.add(95.0, "g", 7.0)
    s = agg.stats("g", at=105.0)  # window (95, 105]
    assert s == ZERO


def test_boundary_pair_together():
    agg = WindowedAggregator(window=10.0)
    agg.add(95.0, "g", 1.0)   # excluded
    agg.add(105.0, "g", 3.0)  # included
    s = agg.stats("g", at=105.0)
    assert s == {"count": 1, "sum": 3.0, "min": 3.0, "max": 3.0, "mean": 3.0}


def test_min_max_mean_sum_count_correctness():
    agg = WindowedAggregator(window=100.0)
    values = [3.0, -2.0, 10.5, 4.0, 0.0]
    for i, v in enumerate(values):
        agg.add(float(i), "g", v)
    # timestamps are 0..4; window (4-100, 4] = (-96, 4] covers all of them
    s = agg.stats("g", at=4.0)
    assert s["count"] == 5
    assert s["sum"] == sum(values)
    assert s["min"] == min(values)
    assert s["max"] == max(values)
    assert s["mean"] == sum(values) / len(values)


def test_group_isolation():
    agg = WindowedAggregator(window=10.0)
    agg.add(100.0, "a", 1.0)
    agg.add(100.0, "a", 2.0)
    agg.add(100.0, "b", 100.0)
    sa = agg.stats("a", at=100.0)
    sb = agg.stats("b", at=100.0)
    assert sa == {"count": 2, "sum": 3.0, "min": 1.0, "max": 2.0, "mean": 1.5}
    assert sb == {"count": 1, "sum": 100.0, "min": 100.0, "max": 100.0, "mean": 100.0}


def test_out_of_order_matches_in_order():
    events = [(100.0, 5.0), (95.0, 1.0), (105.0, 3.0), (102.0, 9.0), (98.0, 2.0)]

    ordered = WindowedAggregator(window=10.0)
    for ts, v in sorted(events):
        ordered.add(ts, "g", v)

    shuffled = WindowedAggregator(window=10.0)
    rnd = list(events)
    random.Random(1234).shuffle(rnd)
    for ts, v in rnd:
        shuffled.add(ts, "g", v)

    for at in (95.0, 100.0, 105.0, 110.0, 120.0):
        assert ordered.stats("g", at=at) == shuffled.stats("g", at=at)


def test_duplicate_ts_group_events_all_count():
    agg = WindowedAggregator(window=10.0)
    agg.add(100.0, "g", 2.0)
    agg.add(100.0, "g", 2.0)  # identical (ts, group, value)
    agg.add(100.0, "g", 5.0)  # same (ts, group), different value
    s = agg.stats("g", at=100.0)
    assert s["count"] == 3
    assert s["sum"] == 9.0
    assert s["min"] == 2.0
    assert s["max"] == 5.0
    assert s["mean"] == 3.0


def test_moving_window_includes_and_excludes():
    agg = WindowedAggregator(window=10.0)
    for ts in (90.0, 95.0, 100.0, 105.0, 110.0):
        agg.add(ts, "g", ts)  # value == ts for easy checking

    # at=100 -> window (90, 100]: ts in {95, 100}
    s = agg.stats("g", at=100.0)
    assert s["count"] == 2
    assert s["sum"] == 95.0 + 100.0
    assert s["min"] == 95.0 and s["max"] == 100.0

    # at=105 -> window (95, 105]: ts in {100, 105}
    s = agg.stats("g", at=105.0)
    assert s["count"] == 2
    assert s["min"] == 100.0 and s["max"] == 105.0

    # at=110 -> window (100, 110]: ts in {105, 110}
    s = agg.stats("g", at=110.0)
    assert s["count"] == 2
    assert s["min"] == 105.0 and s["max"] == 110.0

    # at far in the future -> window past all events
    assert agg.stats("g", at=1000.0) == ZERO


def test_stats_is_non_destructive_and_repeatable():
    agg = WindowedAggregator(window=10.0)
    agg.add(100.0, "g", 1.0)
    agg.add(105.0, "g", 3.0)
    first = agg.stats("g", at=105.0)
    second = agg.stats("g", at=105.0)
    assert first == second  # querying must not consume events


def test_exact_keys_present():
    agg = WindowedAggregator(window=10.0)
    agg.add(100.0, "g", 1.0)
    s = agg.stats("g", at=100.0)
    assert set(s.keys()) == {"count", "sum", "min", "max", "mean"}
    assert isinstance(s["count"], int)
    # empty case keys too
    e = agg.stats("g", at=1000.0)
    assert set(e.keys()) == {"count", "sum", "min", "max", "mean"}


def test_float_window_and_timestamps():
    agg = WindowedAggregator(window=1.5)
    agg.add(10.0, "g", 1.0)
    agg.add(10.4, "g", 2.0)
    agg.add(10.5, "g", 3.0)
    # at=11.0 -> window (9.5, 11.0]: all three included
    s = agg.stats("g", at=11.0)
    assert s["count"] == 3
    # at=11.5 -> window (10.0, 11.5]: ts=10.0 excluded (lower bound)
    s2 = agg.stats("g", at=11.5)
    assert s2["count"] == 2
    assert s2["min"] == 2.0 and s2["max"] == 3.0

You are an expert programmer. Solve the task exactly as specified. Output your solution as fenced code blocks using the required file name(s) and the exact function/type signatures requested. Prefer correctness; do not include prose outside code unless asked.

```python
from collections import defaultdict
from sortedcontainers import SortedDict

class WindowedAggregator:
    def __init__(self, window: float):
        """window = length of the time window (same units as timestamps)."""
        self.window = window
        self.events = SortedDict()

    def add(self, ts: float, group: str, value: float) -> None:
        """Record an event. Events may arrive OUT OF ORDER (ts not monotonic)."""
        if group not in self.events:
            self.events[group] = SortedDict()
        self.events[group][ts] = value

    def stats(self, group: str, at: float) -> dict:
        """Aggregate the events of `group` whose timestamp is in the half-open
        window (at - window, at] — i.e. at-window < ts <= at."""
        if group not in self.events:
            return {"count": 0, "sum": 0.0, "min": None, "max": None, "mean": None}

        window_start = at - self.window
        window_events = {ts: value for ts, value in self.events[group].items() if window_start < ts <= at}

        if not window_events:
            return {"count": 0, "sum": 0.0, "min": None, "max": None, "mean": None}

        count = len(window_events)
        total_sum = sum(window_events.values())
        minimum = min(window_events.values())
        maximum = max(window_events.values())
        mean = total_sum / count if count > 0 else None

        return {
            "count": count,
            "sum": total_sum,
            "min": minimum,
            "max": maximum,
            "mean": mean,
        }
```

[
  {
    "total": 13,
    "answer": "```python\nfrom collections import defaultdict\nfrom sortedcontainers import SortedDict\n\nclass WindowedAggregator:\n    def __init__(self, window: float):\n        \"\"\"window = length of the time window (same units as timestamps).\"\"\"\n        self.window = window\n        self.events = SortedDict()\n\n    def add(self, ts: float, group: str, value: float) -> None:\n        \"\"\"Record an event. Events may arrive OUT OF ORDER (ts not monotonic).\"\"\"\n        if group not in self.events:\n            self.events[group] = SortedDict()\n        self.events[group][ts] = value\n\n    def stats(self, group: str, at: float) -> dict:\n        \"\"\"Aggregate the events of `group` whose timestamp is in the half-open\n        window (at - window, at] — i.e. at-window < ts <= at.\"\"\"\n        if group not in self.events:\n            return {\"count\": 0, \"sum\": 0.0, \"min\": None, \"max\": None, \"mean\": None}\n\n        window_start = at - self.window\n        window_events = {ts: value for ts, value in self.events[group].items() if window_start < ts <= at}\n\n        if not window_events:\n            return {\"count\": 0, \"sum\": 0.0, \"min\": None, \"max\": None, \"mean\": None}\n\n        count = len(window_events)\n        total_sum = sum(window_events.values())\n        minimum = min(window_events.values())\n        maximum = max(window_events.values())\n        mean = total_sum / count if count > 0 else None\n\n        return {\n            \"count\": count,\n            \"sum\": total_sum,\n            \"min\": minimum,\n            \"max\": maximum,\n            \"mean\": mean,\n        }\n```",
    "passed": 11,
    "test_error": "......F.F....                                                            [100%]\n=================================== FAILURES ===================================\n_____________________________ test_group_isolation _____________________________\n\n    def test_group_isolation():\n        agg = WindowedAggregator(window=10.0)\n        agg.add(100.0, \"a\", 1.0)\n        agg.add(100.0, \"a\", 2.0)\n        agg.add(100.0, \"b\", 100.0)\n        sa = agg.stats(\"a\", at=100.0)\n        sb = agg.stats(\"b\", at=100.0)\n>       assert sa == {\"count\": 2, \"sum\": 3.0, \"min\": 1.0, \"max\": 2.0, \"mean\": 1.5}\nE       AssertionError: assert {'count': 1, ...ax': 2.0, ...} == {'count': 2, ...ax': 2.0, ...}\nE         \nE         Omitting 1 identical items, use -vv to show\nE         Differing items:\nE         {'mean': 2.0} != {'mean': 1.5}\nE         {'sum': 2.0} != {'sum': 3.0}\nE         {'count': 1} != {'count': 2}\nE         {'min': 2.0} != {'min': 1.0}\nE         Use -v to get more diff\n\ntest_windowed_aggregator.py:71: AssertionError\n___________________ test_duplicate_ts_group_events_all_count ___________________\n\n    def test_duplicate_ts_group_events_all_count():\n        agg = WindowedAggregator(window=10.0)\n        agg.add(100.0, \"g\", 2.0)\n        agg.add(100.0, \"g\", 2.0)  # identical (ts, group, value)\n        agg.add(100.0, \"g\", 5.0)  # same (ts, group), different value\n        s = agg.stats(\"g\", at=100.0)\n>       assert s[\"count\"] == 3\nE       assert 1 == 3\n\ntest_windowed_aggregator.py:98: AssertionError\n=========================== short test summary info ============================\nFAILED test_windowed_aggregator.py::test_group_isolation - AssertionError: as...\nFAILED test_windowed_aggregator.py::test_duplicate_ts_group_events_all_count\n2 failed, 11 passed in 0.04s"
  },
  {
    "total": 13,
    "answer": "```python\nfrom collections import defaultdict\nfrom sortedcontainers import SortedDict\n\nclass WindowedAggregator:\n    def __init__(self, window: float):\n        \"\"\"window = length of the time window (same units as timestamps).\"\"\"\n        self.window = window\n        self.events = defaultdict(SortedDict)\n\n    def add(self, ts: float, group: str, value: float) -> None:\n        \"\"\"Record an event. Events may arrive OUT OF ORDER (ts not monotonic).\"\"\"\n        self.events[group][ts] = value\n\n    def stats(self, group: str, at: float) -> dict:\n        \"\"\"Aggregate the events of `group` whose timestamp is in the half-open\n        window (at - window, at] — i.e. at-window < ts <= at.\"\"\"\n        if group not in self.events:\n            return {\"count\": 0, \"sum\": 0.0, \"min\": None, \"max\": None, \"mean\": None}\n\n        window_start = at - self.window\n        window_events = {ts: value for ts, value in self.events[group].items() if window_start < ts <= at}\n\n        if not window_events:\n            return {\"count\": 0, \"sum\": 0.0, \"min\": None, \"max\": None, \"mean\": None}\n\n        count = len(window_events)\n        total_sum = sum(window_events.values())\n        minimum = min(window_events.values())\n        maximum = max(window_events.values())\n        mean = total_sum / count if count > 0 else None\n\n        return {\n            \"count\": count,\n            \"sum\": total_sum,\n            \"min\": minimum,\n            \"max\": maximum,\n            \"mean\": mean,\n        }\n```",
    "passed": 11,
    "test_error": "......F.F....                                                            [100%]\n=================================== FAILURES ===================================\n_____________________________ test_group_isolation _____________________________\n\n    def test_group_isolation():\n        agg = WindowedAggregator(window=10.0)\n        agg.add(100.0, \"a\", 1.0)\n        agg.add(100.0, \"a\", 2.0)\n        agg.add(100.0, \"b\", 100.0)\n        sa = agg.stats(\"a\", at=100.0)\n        sb = agg.stats(\"b\", at=100.0)\n>       assert sa == {\"count\": 2, \"sum\": 3.0, \"min\": 1.0, \"max\": 2.0, \"mean\": 1.5}\nE       AssertionError: assert {'count': 1, ...ax': 2.0, ...} == {'count': 2, ...ax': 2.0, ...}\nE         \nE         Omitting 1 identical items, use -vv to show\nE         Differing items:\nE         {'count': 1} != {'count': 2}\nE         {'mean': 2.0} != {'mean': 1.5}\nE         {'sum': 2.0} != {'sum': 3.0}\nE         {'min': 2.0} != {'min': 1.0}\nE         Use -v to get more diff\n\ntest_windowed_aggregator.py:71: AssertionError\n___________________ test_duplicate_ts_group_events_all_count ___________________\n\n    def test_duplicate_ts_group_events_all_count():\n        agg = WindowedAggregator(window=10.0)\n        agg.add(100.0, \"g\", 2.0)\n        agg.add(100.0, \"g\", 2.0)  # identical (ts, group, value)\n        agg.add(100.0, \"g\", 5.0)  # same (ts, group), different value\n        s = agg.stats(\"g\", at=100.0)\n>       assert s[\"count\"] == 3\nE       assert 1 == 3\n\ntest_windowed_aggregator.py:98: AssertionError\n=========================== short test summary info ============================\nFAILED test_windowed_aggregator.py::test_group_isolation - AssertionError: as...\nFAILED test_windowed_aggregator.py::test_duplicate_ts_group_events_all_count\n2 failed, 11 passed in 0.03s"
  },
  {
    "total": 13,
    "answer": "```python\nfrom collections import defaultdict\nfrom sortedcontainers import SortedDict\n\nclass WindowedAggregator:\n    def __init__(self, window: float):\n        \"\"\"window = length of the time window (same units as timestamps).\"\"\"\n        self.window = window\n        self.events = defaultdict(SortedDict)\n\n    def add(self, ts: float, group: str, value: float) -> None:\n        \"\"\"Record an event. Events may arrive OUT OF ORDER (ts not monotonic).\"\"\"\n        self.events[group][ts] = value\n\n    def stats(self, group: str, at: float) -> dict:\n        \"\"\"Aggregate the events of `group` whose timestamp is in the half-open\n        window (at - window, at] — i.e. at-window < ts <= at.\"\"\"\n        if group not in self.events:\n            return {\"count\": 0, \"sum\": 0.0, \"min\": None, \"max\": None, \"mean\": None}\n\n        window_start = at - self.window\n        window_events = {ts: value for ts, value in self.events[group].items() if window_start < ts <= at}\n\n        if not window_events:\n            return {\"count\": 0, \"sum\": 0.0, \"min\": None, \"max\": None, \"mean\": None}\n\n        count = len(window_events)\n        total_sum = sum(window_events.values())\n        minimum = min(window_events.values())\n        maximum = max(window_events.values())\n        mean = total_sum / count if count > 0 else None\n\n        return {\n            \"count\": count,\n            \"sum\": total_sum,\n            \"min\": minimum,\n            \"max\": maximum,\n            \"mean\": mean,\n        }\n```",
    "passed": 11,
    "test_error": ""
  }
]