前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。
今回はCursorを使ってバックテスト用のコードを書かせてみました。
Yodaka
bot開発におけるAIツールの活用方法の一例です。
MMbotのバックテスト用コード
import pandas as pd
from decimal import Decimal
import matplotlib.pyplot as plt
import ccxt
import os
import random
from typing import Dict, Any
class BacktestMarketMaker:
def __init__(self, data_file, initial_balance=1000000):
self.data = pd.read_csv(data_file, index_col='timestamp', parse_dates=True)
if 'close' not in self.data.columns:
self.data['close'] = (self.data['bid'] + self.data['ask']) / 2
self.balance = Decimal(str(initial_balance))
self.position = Decimal('0')
self.trades = []
self.equity_curve = [initial_balance]
def calculate_lot_size(self, price):
return min(Decimal('0.01'), self.balance / (price * Decimal('100')))
def run_backtest(self, SPREAD_ENTRY, SPREAD_CANCEL, LOT, DELTA):
max_drawdown = Decimal('0.02') # 最大2%のドローダウンを許容
initial_equity = self.calculate_equity(self.data.iloc[0]['bid'], self.data.iloc[0]['ask'])
for i in range(len(self.data)):
row = self.data.iloc[i]
current_equity = self.calculate_equity(row['bid'], row['ask'])
if current_equity < initial_equity * (1 - max_drawdown):
self.close_position(row['bid'], row['ask'])
break
ask = Decimal(str(row['ask']))
bid = Decimal(str(row['bid']))
spread = (ask - bid) / bid
if spread > SPREAD_ENTRY:
# Place orders
if self.position == 0:
lot_size = self.calculate_lot_size(bid)
self.place_order('buy', lot_size, bid + DELTA)
self.place_order('sell', lot_size, ask - DELTA)
elif spread < SPREAD_CANCEL:
# Cancel orders and close position
if self.position != 0:
self.close_position(bid, ask)
if i >= 20: # 簡単な移動平均を使用
ma = Decimal(str(self.data['close'].iloc[i-20:i+1].mean()))
if Decimal(str(row['close'])) > ma and self.position <= 0:
self.place_order('buy', LOT, ask)
elif Decimal(str(row['close'])) < ma and self.position >= 0:
self.place_order('sell', LOT, bid)
self.equity_curve.append(float(self.calculate_equity(bid, ask)))
def place_order(self, side, size, price):
fee = size * price * Decimal('0.001') # 0.1%の取引手数料
if side == 'buy':
cost = size * price + fee
if cost <= self.balance:
self.balance -= cost
self.position += size
self.trades.append(('buy', size, price))
else: # sell
revenue = size * price - fee
self.balance += revenue
self.position -= size
self.trades.append(('sell', size, price))
def close_position(self, bid, ask):
if self.position > 0:
self.balance += self.position * bid
elif self.position < 0:
self.balance -= abs(self.position) * ask
self.trades.append(('close', abs(self.position), bid if self.position > 0 else ask))
self.position = Decimal('0')
def calculate_equity(self, bid, ask):
bid = Decimal(str(bid))
ask = Decimal(str(ask))
if self.position > 0:
return self.balance + self.position * bid
elif self.position < 0:
return self.balance - abs(self.position) * ask
else:
return self.balance
def plot_results(self):
plt.figure(figsize=(12, 6))
plt.plot(self.equity_curve)
plt.title('Equity Curve')
plt.xlabel('Time')
plt.ylabel('Equity')
plt.show()
def print_results(self):
print(f"Final Balance: {self.balance}")
print(f"Final Position: {self.position}")
print(f"Number of Trades: {len(self.trades)}")
print(f"Final Equity: {self.equity_curve[-1]}")
def fetch_market_data(symbol='BTC/USDT', timeframe='1m', limit=1000):
exchange = ccxt.binance()
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
df['bid'] = df['close'] - (df['close'] * 0.0005) # 仮のbidを計算
df['ask'] = df['close'] + (df['close'] * 0.0005) # 仮のaskを計算
return df # 全てのカラムを返す
def create_csv_file(df, filename='market_data.csv'):
df.to_csv(filename)
return filename
def run_optimization(num_trials: int = 100) -> Dict[str, Any]:
best_params = None
best_profit = float('-inf')
for _ in range(num_trials):
# Generate random parameters
params = {
'SPREAD_ENTRY': Decimal(str(random.uniform(0.0005, 0.002))),
'SPREAD_CANCEL': Decimal(str(random.uniform(0.0001, 0.001))),
'LOT': Decimal(str(random.uniform(0.001, 0.01))),
'DELTA': Decimal(str(random.uniform(0.1, 1.0)))
}
# Fetch market data and create CSV file
market_data = fetch_market_data()
data_file = create_csv_file(market_data)
# Run backtest with current parameters
backtest = BacktestMarketMaker(data_file, initial_balance=1000000)
backtest.run_backtest(**params)
final_equity = backtest.equity_curve[-1]
profit = final_equity - 1000000
# Update best parameters if current profit is higher
if profit > best_profit:
best_profit = profit
best_params = params.copy()
best_params['profit'] = best_profit
# Clean up: remove the temporary CSV file
os.remove(data_file)
return best_params
if __name__ == "__main__":
# Constants from the original code
SPREAD_ENTRY = Decimal('0.0008') # スプレッドの閾値を上げる
SPREAD_CANCEL = Decimal('0.0002') # キャンセルの閾値を下げる
LOT = Decimal('0.005') # 取引量を増やす
DELTA = Decimal('0.5') # デルタを調整
# Run optimization
best_params = run_optimization(num_trials=100)
print("Best parameters found:")
for key, value in best_params.items():
print(f"{key}: {value}")
# Run final backtest with best parameters
market_data = fetch_market_data()
data_file = create_csv_file(market_data)
backtest = BacktestMarketMaker(data_file)
backtest.run_backtest(
SPREAD_ENTRY=best_params['SPREAD_ENTRY'],
SPREAD_CANCEL=best_params['SPREAD_CANCEL'],
LOT=best_params['LOT'],
DELTA=best_params['DELTA']
)
backtest.print_results()
backtest.plot_results()
# Clean up: remove the temporary CSV file
os.remove(data_file)
このコードは、仮想通貨市場において、市場メーカー戦略をバックテストするプログラムです。具体的な動作は以下の通りです：
- データの取得と整理：
fetch_market_data関数は、CCXTライブラリを用いてBinanceから特定の通貨ペアの価格データを取得し、Pandas DataFrameに整形しています。- 取得したデータには、
bid価格とask価格を仮定した列が追加されます。
- バックテストクラスの定義：
BacktestMarketMakerクラスでは、CSVファイルから市場データを読み込み、初期資金として1,000,000を設定します。- 取引のロジックを実装しており、スプレッドが一定の閾値を超えた場合に注文を置き、逆に閾値以下になった場合に注文をキャンセルまたはポジションを閉じます。
- マーケットデータの移動平均を用いてトレードの決定を行う部分もあります。
- 各取引の結果を記録し、最終的に資金曲線（Equity Curve）を描画します。
- パラメータの最適化：
run_optimization関数では、ランダムに生成した複数のパラメータセットを使ってバックテストを行い、最も利益が高かったパラメータセットを選択します。
- 結果の出力と可視化：
- 最適なパラメータを使用して最終的なバックテストを行い、結果を出力し、資金曲線をプロットします。
コードはPythonで書かれており、pandas、matplotlib、ccxt、decimal などのライブラリが使用されています。また、データ処理、ファイル操作、数値計算、可視化などの技術が組み合わされています。
詳細な解説
このコードについてもっと詳細に解説します。特にクラス BacktestMarketMaker のメソッドとその動作に焦点を当てます。
クラス BacktestMarketMaker
初期化
__init__メソッド:- 引数で渡されたCSVファイルから市場データを読み込みます。
- データにはタイムスタンプがインデックスとして設定され、
parse_dates=Trueにより日付として解釈されます。 - 'close' カラムが存在しない場合は、'bid' と 'ask' の平均から算出し追加します。
- 初期資金として100万単位を設定し、現在のポジションを0として保持します。
- 取引記録と資金曲線の初期値をリストで保持します。
ロットサイズの計算
calculate_lot_sizeメソッド:- 取引する際のロットサイズを計算します。これは資金に依存し、価格の100倍までの資金を使って最大0.01のロットを取引できるように設計されています。
バックテストの実行
run_backtestメソッド:- 引数としてスプレッドの入力・キャンセル閾値、ロットサイズ、価格調整値（DELTA）を受け取ります。
- 各データポイントに対して現在の資産を計算し、最大ドローダウン（資金の減少許容値）を超えた場合はポジションを閉じます。
- スプレッドが閾値を超えると注文を置き、下回ると注文をキャンセルします。
- 20ポイントの移動平均を基にして、市場価格がこれを上回るか下回るかによって買いまたは売りの注文を行います。
- 取引結果を資金曲線に追加していきます。
注文の実行
place_orderメソッド:- 注文の種類（買いまたは売り）、ロットサイズ、価格を引数として受け取ります。
- 取引手数料を考慮した上で、実際の買い注文または売り注文を行い、資金とポジションを更新します。
- 各取引の詳細を記録します。
ポジションのクローズ
close_positionメソッド:- 現在のポジションをクローズし、その際の売買価格を元に資金を更新します。
- 取引記録にクローズの詳細を追加します。
資金の計算
calculate_equityメソッド:- 現在の資金とポジションから総資産を計算します。
結果の出力とグラフの表示
print_resultsとplot_resultsメソッド:- 最終的な資金、ポジション、取引数、最終的な資金曲線の値を出力します。
- 資金曲線をグラフで表示します。
実行パート
if __name__ == "__main__"::- 最適化を行い、得られた最適なパラメータを使用して最終的なバックテストを実行します。
- 結果を出力し、資金曲線をプロットします。
- 一時的に作成されたCSVファイルを削除します。
このコードは、市場メーカーとしての戦略をシミュレートし、異なる市場条件やパラメータ下での戦略の効果を評価するためのものです。バックテストにより、戦略の改善点を見つけることができます。
今いじっているbot。ダメな原因を一つずつ潰す。検証プロセスをもっと高速化する工夫も必要。 pic.twitter.com/j1SWIHDzfJ
— よだか(夜鷹/yodaka) (@yodakablog) September 1, 2024
修正案
Yodaka
このコードは市場メーカー戦略のバックテストを行うために設計されていますが、いくつかの問題点や改善点が考えられます。以下に主要な点を挙げてみます。
1. データの扱い
- データの範囲: コード中で
fetch_market_data関数において、固定されたデータ量（limit=1000）を取得しています。これはシミュレーションの精度に影響を与える可能性があり、実際の市場条件に基づいてデータ量を動的に調整することが望ましいです。 - バイアスの可能性: コードでは
close価格の算出がbidとaskの単純な平均で行われていますが、これは市場の流動性や売買圧力を正確に反映しない可能性があります。
2. エラーハンドリング
- エラー処理の欠如:
fetch_ohlcvを使用してデータを取得する際、APIからの応答が失敗した場合のエラーハンドリングが実装されていません。通信エラーやAPIの利用制限によるエラーが考えられます。
3. コードの効率性
- ループの効率:
run_backtestメソッドでのデータポイント毎のループは非効率で、ベクトル化された操作やより効率的なデータ処理方法を検討する価値があります。 - ファイルIOの頻度: テスト毎にCSVファイルを生成し削除していますが、これはIOコストが高く、特に大規模なデータセットや高頻度のテストではパフォーマンスに影響を与えます。
4. 資源の利用
- 一時ファイルの管理: 最適化プロセスで生成される一時ファイルの扱いが不適切で、プログラムの終了時やエラー発生時に不要ファイルが残るリスクがあります。これはディスク使用量を不必要に増加させる可能性があります。
5. 実行時のパラメータ
- パラメータのランダム性: 最適化関数
run_optimizationでパラメータがランダムに生成されますが、これにより再現性が欠けます。実験の再現性を高めるために、乱数のシードを設定することが推奨されます。
6. プログラム構造
- モジュール性と再利用性: データの取得やバックテストのロジックなど、プログラムの各部分が密接に結合されており、再利用性やテストのしやすさが低下しています。関数やクラスをより独立させ、インターフェースを明確にすることでメンテナンス性を向上させることが可能です。
これらの問題を解決するためには、プログラムの再設計や、データ処理の効率化、エラーハンドリングの強化など、複数のアプローチが考えられます。
宿題
- バックテストの実行環境を整える
- ロジックの考案→市場データでの検証→最適なパラメータ検出→ロジックの評価→デプロイまでの流れを一連のプログラムに落とし込む方法を調べて、それらを身につける具体的な学習プロセスを明確にする
まとめ
今回は「MMbotのバックテストの概要を掴む」ことが目的だったため、詳細に詰めていく部分については省略しています。
本記事のコードを雛形にして、バックテストを行い、理論上の利益が上がりそうなパラメータを検出した後、フォワードテストにかけていくという流れがスタンダードです。
Yodaka
バックテストの方法については、今後も様々な方法を試す中で効率がよく検証性が高いものを残していきたいです。
今後もこの調子でbot開発の状況を発信していきます。