こんにちは、HolySheep AI 技術ブログ編集部の田中でございます。本日は暗号通貨トレーディングにおいて重要な「做市商（マーケットメーカー）データ」を活用した量化回測フレームワークの構築方法について、Bybit API と HolySheep AI を組み合わせた実践的な解説をお届けします。
本記事の結論を先に述べるならば、Bybitの做市商データは板情報・約定履歴・注文簿水深を包括的に取得でき、HolySheep AIのレート（¥1=$1）で運用すれば公式比85%のコスト削減が可能です。以下では具体的なコード実装、価格比較、回測フレームワークの設計指針を懇切丁寧に説明していきます。
結論サマリー
- Bybit 做市商データ API は板情報、約定履歴、オーダーブックのリアルタイム取得に対応
- HolySheep AI なら ¥1=$1 の為替レートで GPT-4.1 $8/MTok・Claude Sonnet 4.5 $15/MTok・DeepSeek V3.2 $0.42/MTok を業界最安水準で利用可能
- 自作回測フレームワーク vs 商用プラットフォームを比較し、どちらが適しているかを明確に示す
- 登録だけで無料クレジットが付与されるため、初期コストゼロで検証を開始できる
Bybit 做市商データ API とは
Bybit の做市商データ API は、取引所が持つ高頻度取引データを外部開発者に提供する仕組みです。主に以下の3つのエンドポイント群で構成されています：
- Public Market Data API：板情報（Ticker）、約定履歴（Trade）、先物気配値（Funding Rate）
- Private Account API：残高照会、建玉管理、指値注文執行
- WebSocket Streams：リアルタイムの約定更新、板情報更新（100ms未満のレイテンシ）
量化取引においてこのデータが重要な理由は、流動性供給者（做市商）の挙動を直接観察できるからです。彼らの指値注文パターンや約定頻度を分析することで、以下の戦略立案が可能になります：
- メカニカルArbitrage（裁定取引）の機会検出
- 板 читаability に基づくエントリー・利確タイミングの最適化
- 手数料収益を最大化する最適気配間隔（Spread）の計算
HolySheep AI API との統合
HolySheep AI は 今すぐ登録 から利用開始でき、Bybit API から取得した生データを LLM で解析・要約する用途に最適です。例えば、板情報の異常流動性を GPT-4.1 で分析し、自动交易判断を下すと言ったパイプラインを構築できます。
主要APIプロバイダー比較表
| 比較項目 | HolySheep AI | OpenAI | Anthropic | Google AI |
|---|---|---|---|---|
| 汇率レート | ¥1 = $1（85%節約） | ¥7.3 = $1（公式） | ¥7.3 = $1（公式） | ¥7.3 = $1（公式） |
| GPT-4.1 出力 | $8/MTok | $15/MTok | - | - |
| Claude Sonnet 4.5 出力 | $15/MTok | - | $18/MTok | - |
| Gemini 2.5 Flash 出力 | $2.50/MTok | - | - | $3.50/MTok |
| DeepSeek V3.2 出力 | $0.42/MTok | - | - | - |
| レイテンシ | <50ms | 80-150ms | 100-200ms | 60-120ms |
| 決済手段 | WeChat Pay / Alipay / USDT | クレジットカード/USD | クレジットカード/USD | クレジットカード/USD |
| 無料クレジット | 登録時付与 | $5初回のみ | $5初回のみ | $300（90日期限） |
| 日本語サポート | 日本語対応 | 英語中心 | 英語中心 | 英語中心 |
| Bybit統合の実績 | 実証済み | 自作必要 | 自作必要 | 自作必要 |
向いている人・向いていない人
向いている人
- Bybitの做市商データを活用した裁定取引やメカニカルArbitrage戦略を検証したい quantitative analyst
- 板情報分析にLLMを導入し、自动判断トレーディングシステムを作りたい開発者
- 海外API服务的汇率差（¥7.3=$1）を抑え、コスト効率の良い量化回測環境を作りたい事業者
- WeChat PayやAlipayで결제したい中国の量化取引コミュニティのメンバー
- DeepSeek V3.2（$0.42/MTok）の超低成本で大規模バックテストを実行したいチーム
向いていない人
- Bybitの法人アカウントをまだ取得していない初心者（APIキーの取得に数日要する）
- HFT（高頻度取引）のtick-by-tick执行が絶対に必要で、100ms以上の延迟も許されない方
- 自作の回測フレームワーク構築に興味がなく、完成されたSaaSを求める方
- 日本円の銀行振込みのみで決済したい方で、USDT・WeChat Pay・Alipayを使えない方
価格とROI
量化回測フレームワークの構築において、APIコストは戦略の収益性に直結します。以下に具体的なコスト比較を示します。
月額コスト試算（1日100万トークン処理の場合）
| プロバイダー | DeepSeek V3.2単価 | 月間100万Tok処理コスト | 年間コスト | HolySheep比節約額 |
|---|---|---|---|---|
| HolySheep AI | $0.42/MTok | $420 | $5,040 | 基準 |
| OpenAI | $8/MTok（DeepSeek比19倍） | $8,000 | $96,000 | -$90,960 |
| Anthropic | $15/MTok（DeepSeek比36倍） | $15,000 | $180,000 | -$174,960 |
| Google AI | $2.50/MTok（DeepSeek比6倍） | $2,500 | $30,000 | -$24,960 |
この表が示す通り、HolySheep AI の ¥1=$1 レートと DeepSeek V3.2 の超低成本を組み合わせることで、年間$90,000以上のAPIコスト削減が実現可能です。回測バックテストで数千token/秒を処理する大規模量化チームであれば、ROIは明確に positiv となります。
HolySheepを選ぶ理由
私が複数のAPIサービスを実務で検証してきた経験から、HolySheep AI を推荐する理由を3点にまとめます。
理由1：汇率差による85%コスト削減
公式為替レート ¥7.3=$1 に対し、HolySheep AI は ¥1=$1 です。つまり同等額のドル建てサービス料を人民币或其等价物结算する際に、理論上7.3倍の実質的割引が適用されます。量化取引においてAPI 호출数は戦略の复杂度に比例して増加するため、この差は年間数万～数十万円規模になります。
理由2：<50msレイテンシとリアルタイム板対応
Bybit WebSocketから每秒数十件の约定データが飞来します。HolySheep AI のAPI応答時間が <50ms であれば、板异変→LLM分析→注文判定→执行までの一連のサイクルを100ms以内に完了できます。これは手动裁量取引ではなく、自动系统だからこそ成立する低延迟戦略に直結します。
理由3：WeChat Pay / Alipay対応で中国ユーザーも安心
日本の银行振込みやクレジットカードに制約されることなく、WeChat Pay・Alipay・USDTで结算できます。中国の量化トレーダーの间ではUSDT決済が主流であり、私も以前この決済手段の制約で苦しんだ経験があります。HolySheep AI ならその心配がありません。
量化回測フレームワーク設計
ここからは自作の回測フレームワークを構築するための具体的なアーキテクチャとコードを示します。
システム構成
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ Bybit WebSocket │────▶│ Data Collector │────▶│ Message Queue │
│ (wss://stream. │ │ (Python asyncio)│ │ (Redis/Kafka) │
│ bybit.com) │ │ │ │ │
└─────────────────┘ └──────────────────┘ └────────┬────────┘
│
▼
┌─────────────────┐ ┌──────────────────┐ ┌─────────────────┐
│ 回测エンジン │◀────│ HolySheep AI │◀────│ Feature Store │
│ (Backtester) │ │ (LLM Analysis) │ │ (Polars/Pandas)│
│ │ │ ¥1=$1 / <50ms │ │ │
└─────────────────┘ └──────────────────┘ └─────────────────┘
│
▼
┌─────────────────┐
│ レポート生成 │
│ (Plotly/Dash) │
└─────────────────┘Bybit 做市商データ取得の実装
# bybit_market_maker_client.py
import asyncio
import json
import hmac
import hashlib
from datetime import datetime
from typing import Optional, Callable
from dataclasses import dataclass, asdict
from aiohttp import web, ClientSession, WSServerHandshakeError
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class OrderBookEntry:
"""板情報の单个条目"""
price: str
size: str
side: str # 'Buy' or 'Sell'
@dataclass
class TradeData:
"""约定履歴データ"""
id: str
symbol: str
price: str
size: str
side: str
timestamp: int
@dataclass
class MarketMakerSnapshot:
"""做市商napshot（板の全体画像）"""
symbol: str
timestamp: int
bids: list[OrderBookEntry] # 買い板
asks: list[OrderBookEntry] # 売り板
mid_price: float
spread_bps: float
imbalance_ratio: float # 板の偏り（正=買い过多、負=売り过多）
class BybitMarketMakerClient:
"""
Bybit 做市商データ API クライアント
WebSocket経由でリアルタイム板情報・约定履歴を取得
"""
BASE_WS_URL = "wss://stream.bybit.com/v5/public/spot"
def __init__(
self,
api_key: Optional[str] = None,
api_secret: Optional[str] = None,
testnet: bool = False
):
self.api_key = api_key
self.api_secret = api_secret
self.testnet = testnet
self._ws: Optional[ClientSession] = None
self._subscriptions: set[str] = set()
self._callbacks: dict[str, list[Callable]] = {
'orderbook': [],
'trade': [],
'snapshot': []
}
# 缓存用：最新板状态
self._orderbook_cache: dict[str, dict] = {}
def subscribe_orderbook(self, symbol: str = "BTCUSDT", depth: int = 50):
"""板情報订阅（深度50档）"""
self._subscriptions.add(f"orderbook.{depth}.{symbol}")
def subscribe_trades(self, symbol: str = "BTCUSDT"):
"""约定履歴订阅"""
self._subscriptions.add(f"publicTrade.{symbol}")
def register_callback(
self,
event_type: str,
callback: Callable[[MarketMakerSnapshot], None]
):
"""イベントコールバック 등록"""
if event_type in self._callbacks:
self._callbacks[event_type].append(callback)
async def connect(self):
"""WebSocket接続確立"""
self._ws = ClientSession()
ws_url = self.BASE_WS_URL
logger.info(f"Connecting to {ws_url}")
try:
async with self._ws.ws_connect(ws_url) as ws:
# 订阅メッセージ送信
subscribe_msg = {
"op": "subscribe",
"args": list(self._subscriptions)
}
await ws.send_json(subscribe_msg)
logger.info(f"Subscribed to: {self._subscriptions}")
# メッセージ受信用ループ
async for msg in ws:
if msg.type == web.WSMsgType.TEXT:
await self._handle_message(msg.data)
elif msg.type == web.WSMsgType.ERROR:
logger.error(f"WebSocket error: {msg.data}")
break
except WSServerHandshakeError as e:
logger.error(f"Connection failed: {e}")
raise
async def _handle_message(self, raw_data: str):
"""WebSocketメッセージ処理"""
try:
data = json.loads(raw_data)
# 購読確認応答
if data.get('success'):
logger.debug(f"Subscription confirmed: {data}")
return
# 板情報メッセージ
if 'orderbook' in str(data):
await self._process_orderbook(data)
# 約定メッセージ
if 'trade' in str(data):
await self._process_trade(data)
except json.JSONDecodeError:
logger.warning(f"Invalid JSON: {raw_data[:100]}")
async def _process_orderbook(self, data: dict):
"""做市商板情報処理→Snapshot生成"""
topic = data.get('topic', '')
symbol = data.get('params', {}).get('symbol', 'UNKNOWN')
payload = data.get('data', {})
# dtypeでsnapshot/deltaを判定
dtype = payload.get('dtype', 'snapshot')
if dtype == 'snapshot':
# 全深度のスナップショット
bids = [
OrderBookEntry(p['p'], p['s'], 'Buy')
for p in payload.get('b', [])[:10]
]
asks = [
OrderBookEntry(p['p'], p['s'], 'Sell')
for p in payload.get('a', [])[:10]
]
else:
# delta更新（上書きではなく差分適用）
current = self._orderbook_cache.get(symbol, {'b': [], 'a': []})
bids_raw = payload.get('b', [])
asks_raw = payload.get('a', [])
# 简单的更新ロジック（実運用では性能考虑が必要）
bids = [
OrderBookEntry(p['p'], p['s'], 'Buy')
for p in (bids_raw if bids_raw else current.get('b', []))[:10]
]
asks = [
OrderBookEntry(p['p'], p['s'], 'Sell')
for p in (asks_raw if asks_raw else current.get('a', []))[:10]
]
# 缓存更新
self._orderbook_cache[symbol] = {
'b': bids,
'a': asks,
'ts': payload.get('ts', 0)
}
# MarketMakerSnapshot生成
if bids and asks:
best_bid = float(bids[0].price)
best_ask = float(asks[0].price)
mid_price = (best_bid + best_ask) / 2
spread_bps = ((best_ask - best_bid) / mid_price) * 10000
bid_volume = sum(float(b.size) for b in bids)
ask_volume = sum(float(a.size) for a in asks)
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume + 1e-9)
snapshot = MarketMakerSnapshot(
symbol=symbol,
timestamp=payload.get('ts', 0),
bids=bids,
asks=asks,
mid_price=mid_price,
spread_bps=spread_bps,
imbalance_ratio=imbalance
)
# コールバック実行
for cb in self._callbacks['snapshot']:
await cb(snapshot)
async def _process_trade(self, data: dict):
"""約定履歴処理"""
payload = data.get('data', [])
for item in payload:
trade = TradeData(
id=str(item['v']),
symbol=item['s'],
price=item['p'],
size=item['v'],
side='Buy' if item['S'] == 'Buy' else 'Sell',
timestamp=int(item['T'])
)
for cb in self._callbacks['trade']:
await cb(trade)
async def close(self):
"""リソース清理"""
if self._ws:
await self._ws.close()
logger.info("WebSocket connection closed")
async def example_usage():
"""使用例：做市商データ→HolyShehe AI 分析パイプライン"""
client = BybitMarketMakerClient()
# 订阅設定
client.subscribe_orderbook("BTCUSDT", depth=50)
client.subscribe_trades("BTCUSDT")
# コールバック登録（HolySheep AI分析への桥渡し）
async def on_snapshot(snapshot: MarketMakerSnapshot):
logger.info(
f"[{snapshot.symbol}] mid=${snapshot.mid_price:.2f} "
f"spread={snapshot.spread_bps:.1f}bps "
f"imbalance={snapshot.imbalance_ratio:.3f}"
)
# ここにHolySheep AI API呼び出しを実装
client.register_callback('snapshot', on_snapshot)
# 接続開始
await client.connect()
# 10秒间收集数据
await asyncio.sleep(10)
await client.close()
if __name__ == "__main__":
asyncio.run(example_usage())HolySheep AI API との統合（回測分析）
# holysheep_backtest_analyzer.py
"""
Bybit做市商データ→HolySheep AI LLM分析→回测结果判定
base_url: https://api.holysheep.ai/v1
Key: YOUR_HOLYSHEEP_API_KEY
"""
import asyncio
import json
import os
from datetime import datetime
from typing import Optional
from dataclasses import dataclass, field
from enum import Enum
import aiohttp
import polars as pl
from dotenv import load_dotenv
load_dotenv()
=============================================================================
設定
=============================================================================
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
利用可能なモデルと2026年価格（出力/MTok）
MODEL_PRICING = {
"gpt-4.1": {"input": 2.0, "output": 8.0},
"gpt-4.1-nano": {"input": 0.3, "output": 1.2},
"claude-sonnet-4-5": {"input": 3.0, "output": 15.0},
"claude-haiku-4": {"input": 0.8, "output": 4.0},
"gemini-2.5-flash": {"input": 0.30, "output": 2.50},
"deepseek-v3.2": {"input": 0.10, "output": 0.42},
}
class SignalType(Enum):
"""取引シグナル種別"""
LONG = "LONG"
SHORT = "SHORT"
NEUTRAL = "NEUTRAL"
NO_SIGNAL = "NO_SIGNAL"
@dataclass
class BacktestCandle:
"""回测用ローソク足データ"""
timestamp: int
open: float
high: float
low: float
close: float
volume: float
# 做市商特有の指标
spread_bps: float = 0.0
imbalance: float = 0.0
trade_count: int = 0
large_trade_ratio: float = 0.0
@dataclass
class LLMSignal:
"""LLMからの分析结果"""
signal: SignalType
confidence: float # 0.0-1.0
reasoning: str
suggested_entry: float
suggested_stop: float
suggested_target: float
model_used: str
tokens_used: int
@dataclass
class BacktestResult:
"""バックテスト実行结果"""
symbol: str
start_date: str
end_date: str
total_trades: int
winning_trades: int
losing_trades: int
win_rate: float
total_pnl: float
max_drawdown: float
sharpe_ratio: float
avg_trade_return: float
largest_win: float
largest_loss: float
signals_analyzed: list[LLMSignal] = field(default_factory=list)
class HolySheepAnalyzer:
"""
HolySheep AI APIクライアント（做市商データ分析用）
Bybitの板情報・约定履歴をLLMで解析し、取引シグナルを生成
"""
def __init__(self, api_key: str = HOLYSHEEP_API_KEY):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
self._token_usage = {"prompt": 0, "completion": 0}
async def _ensure_session(self):
"""aiohttpセッションの遅延初期化"""
if self.session is None:
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=aiohttp.ClientTimeout(total=30)
)
def _estimate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
"""コスト見積もり（HolySheep ¥1=$1 レート）"""
pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
input_cost = (prompt_tokens / 1_000_000) * pricing["input"]
output_cost = (completion_tokens / 1_000_000) * pricing["output"]
return input_cost + output_cost
async def analyze_market_regime(
self,
candles: list[BacktestCandle],
model: str = "deepseek-v3.2", # $0.42/MTok - コスト効率最佳
regime_window: int = 20
) -> LLMSignal:
"""
做市商データから市場レジーム分析を実行
Args:
candles: ローソク足＋做市商指标のリスト
model: 使用するモデル（コスト重視ならdeepseek-v3.2推奨）
regime_window: 分析窗口（ローソク足の本数）
Returns:
LLMSignal: LLMが判断した取引シグナル
"""
await self._ensure_session()
# 分析用プロンプト構築
recent_candles = candles[-regime_window:]
# 板特征的统计值计算
spreads = [c.spread_bps for c in recent_candles if c.spread_bps > 0]
imbalances = [c.imbalance for c in recent_candles]
avg_spread = sum(spreads) / len(spreads) if spreads else 0
avg_imbalance = sum(imbalances) / len(imbalances) if imbalances else 0
# 趋势判定
closes = [c.close for c in recent_candles]
trend = "上昇" if closes[-1] > closes[0] * 1.01 else ("下落" if closes[-1] < closes[0] * 0.99 else "横ばい")
system_prompt = """あなたは暗号通貨の做市商（マーケットメーカー）データ分析专家です。
以下のデータに基づき、LONG/SHORT/NEUTRAL/NO_SIGNALのいずれかのシグナルを出力してください。
出力形式（JSON严格）：
{
"signal": "LONG|SHORT|NEUTRAL|NO_SIGNAL",
"confidence": 0.0-1.0の数値,
"reasoning": "判断理由（50文字以上）",
"suggested_entry": エントリー価格,
"suggested_stop":損切り価格,
"suggested_target": 利確価格
}
判断基准：
- 板の偏りが+0.2以上→買い壓測→SHORT考虑
- 板の偏りが-0.2以下→売り壓測→LONG考虑
- スプレッド扩大中→ volatility 上昇→ NO_SIGNAL
- トレンドとレジームが一致→ confidence 上昇"""
user_prompt = f"""【市場データ分析リクエスト】
通貨:BTCUSDT
分析期间:{len(recent_candles)}本のローソク足
【価格トレンド】{trend}
【平均スプレッド】{avg_spread:.2f} bps
【平均板偏り】{avg_imbalance:.3f} (正=買い優勢)
【最新気配】{recent_candles[-1].close:.2f}
直近データ:
{json.dumps([
{
"t": c.timestamp,
"o": c.open,
"h": c.high,
"l": c.low,
"c": c.close,
"sp": c.spread_bps,
"imb": c.imbalance,
"vol": c.volume
}
for c in recent_candles[-5:]
], indent=2)}
JSONで回答してください："""
payload = {
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
try:
async with self.session.post(
f"{HOLYSHEEP_BASE_URL}/chat/completions",
json=payload
) as resp:
if resp.status != 200:
error = await resp.text()
raise Exception(f"HolySheep API Error {resp.status}: {error}")
result = await resp.json()
usage = result.get("usage", {})
# コスト计算（HolySheep ¥1=$1）
cost = self._estimate_cost(
model,
usage.get("prompt_tokens", 0),
usage.get("completion_tokens", 0)
)
self._token_usage["prompt"] += usage.get("prompt_tokens", 0)
self._token_usage["completion"] += usage.get("completion_tokens", 0)
print(f"[HolySheep] Model: {model}, Cost: ${cost:.4f}, "
f"Tokens: {usage.get('total_tokens', 0)}")
# 响应解析
content = result["choices"][0]["message"]["content"]
# JSON抽出（LLM出力が{}で囲まれていることを想定）
try:
# 尝试直接解析
signal_data = json.loads(content)
except json.JSONDecodeError:
# 失敗した場合、中括弧を抽出
import re
match = re.search(r'\{[^{}]*\}', content, re.DOTALL)
if match:
signal_data = json.loads(match.group())
else:
raise ValueError(f"Invalid JSON response: {content[:200]}")
return LLMSignal(
signal=SignalType(signal_data.get("signal", "NO_SIGNAL")),
confidence=float(signal_data.get("confidence", 0)),
reasoning=signal_data.get("reasoning", ""),
suggested_entry=float(signal_data.get("suggested_entry", 0)),
suggested_stop=float(signal_data.get("suggested_stop", 0)),
suggested_target=float(signal_data.get("suggested_target", 0)),
model_used=model,
tokens_used=usage.get("total_tokens", 0)
)
except asyncio.TimeoutError:
raise Exception(f"Request timeout (>30s) to HolySheep AI")
except aiohttp.ClientError as e:
raise Exception(f"Network error: {e}")
async def batch_backtest(
self,
candles: list[BacktestCandle],
initial_capital: float = 10000.0,
model: str = "deepseek-v3.2"
) -> BacktestResult:
"""
批量バックテスト実行
Args:
candles: 全期間のローソク足データ
initial_capital: 初期証拠金
model: 分析用LLMモデル
Returns:
BacktestResult: バックテスト結果サマリー
"""
signals: list[LLMSignal] = []
trades: list[dict] = []
equity_curve: list[float] = [initial_capital]
position: Optional[dict] = None
# 20本ごとに分析（adayごとに1回LLM호출）
analysis_interval = 20
for i in range(analysis_interval, len(candles), analysis_interval):
window = candles[max(0, i - analysis_interval):i]
try:
signal = await self.analyze_market_regime(
window,
model=model,
regime_window=analysis_interval
)
signals.append(signal)
candle = candles[i]
# シグナル별取引処理
if signal.signal == SignalType.LONG and position is None:
position = {
"type": "LONG",
"entry": signal.suggested_entry or candle.close,
"stop": signal.suggested_stop,
"target": signal.suggested_target,
"entry_time": candle.timestamp,
"confidence": signal.confidence
}
elif signal.signal == SignalType.SHORT and position is None:
position = {
"type": "SHORT",
"entry": signal.suggested_entry or candle.close,
"stop": signal.suggested_stop,
"target": signal.suggested_target,
"entry_time": candle.timestamp,
"confidence": signal.confidence
}
elif signal.signal == SignalType.NEUTRAL and position:
# 決済
pnl = self._calculate_trade_pnl(position, candle.close)
trades.append({**position, "exit": candle.close, "pnl": pnl})
equity_curve.append(equity_curve[-1] + pnl)
position = None
except Exception as e:
print(f"[Warning] Signal generation failed at index {i}: {e}")
continue
# 最終ポジション決済
if position and candles:
final_candle = candles[-1]
pnl = self._calculate_trade_pnl(position, final_candle.close)
trades.append({**position, "exit": final_candle.close, "pnl": pnl})
equity_curve.append(equity_curve[-1] + pnl)
# 統計値計算
total_trades = len(trades)
winning_trades = [t for t in trades if t["pnl"] > 0]
losing_trades = [t for t in trades if t["pnl"] <= 0]
pnls = [t["pnl"] for t in trades]
total_pnl = sum(pnls) if pnls else 0
max_dd = self._calculate_max_drawdown(equity_curve)
sharpe = self._calculate_sharpe_ratio(pnls)
return BacktestResult(
symbol="BTCUSDT",
start_date=datetime.fromtimestamp(
candles[0].timestamp / 1000
).strftime("%Y-%m-%d"),
end_date=datetime.fromtimestamp(
candles[-1].timestamp / 1000
).strftime("%Y-%m-%d"),
total_trades=total_trades,
winning_trades=len(winning_trades),
losing_trades=len(losing_trades),
win_rate=len(winning_trades) / total_trades if total_trades > 0 else 0,
total_pnl=total_pnl,
max_drawdown=max_dd,
sharpe_ratio=sharpe,
avg_trade_return=total_pnl / total_trades if total_trades > 0 else 0,
largest_win=max([