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
import bisect

class WindowedAggregator:
    def __init__(self, window: float):
        """window = length of the time window (same units as timestamps)."""
        self.window = window
        # Store events per group as parallel lists of timestamps and values, kept sorted by timestamp.
        self.events = {}

    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] = {"timestamps": [], "values": []}
        
        timestamps = self.events[group]["timestamps"]
        values = self.events[group]["values"]
        
        # Insert in sorted order by timestamp to enable efficient range queries
        idx = bisect.bisect_left(timestamps, ts)
        timestamps.insert(idx, ts)
        values.insert(idx, 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."""
        empty_result = {
            "count": 0,
            "sum": 0.0,
            "min": None,
            "max": None,
            "mean": None
        }
        
        if group not in self.events or not self.events[group]["timestamps"]:
            return empty_result
        
        timestamps = self.events[group]["timestamps"]
        values = self.events[group]["values"]
        
        start = at - self.window
        end = at
        
        # Find range of events where start < ts <= end
        # bisect_right returns the first index where ts > x
        left_idx = bisect.bisect_right(timestamps, start)
        right_idx = bisect.bisect_right(timestamps, end)
        
        window_values = values[left_idx:right_idx]
        
        if not window_values:
            return empty_result
        
        count = len(window_values)
        total = sum(window_values)
        min_val = min(window_values)
        max_val = max(window_values)
        mean_val = total / count
        
        return {
            "count": count,
            "sum": total,
            "min": min_val,
            "max": max_val,
            "mean": mean_val
        }
```

The user wants a Python class `WindowedAggregator` that:
1. Stores timestamped, grouped events.
2. Computes aggregate statistics over a sliding time window.
3. Handles out-of-order events.
4. Uses a half-open window `(at - window, at]`.
5. Returns specific statistics: count, sum, min, max, mean.

Let's design the data structure:
- We need to store events by group.
- For each group, we need to efficiently query events within a time range.
- Since events can be out of order, we can't just maintain a simple queue.
- A list of events per group is simple but querying would be O(n).
- A sorted list or a balanced BST would allow O(log n) or O(k) queries where k is the number of events in the window.
- Given Python's standard library, we can use `bisect` on a sorted list of timestamps.

Let's refine the approach:
- Store events in a dictionary: `self.events = {group: [(ts, value), ...]}`.
- Keep each group's events sorted by timestamp. When adding an event, insert it in sorted order.
- For `stats(group, at)`:
  - If group not in `self.events`, return empty stats.
  - Use `bisect` to find the range of events where `at - window < ts <= at`.
  - Compute aggregates from the events in that range.

Wait, if we insert in sorted order, insertion is O(n) due to list shifting. But for typical use cases, this might be acceptable. If performance is critical, we might need a more complex structure, but the problem doesn't specify performance constraints. Let's assume correctness and simplicity are prioritized.

Actually, we can just append and sort when needed, or maintain sorted order. Maintaining sorted order with `bisect.insort` is straightforward.

Let's check the constraints:
- `window` is fixed at initialization.
- `ts`, `value` are floats.
- `group` is a string.

Implementation details:
- `__init__(self, window: float)`: Store `window`. Initialize `self.events = {}`.
- `add(self, ts: float, group: str, value: float)`:
  - If `group` not in `self.events`, create an empty list.
  - Insert `(ts, value)` into `self.events[group]` maintaining sorted order by `ts`.
- `stats(self, group: str, at: float)`:
  - If `group` not in `self.events` or list is empty, return empty stats.
  - Define window bounds: `start = at - window`, `end = at`.
  - We need events where `start < ts <= end`.
  - Use `bisect_right` to find the index of the first event with `ts > end`. Let's call it `right_idx`.
  - Use `bisect_right` to find the index of the first event with `ts > start`. Let's call it `left_idx`.
    - Wait, `bisect_right` on `start` gives the first index where `ts > start` if all `ts` are distinct? No.
    - `bisect_right(a, x)` returns the insertion point after any existing entries of `x`. So all elements to the left are `<= x`.
    - We want `ts > start`. So `bisect_right` on `start` gives the first index where `ts > start` is NOT guaranteed. It gives the index after all `ts <= start`. So elements from that index onwards are `> start`. Yes, that's correct.
    - Let's verify: `a = [1, 2, 3]`, `start = 1`. `bisect_right(a, 1)` returns 1. `a[1:]` is `[2, 3]`, which are `> 1`. Correct.
    - `a = [1, 2, 3]`, `start = 1.5`. `bisect_right(a, 1.5)` returns 1. `a[1:]` is `[2, 3]`, which are `> 1.5`. Correct.
    - So `left_idx = bisect_right(timestamps, start)`.
  - For `end`, we want `ts <= end`. `bisect_right` on `end` gives the first index where `ts > end`. So elements before that index are `<= end`.
    - `right_idx = bisect_right(timestamps, end)`.
  - The slice `self.events[group][left_idx:right_idx]` contains all events in the window.
  - Compute aggregates from this slice.

Wait, `bisect` works on a list of comparable items. Our list contains tuples `(ts, value)`. `bisect` will compare tuples lexicographically.
- When searching for `start` (a float), comparing `(ts, value)` with `start` will fail because tuple vs float comparison is not allowed in Python 3.
- We need to separate timestamps and values, or use a key. `bisect` doesn't support keys directly.
- Option 1: Store two parallel lists per group: `timestamps` and `values`. Keep them in sync.
- Option 2: Store tuples, but use a custom search or extract timestamps for bisect.
- Option 3: Store tuples, and use `bisect` with a wrapper or just extract timestamps into a separate list for bisecting.

Let's go with Option 1: Two lists per group.
`self.events = {group: {"timestamps": [], "values": []}}`
When adding:
- Use `bisect.insort` on `timestamps` to find the index.
- Insert `value` at the same index in `values`.

This keeps things simple and efficient enough.

Let's refine the `stats` method:
- Get `timestamps` and `values` for the group.
- `left_idx = bisect.bisect_right(timestamps, at - window)`
- `right_idx = bisect.bisect_right(timestamps, at)`
- Slice `window_values = values[left_idx:right_idx]`
- If `window_values` is empty, return empty stats.
- Else, compute count, sum, min, max, mean.

Edge cases:
- Empty group: handled.
- No events in window: handled.
- Out of order: handled by sorted insertion.
- Duplicates: handled (multiple entries).
- Half-open window: `at - window < ts <= at`.
  - `bisect_right(timestamps, at - window)` gives index after all `ts <= at - window`. So `timestamps[left_idx]` is the first `ts > at - window`. Correct.
  - `bisect_right(timestamps, at)` gives index after all `ts <= at`. So `timestamps[right_idx - 1]` is the last `ts <= at`. Correct.
  - Slice `[left_idx:right_idx]` includes all `ts` where `at - window < ts <= at`. Correct.

Let's write the code.

```python
import bisect

class WindowedAggregator:
    def __init__(self, window: float):
        self.window = window
        self.events = {}  # group -> {"timestamps": [], "values": []}

    def add(self, ts: float, group: str, value: float) -> None:
        if group not in self.events:
            self.events[group] = {"timestamps": [], "values": []}
        
        timestamps = self.events[group]["timestamps"]
        values = self.events[group]["values"]
        
        idx = bisect.bisect_left(timestamps, ts)
        timestamps.insert(idx, ts)
        values.insert(idx, value)

    def stats(self, group: str,
…[5907 chars elided]…
: 0,
            "sum": 0.0,
            "min": None,
            "max": None,
            "mean": None
        }
        
        if group not in self.events or not self.events[group]["timestamps"]:
            return empty_result
        
        timestamps = self.events[group]["timestamps"]
        values = self.events[group]["values"]
        
        start = at - self.window
        end = at
        
        # Find range of events where start < ts <= end
        # bisect_right returns the first index where ts > x
        left_idx = bisect.bisect_right(timestamps, start)
        right_idx = bisect.bisect_right(timestamps, end)
        
        window_values = values[left_idx:right_idx]
        
        if not window_values:
            return empty_result
        
        count = len(window_values)
        total = sum(window_values)
        min_val = min(window_values)
        max_val = max(window_values)
        mean_val = total / count
        
        return {
            "count": count,
            "sum": total,
            "min": min_val,
            "max": max_val,
            "mean": mean_val
        }
```

Wait, should I use `bisect_left` or `bisect_right` for insertion?
If I use `bisect_left`, I insert before equal elements.
If I use `bisect_right`, I insert after.
It doesn't affect correctness of the window query because the window query uses `bisect_right` on both ends, which correctly handles the boundaries regardless of insertion order for equal timestamps.
Let's stick with `bisect_left` for insertion.

One small optimization: we don't need `from typing import ...` if we just use built-in types in the signature as given in the prompt. The prompt uses `float`, `str`, `dict`, `None`. I'll keep it simple.

Final check of the prompt requirements:
- File name: `solution.py`
- Class name: `WindowedAggregator`
- Methods: `__init__`, `add`, `stats`
- Signatures match.
- Semantics match.

Looks good.

One thing: `bisect` module is standard library, so it's fine to import.

Let's produce the final output.