Event Compactor

EventCompactor is an incremental filter and lifecycle tracker for the Event stream produced by Instance::run(). It is designed for server-side integrations that need to track when realtime bars are confirmed (closed) and optionally suppress Log/Alert noise before persisting events.

When to Use

Use EventCompactor when:

• You are persisting events and want to discard intermediate realtime updates. Each realtime bar produces a full event cycle on every tick (BarStart → draws → BarEnd). For most persistence use cases only the final cycle matters; event_count tells you exactly how many trailing entries in your buffer belong to the just-confirmed bar so you can overwrite or discard the earlier ones.
• You are persisting events to a database or message queue and need a clear per-bar commit point.
• You are forwarding events to downstream consumers and want to strip noisy Log/Alert events.
• You need to know when a realtime bar is closed so you can trigger batch operations.

For simple local use, you do not need EventCompactor — just consume the RunHandle stream directly.

Core Types

EventCompactionRules

pub struct EventCompactionRules {
    /// Drop `Event::Log` events instead of forwarding them. Default: `false`.
    pub filter_log: bool,
    /// Drop `Event::Alert` events instead of forwarding them. Default: `false`.
    pub filter_alert: bool,
}

RealtimeConfirmReason

pub enum RealtimeConfirmReason {
    /// A new realtime bar with a different timestamp arrived, confirming the previous one.
    Rollover,
    /// The stream ended while a realtime bar was still open; confirmed at `finish()`.
    StreamEnd,
}

RealtimeConfirmedInfo

Passed to EventCompactionSink::on_realtime_confirmed when a realtime bar is confirmed:

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,
    /// Why the bar was confirmed.
    pub reason: RealtimeConfirmReason,
    /// Number of events forwarded to `on_event` for this bar since its
    /// `RealtimeNew` phase began (not counting stream-level events like
    /// `ScriptInfo`, `Warning`, or `HistoryEnd`).
    pub event_count: usize,
}

event_count is the key metric for managing intermediate state. See Use cases below.

EventCompactionSink

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);
}

Creating a Compactor

use navi_vm::{EventCompactor, EventCompactionRules};

let rules = EventCompactionRules {
    filter_log: true,    // suppress log.*() noise
    filter_alert: false, // keep alert() events
};
let mut compactor = EventCompactor::new(rules);

Feeding Events

Call push for each event from RunHandle, then finish when the stream ends:

use navi_vm::{Event, EventCompactor, EventCompactionRules, EventCompactionSink,
                  RealtimeConfirmedInfo};

struct MySink {
    buffer: Vec<Event>,
}

impl EventCompactionSink for MySink {
    fn on_event(&mut self, event: Event) {
        self.buffer.push(event);
    }
    fn on_realtime_confirmed(&mut self, info: RealtimeConfirmedInfo) {
        println!(
            "bar {} confirmed ({:?}), {} events",
            info.bar_index, info.reason, info.event_count
        );
        // commit or archive the last `info.event_count` entries in self.buffer
    }
}

let mut sink = MySink { buffer: Vec::new() };
let mut compactor = EventCompactor::new(EventCompactionRules::default());

let mut handle = instance.run(symbol, tf);
while let Some(result) = handle.next_event().await {
    compactor.push(result?, &mut sink);
}
compactor.finish(&mut sink);

Event Routing

The compactor routes events as follows:

Event	Forwarded	Counted toward event_count
ScriptInfo	Always	No — stream-level
Warning	Always	No — stream-level
HistoryEnd	Always	No — stream-level
BarStart (History)	Always	No
BarStart (RealtimeNew/Update/Confirmed)	Always	Yes
BarEnd	Always	Yes (when in realtime)
Draw	Always	Yes (when in realtime)
Strategy	Always	Yes (when in realtime)
Log	Only if filter_log = false	Yes if forwarded
Alert	Only if filter_alert = false	Yes if forwarded

Rollover Behaviour

When a BarStart with bar_state = RealtimeNew arrives with a different timestamp from the current realtime bar, the compactor:

1. Calls on_realtime_confirmed with reason = Rollover for the previous bar.
2. Starts tracking the new bar.

This mirrors the VM's own rollover logic: a new realtime tick with a later timestamp implicitly closes the previous bar.

push BarStart(N, RealtimeNew, t=T0)   → on_event, start tracking bar N
push BarEnd                            → on_event, event_count = 2
push BarStart(N, RealtimeUpdate, t=T0) → on_event, event_count = 3
push BarEnd                            → on_event, event_count = 4
push BarStart(N+1, RealtimeNew, t=T1)  → on_realtime_confirmed(bar N, Rollover, count=4)
                                         on_event, start tracking bar N+1

Use Cases

Trimming Intermediate State

When appending events to a buffer or log, event_count tells you how many trailing entries belong to the just-confirmed bar. You can delete or overwrite them to keep only the final state:

fn on_realtime_confirmed(&mut self, info: RealtimeConfirmedInfo) {
    let keep_from = self.buffer.len().saturating_sub(info.event_count);
    // discard intermediate updates, keep only the confirmed bar's events
    self.buffer.drain(..keep_from);
}

Batch Database Commit

After confirmation, atomically commit the last event_count rows:

fn on_realtime_confirmed(&mut self, info: RealtimeConfirmedInfo) {
    let start = self.pending.len().saturating_sub(info.event_count);
    let confirmed_rows = &self.pending[start..];
    db.commit_batch(confirmed_rows, info.bar_index, info.bar_time);
    self.pending.truncate(start);
}

Audit Trail

Partition stored events by bar using bar_index and event_count for later replay or debugging.

Combining with ChartAccumulator

EventCompactor and ChartAccumulator compose naturally — use the compactor's sink to drive the accumulator while also tracking lifecycle:

use navi_vm::{
    Event, EventCompactor, EventCompactionRules, EventCompactionSink, RealtimeConfirmedInfo,
    visuals::{ChartAccumulator, ChartEvent},
};

struct ChartSink {
    acc: ChartAccumulator,
}

impl EventCompactionSink for ChartSink {
    fn on_event(&mut self, event: Event) {
        if let Ok(chart_event) = ChartEvent::try_from(event) {
            self.acc.apply(chart_event);
        }
    }
    fn on_realtime_confirmed(&mut self, info: RealtimeConfirmedInfo) {
        println!("realtime bar {} closed ({:?})", info.bar_index, info.reason);
    }
}

Next Steps

• Strategy Report — querying trade statistics and performance metrics
• Error Handling — handling compile and runtime errors from the VM