tawasuyu/arje

1//! Observador estadístico. Mantiene marginales y co-ocurrencias dentro de una
2//! ventana deslizante. Calcula entropía de Shannon e información mutua para
3//! identificar correlaciones significativas.
4//!
5//! Diseño:
6//!   - Counters incrementales: cada `record()` es O(window_size) en el peor
7//!     caso (actualiza co-ocurrencias con cada evento del window).
8//!   - Sin recomputaciones globales: marginales y joint counts son state.
9//!   - El cálculo de H(X), P(B|A), I(A;B) es O(|distinct events|).
10
11use crate::rules::EventKind;
12use std::collections::{HashMap, VecDeque};
13use std::time::Instant;
14
15/// Evento timestamped. El timestamp se conserva para futuras políticas de
16/// expiración por tiempo (no sólo por count).
17#[derive(Debug, Clone)]
18pub struct TimedEvent {
19    pub kind: EventKind,
20    pub at: Instant,
21}
22
23/// Histograma de gaps temporales con buckets exponenciales en segundos.
24/// Cubre 6 órdenes de magnitud: 1ms hasta 1000s.
25#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
26pub struct GapHistogram {
27    /// Buckets cumulativos (Prometheus-style): cada índice cuenta eventos
28    /// con gap ≤ ese límite. Limites: 1ms, 10ms, 100ms, 1s, 10s, 100s, 1000s.
29    pub buckets: [u64; 7],
30    pub count: u64,
31    pub sum_secs: f64,
32    /// Suma de cuadrados — permite calcular varianza/stddev en O(1).
33    pub sum_squares_secs: f64,
34    pub max_secs: f64,
35}
36
37/// Estadísticas resumidas de un GapHistogram, usables en cristales temporales.
38#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
39pub struct GapStats {
40    pub count: u64,
41    pub mean_secs: f64,
42    pub stddev_secs: f64,
43    pub max_secs: f64,
44}
45
46const GAP_BUCKET_LIMITS_SECS: [f64; 7] = [
47    0.001, 0.01, 0.1, 1.0, 10.0, 100.0, 1000.0,
48];
49
50impl GapHistogram {
51    pub fn observe(&mut self, gap_secs: f64) {
52        for (i, &limit) in GAP_BUCKET_LIMITS_SECS.iter().enumerate() {
53            if gap_secs <= limit {
54                self.buckets[i] += 1;
55            }
56        }
57        self.count += 1;
58        self.sum_secs += gap_secs;
59        self.sum_squares_secs += gap_secs * gap_secs;
60        if gap_secs > self.max_secs { self.max_secs = gap_secs; }
61    }
62
63    pub fn mean_secs(&self) -> f64 {
64        if self.count == 0 { 0.0 } else { self.sum_secs / self.count as f64 }
65    }
66
67    /// Desviación estándar muestral. Computada vía `sum_squares - n*mean²`
68    /// para precisión razonable sin almacenar las muestras.
69    pub fn stddev_secs(&self) -> f64 {
70        if self.count < 2 { return 0.0; }
71        let n = self.count as f64;
72        let mean = self.mean_secs();
73        let var = (self.sum_squares_secs - n * mean * mean) / (n - 1.0);
74        // Numerical floor: var puede ser ligeramente negativo por float ε.
75        if var <= 0.0 { 0.0 } else { var.sqrt() }
76    }
77
78    pub fn stats(&self) -> GapStats {
79        GapStats {
80            count: self.count,
81            mean_secs: self.mean_secs(),
82            stddev_secs: self.stddev_secs(),
83            max_secs: self.max_secs,
84        }
85    }
86
87    pub fn bucket_limits() -> &'static [f64; 7] { &GAP_BUCKET_LIMITS_SECS }
88}
89
90pub struct Observer {
91    window: VecDeque<TimedEvent>,
92    window_size: usize,
93    marginal: HashMap<EventKind, u64>,
94    cooccur: HashMap<(EventKind, EventKind), u64>,
95    total: u64,
96    /// Last-seen timestamps para aplicar decay en query time. None = sin
97    /// time-decay (modo tradicional).
98    last_seen_marginal: HashMap<EventKind, Instant>,
99    last_seen_cooccur: HashMap<(EventKind, EventKind), Instant>,
100    /// Half-life del decay exponencial en segundos. None = sin decay
101    /// (las consultas devuelven los counts crudos).
102    half_life_secs: Option<f64>,
103    /// Histograma de gaps temporales por par (a, b). Capturado al `record()`.
104    gap_histograms: HashMap<(EventKind, EventKind), GapHistogram>,
105}
106
107impl Observer {
108    pub fn new(window_size: usize) -> Self {
109        Self {
110            window: VecDeque::with_capacity(window_size),
111            window_size,
112            marginal: HashMap::new(),
113            cooccur: HashMap::new(),
114            total: 0,
115            last_seen_marginal: HashMap::new(),
116            last_seen_cooccur: HashMap::new(),
117            half_life_secs: None,
118            gap_histograms: HashMap::new(),
119        }
120    }
121
122    /// Activa decay exponencial con half-life en segundos. λ = ln(2)/half_life.
123    /// Aplicado en query time sobre los counts crudos usando last_seen.
124    pub fn with_half_life(mut self, half_life_secs: f64) -> Self {
125        if half_life_secs > 0.0 {
126            self.half_life_secs = Some(half_life_secs);
127        }
128        self
129    }
130
131    pub fn half_life(&self) -> Option<f64> { self.half_life_secs }
132
133    /// Registra un evento. Actualiza marginales y co-ocurrencias contra todo
134    /// evento aún en la ventana.
135    pub fn record(&mut self, kind: EventKind) {
136        let now = Instant::now();
137        let timed = TimedEvent { kind: kind.clone(), at: now };
138
139        // Co-ocurrencias: este evento con cada uno previo en ventana.
140        // Capturamos también el gap temporal (now - w.at) para histograma.
141        for w in &self.window {
142            let key = (w.kind.clone(), kind.clone());
143            *self.cooccur.entry(key.clone()).or_insert(0) += 1;
144            self.last_seen_cooccur.insert(key.clone(), now);
145            let gap_secs = now.duration_since(w.at).as_secs_f64();
146            self.gap_histograms.entry(key).or_default().observe(gap_secs);
147        }
148
149        self.window.push_back(timed);
150        if self.window.len() > self.window_size {
151            self.window.pop_front();
152        }
153
154        *self.marginal.entry(kind.clone()).or_insert(0) += 1;
155        self.last_seen_marginal.insert(kind, now);
156        self.total += 1;
157    }
158
159    /// Aplica el decay sobre un count crudo dado el `last_seen` correspondiente.
160    /// Si half_life es None, devuelve el count tal cual (sin decay).
161    fn decay(&self, count: u64, last_seen: Option<Instant>) -> f64 {
162        let raw = count as f64;
163        let (hl, last) = match (self.half_life_secs, last_seen) {
164            (Some(hl), Some(t)) => (hl, t),
165            _ => return raw,
166        };
167        let age_secs = Instant::now().duration_since(last).as_secs_f64();
168        raw * 0.5_f64.powf(age_secs / hl)
169    }
170
171    /// Marginal con decay aplicado.
172    pub fn marginal_decayed(&self, k: &EventKind) -> f64 {
173        let raw = self.marginal.get(k).copied().unwrap_or(0);
174        let last = self.last_seen_marginal.get(k).copied();
175        self.decay(raw, last)
176    }
177
178    /// Cooccurrence con decay aplicado.
179    pub fn cooccur_decayed(&self, a: &EventKind, b: &EventKind) -> f64 {
180        let raw = self.cooccur.get(&(a.clone(), b.clone())).copied().unwrap_or(0);
181        let last = self.last_seen_cooccur.get(&(a.clone(), b.clone())).copied();
182        self.decay(raw, last)
183    }
184
185    /// Entropía de Shannon de la distribución marginal de eventos.
186    /// H(X) = −Σ p(x) log₂ p(x). Unidad: bits.
187    pub fn shannon_entropy(&self) -> f64 {
188        if self.total == 0 { return 0.0; }
189        let total = self.total as f64;
190        self.marginal.values()
191            .map(|&c| {
192                let p = c as f64 / total;
193                if p > 0.0 { -p * p.log2() } else { 0.0 }
194            })
195            .sum()
196    }
197
198    /// P(b | a) = "dado que algo siguió a `a` dentro del window, qué fracción
199    /// fue `b`". Suma 1 sobre todos los b posibles para un a fijo.
200    ///
201    /// Implementación: cooccur_decayed(a, b) / Σ_x cooccur_decayed(a, x).
202    /// Si half_life is None, los decayed values son los counts crudos.
203    pub fn conditional_prob(&self, a: &EventKind, b: &EventKind) -> f64 {
204        let joint = self.cooccur_decayed(a, b);
205        let row_total: f64 = self.cooccur.keys()
206            .filter(|(x, _)| x == a)
207            .map(|(x, y)| self.cooccur_decayed(x, y))
208            .sum();
209        if row_total <= 0.0 { 0.0 } else { joint / row_total }
210    }
211
212    /// Información mutua puntual entre `a` y `b` con decay aplicado:
213    /// PMI(a, b) = log₂( P(a, b) / (P(a) · P(b)) ).
214    /// Positivo → más correlacionados de lo que sugiere independencia.
215    pub fn pmi(&self, a: &EventKind, b: &EventKind) -> f64 {
216        // Total decayed: suma de marginales con decay (no usamos self.total
217        // directo porque debería ser consistente con los decayed values).
218        let total_decayed: f64 = self.marginal.keys()
219            .map(|k| self.marginal_decayed(k))
220            .sum();
221        if total_decayed <= 0.0 { return 0.0; }
222        let joint = self.cooccur_decayed(a, b) / total_decayed;
223        let pa = self.marginal_decayed(a) / total_decayed;
224        let pb = self.marginal_decayed(b) / total_decayed;
225        if joint <= 0.0 || pa <= 0.0 || pb <= 0.0 { return 0.0; }
226        (joint / (pa * pb)).log2()
227    }
228
229    /// Información mutua acumulada de la pareja (a, b) ponderada por su
230    /// probabilidad conjunta. Útil como medida de "interés" del par.
231    pub fn weighted_pmi(&self, a: &EventKind, b: &EventKind) -> f64 {
232        if self.total == 0 { return 0.0; }
233        let joint = self.cooccur
234            .get(&(a.clone(), b.clone()))
235            .copied()
236            .unwrap_or(0) as f64 / self.total as f64;
237        joint * self.pmi(a, b)
238    }
239
240    pub fn marginals(&self) -> &HashMap<EventKind, u64> { &self.marginal }
241    pub fn cooccurrences(&self) -> &HashMap<(EventKind, EventKind), u64> { &self.cooccur }
242    pub fn total(&self) -> u64 { self.total }
243    pub fn window_size(&self) -> usize { self.window_size }
244    pub fn current_window(&self) -> usize { self.window.len() }
245
246    /// Últimos N eventos del window, en orden cronológico (más viejo primero).
247    /// Si N > window.len(), devuelve todo el window.
248    pub fn recent(&self, n: usize) -> impl Iterator<Item = &TimedEvent> {
249        let start = self.window.len().saturating_sub(n);
250        self.window.range(start..)
251    }
252
253    pub fn gap_histograms(&self) -> &HashMap<(EventKind, EventKind), GapHistogram> {
254        &self.gap_histograms
255    }
256
257    /// Top-K pares por count del histograma (más frecuentes primero).
258    /// Útil para limitar cardinalidad de métricas exportadas.
259    pub fn top_gap_pairs(&self, k: usize) -> Vec<(&(EventKind, EventKind), &GapHistogram)> {
260        let mut pairs: Vec<_> = self.gap_histograms.iter().collect();
261        pairs.sort_by(|a, b| b.1.count.cmp(&a.1.count));
262        pairs.truncate(k);
263        pairs
264    }
265
266    /// Snapshot serializable del estado estadístico (sin Instants — no son
267    /// portables a través de reboots). El window deslizante se descarta —
268    /// se reconstruye desde cero al restore.
269    pub fn snapshot(&self) -> ObserverSnapshot {
270        ObserverSnapshot {
271            schema_version: OBSERVER_SCHEMA_VERSION,
272            window_size: self.window_size,
273            half_life_secs: self.half_life_secs,
274            total: self.total,
275            marginal: self.marginal.iter()
276                .map(|(k, v)| (k.clone(), *v))
277                .collect(),
278            cooccur: self.cooccur.iter()
279                .map(|((a, b), c)| (a.clone(), b.clone(), *c))
280                .collect(),
281            gap_histograms: self.gap_histograms.iter()
282                .map(|((a, b), h)| (a.clone(), b.clone(), h.clone()))
283                .collect(),
284        }
285    }
286
287    /// Reconstruye Observer desde un snapshot. El window queda vacío;
288    /// last_seen_* se inicializa en `now()` para que el decay arranque
289    /// "ahora" para todos los counts (aproximación razonable post-reboot).
290    pub fn from_snapshot(snap: ObserverSnapshot) -> Self {
291        let now = Instant::now();
292        let mut marginal = HashMap::new();
293        let mut last_seen_marginal = HashMap::new();
294        for (k, v) in snap.marginal {
295            last_seen_marginal.insert(k.clone(), now);
296            marginal.insert(k, v);
297        }
298        let mut cooccur = HashMap::new();
299        let mut last_seen_cooccur = HashMap::new();
300        for (a, b, c) in snap.cooccur {
301            last_seen_cooccur.insert((a.clone(), b.clone()), now);
302            cooccur.insert((a, b), c);
303        }
304        let gap_histograms = snap.gap_histograms.into_iter()
305            .map(|(a, b, h)| ((a, b), h))
306            .collect();
307        Self {
308            window: VecDeque::with_capacity(snap.window_size),
309            window_size: snap.window_size,
310            marginal,
311            cooccur,
312            total: snap.total,
313            last_seen_marginal,
314            last_seen_cooccur,
315            half_life_secs: snap.half_life_secs,
316            gap_histograms,
317        }
318    }
319}
320
321const OBSERVER_SCHEMA_VERSION: u16 = 1;
322
323/// Snapshot serializable. Se persiste a JSON en disco y se restaura al
324/// reboot para preservar contadores, co-ocurrencias e histogramas.
325#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
326pub struct ObserverSnapshot {
327    pub schema_version: u16,
328    pub window_size: usize,
329    pub half_life_secs: Option<f64>,
330    pub total: u64,
331    /// Marginales serializados como Vec porque HashMap<EventKind, _> usa
332    /// EventKind como key — y EventKind tiene variantes con payloads que
333    /// no son JSON-key-serializable (BusInvokeOf, Custom).
334    pub marginal: Vec<(EventKind, u64)>,
335    pub cooccur: Vec<(EventKind, EventKind, u64)>,
336    pub gap_histograms: Vec<(EventKind, EventKind, GapHistogram)>,
337}
338
339#[cfg(test)]
340mod tests {
341    use super::*;
342    use crate::rules::EventKind::*;
343
344    #[test]
345    fn entropy_zero_for_single_event() {
346        let mut obs = Observer::new(10);
347        for _ in 0..5 { obs.record(EnteSpawned); }
348        // Distribución degenerada: una sola observación posible → H = 0.
349        assert!(obs.shannon_entropy() < 1e-9);
350    }
351
352    #[test]
353    fn entropy_one_for_balanced_binary() {
354        let mut obs = Observer::new(100);
355        for _ in 0..10 { obs.record(EnteSpawned); }
356        for _ in 0..10 { obs.record(EnteDied); }
357        // Bernoulli(0.5) → H = 1 bit.
358        assert!((obs.shannon_entropy() - 1.0).abs() < 1e-9);
359    }
360
361    #[test]
362    fn conditional_prob_perfect_dependency() {
363        let mut obs = Observer::new(100);
364        // Spawned siempre seguido por Died.
365        for _ in 0..5 {
366            obs.record(EnteSpawned);
367            obs.record(EnteDied);
368        }
369        let p = obs.conditional_prob(&EnteSpawned, &EnteDied);
370        assert!(p > 0.0, "esperamos correlación positiva, got {p}");
371    }
372}