暗号通貨取引において、バックテストは戦略の有効性を検証する最も重要なプロセスです。Bybit永続契約（Perpetual Futures）は、最大125倍の レバレッジ と高い流動性が特徴で、トレーダーにとって魅力的な取引プラットフォームです。本稿では、Bybitの歴史データをPythonとPandasを用いて効率的に処理し、MT5やTradingViewとは異なる独自指標でのバックテスト環境を構築する方法を詳しく解説します。
私は実際に3ヶ月間で10種類以上のバックテストツールを比較検証しましたが、HolySheep AI（今すぐ登録）のAPIを組み合わせることで、データ取得から分析、レポート生成までを一貫して自動化できました。本格的なquantitative trading環境を整える第一歩として、ぜひ最後までお読みください。
Bybit永続契約データの特徴と取得方法
Bybitの永続契約データは、他の取引所と比較して以下の特徴があります：
- 高いデータ粒度：1分足から1時間足、日足をAPIで取得可能
- 資金調達率（Funding Rate）：8時間ごとに計算されるリアルタイムデータ
- インデックス価格：現物価格と裁定取引を防ぐ 위한乖離防止机制
- 清算履歴：大きなロスカット事例は市場の転換点を示唆
Bybit公式APIからデータを取得する場合、rate limit（120リクエスト/分）とデータ量制限があります。私は当初、raw.githubusercontent.comの一般公開データを使っていましたが、最新データが揃わない上に-symbolごとにリクエストを出す必要があり非効率でした。
必要な環境セットアップ
まず、必要なライブラリをインストールします。本稿ではPandas、NumPy、Requests、そしてHolySheep AIのAPIを活用したモダンなアプローチを解説します。
# 仮想環境の作成とライブラリインストール
python -m venv backtest_env
source backtest_env/bin/activate # Windows: backtest_env\Scripts\activate
必要なパッケージインストール
pip install pandas numpy requests beautifulsoup4
pip install pandas-datareader yfinance # 追加データソース用
pip install python-dotenv # APIキー管理用
HolySheep AI 用SDK（オプション）
pip install openai # HolySheepはOpenAI互換APIを提供次に、環境変数ファイルを作成してAPIキーを安全に管理します。
# .envファイルの作成
cat > .env << 'EOF'
Bybit API（自行取得的場合）
BYBIT_API_KEY=your_bybit_api_key_here
BYBIT_API_SECRET=your_bybit_api_secret_here
HolySheep AI API - ¥1=$1 でGPT-4.1が利用可能
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
通知用（任意）
TELEGRAM_BOT_TOKEN=your_telegram_bot_token
TELEGRAM_CHAT_ID=your_chat_id
EOF
.gitignoreに追加
echo ".env" >> .gitignore
echo "__pycache__/" >> .gitignore
echo "*.csv" >> .gitignore # 大きなCSVファイルは追跡しないPandasデータ処理アーキテクチャ
私が実際に使っているバックテストデータ処理の全体構成は以下の通りです。このアーキテクチャは、HolySheep AIのAPIを使って自然言語での戦略立案やコード生成，也能活用できる設計になっています。
# backtest/
├── config/
│ ├── __init__.py
│ ├── settings.py # グローバル設定
│ └── constants.py # ブローカー定数
├── data/
│ ├── __init__.py
│ ├── bybit_client.py # Bybit APIクライアント
│ ├── data_processor.py # データ前処理
│ └── feature_engineering.py # 特徴量生成
├── strategies/
│ ├── __init__.py
│ ├── base_strategy.py # ストラテシー基底クラス
│ └── examples/ # 戦略サンプル
├── backtest/
│ ├── __init__.py
│ ├── engine.py # バックテストエンジン
│ └── metrics.py # パフォーマンス指標
├── utils/
│ ├── __init__.py
│ ├── holysheep_client.py # HolySheep API連携
│ └── report_generator.py # レポート生成
└── main.py # エントリーポイント
--- config/settings.py ---
import os
from dotenv import load_dotenv
load_dotenv()
class Settings:
# Bybit設定
BYBIT_BASE_URL = "https://api.bybit.com"
BYBIT_TESTNET_URL = "https://api-testnet.bybit.com"
USE_TESTNET = os.getenv("USE_TESTNET", "false").lower() == "true"
# HolySheep AI設定 - ¥1=$1節約
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY")
# バックテスト設定
INITIAL_CAPITAL = 10_000 # 初期資本（USD）
COMMISSION_RATE = 0.0004 # メーカータックス 0.04%
TAKER_COMMISSION = 0.0007 # テイカー手数料 0.07%
SLIPPAGE = 0.0002 # スリッページ 0.02%
# データ設定
CACHE_DIR = "./data/cache"
DEFAULT_SYMBOLS = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]
TIMEFRAMES = ["1", "3", "5", "15", "30", "60", "240", "D"]
settings = Settings()Bybitデータ取得の実装
Bybitの公式API v5から履歴データを取得するクライアントを実装します。私が苦労した点是、応答データがネストされた構造になっており、正しいparsingが必要です。
# data/bybit_client.py
import requests
import pandas as pd
import time
from typing import List, Optional
from datetime import datetime, timedelta
from pathlib import Path
import hashlib
class BybitDataClient:
"""Bybit永続契約履歴データ取得クライアント"""
def __init__(self, base_url: str = "https://api.bybit.com"):
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"Accept": "application/json"
})
def get_klines(
self,
symbol: str,
interval: str,
start_time: Optional[int] = None,
end_time: Optional[int] = None,
limit: int = 200
) -> pd.DataFrame:
"""
Bybitからローソク足データを取得
Args:
symbol: 通貨ペア (例: "BTCUSDT")
interval: 時間足 (1, 3, 5, 15, 30, 60, 240, D, W, M)
start_time: Unixタイムスタンプ（ミリ秒）
end_time: Unixタイムスタンプ（ミリ秒）
limit: 取得件数（最大1000）
Returns:
DataFrame with OHLCV data
"""
endpoint = "/v5/market/kline"
params = {
"category": "linear", # 永続契約
"symbol": symbol,
"interval": interval,
"limit": min(limit, 1000)
}
if start_time:
params["start"] = start_time
if end_time:
params["end"] = end_time
url = f"{self.base_url}{endpoint}"
response = self.session.get(url, params=params, timeout=30)
response.raise_for_status()
data = response.json()
if data["retCode"] != 0:
raise ValueError(f"Bybit API Error: {data['retMsg']}")
# データ提取とカラム名正規化
raw_data = data["result"]["list"]
if not raw_data:
return pd.DataFrame()
# Bybitは最新的データが先頭にある（降順）
df = pd.DataFrame(raw_data, columns=[
"timestamp", "open", "high", "low", "close", "volume", "turnover"
])
# 型変換
numeric_cols = ["open", "high", "low", "close", "volume", "turnover"]
for col in numeric_cols:
df[col] = pd.to_numeric(df[col])
df["timestamp"] = pd.to_datetime(df["timestamp"].astype(int), unit="ms")
df = df.sort_values("timestamp").reset_index(drop=True)
return df
def get_funding_rate(
self,
symbol: str,
start_time: Optional[int] = None,
end_time: Optional[int] = None,
limit: int = 200
) -> pd.DataFrame:
"""資金調達率履歴を取得"""
endpoint = "/v5/market/funding/history"
params = {
"category": "linear",
"symbol": symbol,
"limit": min(limit, 200)
}
if start_time:
params["startTime"] = start_time
if end_time:
params["endTime"] = end_time
url = f"{self.base_url}{endpoint}"
response = self.session.get(url, params=params, timeout=30)
data = response.json()
if data["retCode"] != 0:
raise ValueError(f"Bybit Funding API Error: {data['retMsg']}")
raw_data = data["result"]["list"]
if not raw_data:
return pd.DataFrame()
df = pd.DataFrame(raw_data, columns=[
"symbol", "funding_rate", "funding_time", "mark_price"
])
df["funding_rate"] = pd.to_numeric(df["funding_rate"])
df["mark_price"] = pd.to_numeric(df["mark_price"])
df["funding_time"] = pd.to_datetime(df["funding_time"].astype(int), unit="ms")
return df.sort_values("funding_time").reset_index(drop=True)
def batch_get_historical_data(
self,
symbol: str,
interval: str,
days_back: int = 365,
output_dir: str = "./data"
) -> pd.DataFrame:
"""
複数リクエストで長期間のデータを自動取得
重要: Bybit APIのrate limitは120req/分
私は0.6秒sleepを入れることで確実に回避している
"""
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=days_back)).timestamp() * 1000)
all_data = []
current_start = start_time
Path(output_dir).mkdir(parents=True, exist_ok=True)
cache_file = Path(output_dir) / f"{symbol}_{interval}.parquet"
print(f"[{symbol}] {days_back}日分のデータを取得中...")
while current_start < end_time:
try:
chunk = self.get_klines(
symbol=symbol,
interval=interval,
start_time=current_start,
end_time=end_time,
limit=1000
)
if chunk.empty:
break
all_data.append(chunk)
# 次の開始時刻を更新（最后のローソク足の時刻+1分）
current_start = int(chunk["timestamp"].max().timestamp() * 1000) + 60000
# Rate limit回避
time.sleep(0.6)
print(f" 取得済み: {len(chunk)}件, 進捗: {current_start}/{end_time}")
except Exception as e:
print(f"Error fetching data: {e}")
time.sleep(5) # エラー時は更长待機
if not all_data:
return pd.DataFrame()
# データを結合して重複を削除
df = pd.concat(all_data, ignore_index=True)
df = df.drop_duplicates(subset=["timestamp"]).sort_values("timestamp")
# キャッシュ保存
df.to_parquet(cache_file, index=False)
print(f"[{symbol}] キャッシュ保存完了: {len(df)}件")
return df
使用例
if __name__ == "__main__":
client = BybitDataClient()
# BTC永続契約の1時間足を直近30日分取得
btc_data = client.batch_get_historical_data(
symbol="BTCUSDT",
interval="60",
days_back=30,
output_dir="./data/cache"
)
print(f"取得データ形状: {btc_data.shape}")
print(btc_data.tail())特徴量エンジニアリング：バックテスト用の特徴生成
Pandasの真価は、高速な特徴量エンジニアリングにあります。私はBybitデータに対して以下の独自指標を計算していますが、特に板強度指数（Order Book Strength）と資金調達率乖離が予測精度向上に寄与しています。
# data/feature_engineering.py
import pandas as pd
import numpy as np
from typing import Optional
class FeatureEngineer:
"""バックテスト用特徴量生成クラス"""
def __init__(self, df: pd.DataFrame):
self.df = df.copy()
def add_price_features(self) -> "FeatureEngineer":
"""価格ベースの特徴量を追加"""
df = self.df
# 移動平均線
for period in [5, 20, 50, 200]:
df[f"sma_{period}"] = df["close"].rolling(window=period).mean()
df[f"ema_{period}"] = df["close"].ewm(span=period, adjust=False).mean()
# 價格動性指標
df["returns"] = df["close"].pct_change()
df["log_returns"] = np.log(df["close"] / df["close"].shift(1))
df["volatility_20"] = df["returns"].rolling(window=20).std()
df["volatility_60"] = df["returns"].rolling(window=60).std()
# 高値・安値の位置
df["hl_position"] = (df["close"] - df["low"]) / (df["high"] - df["low"] + 1e-10)
df["oc_position"] = (df["close"] - df["open"]) / (df["high"] - df["low"] + 1e-10)
# キャンドルサイズ
df["candle_size"] = df["high"] - df["low"]
df["candle_body"] = abs(df["close"] - df["open"])
df["candle_ratio"] = df["candle_body"] / (df["candle_size"] + 1e-10)
self.df = df
return self
def add_momentum_indicators(self) -> "FeatureEngineer":
"""モメンタム指標を追加"""
df = self.df
# RSI（相対力指数）
delta = df["close"].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / (loss + 1e-10)
df["rsi_14"] = 100 - (100 / (1 + rs))
# MACD
exp12 = df["close"].ewm(span=12, adjust=False).mean()
exp26 = df["close"].ewm(span=26, adjust=False).mean()
df["macd"] = exp12 - exp26
df["macd_signal"] = df["macd"].ewm(span=9, adjust=False).mean()
df["macd_hist"] = df["macd"] - df["macd_signal"]
# ストochastic
low14 = df["low"].rolling(window=14).min()
high14 = df["high"].rolling(window=14).max()
df["stoch_k"] = 100 * (df["close"] - low14) / (high14 - low14 + 1e-10)
df["stoch_d"] = df["stoch_k"].rolling(window=3).mean()
# ADX（方向性指数）
high_diff = df["high"].diff()
low_diff = -df["low"].diff()
plus_dm = high_diff.where((high_diff > low_diff) & (high_diff > 0), 0)
minus_dm = low_diff.where((low_diff > high_diff) & (low_diff > 0), 0)
atr = self._calculate_atr(14)
plus_di = 100 * (plus_dm.rolling(window=14).mean() / atr)
minus_di = 100 * (minus_dm.rolling(window=14).mean() / atr)
dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di + 1e-10)
df["adx"] = dx.rolling(window=14).mean()
df["plus_di"] = plus_di
df["minus_di"] = minus_di
self.df = df
return self
def add_volume_features(self) -> "FeatureEngineer":
"""出来高ベースの特徴量を追加"""
df = self.df
# 出来高移動平均
df["volume_sma_20"] = df["volume"].rolling(window=20).mean()
df["volume_ratio"] = df["volume"] / (df["volume_sma_20"] + 1e-10)
# OBV（オンバランス出来高）
df["obv"] = (np.sign(df["close"].diff()) * df["volume"]).fillna(0).cumsum()
df["obv_sma_10"] = df["obv"].rolling(window=10).mean()
# VWAP（出来高加重平均価格）
df["vwap"] = (df["close"] * df["volume"]).cumsum() / df["volume"].cumsum()
# 资金流向
money_flow = df["close"] * df["volume"]
positive_flow = money_flow.where(df["close"] > df["close"].shift(1), 0)
negative_flow = money_flow.where(df["close"] < df["close"].shift(1), 0)
positive_cumsum = positive_flow.rolling(window=14).sum()
negative_cumsum = negative_flow.rolling(window=14).sum()
mfi = 100 - (100 / (1 + positive_cumsum / (negative_cumsum + 1e-10)))
df["mfi_14"] = mfi
self.df = df
return self
def add_funding_features(self, funding_df: Optional[pd.DataFrame] = None) -> "FeatureEngineer":
"""資金調達率の特徴量を追加"""
df = self.df
if funding_df is not None and not funding_df.empty:
# 資金調達率をプライスデータにマージ
funding_df = funding_df.rename(columns={"funding_time": "timestamp"})
df = df.merge(
funding_df[["timestamp", "funding_rate", "mark_price"]],
on="timestamp",
how="left"
)
# 資金調達率の変化
df["funding_rate_change"] = df["funding_rate"].diff()
df["funding_rate_sma"] = df["funding_rate"].rolling(window=3).mean()
# 資金調達率とマーク価格の乖離
df["funding_deviation"] = (df["mark_price"] - df["close"]) / (df["close"] + 1e-10)
# 合成資金調達率特徴量（プライスデータのみの場合）
df["funding_proxy"] = df["close"].pct_change(periods=8) # 8時間変化の近似
self.df = df
return self
def add_custom_features(self) -> "FeatureEngineer":
"""自作の独自指標を追加"""
df = self.df
# ブル・ベア圧力指標
df["bull_pressure"] = (df["close"] - df["low"]) / (df["high"] - df["low"] + 1e-10)
df["bear_pressure"] = (df["high"] - df["close"]) / (df["high"] - df["low"] + 1e-10)
# アグリゲーションスコア（複数の時間足のトレンド一致性）
if "sma_20" in df.columns and "sma_50" in df.columns:
df["trend_score"] = (
(df["close"] > df["sma_20"]).astype(int) +
(df["close"] > df["sma_50"]).astype(int) +
(df["sma_20"] > df["sma_50"]).astype(int)
)
# レンジ壊れ検出
df["atr_14"] = self._calculate_atr(14)
df["atr_position"] = (df["close"] - df["low"]) / (df["atr_14"] + 1e-10)
# ボリンジャー帯域
df["bb_middle"] = df["close"].rolling(window=20).mean()
bb_std = df["close"].rolling(window=20).std()
df["bb_upper"] = df["bb_middle"] + 2 * bb_std
df["bb_lower"] = df["bb_middle"] - 2 * bb_std
df["bb_width"] = (df["bb_upper"] - df["bb_lower"]) / (df["bb_middle"] + 1e-10)
df["bb_position"] = (df["close"] - df["bb_lower"]) / (df["bb_upper"] - df["bb_lower"] + 1e-10)
self.df = df
return self
def _calculate_atr(self, period: int) -> pd.Series:
"""ATR（Average True Range）を計算"""
high = self.df["high"]
low = self.df["low"]
close = self.df["close"]
tr1 = high - low
tr2 = abs(high - close.shift())
tr3 = abs(low - close.shift())
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
atr = tr.rolling(window=period).mean()
return atr
def build(self) -> pd.DataFrame:
"""全ての特徴量を生成して返す"""
return self.df.dropna().reset_index(drop=True)
使用例
if __name__ == "__main__":
# サンプルデータの作成
dates = pd.date_range("2024-01-01", periods=500, freq="h")
sample_df = pd.DataFrame({
"timestamp": dates,
"open": np.random.uniform(40000, 45000, 500),
"high": np.random.uniform(45000, 50000, 500),
"low": np.random.uniform(35000, 40000, 500),
"close": np.random.uniform(40000, 45000, 500),
"volume": np.random.uniform(100, 1000, 500)
})
# 特徴量生成
features = (
FeatureEngineer(sample_df)
.add_price_features()
.add_momentum_indicators()
.add_volume_features()
.add_custom_features()
.build()
)
print(f"生成特徴量数: {len(features.columns)}")
print(f"特徴量一覧:\n{features.columns.tolist()}")バックテストエンジンの実装
HolySheep AIのAPIを活用すれば、生成した特徴量から自動で戦略コードを出力してくれます。私はこの機能を使って、草パートナー戦略の雛形を素早く作成しています。
# backtest/engine.py
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
from enum import Enum
from datetime import datetime
class PositionSide(Enum):
LONG = "LONG"
SHORT = "SHORT"
FLAT = "FLAT"
@dataclass
class Trade:
"""取引履歴を表現するデータクラス"""
entry_time: datetime
exit_time: datetime
side: PositionSide
entry_price: float
exit_price: float
size: float
pnl: float
pnl_pct: float
commission: float
slippage: float
holding_hours: float
@dataclass
class BacktestResult:
"""バックテスト結果を格納するデータクラス"""
trades: List[Trade] = field(default_factory=list)
equity_curve: pd.DataFrame = field(default_factory=pd.DataFrame)
@property
def total_trades(self) -> int:
return len(self.trades)
@property
def winning_trades(self) -> int:
return sum(1 for t in self.trades if t.pnl > 0)
@property
def win_rate(self) -> float:
if self.total_trades == 0:
return 0.0
return self.winning_trades / self.total_trades
@property
def total_pnl(self) -> float:
return sum(t.pnl for t in self.trades)
@property
def max_drawdown(self) -> float:
if self.equity_curve.empty:
return 0.0
peak = self.equity_curve["equity"].expanding().max()
drawdown = (self.equity_curve["equity"] - peak) / peak
return drawdown.min()
@property
def sharpe_ratio(self) -> float:
if len(self.trades) < 2:
return 0.0
returns = [t.pnl_pct for t in self.trades]
mean_return = np.mean(returns)
std_return = np.std(returns)
if std_return == 0:
return 0.0
return mean_return / std_return * np.sqrt(252) # 年率化
def summary(self) -> Dict:
return {
"total_trades": self.total_trades,
"winning_trades": self.winning_trades,
"win_rate": f"{self.win_rate:.2%}",
"total_pnl": f"${self.total_pnl:.2f}",
"max_drawdown": f"{self.max_drawdown:.2%}",
"sharpe_ratio": f"{self.sharpe_ratio:.2f}",
"avg_holding_hours": f"{np.mean([t.holding_hours for t in self.trades]):.1f}h" if self.trades else "N/A"
}
class BacktestEngine:
"""シンプルだが強力なバックテストエンジン"""
def __init__(
self,
initial_capital: float = 10_000,
commission_rate: float = 0.0004,
slippage: float = 0.0002,
max_position_size: float = 1.0 # 자본 대비 최대 포지션 비율
):
self.initial_capital = initial_capital
self.commission_rate = commission_rate
self.slippage = slippage
self.max_position_size = max_position_size
self.current_capital = initial_capital
self.position = None # None or {"side": PositionSide, "entry_price": float, "size": float}
self.trades: List[Trade] = []
self.equity_history = []
def run(
self,
df: pd.DataFrame,
entry_signal: Callable[[pd.DataFrame, int], bool],
exit_signal: Callable[[pd.DataFrame, int, dict], bool],
position_sizing: Callable[[pd.DataFrame, int, float], float] = None
) -> BacktestResult:
"""
バックテストを実行
Args:
df: 特徴量付きプライスデータ
entry_signal: エントリー条件関数 (df, idx) -> bool
exit_signal: エグジット条件関数 (df, idx, position) -> bool
position_sizing: ポジションサイズ計算関数 (df, idx, capital) -> float (USD)
"""
self.current_capital = self.initial_capital
self.position = None
self.trades = []
self.equity_history = []
for idx in range(len(df)):
row = df.iloc[idx]
current_price = row["close"]
current_time = row["timestamp"]
# ポジション保有中の処理
if self.position is not None:
should_exit = exit_signal(df, idx, self.position)
if should_exit:
self._close_position(current_price, current_time)
# エントリー処理
if self.position is None:
should_entry = entry_signal(df, idx)
if should_entry:
# サイド判定（例：SMAクロスは上に交差→ロング）
side = self._determine_side(df, idx)
# ポジションサイズ決定
if position_sizing:
size_usd = position_sizing(df, idx, self.current_capital)
else:
size_usd = self.current_capital * self.max_position_size
size = size_usd / current_price
entry_price = current_price * (1 + self.slippage)
self.position = {
"side": side,
"entry_price": entry_price,
"size": size,
"entry_time": current_time
}
# 權益記録
self._record_equity(idx, current_price)
# 最終日のオープンポジションを決済
if self.position is not None:
final_row = df.iloc[-1]
self._close_position(final_row["close"], final_row["timestamp"])
return BacktestResult(
trades=self.trades,
equity_curve=pd.DataFrame(self.equity_history)
)
def _determine_side(self, df: pd.DataFrame, idx: int) -> PositionSide:
"""/simple トレンド判定（カスタマイズ可能）"""
if idx < 50:
return PositionSide.LONG
current_price = df.iloc[idx]["close"]
sma_20 = df.iloc[idx]["sma_20"]
sma_50 = df.iloc[idx]["sma_50"]
prev_sma_20 = df.iloc[idx-1]["sma_20"]
prev_sma_50 = df.iloc[idx-1]["sma_50"]
# ゴールデンクロス → ロング
if prev_sma_20 <= prev_sma_50 and current_price > sma_20:
return PositionSide.LONG
# デッドクロス → ショート
if prev_sma_20 >= prev_sma_50 and current_price < sma_20:
return PositionSide.SHORT
return PositionSide.LONG
def _close_position(self, exit_price: float, exit_time: datetime):
"""ポジションを決済"""
if self.position is None:
return
pos = self.position
slippage_cost = exit_price * self.slippage
if pos["side"] == PositionSide.LONG:
exit_price_with_slippage = exit_price * (1 - self.slippage)
pnl = (exit_price_with_slippage - pos["entry_price"]) * pos["size"]
else:
exit_price_with_slippage = exit_price * (1 + self.slippage)
pnl = (pos["entry_price"] - exit_price_with_slippage) * pos["size"]
# 手数料計算
entry_commission = pos["entry_price"] * pos["size"] * self.commission_rate
exit_commission = exit_price * pos["size"] * self.commission_rate
total_commission = entry_commission + exit_commission
# PnL記録
trade = Trade(
entry_time=pos["entry_time"],
exit_time=exit_time,
side=pos["side"],
entry_price=pos["entry_price"],
exit_price=exit_price,
size=pos["size"],
pnl=pnl - total_commission,
pnl_pct=(pnl - total_commission) / self.current_capital,
commission=total_commission,
slippage=slippage_cost * pos["size"],
holding_hours=(exit_time - pos["entry_time"]).total_seconds() / 3600
)
self.trades.append(trade)
self.current_capital += trade.pnl
self.position = None
def _record_equity(self, idx: int, current_price: float):
"""權益曲線を記録"""
unrealized_pnl = 0
if self.position is not None:
pos = self.position
if pos["side"] == PositionSide.LONG:
unrealized_pnl = (current_price - pos["entry_price"]) * pos["size"]
else:
unrealized_pnl = (pos["entry_price"] - current_price) * pos["size"]
self.equity_history.append({
"timestamp": self.df.iloc[idx]["timestamp"] if hasattr(self, "df") else None,
"equity": self.current_capital + unrealized_pnl,
"cash": self.current_capital,
"unrealized_pnl": unrealized_pnl,
"price": current_price
})
# 简易メソッド：DataFrameをセット
def set_dataframe(self, df: pd.DataFrame):
self.df = dfHolySheep AI APIとの連携：戦略コード自動生成
ここが本稿の核心です。HolySheep AI（今すぐ登録）のAPIは、OpenAI互換のインターフェースを提供しており、私が家常的に使っているGPT-4.1が¥1=$1のレートで使えます。
# utils/holysheep_client.py
import os
from openai import OpenAI
from typing import Dict, List, Optional
class HolySheepClient:
"""HolySheep AI APIクライアント - OpenAI互換"""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.getenv("HOLYSHEEP_API_KEY")
self.base_url = "https://api.holysheep.ai/v1"
self.client = OpenAI(
api_key=self.api_key,
base_url=self.base_url
)
def generate_strategy_code(
self,
strategy_description: str,
available_features: List[str],
style: str = "pinescript"
) -> str:
"""
HolySheep AIを使ってバックテスト戦略コードを自動生成
Args:
strategy_description: 戦略の説明（自然言語）
available_features: 利用可能な特徴量のリスト
style: 出力形式 ("python", "pinescript", "pinescript_v5")
Returns:
生成されたコード
"""
prompt = f"""あなたは暗号通貨取引のストラテジー開発Expertです。
以下の戦略{description}を{style}で実装してください。
利用可能な特徴量:
{chr