時計付きで届くデータ
ティック、気配、リスク指標、サーバーのメトリクスは同じ形をしています。タイムスタンプ、いくつかのラベル、いくつかの測定値、そしてそれらの終わりのない流れです。時系列データベース（InfluxDB、TimescaleDB、QuestDB、kdb+）は、まさにこのために作られています。書き込みはほとんど常に最新時刻への追記なので高速で、データは時間順のチャンクに保存されるため「直近1時間」は少量にしか触れず、古い詳細データはより粗いデータに要約して最後には削除できます。InfluxDB のデータモデルは典型的です。メジャーメント（テーブルに相当）、タグ（銘柄や市場のようなインデックス付きラベル）、フィールド（測定値）、タイムスタンプで構成されます。
# 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ダウンサンプリング：ティックからバーへ
すべてのティックを永遠に保存するのは高価で、ほとんど必要ありません。ダウンサンプリングは点を固定の時間バケットにまとめ、各バケットの要約を残します。価格の標準的な要約が OHLCV で、始値（バケット内の最初の価格）、高値、安値、終値（最後の価格）、出来高です。時刻 ts のティックは ts - ts % bucket で始まるバケットに属します。データベースはこれを継続的に行い、結果をはるかに小さい別テーブル（連続集計）に書きます。チャートやほとんどのバックテストが読むのはこちらで、生のティックは短期間しか保持しません。
# 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日、分足は1年、日足は永久に保持します。
# 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- 時系列のポイントはメジャーメント、インデックス付きタグ、測定フィールド、タイムスタンプでできている。書き込みは追記で、データは時間順に保存される
- ティックを OHLCV バー（
ts - ts % bucket）にダウンサンプリングし、バーを読む。生のティックは短期間だけ保持する - 過去方向だけを見る as-of 結合（二分探索）で系列を結び、保持ポリシーを適用し、タグのカーディナリティと時間の単位を管理する
次は Python ライブラリのパッケージングです。あなたのコードを他の人がインストールできる形にします。
ラボ：採点される内容
ohlcv_barsでティックをバーにまとめる25点asof_joinで各約定時点の気配を求める25点line_protocolでポイントを書き出す25点apply_retentionで期限切れのデータを捨てる25点