h5i-db で始める超高速な時系列解析・クオンツ入門 ② - スキーマの作成

コロンビア大学の博士課程でコンピューターシステム・AIエージェントの研究をしている Koukyosyumei です。

本連載では、時系列解析やクオンツに特化した超高速データベース「h5i-db」の使い方を詳しく解説します。セグメント分割やキャッシュを用いた高速処理 (PolarsやDuckDBと比べて4~5倍高速) と、Claude CodeやCodexなどAIエージェントから扱いやすい設計を両立しているのが大きな特徴です。連載全体で共通の処理 (デモデータのダウンロードなどなど) は、cookbook_utilsというモジュールに格納されています。

import h5i_db
import pyarrow as pa

import cookbook_utils as cu

データタイプの定義

本稿では、以下のようなデータタイプを用います。

• ts（タイムスタンプ） - timestamp[us, tz=UTC], 非Null（non-nullable:
  すべての生単位の引数（プラン範囲、ギャップフィル間隔、ASOF結合の許容誤差など）を、一貫してマイクロ秒単位で扱います。
• price（価格） - float64
  今回は決済台帳ではなく、分析を行うことが目的なので、より正確なdecimal128ではなく、より高速なfloat64を用います。
• size（数量） - int64
  株数は整数として表現するのが自然です。
• symbol, exchange, side（銘柄、取引所、売買区分） - utf8
  カーディナリティの低い（＝値の種類が少ない）文字列は、Parquetセグメントの内部で自動的に辞書エンコーディング（dictionary-encoded）されます。そのため、スキーマ層でシンプルな utf8 を指定してもディスク容量のコストはかかりません。

trades_schema = pa.schema(
    [
        pa.field("ts", pa.timestamp("us", tz="UTC"), nullable=False),
        pa.field("symbol", pa.string()),
        pa.field("price", pa.float64()),
        pa.field("size", pa.int64()),
        pa.field("exchange", pa.string()),
        pa.field("side", pa.string()),
    ]
)

quotes_schema = pa.schema(
    [
        pa.field("ts", pa.timestamp("us", tz="UTC"), nullable=False),
        pa.field("symbol", pa.string()),
        pa.field("bid", pa.float64()),
        pa.field("ask", pa.float64()),
        pa.field("bid_size", pa.int64()),
        pa.field("ask_size", pa.int64()),
    ]
)

bars_1d_schema = pa.schema(
    [
        pa.field("ts", pa.timestamp("us", tz="UTC"), nullable=False),
        pa.field("symbol", pa.string()),
        pa.field("open", pa.float64()),
        pa.field("high", pa.float64()),
        pa.field("low", pa.float64()),
        pa.field("close", pa.float64()),
        pa.field("volume", pa.int64()),
    ]
)

time_column と sort_key

time_column="ts"と指定することで、h5i-dbが自動でデータベースをセグメント分割し、WHERE ts BETWEEN ...やASOFジョイン(時間が完全に一致していなくても、その瞬間に最も新しかった過去のデータと結合)・time_bucketロールアップ(一定の時間枠ごとに区切って集計)といった時系列系の処理が高速に行えるようになります。

db.create_table("trades", trades_schema, time_column="ts", sort_key=["ts", "symbol"])
db.create_table("quotes", quotes_schema, time_column="ts", sort_key=["ts", "symbol"])
db.create_table("bars_1d", bars_1d_schema, time_column="ts", sort_key=["ts", "symbol"])
db.tables()

['bars_1d', 'quotes', 'trades']

trades = cu.make_trades(days=2, trades_per_day=5_000)
quotes = cu.make_quotes(days=2, quotes_per_day=8_000)
daily = cu.make_daily_prices(symbols=["AAPL", "MSFT", "NVDA"], days=250)

for name, data in [("trades", trades), ("quotes", quotes), ("bars_1d", daily)]:
    commit = db.append(name, data, note="initial load")
    print(f"{name:8s} v{commit['sequence']}: {commit['rows_total']:>7,} rows, "
          f"{commit['segments_total']} segment(s)")

db.schema("trades")

trades   v1:  30,693 rows, 1 segment(s)
quotes   v1:  48,047 rows, 1 segment(s)
bars_1d  v1:     750 rows, 1 segment(s)
ts: timestamp[us, tz=UTC] not null
symbol: string
price: double
size: int64
exchange: string
side: string

db.sql(
    """
    SELECT symbol,
           count(*)                                    AS quotes,
           round(avg((ask - bid) / ((ask + bid) / 2)) * 1e4, 2) AS avg_spread_bps,
           round(min(bid), 2)                          AS min_bid,
           round(max(ask), 2)                          AS max_ask
    FROM quotes
    GROUP BY symbol
    ORDER BY symbol
    """
).to_pandas()

	symbol	quotes	avg_spread_bps	min_bid	max_ask
0	AAPL	16564	1.76	86.04	87.76
1	MSFT	14687	1.76	216.74	224.45
2	NVDA	16796	1.75	251.23	262.48

appendが失敗する場合

appendに追加されるバッチは、宣言されたスキーマと一致していることはもちろん、時間の列がソート済みであり、かつテーブルの現在の最大タイムスタンプ以降から始まっている必要があります。これらの条件が満たされていないとき、appendメソッドは失敗します。

拒否パターン1: スキーマの不一致

bad_schema_batch = pa.table(
    {
        "ts": trades["ts"][:100],
        "symbol": trades["symbol"][:100],
        "price": trades["price"][:100].cast(pa.float32()),   # wrong width
        "size": trades["size"][:100],
        "exchange": trades["exchange"][:100],
        # 'side' column missing entirely
    }
)
try:
    db.append("trades", bad_schema_batch)
except h5i_db.InvalidInputError as e:
    print(f"rejected  code={e.code}")
    print(f"message   {e}")
    print(f"hint      {e.hint or '(none - the message says it all)'}")

rejected  code=schema_mismatch
message   [schema_mismatch] schema mismatch: expected 6 columns, got 5
hint      (none - the message says it all)

拒否パターン2: ソートされていないデータ

import numpy as np

rng = np.random.default_rng(0)
shuffled = trades.take(rng.permutation(len(trades)))

try:
    db.append("trades", shuffled)
except h5i_db.InvalidInputError as e:
    print(f"rejected  code={e.code}")
    print(f"hint      {e.hint}")

rejected  code=sort_order_violation
hint      append requires input sorted by the time column with min >= current table max; use `write` or sort the input

拒否パターン3: 時間範囲の重複

try:
    db.append("trades", trades)  # すでに取り込み済みのものと全く同じバッチ
except h5i_db.InvalidInputError as e:
    print(f"rejected  code={e.code}")
    print(f"hint      {e.hint}")

rejected  code=sort_order_violation
hint      append requires input sorted by the time column with min >= current table max; use `write` or sort the input

スキーマ変更のコスト

セグメントは不変であるため、スキーマの変更は意図的に難しく設計されてます。

• 低コストな変更: 末尾への Nullable（Null許容）カラムの追加（古いセグメントはNULLとして読み込まれます）、および数値型の拡張（int32 → int64、float32 → float64）。
• 高コストな変更: 前の変更、型の縮小、並べ替え、カラムの削除。これらを行うには、新しいテーブルを作成し、バックフィル（データの再充填）を行う必要があります。

参考

• https://github.com/h5i-dev/h5i-db
• https://github.com/h5i-dev/h5i-db-cookbook