仮想通貨botの開発を本格的に始めてみる#25(2023/11/1)「価格取得とローソク足生成の実験」

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

今回は、ヒストリカルデータに対応したバックテスト用コードをChatGPTに書かせてみました。

なかなかうまくいかないものですが、開発途中のコードを備忘録として残しておきます。

価格取得用のコード

実際に稼働させつつ、過去データを蓄積するコード。

ビットフライヤーFXのAPIを利用して、1分足・5分足・15分足・1時間足でそれぞれローソク足データを蓄積するためのコードです。

import requests
import json
from datetime import datetime, timedelta  # timedeltaを追加
import time

# APIエンドポイントのURL
api_url = "https://api.bitflyer.com/v1/getexecutions?product_code=FX_BTC_JPY&count=500"

# ローソク足データを生成する関数
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"]
            }
            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"]
    
    # 最後の足を追加
    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)
        # ローソク足データを生成（例: 5分足）
        five_minute_candlestick_data = generate_candlestick_data(executions, timeframe_minutes=5)
        # ローソク足データを生成（例: 15分足）
        fifteen_minute_candlestick_data = generate_candlestick_data(executions, timeframe_minutes=15)
        # ローソク足データを生成（例: 1時間足）
        hour_candlestick_data = generate_candlestick_data(executions, timeframe_minutes=60)
        
        # データをJSONファイルに保存（追記モード）
        with open('minute_candlestick_data.json', 'a') as json_file:
            json.dump(minute_candlestick_data, json_file)
            json_file.write('\n')
        
        with open('five_minute_candlestick_data.json', 'a') as json_file:
            json.dump(five_minute_candlestick_data, json_file)
            json_file.write('\n')
        
        with open('fifteen_minute_candlestick_data.json', 'a') as json_file:
            json.dump(fifteen_minute_candlestick_data, json_file)
            json_file.write('\n')
        
        with open('hour_candlestick_data.json', 'a') as json_file:
            json.dump(hour_candlestick_data, json_file)
            json_file.write('\n')
        
        print("Candlestick data saved.")
    else:
        print("Failed to fetch data or generate candlestick data.")
    
    # 10秒待機
    time.sleep(10)

ヒストリカルデータに対応するバックテスト用コード（開発中）

開発中のものなので、実用性は低い。

import json
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime
from itertools import chain
from collections import deque

# JSONファイルからローソク足データを読み込む関数
def load_candlestick_data(file_path):
    with open(file_path, "r") as file:
        candlestick_data = [json.loads(line) for line in file.readlines()]
    # リスト内の各要素を展開して一つのリストにする
    return list(chain.from_iterable(candlestick_data))

# ドンチャンブレイク戦略の関数
def donchian_strategy(candlestick_data, term):
    signals = []
    for i in range(term, len(candlestick_data)):
        highest_high = max(candle["high"] for candle in candlestick_data[i - term:i])
        lowest_low = min(candle["low"] for candle in candlestick_data[i - term:i])
        current_candle = candlestick_data[i]
        
        if current_candle["high"] > highest_high:
            signals.append((current_candle["timestamp"], "Buy"))
        elif current_candle["low"] < lowest_low:
            signals.append((current_candle["timestamp"], "Sell"))
        else:
            signals.append((current_candle["timestamp"], "Hold"))
    return signals

# JSONファイルからローソク足データを読み込む
file_path = "minute_candlestick_data.json"
candlestick_data = load_candlestick_data(file_path)

# ドンチャンブレイク戦略の実行
term = 30  # ドンチャンブレイクの期間
signals = donchian_strategy(candlestick_data, term)

# バックテストの結果をPandas DataFrameに変換
df = pd.DataFrame(signals, columns=["Date", "Signal"])
df["Date"] = pd.to_datetime(df["Date"])

# 折れ線グラフでバックテストの結果を視覚化
plt.figure(figsize=(12, 6))
plt.plot(df["Date"], df["Signal"], marker='o', linestyle='-', color='b', label='Signal')
plt.xlabel("Date")
plt.ylabel("Signal")
plt.title("Donchian Breakout Backtest Signals")
plt.legend()
plt.grid(True)
plt.show()

利益率算出用（お試し）

こちらもお試し版なので実用性は低い。

import json
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime
from itertools import chain
from collections import deque

# JSONファイルからローソク足データを読み込む関数
def load_candlestick_data(file_path):
    with open(file_path, "r") as file:
        candlestick_data = [json.loads(line) for line in file.readlines()]
    # リスト内の各要素を展開して一つのリストにする
    return list(chain.from_iterable(candlestick_data))

# ドンチャンブレイク戦略の関数
def donchian_strategy(candlestick_data, term):
    signals = []
    for i in range(term, len(candlestick_data)):
        highest_high = max(candle["high"] for candle in candlestick_data[i - term:i])
        lowest_low = min(candle["low"] for candle in candlestick_data[i - term:i])
        current_candle = candlestick_data[i]
        
        if current_candle["high"] > highest_high:
            signals.append((current_candle["timestamp"], "Buy"))
        elif current_candle["low"] < lowest_low:
            signals.append((current_candle["timestamp"], "Sell"))
        else:
            signals.append((current_candle["timestamp"], "Hold"))
    return signals

# JSONファイルからローソク足データを読み込む
file_path = "minute_candlestick_data.json"
candlestick_data = load_candlestick_data(file_path)

# ドンチャンブレイク戦略の実行
term = 30  # ドンチャンブレイクの期間
signals = donchian_strategy(candlestick_data, term)

# 利益率を計算する関数
def calculate_profit_rate(signals):
    position = None
    balance = 2.0  # 初期資金
    for _, signal in signals:
        if signal == "Buy" and position is None:
            position = balance
        elif signal == "Sell" and position is not None:
            balance = balance * (position / balance)
            position = None
    # 最終的な利益率を計算
    if position is not None:
        balance = balance * (position / balance)
    return balance

# 利益率を計算
profit_rate = calculate_profit_rate(signals)

# 結果を出力
print("利益率: {:.2f}%".format((profit_rate - 1.0) * 100))

# バックテストの結果をPandas DataFrameに変換
df = pd.DataFrame(signals, columns=["Date", "Signal"])
df["Date"] = pd.to_datetime(df["Date"])

# 折れ線グラフでバックテストの結果を視覚化
plt.figure(figsize=(12, 6))
plt.plot(df["Date"], df["Signal"], marker='o', linestyle='-', color='b', label='Signal')
plt.xlabel("Date")
plt.ylabel("Signal")
plt.title("Donchian Breakout Backtest Signals")
plt.legend()
plt.grid(True)
plt.show()

バックテスト用に稼働するコード

ドンチャンブレイクアウトの損益額を視覚化するコード。

import json
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime
from itertools import chain

# JSONファイルからローソク足データを読み込む関数
def load_candlestick_data(file_path):
    with open(file_path, "r") as file:
        candlestick_data = [json.loads(line) for line in file.readlines()]
    # リスト内の各要素を展開して一つのリストにする
    return list(chain.from_iterable(candlestick_data))

# ドンチャンブレイク戦略の関数
def donchian_strategy(candlestick_data, term):
    signals = []
    balance = 1.0  # 初期資金
    position = None  # position変数を最初にNoneで初期化
    for i in range(term, len(candlestick_data)):
        highest_high = max(candle["high"] for candle in candlestick_data[i - term:i])
        lowest_low = min(candle["low"] for candle in candlestick_data[i - term:i])
        current_candle = candlestick_data[i]
        
        if current_candle["high"] > highest_high:
            signals.append((current_candle["timestamp"], "Buy"))
            position = current_candle["close"]  # Buyシグナルの場合、positionに購入価格を代入
        elif current_candle["low"] < lowest_low:
            if position is not None:  # positionがNoneでない場合のみSellシグナルの処理を行う
                signals.append((current_candle["timestamp"], "Sell"))
                balance = balance * (current_candle["close"] / position)  # Sellシグナルの場合、利益を計算
                position = None  # positionをリセット
        else:
            signals.append((current_candle["timestamp"], "Hold"))
            if position is not None:
                balance = balance * (current_candle["close"] / position)  # Holdシグナルの場合、損益を計算
        
    # ループ終了後にポジションが残っていれば、最後の足で決済する
    if position is not None:
        balance = balance * (current_candle["close"] / position)  # 最後のポジションを決済
    return signals, balance - 1.0

# JSONファイルからローソク足データを読み込む
file_path = "minute_candlestick_data.json"
candlestick_data = load_candlestick_data(file_path)

# ドンチャンブレイク戦略の実行
term = 30  # ドンチャンブレイクの期間
signals, profit = donchian_strategy(candlestick_data, term)

# 結果を出力
print("損益額: {:.2f} JPY".format(profit))

# バックテストの結果をPandas DataFrameに変換
df = pd.DataFrame(signals, columns=["Date", "Signal"])
df["Date"] = pd.to_datetime(df["Date"])

# 損益の推移を計算
profits = []
balance = 1.0
for _, signal in signals:
    if signal == "Buy":
        position = balance
    elif signal == "Sell":
        balance = balance * (position / balance)
    profits.append(balance - 1.0)

# 折れ線グラフで損益の推移を表示
plt.figure(figsize=(12, 6))
plt.plot(df["Date"], profits, marker='o', linestyle='-', color='b', label='Profit/Loss')
plt.xlabel("Date")
plt.ylabel("Profit/Loss")
plt.title("Profit/Loss Transition")
plt.legend()
plt.grid(True)
plt.show()

AIに使い方の解説も書かせてみました。

指示１：上記のコードを使って戦略検証する際に、変更するべきパラメータとそこから予測される結果について具体的に説明しなさい。

戦略検証を行う際には、いくつかのパラメータを調整することが重要です。以下に、ドンチャンブレイク戦略を検証する際に変更できる主なパラメータと、それらを変更した場合の予測される結果について説明します。

1. 期間 (term): ドンチャンブレイク戦略では、一定期間内の最高値と最安値を計算します。この期間を変更すると、戦略の感度が変わります。期間を短くすると、市場の瞬時の変動に敏感に反応しますが、ノイズによって誤ったシグナルが生成されやすくなります。一方、期間を長くすると、より長期的なトレンドを捉えることができますが、適切なタイミングでのエントリーポイントが遅れる可能性があります。
2. 初期資金 (initial balance): 戦略の初期資金額を設定します。初期資金の変化は、最終的な利益に大きな影響を与えます。資金が大きいほど、取引量も大きくなり、利益や損失も増減します。
3. 取引手数料 (transaction fee): 実際の取引所で取引を行う際には、取引手数料がかかります。取引手数料を考慮しないと、戦略の収益が過大評価される可能性があります。取引手数料を適切に設定することで、実際の取引状況に近い結果を得ることができます。
4. 市場の変動性: 市場の変動性が高いほど、価格が急激に変動するため、戦略の利益や損失も大きくなります。逆に、市場の変動性が低い場合、小さな変動しかないため、利益も限られる可能性があります。

これらのパラメータを変更することで、戦略の性能を検証できます。例えば、期間を短くすると、より頻繁に取引が行われるため、市場の小さな変動にも反応しやすくなります。一方で、初期資金を増やすと、取引量が増え、大きな利益を狙える一方で、リスクも高まります。

戦略の検証を行う際には、これらのパラメータを様々に変更して、過去のデータに対して戦略を適用し、利益や損失の推移を評価することが重要です。また、過去のデータだけでなく、将来の予測に対しても戦略の性能をテストする必要があります。これをバックテストと呼び、信頼性のある戦略を見つけるための一つの手法です。

指示２：これらの検証を行うために、上記のコード内で操作・変更する必要がある箇所を具体的に示しなさい。

戦略の検証を行うために、上記のコード内で操作・変更する必要がある箇所を以下に具体的に示します。

1.期間 (term) の変更

term 変数を変更して、ドンチャンブレイク戦略の期間を調整できます。

term = 30  # 期間を変更する値に設定

2.初期資金 (initial balance) の変更:

balance 変数を変更して、戦略の初期資金を設定できます。

balance = 1.0  # 初期資金を変更する値に設定

3.取引手数料 (transaction fee) の考慮:

実際の取引所での手数料を考慮するため、トレードごとに手数料を差し引く処理を追加します。

# Buyの場合の手数料を差し引く（例として手数料を0.1%としています）
if signal == "Buy":
    position = balance
    # 手数料を考慮して購入額を計算
    position = position * 0.999  # 0.1%の手数料を差し引く（実際の手数料率に合わせて調整）

4.市場の変動性の考慮:

戦略の性能を評価する際に、市場の変動性に対応するため、適切な取引手数料や利益計算方法を選択することが重要です。変動性が高い場合は、損益が大きくなる傾向があるため、リスク管理がより重要になります。

これらの変更を行うことで、戦略の検証をより正確に行うことができます。変動性や取引手数料の変化に対して戦略がどのように反応するかを確認することで、戦略の信頼性や適用範囲を把握できます。

まとめ

今回は、試験運用なのでこれらのコードはバックテスト用としては不十分です。

もう一度コードを組み直す方が良いかもしれません。

使える部分は使いながら、検証を進めていきます。