04日目: ローカルデータレイクの構築 —— DuckDBの活用

04日目: ローカルデータレイクの構築 —— DuckDBの活用

第4回は、データの保存場所についてだ。 開発初期においては、データの管理はとりあえずCSVファイルで保存しておいて、YAGNIの原則だ。ここでは、実際に対応が必要になるほどファイル数やサイズに困ったとき、どうすれば良いか記す。要するにデータベースを使うということだ。

PostgreSQLやMySQLをDockerで立ち上げて、ポート管理して、ユーザー権限を設定するのは、個人開発のスコープでは重すぎる。そこで使うのが DuckDB である。

CSVファイル管理の限界とDuckDB

DuckDBは「分析のためのSQLite」と呼ばれている。¹ 以下の特徴がある。

• サーバーレス: プロセス組み込み型なので、uv add duckdb だけで使える。サーバー構築が不要だ。
• 列指向（Columnar）: 集計処理が速い。時系列データの分析と相性が良い。
• 単一ファイル: すべてのデータは trading.duckdb という1つのファイルに収まる。バックアップはコピーするだけである。

Pythonからの操作

DuckDBとPolarsの相性は良い。どちらも Apache Arrow の列指向フォーマットを下敷きにしており、データを受け渡す際のコピーを極力減らせる。² ただし この連携には pyarrow が必要 なので、uv add duckdb polars pyarrow でまとめて入れておく（これを忘れると動かない）。

# duckdb_store.py（抜粋）
import duckdb
import polars as pl

# DBに接続（ファイルが無ければ作られる）。専用の con を使い回す。
con = duckdb.connect("trading.duckdb")

# サンプルデータ（Polars）
df = pl.DataFrame({
    "date":   ["2026-01-01", "2026-01-02"],
    "code":   ["285A", "285A"],
    "open":   [1990, 2040],
    "high":   [2010, 2060],
    "low":    [1985, 2035],
    "close":  [2000, 2050],
    "volume": [10000, 12000],
})

# Polars フレームをビューとして明示登録し、SQLで格納する
con.register("new_prices", df)
con.execute("""
    INSERT INTO stock_prices
    SELECT date, code, open, high, low, close, volume FROM new_prices
""")

DuckDBは con.sql("SELECT * FROM df") のように Python のローカル変数を自動で参照してくれる。この機能を"replacement scan"という。³ ただし呼び出し元のフレーム走査に依存するため、今回はより堅牢な con.register() を使う。グローバルの duckdb.sql(...) と、con のメソッドを混ぜないこと（別の接続を見てしまう）。

SQL in Python

さらに強力なのが、Polarsのデータフレームに対してSQLが打てることだ。結果は .pl() で Polars で受け取れる。

# duckdb_store.py（sql_on_frame 相当）
# Polars フレームを登録して SQL を走らせ、結果を Polars で受け取る
con.register("stock_data", df)
result = con.execute("""
    SELECT code, AVG(close) AS avg_price
    FROM stock_data
    GROUP BY code
""").pl()

print(result)
con.unregister("stock_data")

Polars の API で書くより、複雑な結合や集計は SQL の方が書きやすい場合がある。Python のロジックと SQL の表現力、いいとこ取りができる。

株価データスキーマの設計

実際にテーブルを設計しよう。今回はシンプルに、銘柄マスタ と 時系列データ を分ける。

1. 銘柄マスタ (tickers)

-- schema.sql
CREATE TABLE IF NOT EXISTS tickers (
    code    VARCHAR PRIMARY KEY,
    name    VARCHAR,
    market  VARCHAR,
    sector  VARCHAR
);

2. 時系列データ (stock_prices)

-- schema.sql
CREATE TABLE IF NOT EXISTS stock_prices (
    date    DATE    NOT NULL,
    code    VARCHAR NOT NULL,
    open    DOUBLE,
    high    DOUBLE,
    low     DOUBLE,
    close   DOUBLE,
    volume  DOUBLE,
    PRIMARY KEY (date, code)
);

OHLCV は DOUBLE（64bit）にする。DuckDB の FLOAT は REAL（32bit）で、株価の計算を重ねると誤差が積もりやすい。Polars の f64 とも自然に対応する。

PRIMARY KEY (date, code) は重複排除の基準である（次節の upsert で使う）。ただし DuckDB では主キー制約のために ART 索引 が暗黙に作られる。点検索には効くが、バルク取り込みを重視するなら「主キーを外してアプリ側で重複を弾く」選択もある。⁴

重複排除（Upsert）の戦略

毎日データを追加する際、同じ (date, code) が重複して入らないようにする。DuckDB は INSERT ... ON CONFLICT DO UPDATE で upsert できる。⁵

# duckdb_store.py（insert_prices 相当）
# 新データをビュー登録してから upsert する
con.register("new_prices", df)
con.execute("""
    INSERT INTO stock_prices (date, code, open, high, low, close, volume)
    SELECT date, code, open, high, low, close, volume FROM new_prices
    ON CONFLICT (date, code) DO UPDATE SET
        open   = EXCLUDED.open,
        high   = EXCLUDED.high,
        low    = EXCLUDED.low,
        close  = EXCLUDED.close,
        volume = EXCLUDED.volume
""")

EXCLUDED は「新しく挿入しようとした値」を指す。株式分割の反映などで過去データが修正されたときも、最新値で上書きされる。 昔の記事では一時テーブルにロードしてマージする方法も紹介されるが、con.register() で Polars フレームをビュー化すれば、一時テーブルを実体化せずに同等の upsert ができる（大量データでも速い）。なお SET に列を漏らすと、その列は一時的に NULL にされるため、NOT NULL 列は必ず含めること。

今日の成果

• DuckDB を導入し、サーバーレスなデータベース環境を構築した。
• Polars と DuckDB を Apache Arrow 経由で連携させた（pyarrow 必須）。
• 株価データを格納するスキーマを設計した（OHLCV は DOUBLE）。
• ON CONFLICT DO UPDATE で重複排除の upsert を実装した。

これで「データ基盤」ができた。Polarsで加工し、DuckDBに蓄積する。次回は、このDBに入れるための「データそのもの」を集める。株価のような構造化データは公式API（J-Quants）から、ニュースや開示のような非構造化データは公開データから取得する。

検証について

ただし掲載するコードは、実際に実行して結果を確認したものだけを載せている。今回は DuckDB と Polars の連携が公式docどおりに動くかを、テストを書いて確かめるところまでやった。

テストの抜粋を載せておく（tests/test_duckdb_store.py より）。Polars で作ったデータを DuckDB に格納し、取り出した値が元と一致するか確認するラウンドトリップテストだ。

# tests/test_duckdb_store.py（抜粋）
def test_roundtrip_preserves_data(tmp_path):
    """Polars → DuckDB 格納 → Polars 取得 で値が一致する。"""
    con = connect(tmp_path / "lake.duckdb")
    df = make_sample_ohlcv(codes=("285A", "9984"), rows_per_code=30, seed=7).select(OHLCV_COLS)
    insert_prices(con, df)
    got = fetch_prices(con)

    assert got.height == df.height
    exp = df.sort(["code", "date"])
    got = got.sort(["code", "date"])
    for col in ("open", "high", "low", "close"):
        assert got[col].to_list() == exp[col].to_list()
    assert got["date"].dtype == pl.Date

同じファイルに upsert のテスト（同一キーの上書き確認）も書いている。これらが通れば、格納・取得・重複排除の基本操作は保証される。

---

1. DuckDB は列指向の組み込み型 OLAP データベースである。概要は DuckDB 公式。↩
2. DuckDB は Polars と Arrow 経由で効率的に連携する。公式ガイドは Integration with Polars。なお、公式ガイドはこの連携には pyarrow ライブラリのインストールが必要であると明記している。↩
3. "replacement scan" により、DuckDB は Python プロセス内の DataFrame を名前で見つけられる。詳しくは Import from Pandas — DuckDB。↩
4. DuckDB は主キー/UNIQUE 制約に ART（Adaptive Radix Tree）索引を暗黙生成する。バルク取り込み時のファイル膨張報告もあるため、ワークロードに応じて判断したい。参照: DuckDB Indexes・duckdb/duckdb#19468。↩
5. DuckDB の upsert 構文は INSERT Statement。INSERT OR REPLACE INTO（全列置換）と ON CONFLICT DO UPDATE SET（列指定）の両方をサポートする。↩