仮想通貨botの開発記録#60(2024/2/25)「回転Bot(プロトタイプ)の作成①(バブル相場の戦略)」

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

今回は「回転Bot(プロトタイプ)」を作成しました。

今ならこの考え方は有効かも。テストしてみよう。

バブル相場用の回転ボットのコンセプトとQuantZoneロジック(追記あり)｜Hoheto (仮想通貨botter) #note https://t.co/m8IE1gtjbz

— よだか(夜鷹/yodaka) (@yodakablog) February 24, 2024

Yodaka

バブルの上昇相場に強い戦略を実行するタイプのBotです。

今回のコードはプロトタイプです。実際の使用には改良(パラメータの調整・取引所の仕様にあわせた関数の設定など)が必要です。

回転Botのコード

Yodaka

このプログラムで実行することは以下の通り。

• エントリーの条件は「直近120分の高値から7％下落した価格でエントリー指値を置く」
• 決済の条件①「ロングポジションを持っていれば決済指値を出す」
• 決済の条件②「直近120分の安値から6%上昇したら、その位置に決裁指値を置く」
• 決済条件③「現在のポジションと同じ数値で全てのポジションを決済する」
• 上記の処理を10秒ごとに繰り返す
• レバレッジの設定は2倍
• 利用する取引所は「KucoinFuturesのETH/USDT先物取引」
• ccxtライブラリを使う
• 取引履歴のログをcsvファイルで出力する

import ccxt
from datetime import datetime, timedelta
import csv
import time

# APIキーの設定
api_key = 'YOUR_API_KEY'
api_secret = 'YOUR_API_SECRET'
password = 'YOUR_PASSWORD'

# ccxtライブラリを使ってKucoin Futuresに接続
exchange = ccxt.kucoinfutures({
    'apiKey': api_key,
    'secret': api_secret,
    'password': password,
})

symbol = 'ETH/USDT'  # 取引する通貨ペア
leverage = 2  # レバレッジを2倍に設定
exchange.set_leverage(leverage, symbol)

csv_file_path = 'trade_log.csv'
try:
    with open(csv_file_path, 'x', newline='') as file:
        writer = csv.writer(file)
        writer.writerow(['timestamp', 'action', 'price', 'amount', 'comment'])
except FileExistsError:
    pass  # ファイルが既に存在する場合は何もしない

def log_to_csv(timestamp, action, price, amount, comment=''):
    with open(csv_file_path, 'a', newline='') as file:
        writer = csv.writer(file)
        writer.writerow([timestamp, action, price, amount, comment])

# ポジションの状態を追跡するフラグ
has_position = False

while True:
    now = datetime.utcnow()
    since = exchange.parse8601((now - timedelta(minutes=120)).isoformat())
    ohlcv = exchange.fetch_ohlcv(symbol, '1m', since, limit=120)

    high_prices = [x[2] for x in ohlcv]
    low_prices = [x[3] for x in ohlcv]
    high = max(high_prices)
    low = min(low_prices)

    entry_price = high * (1 - 0.07)
    exit_price = low * (1 + 0.06)
    amount = 1

    ticker = exchange.fetch_ticker(symbol)
    last_price = ticker['last']

    # エントリー条件を満たす場合のオーダー作成
    if last_price <= entry_price and not has_position:
        print(f"エントリー条件を満たしています。購入価格: {entry_price}")
        # 注文を実際に出す
        #exchange.create_limit_buy_order(symbol, amount, entry_price)
        has_position = True  # ポジション保有フラグを立てる
        log_to_csv(datetime.now().isoformat(), 'Entry', entry_price, amount, 'Buy order placed')

    # ロングポジションを持っている場合の決済条件確認
    if has_position and last_price >= exit_price:
        # 決済オーダーを出す条件を満たしている
        print(f"決済条件を満たしています。販売価格: {exit_price}")
        # 注文を実際に出す
        #exchange.create_limit_sell_order(symbol, amount, exit_price)
        has_position = False  # ポジション保有フラグを初期化
        log_to_csv(datetime.now().isoformat(), 'Exit', exit_price, amount, 'Sell order placed')
    elif has_position:
        # ロングポジションを持っているが、決済条件を満たしていない場合
        print("決済条件を満たしていません。ポジションを維持します。")
    else:
        # ポジションを持っていない場合
        print("ポジションを持っていません。")

    time.sleep(10)  # 10秒待機

Yodaka

上記のコードでは、エントリー条件は一つだけです。

しかし、エントリー指値の条件を複数設定すると、収益機会を増やせる可能性があります。

また、ポジション保有のフラグ管理については、改良の余地があります。

売買注文を出す部分だけをコメントアウトして、稼働状況を確認・修正します。

フラグ管理で混乱しそうなので、一度フローチャートに書き起してみる。実際に、コードを書いているとこういう細かな部分をどうやって管理するのかが気になってくる。

— よだか(夜鷹/yodaka) (@yodakablog) February 25, 2024

実際に稼働するコード(2024/2/27更新)

Yodaka

調整を加えて実際に稼働するようにしたものが以下のプログラムです。(2024/2/27日現在、試験運転中です)

import ccxt
from datetime import datetime, timedelta
import csv
import time

# APIキーの設定
api_key = '***'
api_secret = '***'
password = '***'

# ccxtライブラリを使ってKucoin Futuresに接続
exchange = ccxt.kucoinfutures({
    'apiKey': api_key,
    'secret': api_secret,
    'password': password,
})

symbol = 'ETH/USDT:USDT'  # 取引する通貨ペア

csv_file_path = 'trade_log.csv'
try:
    with open(csv_file_path, 'x', newline='') as file:
        writer = csv.writer(file)
        writer.writerow(['timestamp', 'action', 'price', 'amount', 'comment'])
except FileExistsError:
    pass  # ファイルが既に存在する場合は何もしない

def log_to_csv(timestamp, action, price, amount, comment=''):
    with open(csv_file_path, 'a', newline='') as file:
        writer = csv.writer(file)
        writer.writerow([timestamp, action, price, amount, comment])

# ポジションの状態を追跡するフラグ
has_position = False

while True:
    now = datetime.utcnow()
    since = exchange.parse8601((now - timedelta(minutes=120)).isoformat())
    ohlcv = exchange.fetch_ohlcv(symbol, '1m', since, limit=120)

    high_prices = [x[2] for x in ohlcv]
    low_prices = [x[3] for x in ohlcv]
    high = max(high_prices)
    low = min(low_prices)

    entry_price = high * (1 - 0.07)
    exit_price = low * (1 + 0.06)
    amount = 1

    ticker = exchange.fetch_ticker(symbol)
    last_price = ticker['last']

    # エントリー条件を満たす場合のオーダー作成,レバレッジもここで設定する
    if last_price <= entry_price and not has_position:
        print(f"エントリー条件を満たしています。購入価格: {entry_price}")
        # 注文を実際に出す
        #exchange.create_limit_buy_order(symbol, amount, entry_price,params={'leverage': 3})
        has_position = True  # ポジション保有フラグを立てる
        log_to_csv(datetime.now().isoformat(), 'Entry', entry_price, amount, 'Buy order placed')

    # ロングポジションを持っている場合の決済条件確認
    if has_position and last_price >= exit_price:
        # 決済オーダーを出す条件を満たしている
        print(f"決済条件を満たしています。販売価格: {exit_price}")
        # 注文を実際に出す
        #exchange.create_limit_sell_order(symbol, amount, exit_price,params={'leverage': 3})
        has_position = False  # ポジション保有フラグを初期化
        log_to_csv(datetime.now().isoformat(), 'Exit', exit_price, amount, 'Sell order placed')
    elif has_position:
        # ロングポジションを持っているが、決済条件を満たしていない場合
        print("決済条件を満たしていません。ポジションを維持します。")
    else:
        # ポジションを持っていない場合
        print("ポジションを持っていません。")

    time.sleep(10)  # 10秒待機

Yodaka

ここからは、勝てそうな戦場を選んで実際に運用していきます。

bitFlyer Lightning FXで稼働するコード(2024/3/6更新)

Yodaka

コードを稼働させつつ、効果検証します。

import ccxt
from datetime import datetime, timedelta
import csv
import time
import pandas as pd  # Pandasライブラリをインポート

# bitFlyer APIキーの設定
api_key = '***'
api_secret = '***'

# ccxtライブラリを使ってbitFlyerに接続
exchange = ccxt.bitflyer({
    'apiKey': api_key,
    'secret': api_secret,
})

symbol = 'FX_BTC_JPY'  # bitFlyer Lightning FXの取引ペア

csv_file_path = 'fx_trade_log.csv'
try:
    with open(csv_file_path, 'x', newline='') as file:
        writer = csv.writer(file)
        writer.writerow(['timestamp', 'action', 'price', 'amount', 'comment'])
except FileExistsError:
    pass

def log_to_csv(timestamp, action, price, amount, comment=''):
    with open(csv_file_path, 'a', newline='') as file:
        writer = csv.writer(file)
        writer.writerow([timestamp, action, price, amount, comment])


# ポジションの状態を追跡するフラグ
has_position = False

while True:
    # 直近の取引データを取得
    trades = exchange.fetch_trades(symbol)
    prices = [trade['price'] for trade in trades]
    high = max(prices) if prices else 0
    low = min(prices) if prices else 0

    entry_price = high * (1 - 0.07)
    exit_price = low * (1 + 0.06)
    amount = 0.01  # bitFlyer Lightning FXでの取引量はBTC単位で指定

    # 現在価格を取得
    ticker = exchange.fetch_ticker(symbol)
    last_price = ticker['last']

    # エントリー条件を満たす場合のオーダー作成
    if last_price <= entry_price and not has_position:
        print(f"エントリー条件を満たしています。購入価格: {entry_price}")
        # 実際のオーダー出力行（デモなのでコメントアウト）
        # exchange.create_order(symbol, 'limit', 'buy', amount, entry_price)
        has_position = True
        log_to_csv(datetime.now().isoformat(), 'Entry', entry_price, amount, 'Buy order placed')

    # ロングポジションを持っている場合の決済条件確認
    if has_position and last_price >= exit_price:
        print(f"決済条件を満たしています。販売価格: {exit_price}")
        # 実際のオーダー出力行（デモなのでコメントアウト）
        # exchange.create_order(symbol, 'limit', 'sell', amount, exit_price)
        has_position = False
        log_to_csv(datetime.now().isoformat(), 'Exit', exit_price, amount, 'Sell order placed')
    elif has_position:
        print("決済条件を満たしていません。ポジションを維持します。")
    else:
        print("ポジションを持っていません。")

    time.sleep(10)  # 10秒待機

バックテスト用のコード

Yodaka

バックテスト用のコードは以下の通り。

import matplotlib.pyplot as plt

# 仮の市場データ
data = [
    [1609459200000, 730.0, 750.0, 725.0, 740.0, 1500],
    [1609462800000, 740.0, 760.0, 735.0, 755.0, 1500],
    # 実際に取得したデータに置き換える
]

# レバレッジ設定
leverage = 2

# バックテストのロジックに適用するために、必要なデータ形式に変換
processed_data = [(x[0], x[1], x[2], x[3], x[4], x[5]) for x in data]

# 初期資金とポジション変数
initial_balance = 10000
balance = initial_balance
positions = 0
cumulative_profit = 0

# 損益記録
cumulative_profits = []

for i in range(1, len(processed_data)):
    timestamp, open_price, high, low, close, volume = processed_data[i]
    
    # エントリー条件：直前の高値から7%下落
    if close <= high * (1 - 0.07) and positions == 0:
        entry_price = close
        # レバレッジを考慮したポジションサイズの計算
        positions = (balance / close) * leverage
        balance = 0  # レバレッジを使用する場合、実際のバランスは変わらないが、ここでは簡略化のため0とする
    
    # 決済条件：直前の安値から6%上昇
    elif close >= low * (1 + 0.06) and positions > 0:
        exit_price = close
        # レバレッジを考慮して利益計算
        balance = positions * exit_price / leverage
        profit = balance - initial_balance
        cumulative_profit += profit
        cumulative_profits.append(cumulative_profit)
        positions = 0
        entry_price = 0

# 損益のグラフを描画
plt.figure(figsize=(10, 6))
plt.plot(cumulative_profits, label='Cumulative Profit over Time')
plt.xlabel('Trade Number')
plt.ylabel('Cumulative Profit')
plt.title('Cumulative Profit Over Time from Backtest')
plt.legend()
plt.grid(True)
plt.show()

Yodaka

実行結果は以下の通り。適用しているのは仮データであるため、グラフ描画はされていません。

Yodaka

実際の稼働に伴って、データを蓄積させていきます。また、このデータ分析の方法は、DEXでのDeFi運用にも転用できそうです。

【バックテストのポイント】

• エントリーの条件をどの程度広く取るのか(収益機会の損失回避)
• 指値を複数(2つ以上)置いた方が収益が上がる期待値が高いかどうか
• 決済の条件をどの程度狭めるのか(利益確定の頻度)
• レバレッジの大きさ(指値が貫通してもロスカットを喰らわないように調整する)
• 指値を複数置くべきか否か

【注意点とアイデア】

・含み損がある状態で決済指値を出してしまう場合がある(直近N分の安値を基準としているため)
→掘り下げた時点で決済指値幅を自動or手動で広げるようにする。例えば「含み損を抱えた状態では決済指値を出さない」「含み益があれば決済指値を出す」など。

ただし、このアプローチは、損失が際限なく膨らんでいく可能性もあるため、結局はどこかで手仕舞いをする必要があります。最初から決済ライン(≒強制ロスカット)を織り込んでおくことで、シンプルで調整しやすいロジックで戦うことができるのでその方が良いかもしれません。バックテストで検証します。

まとめ

今回は「上昇相場」に適した戦略を執行するBotを紹介しました。

このロジックを転用して「下落相場」や「レンジ相場」の戦略実行も考えていきたいです。

宿題

注文を出す部分だけをコメントアウトしてコードのエラー修正をする
→フラグ管理を使うことに慣れる

注文部分だけコメントアウトして高頻度Botの稼働状況をテストするとエラー修正が捗る。実際に注文を出すのは誤発注のリスク等があるので、注文を出したと仮定して適当なprint文を出力させたり、注文を出したログを出力させたりすれば良い。簡単なことほど、自分ではなかなか気づけない。自戒を込めて。

— よだか(夜鷹/yodaka) (@yodakablog) February 25, 2024

【現在の開発フロー】

また、Botの開発フローをパターン化することで、今後の開発スピードを上げて開発自体にも慣れていくようにします。

①トレード戦略を知る→調べる→考える
②AIなどのツールを使ってコードを書かせる
③そのコードを自分でじっくり読む
④部分的に稼働させて構造を理解しつつ自分でも書く

当面の目標は

開発フローをシステム化して、現場での試行回数を最大化すること

です。

Yodaka

今後も実際の戦略考案と実行を繰り返しながら、一つずつ武器を増やしていきます。

プログラミングに関しては初学者の域を出ないので、以下の流れでBotのコードを書いています。手を動かすと早く覚えられるので。
①トレード戦略を知る→調べる→考える
②AIなどのツールを使ってコードを書かせる
③そのコードを自分でじっくり読む
④部分的に稼働させて構造を理解しつつ自分でも書く

— よだか(夜鷹/yodaka) (@yodakablog) February 25, 2024