本文深入解析链上Swap交易数据如何被聚合成K线数据，涵盖数据流处理架构、时间窗口聚合算法、状态持久化等核心技术。
在DeFi领域，K线（蜡烛图）是展示价格走势最直观的方式。然而，链上的Swap交易是离散的、不规则的单笔交易记录，如何将这些零散的交易数据实时聚合成标准K线？这是构建链上行情系统的核心问题。
本文将以实际项目 chains-agg-kline 为例，讲解完整的技术方案。
链上交易 → Substreams → Kafka → Timely Dataflow → K线 → InfluxDB ↓ Redis (状态缓存) ``` 核心组件： - **Kafka**：消息队列，解耦数据生产与消费 - **Timely Dataflow**：分布式数据流处理框架 - **Redis**：存储K线状态，支持断点续传 - **InfluxDB**：时序数据库，存储最终K线数据 --- ## 数据模型设计 ### Trade（原始交易） ```rust pub struct Trade { pub pool_id: String, // 流动性池ID pub signature: String, // 交易签名 pub block: i64, // 区块高度 pub num: i64, // 交易序号 pub price: f64, // 成交价格 pub amount: f64, // 成交数量 pub ms: i64, // 毫秒级时间戳 } ``` ### Kline（K线数据） ```rust pub struct Kline { pub pool_id: String, // 流动性池ID pub period: String, // 周期（1s/1m/1h） pub open: f64, // 开盘价 pub high: f64, // 最高价 pub low: f64, // 最低价 pub close: f64, // 收盘价 pub amount: f64, // 成交量 pub time: i64, // 周期起始时间戳 } ``` --- ## K线聚合算法 ### 时间窗口计算 不同周期的起始时间计算： ```rust pub enum KlinePeriod { Second, // 1秒 Minute, // 1分钟 Hour, // 1小时 } impl KlinePeriod { // 根据时间戳计算周期起始时间 pub fn start_time(&self, timestamp: i64) -> i64 { match self { KlinePeriod::Second => timestamp, KlinePeriod::Minute => timestamp - timestamp % 60, KlinePeriod::Hour => timestamp - timestamp % 3600, } } } ``` ### 聚合逻辑 ```rust impl Kline { // 从第一笔交易创建新K线 pub fn new(trade: &Trade, period: KlinePeriod) -> Self { Self { pool_id: trade.pool_id.clone(), period: period.as_str().to_string(), open: trade.price, // 第一笔价格为开盘价 high: trade.price, low: trade.price, close: trade.price, // 最新价格为收盘价 amount: trade.amount, time: period.start_time(trade.ms / 1000), } } // 后续交易更新K线 pub fn update(&mut self, trade: &Trade) -> Self { self.high = self.high.max(trade.price); // 更新最高价 self.low = self.low.min(trade.price); // 更新最低价 self.close = trade.price; // 更新收盘价 self.amount += trade.amount; // 累加成交量 self.clone() } } ``` ### 核心处理流程 ```rust pub fn process_trades(&mut self, trades: &[Trade]) -> Vec<Kline> { let mut klines = Vec::new(); let periods = KlinePeriod::all(); // [Second, Minute, Hour] for trade in trades { for &period in &periods { if let Some(last_kline) = self.get_last_kline(&trade.pool_id, period) { // 比较当前交易与K线的时间窗口 match period.start_time(last_kline.time) .cmp(&period.start_time(trade.ms / 1000)) { Equal => { // 同一时间窗口，更新现有K线 let new_kline = last_kline.clone().update(trade); self.update_last_kline(new_kline); } Less => { // 新时间窗口，输出旧K线，创建新K线 klines.push(last_kline.clone()); let new_kline = Kline::new(trade, period); self.update_last_kline(new_kline.clone()); klines.push(new_kline); } _ => {} } } else { // 首次创建K线 let new_kline = Kline::new(trade, period); self.update_last_kline(new_kline.clone()); klines.push(new_kline); } } } klines } ``` --- ## 流式处理架构 ### Kafka消费者 ```rust async fn consume_kafka(consumer: StreamConsumer, tx: Arc<broadcast::Sender<Trade>>) { consumer.subscribe(&["trades"]).expect("订阅失败"); loop { match consumer.recv().await { Ok(msg) => { if let Some(payload) = msg.payload() { if let Ok(trade) = serde_json::from_slice::<Trade>(payload) { tx.send(trade)?; // 异步提交偏移量 consumer.commit_message(&msg, CommitMode::Async); } } } Err(e) => eprintln!("Kafka消费错误: {}", e), } } } ``` ### Timely Dataflow处理 ```rust timely::execute(timely::Config::process(worker_num), move |worker| { let mut input: InputSession<u64, (Trade, u64, i32), isize> = InputSession::new(); let mut kline_processor = KlineProcessor::new(); worker.dataflow(|scope| { let collection = input.to_collection(scope); // 按 pool_id 分区 let exchanged = collection.inner.exchange(|((trade, _, _), _, _)| { let mut hasher = DefaultHasher::new(); trade.pool_id.hash(&mut hasher); hasher.finish() }); // 处理并生成K线 let klines = exchanged.map(move |((trade, _, _), _, _)| { kline_processor.process_trades(&[trade]) }); // 写入InfluxDB klines.inspect(|klines: &Vec<Kline>| { if !klines.is_empty() { InfluxDbClient::write_klines(klines.clone()); } }); }); }); ``` --- ## 状态持久化 ### Redis存储结构 ``` Key: kline:{pool_id}:{period}:last Fields: open, high, low, close, amount, time Key: trade:{pool_id}:last Fields: block, num, ms ``` ### 状态恢复机制 ```rust fn get_last_kline(&mut self, pool_id: &str, period: KlinePeriod) -> Option<&Kline> { let key = format!("kline:{}:{}:last", pool_id, period.as_str()); if !self.last_klines.contains_key(&key) { // 从Redis恢复状态 let redis_kline = RedisClient::hmget(&key, &[ "open", "high", "low", "close", "amount", "time" ])?; let kline = Kline { pool_id: pool_id.to_string(), period: period.to_string(), open: redis_kline[0].parse()?, high: redis_kline[1].parse()?, low: redis_kline[2].parse()?, close: redis_kline[3].parse()?, amount: redis_kline[4].parse()?, time: redis_kline[5].parse()?, }; if kline.amount > 0.0 && kline.time > 0 { self.last_klines.insert(key.clone(), kline); } } self.last_klines.get(&key) } ``` --- ## 背压控制 当消费速度跟不上生产速度时，需要背压控制防止内存溢出： ```rust // 信号量限制并发处理数量 let max_inflight = 1000; let semaphore = Arc::new(Semaphore::new(max_inflight)); // 监控队列大小 let queue_size = Arc::new(AtomicUsize::new(0)); let pause_consumer = Arc::new(AtomicBool::new(false)); // 背压监控任务 tokio::spawn(async move { loop { tokio::time::sleep(Duration::from_secs(1)).await; let current_size = queue_size.load(Ordering::Relaxed); if current_size > max_inflight * 8 / 10 { // 超过80%容量，暂停消费 pause_consumer.store(true, Ordering::SeqCst); } else if current_size < max_inflight * 5 / 10 { // 低于50%容量，恢复消费 pause_consumer.store(false, Ordering::SeqCst); } } }); ``` --- ## 性能优化要点 ### 1. 分区并行处理 按 `pool_id` 哈希分区，不同池子的K线计算并行执行： ```rust let exchanged = collection.inner.exchange(|((trade, _, _), _, _)| { let mut hasher = DefaultHasher::new(); trade.pool_id.hash(&mut hasher); hasher.finish() }); ``` ### 2. 批量写入 K线数据批量写入InfluxDB，减少网络开销： ```rust klines.inspect(|klines: &Vec<Kline>| { if !klines.is_empty() { InfluxDbClient::write_klines(klines.clone()); } }); ``` ### 3. 内存缓存 热点K线状态缓存在内存，减少Redis访问： ```rust pub struct KlineProcessor { pub last_klines: HashMap<String, Kline>, // 内存缓存 pub last_trades: HashMap<String, Trade>, } ``` --- ## 总结 链上K线生成是一个典型的流式数据处理场景，核心要点： - **时间窗口聚合**：根据周期计算时间窗口边界 - **状态管理**：Redis持久化 + 内存缓存双层架构 - **流式处理**：Timely Dataflow分布式计算框架 - **背压控制**：信号量 + 队列监控防止过载 - **断点续传**：游标机制保证数据不丢失 这套架构不仅适用于K线生成，也可应用于其他实时聚合场景，如TVL计算、交易量统计等。 --- ## 参考资料 - [Timely Dataflow 文档](https://github.com/TimelyDataflow/timely-dataflow) - [InfluxDB 时序数据库](https://www.influxdata.com/) - [Kafka 流处理](https://kafka.apache.org/documentation/streams/) - [chains-agg-kline 项目](https://github.com/0xpapercut/chains-agg-kline)
本文作者:yowayimono
本文链接:
版权声明:本博客所有文章除特别声明外，均采用 BY-NC-SA 许可协议。转载请注明出处！