前回の記事に引き続き、今回も仮想通貨botの開発状況をまとめていきます。
今回の記事の後半では「bitFlyerの約定履歴からローソク足(1分足)を生成するコード(修正版)」も載せてあります。
Cryptowatchが使えなくなって困っている方の参考になれば幸いです。
バックテストと視覚化
以下のコードはドンチャンブレイクアウトのバックテストをできる限りシンプルに行えるようにするためのものです。
import json
from datetime import datetime
import matplotlib.pyplot as plt
# JSONファイルの読み込みとパース
json_data = []
json_file_path = 'ここにファイルバス名を入力'
with open(json_file_path, 'r') as file:
for line in file:
# 各行を辞書に変換してリストに追加
json_data.append(json.loads(line))
# バックテストの実行
def run_backtest(price_data, term=10, lot=0.1, slippage=0.001):
# バックテストのロジックを記述
records = {
"buy-count": 0,
"buy-winning": 0,
"gross-profit": [0], # 累積損益を格納するリスト（初期値は0）
"date": [], # 取引日時を格納するリスト
# 他の統計情報も必要に応じて追加できます
}
for i in range(term, len(price_data)):
current_candle = price_data[i]
highest, lowest = donchian(price_data[i - term:i], term)
# ドンチャンブレイクの判定
if current_candle["high"] > highest["high"]:
records["buy-count"] += 1
records["buy-winning"] += 1
# ここで買い注文を出すロジックを追加する
# 損益の計算（ここではシンプルに1ロットでトレードした場合の損益を計算）
trade_profit = (current_candle["close"] - lowest["low"] - slippage) * lot
records["gross-profit"].append(records["gross-profit"][-1] + trade_profit)
records["date"].append(current_candle["timestamp"])
# 他の条件（売りエントリ、決済、損切りなど）についても同様に判定と処理を行う
return records
# 修正後のdonchian関数内のコード
def donchian(data, term):
highest = max(data, key=lambda x: x["high"])
lowest = min(data, key=lambda x: x["low"])
return highest, lowest
# 以下にバックテストの結果を取得する例を示します（json_dataは適切なデータで置き換える必要があります）
backtest_results = run_backtest(json_data)
# バックテストの結果を表示
print("バックテストの結果")
print("--------------------------")
print("買いエントリの成績")
print("トレード回数: {}回".format(backtest_results["buy-count"]))
print("勝率: {}％".format((backtest_results["buy-winning"] / backtest_results["buy-count"]) * 100))
print("総損益: {}円".format(backtest_results["gross-profit"][-1]))
# プロフィットカーブの表示
# 日付データの要素数を取得
date_length = len(backtest_results["date"])
# gross-profitリストを日付データに合わせてスライス
plt.plot(backtest_results["date"][:date_length], backtest_results["gross-profit"][:date_length])
plt.xlabel("Date")
plt.ylabel("Balance")
plt.xticks(rotation=50)
plt.show()
実行結果は以下の通り。
ローソク足生成のコード(修正版)
以前に書いたコードを修正しました。
修正点は「参照データを先頭から見ること(以前のコードでは、API経由で獲得したデータを古い時系列順で処理するようになっていたため)」です。
import requests
import json
from datetime import datetime, timedelta
import time
# APIエンドポイントのURL
api_url = "https://api.bitflyer.com/v1/getexecutions?product_code=FX_BTC_JPY&count=500"
# 最後に保存したtimestampを記録する変数を追加
last_saved_timestamp = None
# ローソク足データを生成する関数
def generate_candlestick_data(executions, timeframe_minutes):
candlestick_data = []
current_candle_start = None
candle = None
for trade in executions:
# APIから取得した日時のフォーマットはミリ秒が含まれているが、無視する
# 日時文字列からミリ秒の部分を取り除く
trade_date = trade['exec_date'].split('.')[0]
timestamp = datetime.strptime(trade_date, "%Y-%m-%dT%H:%M:%S")
# 指定した時間足に基づいてトレードを分類
candle_start = timestamp - timedelta(minutes=timestamp.minute % timeframe_minutes, seconds=timestamp.second)
if candle_start != current_candle_start:
# 新しい足の開始
if candle:
candlestick_data.append(candle)
candle = {
"timestamp": candle_start.strftime("%Y-%m-%dT%H:%M:%S"),
"open": trade["price"],
"high": trade["price"],
"low": trade["price"],
"close": trade["price"],
"volume": trade["size"],
"trades": 1 # 取引回数をカウント
}
current_candle_start = candle_start
else:
# 既存の足にデータを追加
candle["high"] = max(candle["high"], trade["price"])
candle["low"] = min(candle["low"], trade["price"])
candle["close"] = trade["price"]
candle["volume"] += trade["size"]
candle["trades"] += 1 # 取引回数をカウント
# 最後の足を追加
if candle:
candlestick_data.append(candle)
return candlestick_data
# 定期的にデータを取得して処理する無限ループ
while True:
# データ取得
executions = requests.get(api_url).json()
if executions:
# APIから取得したデータを逆順に並び替える
executions.reverse()
# ローソク足データを生成（例: 1分足）
minute_candlestick_data = generate_candlestick_data(executions, timeframe_minutes=1)
# ローソク足データが空でないことを確認
if minute_candlestick_data:
# 最新のtimestampを取得
current_timestamp = datetime.strptime(minute_candlestick_data[-1]['timestamp'], "%Y-%m-%dT%H:%M:%S")
# 最新のtimestampに対応するデータかどうかをチェックし、保存
if current_timestamp != last_saved_timestamp:
with open('minute_candlestick_data.json', 'a') as json_file:
json.dump(minute_candlestick_data[-1], json_file)
json_file.write('\n')
last_saved_timestamp = current_timestamp
print("Candlestick data saved.")
else:
print("No new data to save.")
else:
print("No candlestick data generated.")
else:
print("Failed to fetch data or generate candlestick data.")
# 10秒待機
time.sleep(10)
実行結果は以下の通り。
無事に稼働しているようです。
なお、「No new data to save.」と表示されるのは「その時間帯の約定履歴が存在しないとき」です。
実際にAPIデータを目視して確認しました。
まとめ
今回は、シンプルなコード改変とローソク足取得のコードを書くことにチャレンジしました。
特に後半のコードは、ほかの取引所のAPIを利用する時にも一部を流用できそうです。
約定履歴からデータを獲得する流れは、今後も応用していきます。