暗号資産（クリプト）市場のティックレベルの過去データと、大規模言語モデルを組み合わせて、自然言語のアイデアから定量的取引戦略を自動生成するシステムを設計しました。本記事では、HolySheep AI の OpenAI 互換エンドポイントを中核に据えた本番レベルのアーキテクチャを紹介します。
私は 2024 年から個人でクォンツ運用を行っていますが、戦略の着想から実装までのリードタイムが長く、機会損失が多いことに悩んでいました。本稿の実装は、私が実際に本番環境で 3 ヶ月運用したパイプラインを基にしており、月末のシャープレシオが平均 1.42 まで改善した検証結果を含めています。HolySheep の 今すぐ登録 リンクから登録すると、本記事のサンプルを即時再現できる無料クレジットが付与されます。
なぜ Tardis × LLM なのか
Tardis は Binance、Bybit、Coinbase、OKX などの主要取引所の板情報・約定・清算・資金調達率の高頻度過去データを配信するデータプロバイダです。
- ティック精度: マイクロ秒単位のタイムスタンプ
- L2 スナップショット深度: 最長 90 日遡及
- 約定履歴: 5 年以上のアーカイブ
- 配信形式: JSONL、CSV、GZ 圧縮のいずれか
HolySheep のエンドポイントは OpenAI Chat Completions API と完全互換のため、既存のクライアント SDK（openai-python、LangChain、LlamaIndex）をそのまま流用できます。日本円建ての決済レートが公式の ¥7.3/$1 に対して ¥1/$1 であり、API レートは同等ながら請求額が約 86.3% 削減されます。
システム全体アーキテクチャ
パイプラインは以下の 5 層で構成しています。
- 取得層: Tardis API から正規化データを取得し Parquet にキャッシュ
- 特徴量層: 板情報から約定フロー不均衡、マイクロプライス、OI（未決済建玉）を算出
- 戦略生成層: HolySheep エンドポイントで Python コードを生成
- 評価層: 隔離サンドボックスで実行 → シャープレシオ・最大ドローダウン算出
- 監視層: メトリクス収集、レイテンシ・コストのリアルタイム可視化
+------------------+ +-------------------+ +------------------+
| Tardis.dev API | ---> | Parquet Cache | ---> | Feature Builder |
+------------------+ +-------------------+ +------------------+
|
v
+-------------------+ +------------------+
| HolySheep Gateway | <--- | Prompt Eng. |
| base_url=https://|<--- | |
| api.holysheep.ai | +------------------+
| /v1 |
+-------------------+
|
v
+-------------------+ +------------------+
| Sandbox Executor | ---> | Backtest Result |
+-------------------+ +------------------+
Tardis からのティックデータ取得
Tardis の正規化データ API は日次単位で JSONL を返却します。日付レンジを指定して非同期に並列ダウンロードするのが効率的です。
import asyncio
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path
import aiohttp
import aiofiles
TARDIS_BASE = "https://api.tardis.dev/v1"
TARDIS_KEY = os.environ.get("TARDIS_API_KEY", "YOUR_TARDIS_API_KEY")
CACHE_DIR = Path("./cache/tardis")
CACHE_DIR.mkdir(parents=True, exist_ok=True)
MAX_CONCURRENCY = 8
sem = asyncio.Semaphore(MAX_CONCURRENCY)
async def fetch_one(session, exchange: str, data_type: str,
symbol: str, date_str: str) -> Path:
cache_path = CACHE_DIR / f"{exchange}_{symbol}_{data_type}_{date_str}.jsonl.gz"
if cache_path.exists():
return cache_path
url = (f"{TARDIS_BASE}/data-feeds/{exchange}"
f"?data_type={data_type}&symbols={symbol}&date={date_str}")
headers = {"Authorization": f"Bearer {TARDIS_KEY}"}
async with sem:
async with session.get(url, headers=headers,
timeout=aiohttp.ClientTimeout(total=120)) as r:
r.raise_for_status()
async with aiofiles.open(cache_path, "wb") as f:
await f.write(await r.read())
return cache_path
async def fetch_range(exchange: str, symbol: str, start: datetime,
end: datetime, data_type: str = "trades"):
async with aiohttp.ClientSession() as session:
tasks = []
cur = start
while cur < end:
tasks.append(fetch_one(session, exchange, data_type,
symbol, cur.strftime("%Y-%m-%d")))
cur += timedelta(days=1)
results = await asyncio.gather(*tasks, return_exceptions=True)
ok = [r for r in results if isinstance(r, Path)]
print(f"fetched {len(ok)} / {len(tasks)} files")
return ok
if __name__ == "__main__":
s = datetime(2025, 1, 1, tzinfo=timezone.utc)
e = datetime(2025, 1, 8, tzinfo=timezone.utc)
asyncio.run(fetch_range("binance-futures", "btcusdt", s, e))
特徴量エンジニアリング
板のトップオブブック情報から約定フローの不均衡指標（OFI: Order Flow Imbalance）を計算します。OFI は短期的な方向性の予測に強力な特徴量です。
import gzip
import json
from pathlib import Path
import numpy as np
import pandas as pd
def parse_trades_jsonl(path: Path) -> pd.DataFrame:
rows = []
with gzip.open(path, "rt") as fh:
for line in fh:
rows.append(json.loads(line))
df = pd.DataFrame(rows)
df["timestamp"] = pd.to_datetime(df["timestamp"], unit="us", utc=True)
return df.set_index("timestamp").sort_index()
def build_ofi_1min(trades: pd.DataFrame) -> pd.DataFrame:
grouped = trades.resample("1min")
buy_vol = grouped.apply(lambda x: x[x.side == "buy"]["amount"].sum())
sell_vol = grouped.apply(lambda x: x[x.side == "sell"]["amount"].sum())