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
class WindowedAggregator:
    def __init__(self, window: float):
        self.window = window
        self.events = {}

    def add(self, ts: float, group: str, value: float) -> None:
        if group not in self.events:
            self.events[group] = []
        self.events[group].append((ts, value))

    def stats(self, group: str, at: float) -> dict:
        events = self.events.get(group)
        if not events:
            return {"count": 0, "sum": 0.0, "min": None, "max": None, "mean": None}
        
        lower = at - self.window
        values = [v for ts, v in events if lower < ts <= at]
        
        if not values:
            return {"count": 0, "sum": 0.0, "min": None, "max": None, "mean": None}
            
        count = len(values)
        total = sum(values)
        return {
            "count": count,
            "sum": total,
            "min": min(values),
            "max": max(values),
            "mean": total / count
        }
```

The user wants a Python class `WindowedAggregator` that:
1. Takes a `window` size in the constructor.
2. Has an `add(ts, group, value)` method to record events.
3. Has a `stats(group, at)` method to compute aggregates over the window `(at - window, at]`.
4. Handles out-of-order events.
5. Groups are independent.
6. Duplicates count.
7. Returns a specific dictionary format.

Let's design the data structure.
Since events can arrive out of order and we need to query by time window for specific groups, storing events per group in a list or sorted structure makes sense.
Given that `stats` is called with an `at` time, we need to filter events where `at - window < ts <= at`.
If we store events per group as a list of `(ts, value)` tuples, we can filter them on each `stats` call.
Is performance a concern? The problem doesn't specify constraints, but typically for such problems, a simple list per group is fine unless there are millions of events. We can optimize later if needed, but a straightforward approach is usually expected.
Wait, we can store events in a list per group. Since `stats` might be called frequently, sorting the list per group by timestamp could help, but filtering is O(N) per group anyway. Given no constraints, O(N) per `stats` call is acceptable.

Let's stick to a dictionary mapping `group` to a list of `(ts, value)` tuples.
`self.events = defaultdict(list)`
`add`: `self.events[group].append((ts, value))`
`stats`:
  Get list for group. If empty or group not present, return default dict.
  Filter: `[v for ts, v in events if at - window < ts <= at]`
  Compute count, sum, min, max, mean.
  Return dict.

Wait, the window condition is `at - window < ts <= at`.
Let's verify with the example:
`window = 10.0`
`at = 105.0` -> window is `(95.0, 105.0]`
Events for "cpu": `(100.0, 5.0)`, `(95.0, 1.0)`, `(105.0, 3.0)`
`95.0` is excluded because `95.0 <= 95.0` is false for `>`.
`100.0` is included.
`105.0` is included.
Matches example.

Edge cases:
- Empty group or unknown group: returns default dict.
- `count == 0`: returns `min: None, max: None, mean: None`.
- Duplicates: handled naturally by list.
- Out of order: handled naturally by list.

Implementation details:
```python
from collections import defaultdict

class WindowedAggregator:
    def __init__(self, window: float):
        self.window = window
        self.events = defaultdict(list)

    def add(self, ts: float, group: str, value: float) -> None:
        self.events[group].append((ts, value))

    def stats(self, group: str, at: float) -> dict:
        events = self.events.get(group)
        if not events:
            return {"count": 0, "sum": 0.0, "min": None, "max": None, "mean": None}
        
        window_start = at - self.window
        values = [v for ts, v in events if window_start < ts <= at]
        
        if not values:
            return {"count": 0, "sum": 0.0, "min": None, "max": None, "mean": None}
            
        count = len(values)
        total = sum(values)
        return {
            "count": count,
            "sum": total,
            "min": min(values),
            "max": max(values),
            "mean": total / count
        }
```

Wait, should I consider performance? If `stats` is called many times, iterating over all events each time might be slow. But without constraints, this is fine. We could sort events per group and use binary search, but it's probably overkill. I'll stick to the simple list approach unless there's a reason not to. The problem statement is simple.

Let's double check the exact return format.
`"count"`: `int`
`"sum"`: `float`
`"min"`: `float` or `None`
`"max"`: `float` or `None`
`"mean"`: `float` or `None`

The example shows `sum` as `8.0`, `mean` as `4.0`. `sum(values)` returns float if values are floats. `min` and `max` return floats.
One minor thing: `sum` of empty list is `0`, but we return `0.0`. The code handles it by returning early.
What if `values` is not empty? `sum(values)` will be float. `min` and `max` will be float. `mean` will be float.
All good.

Wait, should `sum` be explicitly cast to float? `sum([5.0, 3.0])` is `8.0`. It's fine.
What about `count`? `len(values)` is int.

Let's test mentally with the example:
`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)`
`stats("cpu", 105.0)` -> `window_start = 95.0`. `values = [5.0, 3.0]`. `count=2`, `sum=8.0`, `min=3.0`, `max=5.0`, `mean=4.0`. Matches.
`stats("cpu", 110.0)` -> `window_start = 100.0`. `values = [3.0]` (since 100.0 is not > 100.0). `count=1`, `sum=3.0`, `min=3.0`, `max=3.0`, `mean=3.0`. Matches.
`stats("disk", 105.0)` -> `events` is None or empty. Returns default. Matches.

One detail: `self.events.get(group)` returns `None` if group not in dict. `if not events:` handles both `None` and `[]`.
But `defaultdict(list)` will create an empty list if accessed via `self.events[group]`. Using `.get()` is safer to avoid creating empty lists for unknown groups, though it doesn't matter much. I'll use `.get()`.

Is there any case where `sum` should be `0.0` but `values` is empty? Handled.
What about floating point precision? Not specified, standard float operations are fine.

Could there be a requirement to handle large datasets efficiently? The problem says "Windowed streaming aggregator", which sometimes implies using a deque or sorted list with binary search. But given the simplicity, I'll stick to list filtering. If I want to be slightly more robust, I could sort the list per group on insertion or lazily, but it's not strictly necessary. I'll keep it simple.

Wait, `add` might be called many times, and `stats` might be called many times. Filtering each time is O(N). If N is large, it could be slow. But without constraints, O(N) is acceptable. I'll add a comment or just keep it as is. Actually, sorting the events per group and using `bisect` would make `stats` O(log N + K) where K is number of events in window. That's better practice for "streaming aggregator". Let's implement it with `bisect` to be safe and professional.

Actually, keeping it simp
…[9468 chars elided]…
ndition: `at - window < ts <= at`. Matches spec.
Check out-of-order: list stores all, filters by ts. Matches spec.
Check duplicates: list stores all, counts all. Matches spec.
Check groups: dict keys separate groups. Matches spec.
All correct.
Output generation. 
```python
class WindowedAggregator:
    def __init__(self, window: float):
        self.window = window
        self.events = {}

    def add(self, ts: float, group: str, value: float) -> None:
        if group not in self.events:
            self.events[group] = []
        self.events[group].append((ts, value))

    def stats(self, group: str, at: float) -> dict:
        events = self.events.get(group)
        if not events:
            return {"count": 0, "sum": 0.0, "min": None, "max": None, "mean": None}
        
        lower = at - self.window
        values = [v for ts, v in events if lower < ts <= at]
        
        if not values:
            return {"count": 0, "sum": 0.0, "min": None, "max": None, "mean": None}
            
        count = len(values)
        total = sum(values)
        return {
            "count": count,
            "sum": total,
            "min": min(values),
            "max": max(values),
            "mean": total / count
        }
```
This is correct and complete. I will wrap it in the requested format.
One detail: `sum` might be `int` if values are ints, but type hint says `value: float`. I'll cast to `float` just to be strictly compliant: `float(total)`. It doesn't hurt.
Actually, `sum` of floats is float. I'll leave it.
Done. 
Output matches. 
Proceeds. 
[Final Check]
The problem statement says: `"sum": float — sum of their values (0.0 if there are none).`
My code returns `0.0` for empty, and `sum(values)` otherwise.
All good.
I will generate the response. 
(Self-Correction/Note during generation prep)
I'll add `from collections import defaultdict` just to be safe and clean, but it's not strictly needed. I'll stick to standard dict.
The code is ready.
Output matches request.
Done. 
[Output Generation] -> *Proceeds*