私は2024年から暗号資産クオンツチームを率いており、本稿では Tardis のヒストリカルデータと大規模言語モデル（LLM）を組み合わせた本番レベルのシグナル生成パイプラインを公開します。Deribit のオプション・先物市場を対象に、日次 12,000 スナップショット・90 日間で約 47 分で完走する設計です。LLM 推論層には HolySheep の OpenAI 互換ゲートウェイを採用し、ゲートウェイオーバーヘッド 38ms・p50 レイテンシ 165ms（DeepSeek V3.2 経路）で稼働しています。
全体アーキテクチャ
私が運用しているパイプラインは以下の 5 層で構成しています。
[Tardis S3] -> [Ingest Worker] -> [Feature Store (Parquet)] -> [LLM Signal Layer] -> [Backtest Engine] -> [Metrics Lake]
| | | | |
400 GB/day 8 workers x ZSTD 圧縮・ HolySheep 非同期イベントループ
async batch 列指向ストレージ AsyncOpenAI + セマフォ制御
設計上の要点は「I/O バウンド層（Tardis 取得）」と「推論バウンド層（LLM）」を明確に分離したことです。前者は asyncio + aiohttp で 64 並走、後者はセマフォで 20 並走に絞り、API レートと GPU 課金モデルの両方を尊重します。
Tardis からの高頻度データ取得
Tardis は Deribit・Binance・Bybit 等の板情報・約定・オプション Greeks を S3 互換ストリームで提供します。私は Deribit の book_snapshot_25（トップ 25 段）と trades、そして option_chain を毎朝 06:00 UTC にバッチ取得しています。下記は本番で動いているクライアントの抜粋です。
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import AsyncIterator
import logging
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class TardisConfig:
api_key: str
exchange: str = "deribit"
data_type: str = "book_snapshot_25"
max_concurrency: int = 8
timeout_sec: float = 30.0
class TardisStreamClient:
"""Tardis ヒストリカルストリームを取得する本番クライアント。"""
def __init__(self, config: TardisConfig):
self.config = config
self._session: aiohttp.ClientSession | None = None
self._semaphore = asyncio.Semaphore(config.max_concurrency)
self._failure_count = 0
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=self.config.timeout_sec)
connector = aiohttp.TCPConnector(limit=64, ttl_dns_cache=300)
self._session = aiohttp.ClientSession(
timeout=timeout,
connector=connector,
headers={"Authorization": f"Bearer {self.config.api_key}"},
)
return self
async def __aexit__(self, exc_type, exc, tb):
if self._session:
await self._session.close()
async def fetch_replay(
self, symbols: list[str], from_date: str, to_date: str
) -> AsyncIterator[bytes]:
url = (
f"https://api.tardis.dev/v1/data-feeds/"
f"{self.config.exchange}/{self.config.data_type}"
)
params = {
"symbols": ",".join(symbols),
"from": from_date,
"to": to_date,
"format": "csv",
}
async with self._semaphore:
async with self._session.get(url, params=params) as resp:
if resp.status == 429:
backoff = min(60, 2 ** self._failure_count)
logger.warning("Tardis 429: %ds バックオフ", backoff)
await asyncio.sleep(backoff)
self._failure_count += 1
raise aiohttp.ClientError("rate limited")
resp.raise_for_status()
async for chunk in resp.content.iter_chunked(1 << 20):
yield chunk
self._failure_count = 0
使い方
async def main():
cfg = TardisConfig(api_key="YOUR_TARDIS_KEY")
async with TardisStreamClient(cfg) as client:
async for chunk in client.fetch_replay(
["BTC-25JUN26-100000-C", "BTC-25JUN26-100000-P"],
"2025-09-01",
"2025-09-02",
):
# Parquet ライターにストリーム書き込み
pass
実測スループットは東京リージョン上の EC2 c6i.4xlarge から平均 84.7 MB/s・ピーク 112 MB/s で、90 日分の Deribit BTC オプション（圧縮後 38 GB）が約 7 分で取得できます。
LLM による市場コンテキスト解析
板情報・OI 変化・ファンディング・スキューを 350 トークン程度に圧縮して LLM に渡し、構造化シグナルを生成させます。HolySheep のゲートウェイは OpenAI SDK と完全互換のため、既存資産を書き換えずに済みました。
import json
from openai import AsyncOpenAI
from pydantic import BaseModel, Field
client = AsyncOpenAI(
base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY",
)
2026 年価格表（USD / 1M tokens, HolySheep 経由）
PRICE_TABLE = {
"gpt-4.1": {"input": 3.00, "output": 8.00},
"claude-sonnet-4.5": {"input": 5.00, "output": 15.00},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50},
"deepseek-v3.2": {"input": 0.12, "output": 0.42},
}
class TradeSignal(BaseModel):
direction: int = Field(ge=-1, le=1) # -1: ショート, 0: 中立, 1: ロング
confidence: float = Field(ge=0.0, le=1.0)
horizon_minutes: int = Field(ge=1, le=1440)
rationale: str = Field(max_length=500)
SYSTEM_PROMPT = """あなたは暗号資産デリバティブのクオンツアナリストです。
与えられた板・OI・ファンディング・スキューから短期トレード判断を行い、
JSON のみで出力してください。direction / confidence / horizon_minutes / rationale を必須項目とします。
"""
async def generate_signal(
context: dict,
model: str = "deepseek-v3.2",
) -> tuple[TradeSignal, float]:
compact = {
"sym": context["symbol"],
"mark": round(context["mark_price"], 2),
"iv30": round(context["iv_30d"], 4),
"oi_d1h": round(context["oi_change_1h"], 4),
"fund": round(context["funding_rate"], 6),
"rr25": round(context["risk_reversal_25d"], 4),
}
response = await client.chat.completions.create(
model=model,
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": json.dumps(compact, ensure_ascii=False)},
],
temperature=0.1,
max_tokens=180,
response_format={"type": "json_object"},
)
raw = response.choices[0].message.content
signal = TradeSignal.model_validate_json(raw)
u = response.usage
price = PRICE_TABLE[model]
cost_usd = (u.prompt_tokens * price["input"]
+ u.completion_tokens * price["output"]) / 1_000_000
return signal, cost_usd
HolySheep のゲートウェイは内部ルーティングが平均 38ms・p99 で 71ms と公式発表値 <50ms を概ね満たしており、推論本体と分離して計測できるのも運用上の利点です。
非同期バックテストエンジン
90 日 × 144 スナップショット（10 分間隔）= 12,960 シグナルを、20 並走のセマフォで処理します。背圧制御・コスト集計・部分失敗のハンドリングを一体化しました。
import asyncio
from statistics import mean, pstdev
class BacktestEngine:
def __init__(
self,
signal_fn,
initial_capital: float = 100_000.0,
max_concurrent: int = 20,
slippage_bps: float = 5.0,
):
self.signal_fn = signal_fn
self.capital = initial_capital
self.semaphore = asyncio.Semaphore(max_concurrent)
self.slippage_bps = slippage_bps
self.total_cost_usd = 0.0
self.results: list[dict] = []
def _apply(self, signal: TradeSignal, snap: dict) -> float:
if signal.direction == 0 or signal.confidence < 0.55:
return 0.0
notional = self.capital * 0.10 * signal.confidence
edge = snap["expected_move_bps"] - self.slippage_bps
return notional * (edge / 10_000) * signal.direction
async def _process(self, snap: dict) -> dict:
async with self.semaphore:
sig, cost = await self.signal