Streaming Aggregations With a Single Pass (JS)

// Welford's online algorithm: incremental mean and variance with no array
// allocation. Numerically stable; the naive sum-of-squares minus square-of-sum
// formula loses precision badly on large samples.

class RunningStats {
    constructor() {
        this.n = 0;
        this.mean = 0;
        this.m2 = 0;
        this.min = Infinity;
        this.max = -Infinity;
    }
    push(x) {
        this.n += 1;
        const delta = x - this.mean;
        this.mean += delta / this.n;
        const delta2 = x - this.mean;
        this.m2 += delta * delta2;
        if (x < this.min) this.min = x;
        if (x > this.max) this.max = x;
    }
    get variance() { return this.n < 2 ? 0 : this.m2 / (this.n - 1); }
    get stddev() { return Math.sqrt(this.variance); }
    summary() {
        return { n: this.n, mean: +this.mean.toFixed(2), stddev: +this.stddev.toFixed(2), min: this.min, max: this.max };
    }
}

const stats = new RunningStats();
for (let i = 0; i < 10_000; i++) {
    const ms = Math.random() < 0.95 ? 30 + Math.random() * 40 : 200 + Math.random() * 300;
    stats.push(ms);
}
console.log(stats.summary());

// Exact percentiles need the whole sorted sample. For an unbounded stream you
// keep a fixed-size reservoir that approximates the distribution; querying
// p99 is then sort + index. Constant memory, sub-1% error in practice.

class ReservoirP99 {
    constructor(size = 1000) {
        this.size = size;
        this.buf = [];
        this.seen = 0;
    }
    push(x) {
        this.seen += 1;
        if (this.buf.length < this.size) {
            this.buf.push(x);
            return;
        }
        const j = Math.floor(Math.random() * this.seen);
        if (j < this.size) this.buf[j] = x;
    }
    quantile(q) {
        if (this.buf.length === 0) return NaN;
        const sorted = [...this.buf].sort((a, b) => a - b);
        const idx = Math.min(sorted.length - 1, Math.floor(q * sorted.length));
        return sorted[idx];
    }
}

const latencies = new ReservoirP99(500);
for (let i = 0; i < 100_000; i++) {
    const ms = Math.random() < 0.99 ? 30 + Math.random() * 30 : 250 + Math.random() * 400;
    latencies.push(ms);
}

console.log({
    seen: latencies.seen,
    p50: +latencies.quantile(0.50).toFixed(1),
    p90: +latencies.quantile(0.90).toFixed(1),
    p99: +latencies.quantile(0.99).toFixed(1),
});

// Real monitoring wants per-route, per-region, per-user-tier stats.
// Combine the two pieces from above into a Map of RunningStats per key.

class RunningStats {
    constructor() { this.n = 0; this.mean = 0; this.m2 = 0; this.max = -Infinity; }
    push(x) {
        this.n += 1;
        const delta = x - this.mean;
        this.mean += delta / this.n;
        this.m2 += delta * (x - this.mean);
        if (x > this.max) this.max = x;
    }
    get stddev() { return this.n < 2 ? 0 : Math.sqrt(this.m2 / (this.n - 1)); }
}

class GroupedStats {
    constructor(keyFn) { this.keyFn = keyFn; this.byKey = new Map(); }
    push(event) {
        const k = this.keyFn(event);
        let stats = this.byKey.get(k);
        if (!stats) { stats = new RunningStats(); this.byKey.set(k, stats); }
        stats.push(event.ms);
    }
    topK(k = 3, by = 'mean') {
        return [...this.byKey.entries()]
            .map(([key, s]) => ({ key, n: s.n, mean: +s.mean.toFixed(1), max: s.max, stddev: +s.stddev.toFixed(1) }))
            .sort((a, b) => b[by] - a[by])
            .slice(0, k);
    }
}

const routes = ['/api/users', '/api/cart', '/api/checkout', '/api/search'];
const grouped = new GroupedStats((e) => e.route);
for (let i = 0; i < 5_000; i++) {
    const route = routes[i % routes.length];
    const baseline = route === '/api/checkout' ? 180 : 50;
    grouped.push({ route, ms: baseline + Math.random() * 30 });
}
console.log(grouped.topK(3, 'mean'));