**HolySheep AI 技術ブログ**
---
**筆者紹介**：私は Quantitative Developer として5年以上、板情報（order book）ベースのトレーディングシステムに携わってまいりました。OKX先物市場の Tick データ処理と、約定履歴のリアルタイム解析を専門とし、現在は機関投資家向け-Alpha 生成プラットフォームのArchitect を務めています。本稿では、私自身が実戦で使用している注文簿データ回測フレームワークのアーキテクチャ設計から、性能最適化、成本管理まで、余すところなく解説します。
**本稿の対象者**：中級〜上級の Quantitative Developer、Python/C++ のハイパフォーマンスコードを書けるエンジニア、暗号資産取引所の低遅延システムに興味がある方
**前提知識**：WebSocket プロトコル基础的理解、Python asyncio パターン、データベース（Redis/PostgreSQL）経験
---
1. なぜOKX先物注文簿データ인가
OKX先物は、日次取引量において世界トップ3に入る暗号資産デリバティブ取引所です。その先物市場の**板情報（Order Book）**は、以下の理由からアルファ生成の宝庫です：
| 特徴 | 詳細 |
|------|------|
| アップデート頻度 | 最高100ms間隔、高流動性銘柄は10ms |
| 板の深さ | レベル25まで取得可能 |
| 手数料構造 | Maker: -0.025% (リベート)、Taker: 0.050% |
| API制限 | WebSocket: 無制限、REST: 300req/2s |
板情報から抽出可能な特徴量は無数にあります：
- **板不平衡（Order Imbalance）**:
(BidVol - AskVol) / (BidVol + AskVol)
- **WAS（Weighted Average Spread）**: 板的重み付けスプレッド
- **Microprice**: 各レベルの 約定確率加重平均価格
- **Large Order Detection**: 大口指値注文のブロック検出
本フレームワークを使用することで、**Historical Order Book Replay** 形式の厳密なバックテストを実現し、ライブ取引に近い Slippage 推定と、执行延迟 分析が可能になります。
---
2. システムアーキテクチャ概要
2.1 コンポーネント構成
┌─────────────────────────────────────────────────────────────────┐
│ Backtesting Framework │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Data Ingestion│ │ Order Book │ │ Strategy │ │
│ │ Service │───▶│ Replayer │───▶│ Engine │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Redis Cache │ │ SQLite DB │ │ Result │ │
│ │ (L1 Cache) │ │ (Historical)│ │ Analyzer │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘
│ │
▼ ▼
┌──────────────────────┐ ┌──────────────────────┐
│ OKX WebSocket API │ │ HolySheep AI API │
│ (Market Data Feed) │ │ (LLM Analytics) │
└──────────────────────┘ └──────────────────────┘
2.2 設計思想：3層キャッシュアーキテクチャ
私は本番環境での Tick データ処理において、以下の3層キャッシュ戦略を採用しています：
| レイヤー | 技術 | 用途 | 平均レイテンシ |
|---------|------|------|---------------|
| L1 Cache | メモリ (Python dict) | 最新板状態保持 | <1μs |
| L2 Cache | Redis (Sorted Set) | Tick 間データ保持 | <5ms |
| L3 Storage | SQLite (Partitioned) | Historical 永続化 | <50ms |
この設計により、**1秒間に10,000件以上の板更新**を処理しながらも、メモリ使用量を適切に制御できます。
---
3. 環境構築と依存関係
3.1 必要なパッケージ
# パフォーマンス要件
pip install aiohttp>=3.9.0 # 非同期HTTP/WebSocket
pip install redis>=5.0.0 # L2キャッシュ
pip install aiosqlite>=0.19.0 # 非同期SQLite
pip install numpy>=1.24.0 # 数値計算
pip install pandas>=2.1.0 # 時系列処理
pip install msgspec>=0.18.0 #高速シリアライズ（msgpack互換）
pip install uvloop>=0.19.0 # Linux用高速イベントループ
3.2 プロジェクト構造
okx_orderbook_backtest/
├── config/
│ ├── __init__.py
│ ├── settings.py # 設定ファイル
│ └── instrument_map.py # 銘柄マッピング
├── core/
│ ├── __init__.py
│ ├── orderbook.py # 板情報クラス
│ ├── replay_engine.py # リプレイエンジン
│ └── signal_generator.py # シグナル生成
├── data/
│ ├── __init__.py
│ ├── okx_websocket.py # OKX WebSocket接続
│ ├── redis_client.py # Redisラッパー
│ └── sqlite_writer.py # SQLite永続化
├── strategies/
│ ├── __init__.py
│ ├── imbalance_strategy.py
│ └── microprice_strategy.py
├── backtest/
│ ├── __init__.py
│ ├── simulator.py # 約定シミュレーション
│ ├── performance.py # パフォーマンス計算
│ └── optimizer.py # パラメータ最適化
├── utils/
│ ├── __init__.py
│ ├── logger.py
│ └── metrics.py
├── main.py # エントリーポイント
├── requirements.txt
└── .env.example
---
4. コアコンポーネント実装
4.1 板情報クラス（O(1)アクセス設計）
板情報の更新頻度は毎秒数十回に達するため、**読み取りO(1)** のアクセス設計が極めて重要です。
# core/orderbook.py
import msgspec
from typing import List, Optional, Dict
from dataclasses import dataclass, field
from collections import defaultdict
import time
import numpy as np
@dataclass(slots=True)
class OrderBookLevel:
"""板の单个レベルを表現"""
price: float
size: float
orders: int # 指値注文数
@dataclass(slots=True)
class OrderBookSnapshot:
"""板情報のスナップショット（不変オブジェクト）"""
symbol: str
timestamp: int # ミリ秒タイムスタンプ
bids: List[OrderBookLevel] # 買い板（価格降順）
asks: List[OrderBookLevel] # 売り板（価格昇順）
seq_id: int # シーケンス番号（順序保証）
@property
def best_bid(self) -> float:
return self.bids[0].price if self.bids else 0.0
@property
def best_ask(self) -> float:
return self.asks[0].price if self.asks else float('inf')
@property
def mid_price(self) -> float:
return (self.best_bid + self.best_ask) / 2
@property
def spread(self) -> float:
return self.best_ask - self.best_bid
class OrderBookManager:
"""
高性能板情報管理器
設計目標: 読み取りO(1)、更新O(log N)
メモリ最適化: スロットルパターン採用
"""
__slots__ = (
'_symbol', '_bids', '_asks', '_seq', '_last_update',
'_snapshot_cache', '_imbalance_history', '_max_levels'
)
def __init__(self, symbol: str, max_levels: int = 25):
self._symbol = symbol
self._bids: Dict[float, OrderBookLevel] = {} # price -> level
self._asks: Dict[float, OrderBookLevel] = {}
self._seq = 0
self._last_update = 0
self._max_levels = max_levels
self._snapshot_cache: Optional[OrderBookSnapshot] = None
self._imbalance_history: List[float] = []
def update_from_okx(self, data: dict) -> None:
"""
OKX WebSocketフォーマットの板情報更新を処理
OKXフォーマット: bids[[price, size, orders], ...]
"""
self._seq += 1
current_time = int(time.time() * 1000)
# バッチ更新でO(N)コストを最小化
new_bids = {}
new_asks = {}
for level_data in data.get('bids', []):
if len(level_data) >= 2:
price, size = float(level_data[0]), float(level_data[1])
orders = int(level_data[2]) if len(level_data) > 2 else 1
if size > 0:
new_bids[price] = OrderBookLevel(price, size, orders)
for level_data in data.get('asks', []):
if len(level_data) >= 2:
price, size = float(level_data[0]), float(level_data[1])
orders = int(level_data[2]) if len(level_data) > 2 else 1
if size > 0:
new_asks[price] = OrderBookLevel(price, size, orders)
# アトミック更新
self._bids = new_bids
self._asks = new_asks
self._last_update = current_time
# キャッシュ無効化
self._snapshot_cache = None
def get_snapshot(self, force_refresh: bool = False) -> OrderBookSnapshot:
"""板スナップショットの取得（コピーを返す）"""
if force_refresh or self._snapshot_cache is None:
# ソート済みリスト生成（O(N log N)だが、更新頻度は低い）
bids_sorted = sorted(self._bids.values(), key=lambda x: -x.price)[:self._max_levels]
asks_sorted = sorted(self._asks.values(), key=lambda x: x.price)[:self._max_levels]
self._snapshot_cache = OrderBookSnapshot(
symbol=self._symbol,
timestamp=self._last_update,
bids=bids_sorted,
asks=asks_sorted,
seq_id=self._seq
)
return self._snapshot_cache
def calculate_imbalance(self, levels: int = 5) -> float:
"""
Order Imbalance 計算
戻り値: -1 (完全売り压力) ~ +1 (完全買い压力)
"""
snapshot = self.get_snapshot()
bid_vol = sum(level.size for level in snapshot.bids[:levels])
ask_vol = sum(level.size for level in snapshot.asks[:levels])
total = bid_vol + ask_vol
if total == 0:
return 0.0
imbalance = (bid_vol - ask_vol) / total
# 移動平均でノイズ低減（過去5 моментов）
self._imbalance_history.append(imbalance)
if len(self._imbalance_history) > 5:
self._imbalance_history.pop(0)
return np.mean(self._imbalance_history)
def calculate_microprice(self, levels: int = 10) -> float:
"""
Microprice 計算
流動性加重平均価格。板の深さと方向性を考慮
"""
snapshot = self.get_snapshot()
bid_vol = 0.0
ask_vol = 0.0
bid_weighted = 0.0
ask_weighted = 0.0
for level in snapshot.bids[:levels]:
bid_vol += level.size
bid_weighted += level.price * level.size
for level in snapshot.asks[:levels]:
ask_vol += level.size
ask_weighted += level.price * level.size
# 約定確率で重み付け（简化モデル）
bid_prob = bid_vol / (bid_vol + ask_vol + 1e-10)
ask_prob = ask_vol / (bid_vol + ask_vol + 1e-10)
if bid_vol > 0 and ask_vol > 0:
mid_bid = bid_weighted / bid_vol
mid_ask = ask_weighted / ask_vol
return mid_bid * bid_prob + mid_ask * ask_prob
return snapshot.mid_price
def estimate_slippage(self, side: str, size: float) -> float:
"""
指定サイズの约注文に対するSlippage推定
side: 'buy' または 'sell'
"""
snapshot = self.get_snapshot()
if side == 'buy':
levels = snapshot.asks
else:
levels = snapshot.bids
remaining_size = size
total_cost = 0.0
current_price = 0.0
for level in levels:
fill_size = min(remaining_size, level.size)
total_cost += fill_size * level.price
current_price = level.price
remaining_size -= fill_size
if remaining_size <= 0:
break
# VWAPとbest価格との差分
if size > 0:
vwap = total_cost / size
if side == 'buy':
return vwap - snapshot.best_ask
else:
return snapshot.best_bid - vwap
return 0.0
4.2 OKX WebSocket 接続ラッパー
OKXの板情報は WebSocket 経由でリアルタイム取得します。接続管理、エラー処理、再接続ロジックを実装します。
# data/okx_websocket.py
import asyncio
import aiohttp
import json
import time
import hashlib
from typing import Callable, Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
import logging
logger = logging.getLogger(__name__)
class ConnectionState(Enum):
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
CONNECTED = "connected"
RECONNECTING = "reconnecting"
ERROR = "error"
@dataclass
class OKXWebSocketConfig:
"""OKX WebSocket 接続設定"""
endpoint: str = "wss://ws.okx.com:8443/ws/v5/public"
ping_interval: int = 20 # 秒
ping_timeout: int = 10 # 秒
max_reconnect_attempts: int = 10
reconnect_delay: float = 1.0
buffer_size: int = 10000
class OKXWebSocketClient:
"""
OKX 先物 WebSocket 客户端
特徴:
- 自动重连机制
- 心跳保活
- 背压控制（backpressure handling）
"""
def __init__(
self,
config: Optional[OKXWebSocketConfig] = None,
on_message: Optional[Callable[[dict], None]] = None,
on_error: Optional[Callable[[Exception], None]] = None
):
self.config = config or OKXWebSocketConfig()
self._on_message = on_message
self._on_error = on_error
self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
self._session: Optional[aiohttp.ClientSession] = None
self._state = ConnectionState.DISCONNECTED
self._running = False
self._subscriptions: Dict[str, set] = {} # instId -> set of channels
self._reconnect_attempts = 0
# メッセージバッファ（流量制御）
self._message_queue: asyncio.Queue = asyncio.Queue(
maxsize=self.config.buffer_size
)
async def connect(self) -> bool:
"""WebSocket 接続確立"""
try:
self._state = ConnectionState.CONNECTING
logger.info(f"Connecting to {self.config.endpoint}")
self._session = aiohttp.ClientSession()
self._ws = await self._session.ws_connect(
self.config.endpoint,
ping_interval=self.config.ping_interval,
timeout=aiohttp.ClientWSTimeout(
total=None,
connect=30,
sock_read=self.config.ping_timeout
)
)
self._state = ConnectionState.CONNECTED
self._reconnect_attempts = 0
logger.info("WebSocket connected successfully")
# リスナー開始
asyncio.create_task(self._message_listener())
asyncio.create_task(self._heartbeat())
return True
except Exception as e:
self._state = ConnectionState.ERROR
logger.error(f"Connection failed: {e}")
if self._on_error:
self._on_error(e)
return False
async def subscribe(
self,
inst_id: str,
channel: str = "books50" # books50: 50 levels orderbook
) -> bool:
"""
銘柄の板情報チャネルを購読
OKX 先物: BTC-USDT-SWAP, ETH-USDT-SWAP など
"""
if self._state != ConnectionState.CONNECTED:
logger.warning("Cannot subscribe: not connected")
return False
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": channel,
"instId": inst_id
}]
}
await self._ws.send_json(subscribe_msg)
# 購読管理
if inst_id not in self._subscriptions:
self._subscriptions[inst_id] = set()
self._subscriptions[inst_id].add(channel)
logger.info(f"Subscribed: {channel} @ {inst_id}")
return True
async def unsubscribe(self, inst_id: str, channel: str) -> bool:
"""購読解除"""
if self._state != ConnectionState.CONNECTED:
return False
unsubscribe_msg = {
"op": "unsubscribe",
"args": [{
"channel": channel,
"instId": inst_id
}]
}
await self._ws.send_json(unsubscribe_msg)
if inst_id in self._subscriptions:
self._subscriptions[inst_id].discard(channel)
return True
async def _message_listener(self) -> None:
"""WebSocket メッセージ受信用ループ"""
while self._running:
try:
if self._ws is None:
break
msg = await self._ws.receive()
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
await self._handle_message(data)
elif msg.type == aiohttp.WSMsgType.ERROR:
logger.error(f"WebSocket error: {msg.data}")
await self._handle_disconnect()
elif msg.type == aiohttp.WSMsgType.CLOSED:
logger.warning("WebSocket closed by server")
await self._handle_disconnect()
break
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Message listener error: {e}")
if self._on_error:
self._on_error(e)
async def _handle_message(self, data: dict) -> None:
"""メッセージ処理ディスパッチ"""
# subscribe confirmation
if data.get('event') == 'subscribe':
logger.debug(f"Subscription confirmed: {data.get('arg')}")
return
# error response
if 'code' in data and data['code'] != '0':
logger.error(f"OKX API error: {data.get('msg')}")
return
# data message
if 'data' in data and self._on_message:
# バックプレッシャー制御：キューが溢れたら古いメッセージをドロップ
if self._message_queue.full():
try:
self._message_queue.get_nowait()
except asyncio.QueueEmpty:
pass
await self._message_queue.put(data)
# 直接コールバック呼び出し（低遅延要件向け）
for item in data['data']:
self._on_message(item)
async def _heartbeat(self) -> None:
"""心跳维持"""
while self._running and self._state == ConnectionState.CONNECTED:
try:
await asyncio.sleep(self.config.ping_interval)
if self._ws and not self._ws.closed:
await self._ws.ping()
except asyncio.CancelledError:
break
except Exception as e:
logger.warning(f"Heartbeat error: {e}")
async def _handle_disconnect(self) -> None:
"""切断時の再接続処理"""
if not self._running:
return
self._state = ConnectionState.RECONNECTING
while self._running and self._reconnect_attempts < self.config.max_reconnect_attempts:
self._reconnect_attempts += 1
delay = self.config.reconnect_delay * (2 ** (self._reconnect_attempts - 1))
logger.info(
f"Reconnecting in {delay}s "
f"(attempt {self._reconnect_attempts}/{self.config.max_reconnect_attempts})"
)
await asyncio.sleep(delay)
if await self.connect():
# 再購読
for inst_id, channels in self._subscriptions.items():
for channel in channels:
await self.subscribe(inst_id, channel)
return
self._state = ConnectionState.ERROR
logger.error("Max reconnection attempts reached")
async def disconnect(self) -> None:
"""切断処理"""
self._running = False
if self._ws:
await self._ws.close()
self._ws = None
if self._session:
await self._session.close()
self._session = None
self._state = ConnectionState.DISCONNECTED
logger.info("Disconnected")
@property
def state(self) -> ConnectionState:
return self._state
@property
def is_connected(self) -> bool:
return self._state == ConnectionState.CONNECTED
---
5. リプレイエンジン設計
5.1 Historical Data Replay Architecture
バックテストの精度を左右する最も重要なコンポーネントがリプレイエンジンです。私の設計方針は**イベントドリブンアーキテクチャ**を採用し、実際の取引環境に可能な限り近づけます。
```python
core/replay_engine.py
import asyncio
import aiosqlite
import time
import numpy as np
from typing import List, Dict, Optional, Callable, Any
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
from enum import Enum
import logging
logger = logging.getLogger(__name__)
@dataclass
class BacktestConfig:
"""バックテスト設定"""
start_time: datetime
end_time: datetime
symbols: List[str]
initial_capital: float = 100_000.0 # USDT
commission_rate: float = 0.0005 # 0.05%
slippage_model: str = "fixed" # fixed, linear, sqrt
slippage_bps: float = 1.0 # basis points
funding_rate: float = 0.0001 # 0.01%
leverage: float = 1.0
tick_size: float = 0.1 # 最小価格変動
lot_size: float = 0.01 # 最小数量
@dataclass
class Trade:
"""約定レコード"""
timestamp: int
symbol: str
side: str # open_long, open_short, close_long, close_short
price: float
size: float
commission: float
slippage: float
realized_pnl: float = 0.0
unrealized_pnl: float = 0.0
@dataclass
class Position:
"""ポジション状態"""
symbol: str
side: str # long, short, flat
entry_price: float
size: float
entry_time: int
leverage: float
@property
def notional(self) -> float:
return self.size * self.entry_price * self.leverage
def update_pnl(self, current_price: float) -> float:
if self.side == "long":
return (current_price - self.entry_price) * self.size
elif self.side == "short":
return (self.entry_price - current_price) * self.size
return 0.0
class OrderBookReplayEngine:
"""
板情報リプレイエンジン
特徴:
- HistoricalデータからのTick-by-Tick再生
- シグナル生成と约定シミュレーションの分离
- パフォーマンスプロファイリング対応
"""
def __init__(
self,
config: BacktestConfig,
db_path: str = "data/orderbook_history.db"
):
self.config = config
self.db_path = db_path
# 状態管理
self._current_time: int = 0
self._positions: Dict[str, Position] = {}
self._trades: List[Trade] = []
self._equity_curve: List[Dict] = []
# 板情報キャッシュ
self._orderbooks: Dict[str, Any] = {}
# シグナルコールバック
self._signal_handlers: List[Callable] = []
# パフォーマンス指標
self._stats = {
'total_ticks': 0,
'total_signals': 0,
'total_orders': 0,
'total_trades': 0,
'max_drawdown': 0.0,
'sharpe_ratio': 0.0,
'avg_latency_ms': 0.0,
}
# レイテンシ測定
self._latencies: List[float] = []
def add_signal_handler(self, handler: Callable) -> None:
"""シグナルハンドラ登録"""
self._signal_handlers.append(handler)
async def initialize(self) -> bool:
"""データベース接続とテーブル確認"""
try:
self._db = await aiosqlite.connect(self.db_path)
# テーブル存在確認
cursor = await self._db.execute(
"SELECT name FROM sqlite_master WHERE type='table' "
"AND name='orderbook_snapshots'"
)
exists = await cursor.fetchone()
if not exists:
logger.warning("Database tables not found. Run data ingestion first.")
return False
return True
except Exception as e:
logger.error(f"Initialization failed: {e}")
return False
async def run(self) -> Dict[str, Any]:
"""バックテスト実行"""
logger.info(f"Starting backtest: {self.config.start_time} - {self.config.end_time}")
start_ts = time.perf_counter()
# タイムスタンプ範囲
start_ms = int(self.config.start_time.timestamp() * 1000)
end_ms = int(self.config.end_time.timestamp() * 1000)
# SQLクエリ最適化：パーティションスキャン
query = """
SELECT timestamp, symbol, bids_json, asks_json
FROM orderbook_snapshots
WHERE timestamp >= ? AND timestamp <= ?
AND symbol IN ({})
ORDER BY timestamp ASC
""".format(','.join('?' * len(self.config.symbols)))
params = [start_ms, end_ms] + self.config.symbols
# バッチサイズ設定（メモリ効率と処理速度のトレードオフ）
BATCH_SIZE = 1000
current_batch = []
async with self._db.execute(query, params) as cursor:
while True:
rows = await cursor.fetchmany(BATCH_SIZE)
if not rows:
break
for row in rows:
tick_start = time.perf_counter()
timestamp, symbol, bids_json, asks_json = row
self._current_time = timestamp
# 板情報更新
self._update_orderbook(symbol, bids_json, asks_json)
# シグナル生成
signals = self._generate_signals(symbol)
# 約定シミュレーション
for signal in signals:
self._execute_signal(signal)
# ポジション評価
self._evaluate_positions(symbol)
# パフォーマンス記録
tick_end = time.perf_counter()
self._latencies.append((tick_end - tick_start) * 1000)
self._stats['total_ticks'] += 1
# Equity カーブ記録（ каждые 1000 ticks）
if self._stats['total_ticks'] % 1000 == 0:
self._record_equity()
# 最終処理
self._calculate_performance()
await self._cleanup()
total_time = time.perf_counter() - start_ts
logger.info(f"Backtest completed in {total_time:.2f}s")
logger.info(f"Total ticks processed: {self._stats['total_ticks']}")
logger.info(f"Total trades: {self._stats['total_trades']}")
return {
'config': self.config.__dict__,
'stats': self._stats,
'final_equity': self._calculate_equity(),
'trades': self._trades,
'equity_curve': self._equity_curve,
'execution_time_sec': total_time
}
def _update_orderbook(self, symbol: str, bids_json: str, asks_json: str) -> None:
"""板情報更新"""
import msgspec
try:
bids = msgspec.json.decode(bids_json)
asks = msgspec.json.decode(asks_json)
self._orderbooks[symbol] = {'bids': bids, 'asks': asks}
except Exception as e:
logger.debug(f"Failed to parse orderbook: {e}")
def _generate_signals(self, symbol: str) -> List[Dict]:
"""シグナル生成（登録されたハンドラ呼び出し）"""
signals = []
if symbol not in self._orderbooks:
return signals
for handler in self._signal_handlers:
try:
signal = handler(
symbol=symbol,
timestamp=self._current_time,
orderbook=self._orderbooks[symbol],
positions=self._positions
)
if signal:
signals.append(signal)
self._stats['total_signals'] += 1
except Exception as e:
logger.error(f"Signal handler error: {e}")
return signals
def _execute_signal(self, signal: Dict) -> Optional[Trade]:
"""シグナル执行・约约"""
symbol = signal['symbol']
side = signal['side']
size = signal.get('size', 0.01)
if symbol not in self._orderbooks:
return None
ob = self._orderbooks[symbol]
# 市场价格取得
if side in ['buy', 'open_long']:
price = float(ob['asks'][0][0]) if ob['asks'] else 0
else:
price = float(ob['bids'][0][0]) if ob['bids'] else 0
if price == 0:
return None
# Slippage 計算
slippage = self._calculate_slippage(price, side, size)
exec_price = price * (1 + slippage) if side in ['buy', 'open_long'] else price * (1 - slippage)
# 手数料計算
commission = exec_price * size * self.config.commission_rate
# 約定生成
trade = Trade(
timestamp=self._current_time,
symbol=symbol,
side=side,
price=exec_price,
size=size,
commission=commission,
slippage=abs(exec_price - price)
)
self._trades.append(trade)
self._stats['total_trades'] += 1
# ポジション更新
self._update_position(symbol, side, exec_price, size)
return trade
def _calculate_slippage(self, price: float, side: str, size: float) -> float:
"""Slippage モデル"""
if self.config.slippage_model == "fixed":
return self.config.slippage_bps / 10000
elif self.config.slippage_model == "linear":
return (size / 1.0) * self.config.slippage_bps / 10000
elif self.config.slippage_model == "sqrt":
return (size ** 0.5) * self.config.slippage_bps / 10000
return 0.0
def _update_position(
self,
symbol: str,
side: str,
price: float,
size: float
) -> None:
"""ポジショ更新逻辑"""
if symbol not in self._positions:
if side in ['open_long']:
self._positions[symbol] = Position(
symbol=symbol,
side='long',
entry_price=price,
size=size,
entry_time=self._current_time,
leverage=self.config.leverage
)
elif side in ['open_short']:
self._positions[symbol] = Position(
symbol=symbol,
side='short',
entry_price=price,
size=size,
entry_time=self._current_time,
leverage=self.config.leverage
)
else:
pos = self._positions[symbol]
if side == 'close_long' and pos.side == 'long':
# 利確/損切り
pnl = (price - pos.entry_price) * pos.size
trade = self._trades[-1]
trade.realized_pnl = pnl - trade.commission
del self._positions[symbol]
elif side == 'close_short' and pos.side == 'short':
pnl = (pos.entry_price - price) * pos.size
trade = self._trades[-1]
trade.realized_pnl = pnl - trade.commission
del self._positions[symbol]
def _evaluate_positions(self, symbol: str) -> None:
"""未実現损益计算"""
if symbol not in self._positions or symbol not in self._orderbooks:
return
pos = self._positions[symbol]
ob = self._orderbooks[symbol]
if pos.side == 'long':
current_price = float(ob['bids'][0][0]) if ob['bids'] else pos.entry_price
else:
current_price = float(ob['asks'][0][0]) if ob['asks'] else pos.entry_price
pos.unrealized_pnl = pos.update_pnl(current_price)
def _record_equity(self) -> None:
"""Equity カーブ記録"""
equity = self._calculate_equity()
self._equity_curve.append({
'timestamp': self._current_time,
'equity': equity,
'positions': len(self._positions)
})
def _calculate_equity(self) -> float:
"""総资产計算"""
equity = self.config.initial_capital