DGIM Algorithm and Its Role in Stream Analysis
DGIM is a statistical-flavoured streaming technique widely applied in sensor monitoring, network traffic analysis, clickstream analytics, and other fields where data arrives continuously and cannot be stored in full. The method focuses on estimating how many 1s (events, spikes, threshold breaches) occurred in the last N readings of a binary stream, using only O(log²N) memory instead of O(N).
• The timestamp of a bucket records when its most recent 1 arrived.
• The size of a bucket (always a power of 2) records how many 1s it represents.
• The method guarantees the final estimate is never off by more than 50% from the true count.
Core Idea: Buckets
Instead of storing every bit, DGIM groups the 1s into buckets. Each bucket stores exactly two values — a timestamp and a size — and four rules keep the bucket list small enough to fit in log-squared space:
- Non-decreasing sizes: Bucket sizes never shrink as you scan from the newest bucket toward the oldest.
- At most 2 buckets per size: No size value may be represented by more than two buckets at once.
- On a 0: No update — only buckets that have aged out of the window get dropped.
- On a 1: A new size-1 bucket is created. If this causes 3 buckets of the same size to exist, the two oldest of them merge into one bucket of double the size — a merge that can cascade upward.
Worked Example: Vibration Monitoring on a Production Line
A factory floor runs motors fitted with vibration sensors sampling once per second. Each reading is either under a safety threshold (0) or over it (1, a spike — an early warning sign of bearing wear). A maintenance dashboard continuously asks: "how many spikes has this motor logged in the last 20 seconds?" Storing full spike history for every motor on the line does not scale, so DGIM's bucket summary is used instead.
| t (sec) | 1 | 2 | 3 | 4 | 5 | 6 | 7 | 8 | 9 | 10 | 11 | 12 | 13 | 14 | 15 | 16 | 17 | 18 | 19 | 20 |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| spike? | 1 | 0 | 1 | 1 | 0 | 0 | 1 | 0 | 1 | 1 | 0 | 0 | 1 | 0 | 1 | 0 | 1 | 1 | 0 | 1 |
Walking through the key merge steps
Final bucket state at t = 20
| Bucket | Size | Timestamp |
|---|---|---|
| A | 1 | 20 |
| B | 2 | 18 |
| C | 4 | 15 |
| D | 4 | 7 |
All four buckets are still valid, since N = 20 and the current time is t = 20.
Querying: How Many Spikes in the Last 10 Seconds?
Last 10 seconds = readings 11–20.
- → Bucket A (size 1, ts=20) — fully inside → count 1
- → Bucket B (size 2, ts=18) — fully inside → count 2
- → Bucket C (size 4, ts=15) — the boundary bucket → count only half = 2
- → Bucket D (size 4, ts=7) — outside range → ignore
Estimate = 1 + 2 + 2 = 5
Actual count (readings at t=11–20: 0,0,1,0,1,0,1,1,0,1) = 5 spikes. The estimate lands exactly right here — that won't always happen, but the 50% error bound always holds.
Why It's Efficient
- → Each bucket needs only O(log N) bits to store (timestamp + size).
- → At most O(log N) different sizes exist, and at most 2 buckets per size → total buckets = O(log N).
- → Total space = O(log²N) — a large saving over storing all N raw readings.
Summary Table
| Aspect | Detail |
|---|---|
| Purpose | Count 1s in last N readings with limited memory |
| Storage | O(log²N) bits |
| Structure | Buckets: (timestamp, size = power of 2) |
| Constraint | ≤ 2 buckets per size |
| Update on 1 | New bucket + merge if 3 buckets of same size appear |
| Query | Sum full bucket sizes in range + half of the boundary bucket |
| Error bound | ≤ 50% |
Python Implementation
A compact DGIM class that processes a sensor stream reading by reading and answers rolling-window queries, reproducing the numbers from the walkthrough above.
from dataclasses import dataclass
from typing import List
@dataclass
class Bucket:
"""A DGIM bucket: `size` spikes, the most recent of which arrived at `timestamp`."""
size: int
timestamp: int
class DGIM:
"""
DGIM algorithm for estimating the count of 1s (threshold-breach events)
in the last N readings of a sensor stream, using O(log^2 N) space.
"""
def __init__(self, window_size: int):
self.N = window_size
self.time = 0
self.buckets: List[Bucket] = [] # buckets[0] = newest
def add_reading(self, spike: int) -> None:
self.time += 1
while self.buckets and self.buckets[-1].timestamp <= self.time - self.N:
self.buckets.pop()
if spike == 0:
return
self.buckets.insert(0, Bucket(size=1, timestamp=self.time))
self._merge()
def _merge(self) -> None:
changed = True
while changed:
changed = False
i = 0
while i < len(self.buckets):
j = i
while j < len(self.buckets) and self.buckets[j].size == self.buckets[i].size:
j += 1
if j - i >= 3:
newer, older = self.buckets[j - 2], self.buckets[j - 1]
merged = Bucket(size=newer.size * 2, timestamp=newer.timestamp)
self.buckets[j - 2:j] = [merged]
changed = True
break
i = j
def query(self, k: int) -> int:
boundary = self.time - k
in_range = [b for b in self.buckets if b.timestamp > boundary]
if not in_range:
return 0
oldest_in_range = in_range[-1]
return sum(b.size for b in in_range) - oldest_in_range.size // 2
if __name__ == "__main__":
motor_readings = [1, 0, 1, 1, 0, 0, 1, 0, 1, 1, 0, 0, 1, 0, 1, 0, 1, 1, 0, 1]
monitor = DGIM(window_size=20)
for reading in motor_readings:
monitor.add_reading(reading)
print(monitor.buckets)
print("Estimate (last 10 sec):", monitor.query(10))
print("Actual (last 10 sec):", sum(motor_readings[-10:]))
Estimate (last 10 sec): 5
Actual (last 10 sec): 5
Advantages
- → Sub-linear memory: answers sliding-window queries in O(log²N) instead of O(N).
- → Bounded error: the 50% worst-case guarantee holds regardless of stream shape.
- → Fast updates: each bit triggers only a constant number of amortised merge operations.
Disadvantages
- → Only approximate — not suitable when an exact count is required.
- → Designed for counting 1s specifically; extending to general sums needs variants (e.g. exponential histograms).
- → Error can approach the 50% bound in adversarial or highly bursty streams.