事件压缩器

EventCompactor 是 Instance::run() 产出的 Event 流的增量过滤器与生命周期追踪器。专为服务端集成设计：可在实时 K 线确认（收盘）时通知调用方，并可选过滤 Log/Alert 噪音，再持久化事件。

适用场景

以下场景适合使用 EventCompactor：

• 持久化事件时剔除实时中间态。每根实时 K 线在每个 tick 都会产生一套完整的事件周期（BarStart → 绘图事件 → BarEnd）。大多数持久化场景只需要最终状态；event_count 精确告知缓冲区末尾有多少条属于刚刚确认的 K 线，可据此覆盖或丢弃之前的中间态记录。
• 将事件持久化到数据库或消息队列，需要明确的每根实时 K 线提交点。
• 向下游消费者转发事件，希望剔除 Log/Alert 噪音。
• 需要知道实时 K 线何时收盘，以便触发批量操作。

如果只是本地简单使用，无需 EventCompactor，直接消费 RunHandle 流即可。

核心类型

EventCompactionRules

pub struct EventCompactionRules {
    /// 丢弃 `Event::Log` 事件，不转发。默认：`false`。
    pub filter_log: bool,
    /// 丢弃 `Event::Alert` 事件，不转发。默认：`false`。
    pub filter_alert: bool,
}

RealtimeConfirmReason

pub enum RealtimeConfirmReason {
    /// 收到时间戳不同的新实时 K 线，上一根自动闭合。
    Rollover,
    /// 流结束时仍有未确认的实时 K 线，在 `finish()` 时闭合。
    StreamEnd,
}

RealtimeConfirmedInfo

实时 K 线确认时传递给 EventCompactionSink::on_realtime_confirmed：

pub struct RealtimeConfirmedInfo {
    /// 已确认 K 线的从 0 开始的索引。
    pub bar_index: usize,
    /// 已确认 K 线的 Unix 毫秒时间戳。
    pub bar_time: i64,
    /// K 线确认的原因。
    pub reason: RealtimeConfirmReason,
    /// 自本根 K 线进入 `RealtimeNew` 阶段起，compactor 转发给 `on_event`
    /// 的事件数量（不含 `ScriptInfo`、`Warning`、`HistoryEnd` 等流级别事件）。
    pub event_count: usize,
}

event_count 是管理中间状态的核心指标，详见使用场景。

EventCompactionSink

pub trait EventCompactionSink {
    /// 每个通过过滤规则的事件都会调用此方法。
    fn on_event(&mut self, event: Event);
    /// 实时 K 线确认时调用此方法。
    fn on_realtime_confirmed(&mut self, info: RealtimeConfirmedInfo);
}

创建 Compactor

use navi_vm::{EventCompactor, EventCompactionRules};

let rules = EventCompactionRules {
    filter_log: true,    // 过滤 log.*() 噪音
    filter_alert: false, // 保留 alert() 事件
};
let mut compactor = EventCompactor::new(rules);

推送事件

对 RunHandle 的每个事件调用 push，流结束后调用 finish：

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!(
            "K 线 {} 已确认（{:?}），共 {} 个事件",
            info.bar_index, info.reason, info.event_count
        );
        // 提交或归档 self.buffer 中最后 event_count 条记录
    }
}

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

事件路由

compactor 按如下规则路由事件：

事件	是否转发	是否计入 event_count
ScriptInfo	始终转发	否——流级别
Warning	始终转发	否——流级别
HistoryEnd	始终转发	否——流级别
BarStart（History）	始终转发	否
BarStart（RealtimeNew/Update/Confirmed）	始终转发	是
BarEnd	始终转发	是（实时阶段时）
Draw	始终转发	是（实时阶段时）
Strategy	始终转发	是（实时阶段时）
Log	仅 filter_log = false 时	转发时计入
Alert	仅 filter_alert = false 时	转发时计入

Rollover 行为

当收到 bar_state = RealtimeNew 且时间戳不同于当前实时 K 线的 BarStart 时，compactor：

1. 以 reason = Rollover 调用 on_realtime_confirmed，通知前一根 K 线已确认。
2. 开始追踪新 K 线。

这与 VM 自身的 rollover 逻辑一致：时间戳更晚的新 tick 会隐式关闭前一根 K 线。

push BarStart(N, RealtimeNew, t=T0)    → on_event，开始追踪 K 线 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(K 线 N, Rollover, count=4)
                                          on_event，开始追踪 K 线 N+1

使用场景

清理中间状态

向缓冲区追加事件时，event_count 告知末尾有多少条属于刚刚确认的 K 线，可将其删除或覆盖，只保留最终状态：

fn on_realtime_confirmed(&mut self, info: RealtimeConfirmedInfo) {
    let keep_from = self.buffer.len().saturating_sub(info.event_count);
    // 丢弃中间更新，只保留已确认 K 线的事件
    self.buffer.drain(..keep_from);
}

数据库批量提交

确认后原子提交最后 event_count 行：

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

审计追踪

使用 bar_index 和 event_count 按 K 线分区存储事件，便于后续回放或调试。

与 ChartAccumulator 组合

EventCompactor 和 ChartAccumulator 可自然组合——在 sink 中驱动 accumulator 的同时追踪生命周期：

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!("实时 K 线 {} 已收盘（{:?}）", info.bar_index, info.reason);
    }
}

下一步

• 策略报告 — 查询交易统计与绩效指标
• 错误处理 — 处理 VM 的编译与运行时错误