tawasuyu/pineal

1//! Aggregation de OHLC por bucket de **tiempo** (no de índice).
2//!
3//! Bucket index = `floor((bar.t - t_start) / bucket_duration)`.
4//! Cuando cambia el bucket, commit del anterior:
5//!
6//! - `open` = primer `open` del bucket.
7//! - `close` = último `close` del bucket.
8//! - `high` = max(`high`) del bucket.
9//! - `low` = min(`low`) del bucket.
10//! - `volume` = sum(`volume`) del bucket.
11//! - `t` = timestamp del primer bar del bucket (canónico para
12//!   ploteo; el doc original sugiere usar el inicio del bucket
13//!   pero acá preferimos el sample real para no introducir bias).
14//!
15//! Buckets vacíos no se emiten — el length de salida es ≤ inputs.
16//! Fallback a index-bucketing si el span temporal es cero (todos
17//! los timestamps colapsados, e.g. tick-data).
18
19use crate::ohlc_buffer::{Bar, OhlcBuffer, STRIDE};
20
21/// Agrega `src` en buckets de duración `bucket_duration` (en
22/// unidades de `bar.t`). Escribe el output en `out` extendiéndolo
23/// (no se clearea; el caller decide).
24///
25/// Si `bucket_duration <= 0` o el span del input es cero, hace
26/// fallback a index-bucketing con `samples_per_bucket = 1` (es decir,
27/// copia el input tal cual). Esto evita panic con tick-data
28/// colapsado.
29pub fn aggregate_time_bucketed(src: &OhlcBuffer, bucket_duration: f32, out: &mut OhlcBuffer) {
30    if src.is_empty() {
31        return;
32    }
33    let n = src.len();
34    let (t_first, t_last) = src.time_range().unwrap();
35
36    if bucket_duration <= 0.0 || (t_last - t_first).abs() < f32::EPSILON {
37        // Fallback: copia tal cual.
38        for i in 0..n {
39            out.push_bar(src.bar(i));
40        }
41        return;
42    }
43
44    let mut current_bucket = i64::MIN;
45    let mut acc_t: f32 = 0.0;
46    let mut acc_o: f32 = 0.0;
47    let mut acc_h: f32 = f32::NEG_INFINITY;
48    let mut acc_l: f32 = f32::INFINITY;
49    let mut acc_c: f32 = 0.0;
50    let mut acc_v: f32 = 0.0;
51    let mut has_acc = false;
52
53    for i in 0..n {
54        let b = src.bar(i);
55        let bucket = ((b.t - t_first) / bucket_duration).floor() as i64;
56        if bucket != current_bucket {
57            if has_acc {
58                out.push_bar(Bar {
59                    t: acc_t,
60                    o: acc_o,
61                    h: acc_h,
62                    l: acc_l,
63                    c: acc_c,
64                    v: acc_v,
65                });
66            }
67            current_bucket = bucket;
68            acc_t = b.t;
69            acc_o = b.o;
70            acc_h = b.h;
71            acc_l = b.l;
72            acc_c = b.c;
73            acc_v = b.v;
74            has_acc = true;
75        } else {
76            if b.h > acc_h {
77                acc_h = b.h;
78            }
79            if b.l < acc_l {
80                acc_l = b.l;
81            }
82            acc_c = b.c;
83            acc_v += b.v;
84        }
85    }
86    if has_acc {
87        out.push_bar(Bar {
88            t: acc_t,
89            o: acc_o,
90            h: acc_h,
91            l: acc_l,
92            c: acc_c,
93            v: acc_v,
94        });
95    }
96
97    // Una métrica de cordura: el output nunca puede ser más largo
98    // que el input.
99    debug_assert!(out.bars().len() / STRIDE <= n);
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105
106    fn fixture() -> OhlcBuffer {
107        // 10 bars con t en `[0, 9]`, valores deterministicos.
108        let mut b = OhlcBuffer::with_capacity(10);
109        for i in 0..10 {
110            let t = i as f32;
111            let base = 100.0 + (i as f32) * 0.5;
112            b.push_values(t, base, base + 1.0, base - 1.0, base + 0.2, 10.0);
113        }
114        b
115    }
116
117    #[test]
118    fn bucket_de_3_agrega_a_4_bars() {
119        // 10 inputs con t en `[0, 9]`, bucket 3 → buckets 0-2, 3-5, 6-8, 9.
120        // = 4 buckets.
121        let src = fixture();
122        let mut out = OhlcBuffer::new();
123        aggregate_time_bucketed(&src, 3.0, &mut out);
124        assert_eq!(out.len(), 4);
125    }
126
127    #[test]
128    fn aggregation_preserva_volatilidad() {
129        // Inventamos un bucket donde un bar tiene spike alto y otro
130        // spike bajo. El aggregate debe capturar AMBOS extremos.
131        let mut src = OhlcBuffer::new();
132        src.push_values(0.0, 10.0, 12.0, 9.0, 11.0, 5.0);
133        src.push_values(0.5, 11.0, 20.0, 10.5, 11.5, 5.0); // spike up
134        src.push_values(0.8, 11.5, 12.0, 2.0, 11.0, 5.0); // spike down
135        let mut out = OhlcBuffer::new();
136        aggregate_time_bucketed(&src, 1.0, &mut out);
137        assert_eq!(out.len(), 1);
138        let agg = out.bar(0);
139        assert_eq!(agg.h, 20.0, "max H debe sobrevivir");
140        assert_eq!(agg.l, 2.0, "min L debe sobrevivir");
141        assert_eq!(agg.o, 10.0, "first open");
142        assert_eq!(agg.c, 11.0, "last close");
143        assert_eq!(agg.v, 15.0, "sum volumes");
144    }
145
146    #[test]
147    fn fallback_a_index_si_span_cero() {
148        // Todos los t iguales — fallback copia 1:1.
149        let mut src = OhlcBuffer::new();
150        src.push_values(7.0, 1.0, 2.0, 0.0, 1.5, 1.0);
151        src.push_values(7.0, 1.5, 2.5, 1.0, 2.0, 1.0);
152        src.push_values(7.0, 2.0, 3.0, 1.0, 1.0, 1.0);
153        let mut out = OhlcBuffer::new();
154        aggregate_time_bucketed(&src, 1.0, &mut out);
155        assert_eq!(out.len(), 3, "span 0 ⇒ copy 1:1");
156    }
157
158    #[test]
159    fn empty_no_emite() {
160        let src = OhlcBuffer::new();
161        let mut out = OhlcBuffer::new();
162        aggregate_time_bucketed(&src, 1.0, &mut out);
163        assert_eq!(out.len(), 0);
164    }
165}