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