nx_sketch_stream_stats.nx source
↩ module page · 213 lines · 6663 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
38// nx_safety_envelope:
39// intended_use: AUTO_APPLIED -- primitive-specific tuning queued
40// sil_target: SIL1
41// evidence: [bulk_applied_2026-05-16, see-file-comment-for-detail]
42// verdict: NOT_YET_EVALUATED
43
44import "nx_syscalls.nx"
45import "nx_sketch_types.nx"
46
47struct StreamStats {
48 count: i64,
49 sum: i64,
50 sum_sq: i64,
51 min_val: i64,
52 max_val: i64,
53 has_data: i64, // 0 if empty, 1 if any add happened
54}
55
56// === construction =================================================
57
58func nx_stats_alloc() -> *StreamStats {
59 let raw: *u8 = sys_mmap(56)
60 let s: *StreamStats = raw as *StreamStats
61 s.count = 0
62 s.sum = 0
63 s.sum_sq = 0
64 s.min_val = 0
65 s.max_val = 0
66 s.has_data = 0
67 return s
68}
69
70// === overflow guard =================================================
71//
72// Returns 1 iff adding `value` is overflow-safe given the current
73// accumulator state. Conservative: declares unsafe if sum_sq + v² would
74// hit upper 2 bits of i64 (margin for next-step arithmetic).
75
76const NX_STATS_SAFE_HI: i64 = 0x2000000000000000 // 2^61
77
78func nx_stats_safe_p(s: *StreamStats, value: i64) -> i64 {
79 var v: i64 = value
80 if v < 0 { v = -v }
81 if v >= 0x40000000 { return 0 } // |v| >= 2^30 -> v² overflows i64
82 let v_sq: i64 = v * v
83 if s.sum > NX_STATS_SAFE_HI - v { return 0 }
84 if s.sum_sq > NX_STATS_SAFE_HI - v_sq { return 0 }
85 return 1
86}
87
88// === add ==========================================================
89
90func nx_stats_add(s: *StreamStats, value: i64) -> i64 {
91 if nx_stats_safe_p(s, value) == 0 { return -1 }
92 s.count = s.count + 1
93 s.sum = s.sum + value
94 s.sum_sq = s.sum_sq + value * value
95 if s.has_data == 0 {
96 s.min_val = value
97 s.max_val = value
98 s.has_data = 1
99 }
100 if s.has_data == 1 {
101 if value < s.min_val { s.min_val = value }
102 if value > s.max_val { s.max_val = value }
103 }
104 return 0
105}
106
107// === queries ======================================================
108
109func nx_stats_mean(s: *StreamStats) -> i64 {
110 if s.count == 0 { return 0 }
111 return s.sum / s.count
112}
113
114func nx_stats_variance(s: *StreamStats) -> i64 {
115 if s.count == 0 { return 0 }
116 let m: i64 = nx_stats_mean(s)
117 let e_sq: i64 = s.sum_sq / s.count
118 let m_sq: i64 = m * m
119 if e_sq < m_sq { return 0 } // shouldn't happen but guards against rounding
120 return e_sq - m_sq
121}
122
123// Integer square root via Newton's method. For x >= 0, returns floor(sqrt(x)).
124func nx_stats_isqrt(x: i64) -> i64 {
125 if x < 0 { return 0 }
126 if x == 0 { return 0 }
127 if x < 4 { return 1 }
128 var g: i64 = x
129 // Initial overestimate: x itself (will converge fast).
130 // Better: g = x >> 1 + 1 for rough first guess.
131 g = (x >> 1) + 1
132 var iter: i64 = 0
133 while iter < 64 {
134 let next_g: i64 = (g + x / g) / 2
135 if next_g >= g { iter = 64 }
136 if next_g < g {
137 g = next_g
138 iter = iter + 1
139 }
140 }
141 return g
142}
143
144func nx_stats_stddev(s: *StreamStats) -> i64 {
145 return nx_stats_isqrt(nx_stats_variance(s))
146}
147
148func nx_stats_min(s: *StreamStats) -> i64 {
149 return s.min_val
150}
151
152func nx_stats_max(s: *StreamStats) -> i64 {
153 return s.max_val
154}
155
156func nx_stats_count(s: *StreamStats) -> i64 {
157 return s.count
158}
159
160// === typed queries ================================================
161//
162// All exact under the overflow budget -- envelope conf_ppb = 1e9.
163
164func nx_stats_query_mean(s: *StreamStats) -> *ApproxI64 {
165 let m: i64 = nx_stats_mean(s)
166 return nx_approx_new(m, NX_ENV_ABS, 0, 1000000000,
167 NX_MATURITY_PRODUCTION,
168 NX_ADV_HONEST)
169}
170
171func nx_stats_query_variance(s: *StreamStats) -> *ApproxI64 {
172 let v: i64 = nx_stats_variance(s)
173 return nx_approx_new(v, NX_ENV_ABS, 0, 1000000000,
174 NX_MATURITY_PRODUCTION,
175 NX_ADV_HONEST)
176}
177
178// === merge (Chan 1979 parallel combine, integer variant) ==========
179//
180// Combine two independent streams. All sums add exactly.
181
182func nx_stats_merge(a: *StreamStats, b: *StreamStats) -> *StreamStats {
183 let out: *StreamStats = nx_stats_alloc()
184 out.count = a.count + b.count
185 out.sum = a.sum + b.sum
186 out.sum_sq = a.sum_sq + b.sum_sq
187 if a.has_data == 1 {
188 if b.has_data == 1 {
189 out.min_val = a.min_val
190 if b.min_val < out.min_val { out.min_val = b.min_val }
191 out.max_val = a.max_val
192 if b.max_val > out.max_val { out.max_val = b.max_val }
193 out.has_data = 1
194 }
195 if b.has_data == 0 {
196 out.min_val = a.min_val
197 out.max_val = a.max_val
198 out.has_data = 1
199 }
200 }
201 if a.has_data == 0 {
202 if b.has_data == 1 {
203 out.min_val = b.min_val
204 out.max_val = b.max_val
205 out.has_data = 1
206 }
207 }
208 return out
209}
210
211func nx_stats_memory_bytes(s: *StreamStats) -> i64 {
212 return 56
213}