sketch_stream_stats.nx source
↩ module page · 190 lines · 6178 B
1// sketch_stream_stats.nx -- streaming mean / variance / stddev / min / max.
2//
3// Fundamental single-pass streaming primitive. Maintains six i64 fields
4// across the stream:
5// count = N
6// sum = Σ x_i
7// sum_sq = Σ x_i²
8// min = min over stream
9// max = max over stream
10//
11// Algebraic-identity-based estimators (NOT Welford, which is f64-stable but
12// loses precision when arithmetic is exact-integer):
13// mean = sum / count
14// variance = sum_sq / count - mean² (population)
15// stddev = isqrt(variance)
16//
17// OVERFLOW BUDGET:
18// For sample values bounded by |x| <= V and count N:
19// sum: N * V — must fit i64. Safe to N * V < 2^62.
20// sum_sq: N * V² — same constraint applied to V² instead of V.
21// For V = 2^30 (1 billion): N up to ~2^32 (4 billion) safe for sum,
22// but sum_sq overflows at N * 2^60 -- safe to N=4.
23// For V = 2^20 (~1M): N up to 2^22 (~4M) safe for sum_sq.
24// For V = 2^16 (65K): N up to 2^30 (1B) safe.
25// Caller responsibility to size domain; nx_stats_safe_p returns 1 iff
26// adding `value` would NOT overflow.
27//
28// COMPLEMENTS the sketch suite:
29// - sketch_reservoir: arbitrary-statistic estimation via sample-based math
30// - sketch_kll / sketch_tdigest: quantiles (median, p99)
31// - sketch_stream_stats (here): exact-integer mean/variance over the FULL
32// stream (no sampling, no probabilistic bounds)
33//
34// LOSSLESS-LANGUAGE DISCIPLINE: nx_stats_query_mean and friends return
35// ApproxI64 with envelope_kind = NX_ENV_ABS, param_a = 0 (mean and
36// variance are EXACT under the overflow budget), conf_ppb = 1e9.
37
38import "syscalls.nx"
39import "sketch_types.nx"
40import "nx_vecmath.nx"
41
42struct StreamStats {
43 count: i64,
44 sum: i64,
45 sum_sq: i64,
46 min_val: i64,
47 max_val: i64,
48 has_data: i64, // 0 if empty, 1 if any add happened
49}
50
51// === construction =================================================
52
53func nx_stats_alloc() -> *StreamStats {
54 let raw: *u8 = sys_mmap(56)
55 let s: *StreamStats = raw as *StreamStats
56 s.count = 0
57 s.sum = 0
58 s.sum_sq = 0
59 s.min_val = 0
60 s.max_val = 0
61 s.has_data = 0
62 return s
63}
64
65// === overflow guard =================================================
66//
67// Returns 1 iff adding `value` is overflow-safe given the current
68// accumulator state. Conservative: declares unsafe if sum_sq + v² would
69// hit upper 2 bits of i64 (margin for next-step arithmetic).
70
71const NX_STATS_SAFE_HI: i64 = 0x2000000000000000 // 2^61
72
73func nx_stats_safe_p(s: *StreamStats, value: i64) -> i64 {
74 var v: i64 = value
75 if v < 0 { v = -v }
76 if v >= 0x40000000 { return 0 } // |v| >= 2^30 -> v² overflows i64
77 let v_sq: i64 = v * v
78 if s.sum > NX_STATS_SAFE_HI - v { return 0 }
79 if s.sum_sq > NX_STATS_SAFE_HI - v_sq { return 0 }
80 return 1
81}
82
83// === add ==========================================================
84
85func nx_stats_add(s: *StreamStats, value: i64) -> i64 {
86 if nx_stats_safe_p(s, value) == 0 { return -1 }
87 s.count = s.count + 1
88 s.sum = s.sum + value
89 s.sum_sq = s.sum_sq + value * value
90 if s.has_data == 0 {
91 s.min_val = value
92 s.max_val = value
93 s.has_data = 1
94 }
95 if s.has_data == 1 {
96 if value < s.min_val { s.min_val = value }
97 if value > s.max_val { s.max_val = value }
98 }
99 return 0
100}
101
102// === queries ======================================================
103
104func nx_stats_mean(s: *StreamStats) -> i64 {
105 if s.count == 0 { return 0 }
106 return s.sum / s.count
107}
108
109func nx_stats_variance(s: *StreamStats) -> i64 {
110 if s.count == 0 { return 0 }
111 let m: i64 = nx_stats_mean(s)
112 let e_sq: i64 = s.sum_sq / s.count
113 let m_sq: i64 = m * m
114 if e_sq < m_sq { return 0 } // shouldn't happen but guards against rounding
115 return e_sq - m_sq
116}
117
118// Integer square root via Newton's method. For x >= 0, returns floor(sqrt(x)).
119func nx_stats_isqrt(x: i64) -> i64 { return vm_isqrt(x) }
120
121func nx_stats_stddev(s: *StreamStats) -> i64 {
122 return nx_stats_isqrt(nx_stats_variance(s))
123}
124
125func nx_stats_min(s: *StreamStats) -> i64 {
126 return s.min_val
127}
128
129func nx_stats_max(s: *StreamStats) -> i64 {
130 return s.max_val
131}
132
133func nx_stats_count(s: *StreamStats) -> i64 {
134 return s.count
135}
136
137// === typed queries ================================================
138//
139// All exact under the overflow budget -- envelope conf_ppb = 1e9.
140
141func nx_stats_query_mean(s: *StreamStats) -> *ApproxI64 {
142 let m: i64 = nx_stats_mean(s)
143 return nx_approx_new(m, NX_ENV_ABS, 0, 1000000000,
144 NX_MATURITY_PRODUCTION,
145 NX_ADV_HONEST)
146}
147
148func nx_stats_query_variance(s: *StreamStats) -> *ApproxI64 {
149 let v: i64 = nx_stats_variance(s)
150 return nx_approx_new(v, NX_ENV_ABS, 0, 1000000000,
151 NX_MATURITY_PRODUCTION,
152 NX_ADV_HONEST)
153}
154
155// === merge (Chan 1979 parallel combine, integer variant) ==========
156//
157// Combine two independent streams. All sums add exactly.
158
159func nx_stats_merge(a: *StreamStats, b: *StreamStats) -> *StreamStats {
160 let out: *StreamStats = nx_stats_alloc()
161 out.count = a.count + b.count
162 out.sum = a.sum + b.sum
163 out.sum_sq = a.sum_sq + b.sum_sq
164 if a.has_data == 1 {
165 if b.has_data == 1 {
166 out.min_val = a.min_val
167 if b.min_val < out.min_val { out.min_val = b.min_val }
168 out.max_val = a.max_val
169 if b.max_val > out.max_val { out.max_val = b.max_val }
170 out.has_data = 1
171 }
172 if b.has_data == 0 {
173 out.min_val = a.min_val
174 out.max_val = a.max_val
175 out.has_data = 1
176 }
177 }
178 if a.has_data == 0 {
179 if b.has_data == 1 {
180 out.min_val = b.min_val
181 out.max_val = b.max_val
182 out.has_data = 1
183 }
184 }
185 return out
186}
187
188func nx_stats_memory_bytes(s: *StreamStats) -> i64 {
189 return 56
190}