时序数据库

带着时钟到来的数据

tick、报价、风险数字和服务器指标有着相同的形状：一个时间戳、几个标签、几个测量值，以及无穷无尽的数据流。时序数据库（InfluxDB、TimescaleDB、QuestDB、kdb+）正是为此而建。写入几乎总是在最新时间上追加，所以很快；数据按时间有序的块存放，因此查询“最近一小时”只会碰很少的数据；旧的细节可以汇总成更粗的数据，最后删除。InfluxDB 的数据模型很典型：measurement（相当于表）、tag（被索引的标签，如代码或交易场所）、field（测量值）和时间戳。

# One point in InfluxDB line protocol: measurement,tags fields timestamp
point = "trades,symbol=AAPL,venue=XNAS price=184.25,qty=100i 1700000000000000000"
head, fields, ts = point.split(" ")
measurement, *tag_pairs = head.split(",")
tags = dict(pair.split("=") for pair in tag_pairs)
print(measurement, tags)                             # trades {'symbol': 'AAPL', 'venue': 'XNAS'}
print(fields, int(ts))                               # price=184.25,qty=100i 1700000000000000000

降采样：从 tick 到 K 线

永久保存每个 tick 代价高昂，也很少需要。降采样把数据点分到固定的时间桶里，并为每个桶保留一个汇总。价格的标准汇总是 OHLCV：开盘价（桶内第一个价格）、最高价、最低价、收盘价（最后一个价格）和成交量。时间为 ts 的 tick 属于从 ts - ts % bucket 开始的桶。数据库会持续做这件事，并把结果写入一张小得多的独立表（连续聚合），图表和大多数回测读取的就是它，而原始 tick 只保留很短的时间。

# Rolling ticks (ts, price, size) into 60-second OHLCV bars
def ohlcv_bars(ticks, bucket):
    bars = []
    for ts, price, size in ticks:
        start = ts - ts % bucket
        if bars and bars[-1]["start"] == start:
            bar = bars[-1]
            bar["high"], bar["low"] = max(bar["high"], price), min(bar["low"], price)
            bar["close"] = price
            bar["volume"] += size
        else:
            bars.append({"start": start, "open": price, "high": price, "low": price, "close": price, "volume": size})
    return bars

ticks = [(0, 100.0, 10), (30, 101.0, 5), (61, 99.5, 20), (65, 100.5, 1)]
for bar in ohlcv_bars(ticks, 60):
    print(bar)
# {'start': 0, 'open': 100.0, 'high': 101.0, 'low': 100.0, 'close': 101.0, 'volume': 15}
# {'start': 60, 'open': 99.5, 'high': 100.5, 'low': 99.5, 'close': 100.5, 'volume': 21}

as-of 连接与保留期

不同的序列很少在同一时刻更新，所以按相等时间戳连接几乎找不到什么。你真正问的问题是“这笔成交发生时的报价是多少？”：对每笔成交，取不晚于其时间的最新一条报价。这种 as-of 连接（pandas.merge_asof，kdb+ 和 DuckDB 中的 ASOF JOIN）只能向后看；使用成交之后的报价会把未来泄漏到分析中。因为报价按时间排序，二分查找（bisect_right）能在对数时间内找到正确的那条。最后，保留策略会丢弃超过期限的数据点，让存储保持有界：保留 30 天 tick、一年分钟线，以及永久保存日线。

# For each trade, the latest quote at or before it, found by binary search
from bisect import bisect_right

def asof_join(trades, quotes):
    times = [quote[0] for quote in quotes]
    joined = []
    for ts, price in trades:
        position = bisect_right(times, ts)
        joined.append((ts, price, *quotes[position - 1][1:]) if position else (ts, price, None, None))
    return joined

quotes = [(10, 99.9, 100.1), (40, 100.5, 100.7)]
print(asof_join([(5, 100.0), (10, 100.0), (50, 101.0)], quotes))
# [(5, 100.0, None, None), (10, 100.0, 99.9, 100.1), (50, 101.0, 100.5, 100.7)]

# Retention: keep only the points younger than the limit
def apply_retention(points, now, ttl):
    return [(ts, value) for ts, value in points if ts > now - ttl]

points = [(10, 1.0), (40, 2.0), (50, 3.0), (90, 4.0)]
print(apply_retention(points, 100, 60))              # [(50, 3.0), (90, 4.0)]   the point at 40 is exactly 60 s old: dropped
print(apply_retention(points, 100, 1000))            # everything is younger than the limit

• 时序数据点由 measurement、被索引的 tag、测量 field 和时间戳组成；写入是追加，数据按时间顺序存储。
• 把 tick 降采样为 OHLCV K 线（ts - ts % bucket）并读取 K 线；原始 tick 只短期保留。
• 用只向后看的 as-of 连接（二分查找）连接序列，应用保留策略，并控制好标签基数和时间单位。

下一节：打包 Python 库，把你的代码变成别人可以安装的东西。

实验：评分内容

• 用 ohlcv_bars 把 tick 汇总成 K 线25 分
• 用 asof_join 找到每笔成交时的报价25 分
• 用 line_protocol 写出一个数据点25 分
• 用 apply_retention 丢弃过期数据25 分