Bot

仮想通貨botの開発を本格的に始めてみる#27 (2023/11/4)「ローソク足データ生成(改)」

2023年11月6日

前回の記事に引き続き、今回も仮想通貨ボットの開発状況をまとめていきます。

以前より、生成したローソク足データの読み込みでエラーが出ていたので、その原因を突き止めようとしています。

結論としては、データが整理されていないことが原因だったようです。

コードを修正して1分足のデータを保存・更新するようにしました。

ローソク足データ(1分足)の生成

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:
        # ローソク足データを生成（例: 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)

実行結果は以下の通り。

時々、抜け漏れがありますね。

ひどい時には10分以上も間隔が空いています。

この状態では、完全な1分足データの獲得には至りません。

しかも、稼働から7時間ほど経ったら止まっていました...。

エラーの原因と修正が必要ですね。

1分足データの生成と読み込み

上記のコードで生成された1分足データは以下のコードで読み込みおよび書き出しができます。

import json

# ファイルパス
file_path = 'ここにファイルパス名を入力'

# JSONファイルを読み込んでデータを取得
with open(file_path, 'r') as file:
    data = file.readlines()

# 読み込んだデータを出力
for line in data:
    # 各行のJSON文字列をパースして辞書に変換
    candlestick_data = json.loads(line)
    # 辞書を出力
    print(candlestick_data)

実行結果は以下の通り。

生成されたデータは無事読み込めたので、ひとまずデータの保存形式については問題なさそうです。

今回のコードを元にして、バックテスト用のデータを蓄積していきます。

まとめ

今回は、 「JSON形式のデータの中身を指定すること」の重要性を再確認できました。

コードもChatGPTに書かせっぱなしにするのではなく、自分の目でもきちんと確認していくことが欠かせません。

今後は、今回生成したデータを既存のbotに組み込んでバックテストに使えるかどうかも検証していきます。

-Bot