用Polars构建分钟级K线合成器：从Tick到OHLCV高效转换

不吃草的牛德

发布于 2026-04-23 12:51:46

820

文章被收录于专栏：RustRust

K线是量化的基石

无论你做什么策略，K线都是最基础的数据。

但现实中的问题是：原始数据往往是Tick逐笔数据，不是K线。

你需要把逐笔成交数据转换成各种周期的K线：

• 1分钟K线
• 5分钟K线
• 15分钟K线
• 30分钟K线
• 日K线

这个过程叫「K线合成」或「K线重采样」。

今天我们用Polars实现一个高效的分钟级K线合成器。

K线合成原理

OHLCV是什么？

每个K线包含五个核心字段：

• Open：开盘价
• High：最高价
• Low：最低价
• Close：收盘价
• Volume：成交量

从Tick到K线



1
2
3
4
5
6
7
8

Tick数据：
09:30:01   10.00   1000
09:30:05   10.05   2000
09:30:15   10.08   1500
...
 
合成1分钟K线：
09:30:00  O=10.00  H=10.08  L=10.00  C=10.08  V=4500

核心逻辑：按时间窗口聚合。

Rust + Polars实现

数据结构



1
2
3
4
5
6
7
8

use polars::prelude::*;
 
#[derive(Debug, Clone)]
struct TickData {
    timestamp: String,
    price: f64,
    volume: i64,
}

时间对齐

首先需要把时间对齐到分钟边界：



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

fn align_to_minute(df: &DataFrame, period: &str) -> DataFrame {
    df.lazy()
        .with_column(
            col("timestamp")
                .str()
                .to_datetime(None, None, StrptimeOptions::default(), lit("raise"))
        )
        .with_column(
            col("timestamp")
                .dt()
                .truncate(lit(period))
                .alias("period_start")
        )
        .collect()
        .unwrap()
}

OHLCV聚合



1
2
3
4
5
6
7
8
9
10
11
12
13
14

fn aggregate_ohlcv(df: &DataFrame) -> DataFrame {
    df.lazy()
        .group_by([col("period_start")])
        .agg([
            col("price").first().alias("open"),
            col("price").max().alias("high"),
            col("price").min().alias("low"),
            col("price").last().alias("close"),
            col("volume").sum().alias("volume"),
        ])
        .sort(["period_start"], Default::default())
        .collect()
        .unwrap()
}

完整流水线



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

fn tick_to_ohlcv(df: &DataFrame, period: &str) -> DataFrame {
    df.lazy()
        // 时间对齐
        .with_column(
            col("timestamp")
                .str()
                .to_datetime(None, None, StrptimeOptions::default(), lit("raise"))
        )
        .with_column(
            col("timestamp")
                .dt()
                .truncate(lit(period))
                .alias("period_start")
        )
        // OHLCV聚合
        .group_by([col("period_start")])
        .agg([
            col("price").first().alias("open"),
            col("price").max().alias("high"),
            col("price").min().alias("low"),
            col("price").last().alias("close"),
            col("volume").sum().alias("volume"),
        ])
        .sort(["period_start"], Default::default())
        .collect()
        .unwrap()
}

多周期并行生成



1
2
3
4
5
6
7
8
9
10
11

fn generate_multi_period(df: &DataFrame) -> HashMap<String, DataFrame> {
    let periods = vec!["1m", "5m", "15m", "30m", "1h", "1d"];
    let mut results = HashMap::new();
 
    for period in periods {
        let ohlcv = tick_to_ohlcv(df, period);
        results.insert(period.to_string(), ohlcv);
    }
 
    results
}