navi_vm/

compactor.rs

//! Incremental event compactor for tracking realtime bar lifecycle.
//!
//! [`EventCompactor`] sits between an [`Event`] stream and a downstream sink,
//! forwarding events according to [`EventCompactionRules`] and notifying the
//! sink whenever a realtime bar becomes *confirmed* (closed).

use navi_types::BarStartEvent;

use crate::{BarState, Event};

/// Configuration rules for [`EventCompactor`].
#[derive(Debug, Clone, Default)]
pub struct EventCompactionRules {
    /// When `true`, [`Event::Log`] events are dropped instead of forwarded.
    pub filter_log: bool,
    /// When `true`, [`Event::Alert`] events are dropped instead of forwarded.
    pub filter_alert: bool,
}

/// Reason why a realtime bar was confirmed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RealtimeConfirmReason {
    /// A new realtime bar with a different timestamp arrived, closing the
    /// previous one.
    Rollover,
    /// The stream ended while a realtime bar was still open; confirmed at
    /// [`EventCompactor::finish`].
    StreamEnd,
}

/// Information about a confirmed realtime bar.
#[derive(Debug, Clone)]
pub struct RealtimeConfirmedInfo {
    /// Zero-based bar index of the confirmed bar.
    pub bar_index: usize,
    /// Unix-millisecond timestamp of the confirmed bar.
    pub bar_time: i64,
    /// Reason the bar was confirmed.
    pub reason: RealtimeConfirmReason,
    /// Number of events forwarded to the sink for this bar since its
    /// [`BarState::RealtimeNew`] phase began.
    ///
    /// Does **not** include stream-level events (`ScriptInfo`, `Warning`,
    /// `HistoryEnd`) that were emitted outside of any bar.
    ///
    /// ## Use cases
    ///
    /// - **Trimming intermediate state**: when the sink appends events to a
    ///   buffer, `event_count` tells how many trailing entries belong to this
    ///   bar.  The caller can archive or delete them after confirmation.
    /// - **Batch commit**: after a `RealtimeConfirmed` notification, the sink
    ///   can atomically commit or discard the last `event_count` writes.
    /// - **Audit/replay**: `event_count` gives the total per-bar event count
    ///   for indexing or storage partitioning.
    pub event_count: usize,
}

/// Callback interface for [`EventCompactor`].
pub trait EventCompactionSink {
    /// Called for every event that passes the compaction rules.
    fn on_event(&mut self, event: Event);
    /// Called when a realtime bar transitions to confirmed.
    fn on_realtime_confirmed(&mut self, info: RealtimeConfirmedInfo);
}

/// Per-realtime-bar tracking state.
struct RealtimeBarTracking {
    bar_index: usize,
    bar_time: i64,
    event_count: usize,
}

/// Incremental compactor that tracks realtime bar lifecycle.
///
/// Feed events via [`push`](Self::push) and call [`finish`](Self::finish) when
/// the stream ends.  The compactor:
///
/// - Forwards history events unchanged.
/// - Counts events per realtime bar and notifies
///   [`EventCompactionSink::on_realtime_confirmed`] on rollover or stream end.
/// - Optionally drops [`Event::Log`] and/or [`Event::Alert`] events.
pub struct EventCompactor {
    rules: EventCompactionRules,
    current_realtime: Option<RealtimeBarTracking>,
}

impl EventCompactor {
    /// Create a new compactor with the given rules.
    pub fn new(rules: EventCompactionRules) -> Self {
        Self {
            rules,
            current_realtime: None,
        }
    }

    /// Push one event through the compactor.
    ///
    /// The event is forwarded to `sink.on_event` (unless filtered) and the
    /// realtime bar state is updated accordingly.  When a new realtime bar
    /// rolls over the previous one, `sink.on_realtime_confirmed` is called
    /// before forwarding the new `BarStart`.
    pub fn push<S>(&mut self, event: Event, sink: &mut S)
    where
        S: EventCompactionSink,
    {
        match &event {
            Event::BarStart(bs) => self.handle_bar_start(bs.clone(), event, sink),

            Event::BarEnd => {
                if let Some(rt) = &mut self.current_realtime {
                    rt.event_count += 1;
                }
                sink.on_event(event);
            }

            // Stream-level events — forward, do not count.
            Event::SessionInfo { .. } | Event::HistoryEnd | Event::Warning(_) => {
                sink.on_event(event);
            }

            Event::Log(_) => {
                if !self.rules.filter_log {
                    if let Some(rt) = &mut self.current_realtime {
                        rt.event_count += 1;
                    }
                    sink.on_event(event);
                }
            }

            Event::Alert(_) => {
                if !self.rules.filter_alert {
                    if let Some(rt) = &mut self.current_realtime {
                        rt.event_count += 1;
                    }
                    sink.on_event(event);
                }
            }

            Event::Draw(_) | Event::Strategy(_) => {
                if let Some(rt) = &mut self.current_realtime {
                    rt.event_count += 1;
                }
                sink.on_event(event);
            }
        }
    }

    /// Signal stream completion.
    ///
    /// If a realtime bar is still open, it is confirmed with
    /// [`RealtimeConfirmReason::StreamEnd`].
    pub fn finish<S>(&mut self, sink: &mut S)
    where
        S: EventCompactionSink,
    {
        self.emit_confirmed(RealtimeConfirmReason::StreamEnd, sink);
    }

    fn handle_bar_start<S>(&mut self, bs: BarStartEvent, event: Event, sink: &mut S)
    where
        S: EventCompactionSink,
    {
        match bs.bar_state {
            BarState::History => {
                // History bars are not tracked; just forward.
                sink.on_event(event);
            }

            BarState::RealtimeNew => {
                // Check for rollover: a new realtime bar with a different
                // timestamp means the previous one is confirmed.
                let rollover = self
                    .current_realtime
                    .as_ref()
                    .is_some_and(|rt| rt.bar_time != bs.candlestick.time);
                if rollover {
                    self.emit_confirmed(RealtimeConfirmReason::Rollover, sink);
                }
                // Start tracking the new bar.
                self.current_realtime = Some(RealtimeBarTracking {
                    bar_index: bs.bar_index,
                    bar_time: bs.candlestick.time,
                    event_count: 1, // count this BarStart
                });
                sink.on_event(event);
            }

            BarState::RealtimeUpdate | BarState::RealtimeConfirmed => {
                if let Some(rt) = &mut self.current_realtime {
                    rt.event_count += 1;
                }
                sink.on_event(event);
            }
        }
    }

    fn emit_confirmed<S>(&mut self, reason: RealtimeConfirmReason, sink: &mut S)
    where
        S: EventCompactionSink,
    {
        if let Some(rt) = self.current_realtime.take() {
            sink.on_realtime_confirmed(RealtimeConfirmedInfo {
                bar_index: rt.bar_index,
                bar_time: rt.bar_time,
                reason,
                event_count: rt.event_count,
            });
        }
    }
}

#[cfg(test)]
mod tests {
    use navi_types::{BarStartEvent, BarState, Candlestick};
    use navi_visuals::DrawEvent;

    use super::*;
    use crate::events::{AlertEvent, LogEvent, LogLevel};

    // ── Helpers ─────────────────────────────────────────────────────────────

    fn make_candlestick(time: i64) -> Candlestick {
        use crate::TradeSession;
        Candlestick::new(time, 1.0, 2.0, 0.5, 1.5, 1000.0, 0.0, TradeSession::Regular)
    }

    fn bar_start(bar_index: usize, time: i64, state: BarState) -> Event {
        Event::BarStart(BarStartEvent {
            bar_index,
            candlestick: make_candlestick(time),
            bar_state: state,
        })
    }

    /// A simple collecting sink used in tests.
    #[derive(Default)]
    struct CollectSink {
        events: Vec<Event>,
        confirmations: Vec<RealtimeConfirmedInfo>,
    }

    impl EventCompactionSink for CollectSink {
        fn on_event(&mut self, event: Event) {
            self.events.push(event);
        }
        fn on_realtime_confirmed(&mut self, info: RealtimeConfirmedInfo) {
            self.confirmations.push(info);
        }
    }

    const T0: i64 = 1_700_000_000_000;
    const DAY: i64 = 86_400_000;

    // ── Test: history bars pass through without triggering confirmations ────

    #[test]
    fn history_bars_pass_through() {
        let mut compactor = EventCompactor::new(EventCompactionRules::default());
        let mut sink = CollectSink::default();

        compactor.push(bar_start(0, T0, BarState::History), &mut sink);
        compactor.push(Event::BarEnd, &mut sink);
        compactor.push(bar_start(1, T0 + DAY, BarState::History), &mut sink);
        compactor.push(Event::BarEnd, &mut sink);
        compactor.finish(&mut sink);

        assert_eq!(sink.events.len(), 4);
        assert!(
            sink.confirmations.is_empty(),
            "no realtime bars → no confirmations"
        );
    }

    // ── Test: realtime rollover triggers Rollover confirmation ────────────

    #[test]
    fn realtime_rollover() {
        let mut compactor = EventCompactor::new(EventCompactionRules::default());
        let mut sink = CollectSink::default();

        // Bar 1: RealtimeNew then RealtimeUpdate then RealtimeConfirmed
        compactor.push(bar_start(1, T0 + DAY, BarState::RealtimeNew), &mut sink);
        compactor.push(Event::BarEnd, &mut sink);
        compactor.push(bar_start(1, T0 + DAY, BarState::RealtimeUpdate), &mut sink);
        compactor.push(Event::BarEnd, &mut sink);
        compactor.push(
            bar_start(1, T0 + DAY, BarState::RealtimeConfirmed),
            &mut sink,
        );
        compactor.push(Event::BarEnd, &mut sink);

        // No confirmation yet (still on same bar).
        assert!(sink.confirmations.is_empty());

        // Bar 2: new timestamp → rollover
        compactor.push(bar_start(2, T0 + DAY * 2, BarState::RealtimeNew), &mut sink);
        compactor.push(Event::BarEnd, &mut sink);

        // Exactly 1 Rollover confirmation for bar 1.
        assert_eq!(sink.confirmations.len(), 1);
        let c = &sink.confirmations[0];
        assert_eq!(c.bar_index, 1);
        assert_eq!(c.bar_time, T0 + DAY);
        assert_eq!(c.reason, RealtimeConfirmReason::Rollover);
        // 3 BarStarts + 3 BarEnds = 6 events
        assert_eq!(c.event_count, 6);

        // Finish closes bar 2.
        compactor.finish(&mut sink);
        assert_eq!(sink.confirmations.len(), 2);
        assert_eq!(
            sink.confirmations[1].reason,
            RealtimeConfirmReason::StreamEnd
        );
    }

    // ── Test: stream end triggers StreamEnd confirmation ──────────────────

    #[test]
    fn realtime_stream_end() {
        let mut compactor = EventCompactor::new(EventCompactionRules::default());
        let mut sink = CollectSink::default();

        compactor.push(bar_start(0, T0, BarState::RealtimeNew), &mut sink);
        compactor.push(Event::BarEnd, &mut sink);
        compactor.finish(&mut sink);

        assert_eq!(sink.confirmations.len(), 1);
        assert_eq!(
            sink.confirmations[0].reason,
            RealtimeConfirmReason::StreamEnd
        );
        assert_eq!(sink.confirmations[0].event_count, 2); // BarStart + BarEnd
    }

    // ── Test: filter_log=true drops Log events ────────────────────────────

    #[test]
    fn filter_log_drops_event() {
        let rules = EventCompactionRules {
            filter_log: true,
            ..Default::default()
        };
        let mut compactor = EventCompactor::new(rules);
        let mut sink = CollectSink::default();

        compactor.push(bar_start(0, T0, BarState::RealtimeNew), &mut sink);
        compactor.push(
            Event::Log(LogEvent {
                level: LogLevel::Info,
                message: "hello".into(),
            }),
            &mut sink,
        );
        compactor.push(Event::BarEnd, &mut sink);
        compactor.finish(&mut sink);

        // Log event should be dropped.
        let log_count = sink
            .events
            .iter()
            .filter(|e| matches!(e, Event::Log(_)))
            .count();
        assert_eq!(log_count, 0);

        // event_count should not include the dropped log.
        assert_eq!(sink.confirmations[0].event_count, 2); // BarStart + BarEnd
    }

    // ── Test: bar-outside events are not counted ──────────────────────────

    #[test]
    fn bar_outside_events_not_counted() {
        let mut compactor = EventCompactor::new(EventCompactionRules::default());
        let mut sink = CollectSink::default();

        // HistoryEnd is a stream-level event emitted outside bars.
        compactor.push(Event::HistoryEnd, &mut sink);

        // Now start a realtime bar.
        compactor.push(bar_start(0, T0, BarState::RealtimeNew), &mut sink);
        compactor.push(Event::BarEnd, &mut sink);
        compactor.finish(&mut sink);

        // HistoryEnd is forwarded but not counted.
        let forwarded_outside: Vec<_> = sink
            .events
            .iter()
            .filter(|e| matches!(e, Event::HistoryEnd))
            .collect();
        assert_eq!(forwarded_outside.len(), 1);

        // Only BarStart + BarEnd counted.
        assert_eq!(sink.confirmations[0].event_count, 2);
    }

    // ── Test: filter_alert=true drops Alert events ────────────────────────

    #[test]
    fn filter_alert_drops_event() {
        let rules = EventCompactionRules {
            filter_alert: true,
            ..Default::default()
        };
        let mut compactor = EventCompactor::new(rules);
        let mut sink = CollectSink::default();

        compactor.push(bar_start(0, T0, BarState::RealtimeNew), &mut sink);
        compactor.push(
            Event::Alert(AlertEvent {
                id: None,
                message: "alert!".into(),
            }),
            &mut sink,
        );
        compactor.push(Event::BarEnd, &mut sink);
        compactor.finish(&mut sink);

        let alert_count = sink
            .events
            .iter()
            .filter(|e| matches!(e, Event::Alert(_)))
            .count();
        assert_eq!(alert_count, 0);
        assert_eq!(sink.confirmations[0].event_count, 2);
    }

    // ── Test: Draw events are counted ─────────────────────────────────────

    #[test]
    fn draw_events_counted_in_realtime() {
        let mut compactor = EventCompactor::new(EventCompactionRules::default());
        let mut sink = CollectSink::default();

        compactor.push(bar_start(0, T0, BarState::RealtimeNew), &mut sink);
        compactor.push(Event::Draw(DrawEvent::NewBar { bar_index: 0 }), &mut sink);
        compactor.push(Event::BarEnd, &mut sink);
        compactor.finish(&mut sink);

        // BarStart + Draw + BarEnd = 3
        assert_eq!(sink.confirmations[0].event_count, 3);
    }

    // ── Test: history events do not start realtime tracking ───────────────

    #[test]
    fn history_does_not_start_realtime_tracking() {
        let mut compactor = EventCompactor::new(EventCompactionRules::default());
        let mut sink = CollectSink::default();

        compactor.push(bar_start(0, T0, BarState::History), &mut sink);
        compactor.push(Event::BarEnd, &mut sink);
        compactor.finish(&mut sink);

        assert!(sink.confirmations.is_empty());
        assert!(compactor.current_realtime.is_none());
    }

    // ── Test: filter_log=true does not affect ChartAccumulator output ─────

    #[test]
    fn equivalence_chart_accumulator() {
        use navi_visuals::{ChartAccumulator, ChartEvent};

        let events_with_log = vec![
            Event::HistoryEnd,
            bar_start(0, T0, BarState::RealtimeNew),
            Event::Draw(DrawEvent::NewBar { bar_index: 0 }),
            Event::Log(LogEvent {
                level: LogLevel::Info,
                message: "noise".into(),
            }),
            Event::BarEnd,
        ];
        let events_without_log: Vec<Event> = events_with_log
            .iter()
            .filter(|e| !matches!(e, Event::Log(_)))
            .cloned()
            .collect();

        // Reference: apply events (minus Log) directly to ChartAccumulator.
        let mut ref_acc = ChartAccumulator::new();
        for event in events_without_log {
            if let Ok(ce) = ChartEvent::try_from(event) {
                ref_acc.apply(ce);
            }
        }

        // Compactor path: feed all events with filter_log=true.
        struct ChartSink {
            acc: ChartAccumulator,
        }
        impl EventCompactionSink for ChartSink {
            fn on_event(&mut self, event: Event) {
                if let Ok(ce) = ChartEvent::try_from(event) {
                    self.acc.apply(ce);
                }
            }
            fn on_realtime_confirmed(&mut self, _: RealtimeConfirmedInfo) {}
        }
        let mut sink = ChartSink {
            acc: ChartAccumulator::new(),
        };
        let mut compactor = EventCompactor::new(EventCompactionRules {
            filter_log: true,
            ..Default::default()
        });
        for event in events_with_log {
            compactor.push(event, &mut sink);
        }
        compactor.finish(&mut sink);

        assert_eq!(
            format!("{:?}", ref_acc.chart()),
            format!("{:?}", sink.acc.chart()),
            "chart state must be identical whether Log events are present or filtered"
        );
    }

    // ── Test: filter_log=true does not affect StrategyAccumulator output ──

    #[test]
    fn equivalence_strategy_accumulator() {
        use navi_strategy::{StrategyAccumulator, StrategyEvent};

        let events_with_log = vec![
            Event::HistoryEnd,
            bar_start(0, T0, BarState::RealtimeNew),
            Event::Strategy(StrategyEvent::EquitySnapshot {
                equity: 1000.0,
                buy_hold_equity: 950.0,
            }),
            Event::Log(LogEvent {
                level: LogLevel::Info,
                message: "noise".into(),
            }),
            Event::BarEnd,
        ];
        let events_without_log: Vec<Event> = events_with_log
            .iter()
            .filter(|e| !matches!(e, Event::Log(_)))
            .cloned()
            .collect();

        // Reference: apply events (minus Log) directly to StrategyAccumulator.
        let mut ref_acc = StrategyAccumulator::new();
        for event in events_without_log {
            if let Event::Strategy(e) = event {
                ref_acc.apply(e);
            }
        }

        // Compactor path: feed all events with filter_log=true.
        struct StrategySink {
            acc: StrategyAccumulator,
        }
        impl EventCompactionSink for StrategySink {
            fn on_event(&mut self, event: Event) {
                if let Event::Strategy(e) = event {
                    self.acc.apply(e);
                }
            }
            fn on_realtime_confirmed(&mut self, _: RealtimeConfirmedInfo) {}
        }
        let mut sink = StrategySink {
            acc: StrategyAccumulator::new(),
        };
        let mut compactor = EventCompactor::new(EventCompactionRules {
            filter_log: true,
            ..Default::default()
        });
        for event in events_with_log {
            compactor.push(event, &mut sink);
        }
        compactor.finish(&mut sink);

        assert_eq!(
            format!("{:?}", ref_acc.report()),
            format!("{:?}", sink.acc.report()),
            "strategy report must be identical whether Log events are present or filtered"
        );
    }
}