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 10 órdenes de magnitud: 1ms hasta 1000s.
25#[derive(Debug, Clone, Default)]
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    pub max_secs: f64,
33}
34
35const GAP_BUCKET_LIMITS_SECS: [f64; 7] = [
36    0.001, 0.01, 0.1, 1.0, 10.0, 100.0, 1000.0,
37];
38
39impl GapHistogram {
40    pub fn observe(&mut self, gap_secs: f64) {
41        for (i, &limit) in GAP_BUCKET_LIMITS_SECS.iter().enumerate() {
42            if gap_secs <= limit {
43                self.buckets[i] += 1;
44            }
45        }
46        self.count += 1;
47        self.sum_secs += gap_secs;
48        if gap_secs > self.max_secs { self.max_secs = gap_secs; }
49    }
50
51    pub fn mean_secs(&self) -> f64 {
52        if self.count == 0 { 0.0 } else { self.sum_secs / self.count as f64 }
53    }
54
55    pub fn bucket_limits() -> &'static [f64; 7] { &GAP_BUCKET_LIMITS_SECS }
56}
57
58pub struct Observer {
59    window: VecDeque<TimedEvent>,
60    window_size: usize,
61    marginal: HashMap<EventKind, u64>,
62    cooccur: HashMap<(EventKind, EventKind), u64>,
63    total: u64,
64    /// Last-seen timestamps para aplicar decay en query time. None = sin
65    /// time-decay (modo tradicional).
66    last_seen_marginal: HashMap<EventKind, Instant>,
67    last_seen_cooccur: HashMap<(EventKind, EventKind), Instant>,
68    /// Half-life del decay exponencial en segundos. None = sin decay
69    /// (las consultas devuelven los counts crudos).
70    half_life_secs: Option<f64>,
71    /// Histograma de gaps temporales por par (a, b). Capturado al `record()`.
72    gap_histograms: HashMap<(EventKind, EventKind), GapHistogram>,
73}
74
75impl Observer {
76    pub fn new(window_size: usize) -> Self {
77        Self {
78            window: VecDeque::with_capacity(window_size),
79            window_size,
80            marginal: HashMap::new(),
81            cooccur: HashMap::new(),
82            total: 0,
83            last_seen_marginal: HashMap::new(),
84            last_seen_cooccur: HashMap::new(),
85            half_life_secs: None,
86            gap_histograms: HashMap::new(),
87        }
88    }
89
90    /// Activa decay exponencial con half-life en segundos. λ = ln(2)/half_life.
91    /// Aplicado en query time sobre los counts crudos usando last_seen.
92    pub fn with_half_life(mut self, half_life_secs: f64) -> Self {
93        if half_life_secs > 0.0 {
94            self.half_life_secs = Some(half_life_secs);
95        }
96        self
97    }
98
99    pub fn half_life(&self) -> Option<f64> { self.half_life_secs }
100
101    /// Registra un evento. Actualiza marginales y co-ocurrencias contra todo
102    /// evento aún en la ventana.
103    pub fn record(&mut self, kind: EventKind) {
104        let now = Instant::now();
105        let timed = TimedEvent { kind: kind.clone(), at: now };
106
107        // Co-ocurrencias: este evento con cada uno previo en ventana.
108        // Capturamos también el gap temporal (now - w.at) para histograma.
109        for w in &self.window {
110            let key = (w.kind.clone(), kind.clone());
111            *self.cooccur.entry(key.clone()).or_insert(0) += 1;
112            self.last_seen_cooccur.insert(key.clone(), now);
113            let gap_secs = now.duration_since(w.at).as_secs_f64();
114            self.gap_histograms.entry(key).or_default().observe(gap_secs);
115        }
116
117        self.window.push_back(timed);
118        if self.window.len() > self.window_size {
119            self.window.pop_front();
120        }
121
122        *self.marginal.entry(kind.clone()).or_insert(0) += 1;
123        self.last_seen_marginal.insert(kind, now);
124        self.total += 1;
125    }
126
127    /// Aplica el decay sobre un count crudo dado el `last_seen` correspondiente.
128    /// Si half_life es None, devuelve el count tal cual (sin decay).
129    fn decay(&self, count: u64, last_seen: Option<Instant>) -> f64 {
130        let raw = count as f64;
131        let (hl, last) = match (self.half_life_secs, last_seen) {
132            (Some(hl), Some(t)) => (hl, t),
133            _ => return raw,
134        };
135        let age_secs = Instant::now().duration_since(last).as_secs_f64();
136        raw * 0.5_f64.powf(age_secs / hl)
137    }
138
139    /// Marginal con decay aplicado.
140    pub fn marginal_decayed(&self, k: &EventKind) -> f64 {
141        let raw = self.marginal.get(k).copied().unwrap_or(0);
142        let last = self.last_seen_marginal.get(k).copied();
143        self.decay(raw, last)
144    }
145
146    /// Cooccurrence con decay aplicado.
147    pub fn cooccur_decayed(&self, a: &EventKind, b: &EventKind) -> f64 {
148        let raw = self.cooccur.get(&(a.clone(), b.clone())).copied().unwrap_or(0);
149        let last = self.last_seen_cooccur.get(&(a.clone(), b.clone())).copied();
150        self.decay(raw, last)
151    }
152
153    /// Entropía de Shannon de la distribución marginal de eventos.
154    /// H(X) = −Σ p(x) log₂ p(x). Unidad: bits.
155    pub fn shannon_entropy(&self) -> f64 {
156        if self.total == 0 { return 0.0; }
157        let total = self.total as f64;
158        self.marginal.values()
159            .map(|&c| {
160                let p = c as f64 / total;
161                if p > 0.0 { -p * p.log2() } else { 0.0 }
162            })
163            .sum()
164    }
165
166    /// P(b | a) = "dado que algo siguió a `a` dentro del window, qué fracción
167    /// fue `b`". Suma 1 sobre todos los b posibles para un a fijo.
168    ///
169    /// Implementación: cooccur_decayed(a, b) / Σ_x cooccur_decayed(a, x).
170    /// Si half_life is None, los decayed values son los counts crudos.
171    pub fn conditional_prob(&self, a: &EventKind, b: &EventKind) -> f64 {
172        let joint = self.cooccur_decayed(a, b);
173        let row_total: f64 = self.cooccur.keys()
174            .filter(|(x, _)| x == a)
175            .map(|(x, y)| self.cooccur_decayed(x, y))
176            .sum();
177        if row_total <= 0.0 { 0.0 } else { joint / row_total }
178    }
179
180    /// Información mutua puntual entre `a` y `b` con decay aplicado:
181    /// PMI(a, b) = log₂( P(a, b) / (P(a) · P(b)) ).
182    /// Positivo → más correlacionados de lo que sugiere independencia.
183    pub fn pmi(&self, a: &EventKind, b: &EventKind) -> f64 {
184        // Total decayed: suma de marginales con decay (no usamos self.total
185        // directo porque debería ser consistente con los decayed values).
186        let total_decayed: f64 = self.marginal.keys()
187            .map(|k| self.marginal_decayed(k))
188            .sum();
189        if total_decayed <= 0.0 { return 0.0; }
190        let joint = self.cooccur_decayed(a, b) / total_decayed;
191        let pa = self.marginal_decayed(a) / total_decayed;
192        let pb = self.marginal_decayed(b) / total_decayed;
193        if joint <= 0.0 || pa <= 0.0 || pb <= 0.0 { return 0.0; }
194        (joint / (pa * pb)).log2()
195    }
196
197    /// Información mutua acumulada de la pareja (a, b) ponderada por su
198    /// probabilidad conjunta. Útil como medida de "interés" del par.
199    pub fn weighted_pmi(&self, a: &EventKind, b: &EventKind) -> f64 {
200        if self.total == 0 { return 0.0; }
201        let joint = self.cooccur
202            .get(&(a.clone(), b.clone()))
203            .copied()
204            .unwrap_or(0) as f64 / self.total as f64;
205        joint * self.pmi(a, b)
206    }
207
208    pub fn marginals(&self) -> &HashMap<EventKind, u64> { &self.marginal }
209    pub fn cooccurrences(&self) -> &HashMap<(EventKind, EventKind), u64> { &self.cooccur }
210    pub fn total(&self) -> u64 { self.total }
211    pub fn window_size(&self) -> usize { self.window_size }
212    pub fn current_window(&self) -> usize { self.window.len() }
213
214    /// Últimos N eventos del window, en orden cronológico (más viejo primero).
215    /// Si N > window.len(), devuelve todo el window.
216    pub fn recent(&self, n: usize) -> impl Iterator<Item = &TimedEvent> {
217        let start = self.window.len().saturating_sub(n);
218        self.window.range(start..)
219    }
220
221    pub fn gap_histograms(&self) -> &HashMap<(EventKind, EventKind), GapHistogram> {
222        &self.gap_histograms
223    }
224
225    /// Top-K pares por count del histograma (más frecuentes primero).
226    /// Útil para limitar cardinalidad de métricas exportadas.
227    pub fn top_gap_pairs(&self, k: usize) -> Vec<(&(EventKind, EventKind), &GapHistogram)> {
228        let mut pairs: Vec<_> = self.gap_histograms.iter().collect();
229        pairs.sort_by(|a, b| b.1.count.cmp(&a.1.count));
230        pairs.truncate(k);
231        pairs
232    }
233}
234
235#[cfg(test)]
236mod tests {
237    use super::*;
238    use crate::rules::EventKind::*;
239
240    #[test]
241    fn entropy_zero_for_single_event() {
242        let mut obs = Observer::new(10);
243        for _ in 0..5 { obs.record(EnteSpawned); }
244        // Distribución degenerada: una sola observación posible → H = 0.
245        assert!(obs.shannon_entropy() < 1e-9);
246    }
247
248    #[test]
249    fn entropy_one_for_balanced_binary() {
250        let mut obs = Observer::new(100);
251        for _ in 0..10 { obs.record(EnteSpawned); }
252        for _ in 0..10 { obs.record(EnteDied); }
253        // Bernoulli(0.5) → H = 1 bit.
254        assert!((obs.shannon_entropy() - 1.0).abs() < 1e-9);
255    }
256
257    #[test]
258    fn conditional_prob_perfect_dependency() {
259        let mut obs = Observer::new(100);
260        // Spawned siempre seguido por Died.
261        for _ in 0..5 {
262            obs.record(EnteSpawned);
263            obs.record(EnteDied);
264        }
265        let p = obs.conditional_prob(&EnteSpawned, &EnteDied);
266        assert!(p > 0.0, "esperamos correlación positiva, got {p}");
267    }
268}