仮想通貨botの開発記録#87(2024/8/1)「発展的な取引ロジックの雛形まとめ②」

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

実際にbotで取引をしてみたいけれど、どんなコードを書いたら良いのか分からないという方の参考になると思います。

1.オークション取引の活用

オープニングオークションとクロージングオークションを利用した取引戦略は、特に株式市場でよく見られます。これらのオークションは、市場の開始前と終了時に設定され、一定時間内の注文を集約して、開始価格や終了価格を決定します。これらのオークションの価格動きを利用することで、日中の市場のボラティリティから逃れたり、大量の株を一定価格で取引したりすることが可能です。

以下は、オープニングオークションおよびクロージングオークションの時間に注文を行い、その価格動きを利用する基本的な戦略をPythonで実装する方法の例です。ここではシンプルなスケルトンコードを示しますが、実際の取引では取引所のAPIを使用して具体的な実装が必要になります。

必要なライブラリのインポート

import ccxt
import time
from datetime import datetime, timedelta

取引所の設定と初期化

exchange = ccxt.binance({
    'apiKey': 'YOUR_API_KEY',
    'secret': 'YOUR_API_SECRET',
    'enableRateLimit': True
})
symbol = 'BTC/USDT'

オークションの時間に基づいて注文を設定

def place_order_during_auction(exchange, symbol, order_type, quantity):
    """オークション中に注文を出す関数"""
    if order_type == 'buy':
        print(f"Placing a buy order for {quantity} {symbol} during the auction")
        # exchange.create_market_buy_order(symbol, quantity)
    else:
        print(f"Placing a sell order for {quantity} {symbol} during the auction")
        # exchange.create_market_sell_order(symbol, quantity)

def schedule_auction_orders(exchange, symbol):
    """オークションのスケジュールに従って注文を出す"""
    current_time = datetime.now()
    opening_auction_time = current_time.replace(hour=9, minute=30, second=0, microsecond=0)  # 仮の開場時間
    closing_auction_time = current_time.replace(hour=16, minute=0, second=0, microsecond=0)  # 仮の閉場時間

    # オープニングオークションでの注文
    if current_time < opening_auction_time:
        place_order_during_auction(exchange, symbol, 'buy', 1)
        time.sleep((opening_auction_time - current_time).total_seconds())

    # クロージングオークションでの注文
    if current_time < closing_auction_time:
        place_order_during_auction(exchange, symbol, 'sell', 1)
        time.sleep((closing_auction_time - current_time).total_seconds())

def main():
    schedule_auction_orders(exchange, symbol)

if __name__ == '__main__':
    main()

注意点と実装上の考慮事項

• 取引所のルール: オークションのルールは取引所によって異なります。注文を出す前に、各取引所の具体的なオークションのルールを確認してください。
• タイムゾーンの扱い: サーバーのタイムゾーンと市場のタイムゾーンが異なる場合があります。適切な時間に注文が出されるように、タイムゾーンの調整が必要です。
• ネットワーク遅延とAPIの制限: リアルタイムでの注文実行を想定しているため、ネットワーク遅延やAPIの利用制限に注意が必要です。

2.マイクロストラクチャー分析

マイクロストラクチャー分析戦略は、オーダーブックの詳細な動きを解析し、短期的な価格の動向を予測して取引するアプローチです。この戦略では、注文の流れ、売り買い圧力のバランス、そしてそれが価格に与える影響をリアルタイムで分析します。以下のPythonコードは、オーダーブックのデータを取得し、特定の条件に基づいて売買シグナルを生成する基本的なプロセスを示します。

必要なライブラリのインポート

import ccxt
import time

取引所の設定と初期化

exchange = ccxt.binance({
    'apiKey': 'YOUR_API_KEY',
    'secret': 'YOUR_API_SECRET',
    'enableRateLimit': True
})
symbol = 'BTC/USDT'

オーダーブックからの売買シグナルの生成

def analyze_order_book(exchange, symbol):
    order_book = exchange.fetch_order_book(symbol)
    bids = order_book['bids']
    asks = order_book['asks']

    # 最高買い注文と最低売り注文を取得
    best_bid = bids[0][0] if bids else None
    best_ask = asks[0][0] if asks else None

    # マイクロストラクチャー分析の基本的なロジック
    if best_bid and best_ask:
        mid_price = (best_bid + best_ask) / 2
        bid_ask_spread = best_ask - best_bid

        # 売買圧力の計算（累積）
        total_bid_volume = sum([bid[1] for bid in bids])
        total_ask_volume = sum([ask[1] for ask in asks])

        # 売買シグナルの判断
        if total_bid_volume > total_ask_volume * 1.2:  # 買い圧力が強い場合
            print(f"Buy signal at {best_ask}, Spread: {bid_ask_spread}")
            # 実際に注文を出す場合
            # exchange.create_market_buy_order(symbol, amount)
        elif total_ask_volume > total_bid_volume * 1.2:  # 売り圧力が強い場合
            print(f"Sell signal at {best_bid}, Spread: {bid_ask_spread}")
            # 実際に注文を出す場合
            # exchange.create_market_sell_order(symbol, amount)
        else:
            print("No clear signal, market is balanced.")

メイン関数

def main():
    while True:
        analyze_order_book(exchange, symbol)
        time.sleep(1)  # 1秒ごとにオーダーブックを分析

if __name__ == '__main__':
    main()

注意点と実装上の考慮事項

• データの遅延と整合性: リアルタイムのオーダーブックデータは非常に高速で更新されるため、データの遅延や整合性が取引結果に大きな影響を与える可能性があります。
• 市場の変動性: マイクロストラクチャー戦略は市場の変動性が高い時に特に有効ですが、急激な市場変動には注意が必要です。
• 取引手数料とスリッページ: 小さな価格変動を利用する戦略では、取引手数料やスリッページが利益に大きな影響を与えるため、これらのコストを考慮に入れる必要があります。
• 戦略のバックテストと最適化: 実際に運用する前に、過去データに対して戦略をバックテストし、パラメータの最適化を行うことが推奨されます。

3.ハイブリッドモデル

ハイブリッドモデル戦略では、定量的なアルゴリズム（クォンツモデル）と定性的な人的判断を組み合わせて、より効果的な投資決定を行います。このアプローチは、特に市場が非常に不確実で、単独の定量モデルでは対応が難しい場合に有効です。以下のPythonコードは、定量的なデータ分析に基づいて生成されたシグナルをレビューし、最終的な投資判断に定性的な要素を取り入れるプロセスを示します。

必要なライブラリのインポート

import numpy as np
import pandas as pd

定量的モデルの構築

ここでは、シンプルな移動平均ベースの戦略を例に定量的モデルを構築します。

# データ生成（例として株価の日次リターンを想定）
np.random.seed(0)
dates = pd.date_range(start="2020-01-01", periods=100)
data = np.random.randn(100).cumsum()
price_series = pd.Series(data, index=dates)

# 移動平均の計算
def calculate_moving_averages(price_data, window=5):
    return price_data.rolling(window=window).mean()

# トレーディングシグナルの生成
def generate_trading_signals(price_data):
    short_ma = calculate_moving_averages(price_data, window=5)
    long_ma = calculate_moving_averages(price_data, window=20)
    signals = np.where(short_ma > long_ma, 1, 0) - np.where(short_ma < long_ma, 1, 0)
    return signals

定性的判断の組み込み

投資の意思決定プロセスに定性的要素を組み入れるための関数を定義します。

def qualitative_adjustment(signals, current_news):
    adjusted_signals = signals.copy()
    # 定性的なニュースや情報に基づいたシグナルの調整
    for i in range(len(signals)):
        if current_news[i] == "positive":
            adjusted_signals[i] = max(1, signals[i])  # ポジティブなニュースがあれば買い強化
        elif current_news[i] == "negative":
            adjusted_signals[i] = min(-1, signals[i])  # ネガティブなニュースがあれば売り強化
    return adjusted_signals

実行と結果の表示

def main():
    signals = generate_trading_signals(price_series)
    current_news = ["neutral"] * 95 + ["positive"] * 5  # 最後の5日間にポジティブなニュースがあったと仮定
    adjusted_signals = qualitative_adjustment(signals, current_news)
    print("Original Signals:", signals[-10:])
    print("Adjusted Signals:", adjusted_signals[-10:])

if __name__ == '__main__':
    main()

注意点と実装上の考慮事項

• 定量的モデルの精度: 定量的な分析はデータ品質とモデルの精度に依存します。定期的なモデルの評価と調整が必要です。
• 定性的判断のバイアス: 人間の判断はバイアスがかかることがあります。定性的要素を取り入れる際は、その影響を理解し、可能な限り客観的な判断基準を設けることが重要です。
• ニュースの解釈: 定性的判断に用いる情報（例えば経済ニュース、政治的イベント）は複雑で、その解釈には専門知識が必要な場合があります。

このコードは、定量的アプローチと定性的判断を組み合わせたハイブリッドな投資戦略の一例を提供しますが、実際の運用にはさらに複雑な要素が介入することを理解してください。

実践的なコード

import numpy as np
import pandas as pd
import ccxt
from datetime import datetime

def fetch_crypto_data(exchange_id, symbol, timeframe, since, limit):
    exchange_class = getattr(ccxt, exchange_id)
    exchange = exchange_class({
        'enableRateLimit': True,
    })
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, since, 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)
    return df['close']

def calculate_moving_averages(price_data, window=5):
    return price_data.rolling(window=window).mean()

def generate_trading_signals(price_data):
    short_ma = calculate_moving_averages(price_data, window=5)
    long_ma = calculate_moving_averages(price_data, window=20)
    signals = np.where(short_ma > long_ma, 1, 0) - np.where(short_ma < long_ma, 1, 0)
    return signals

def qualitative_adjustment(signals, current_news):
    adjusted_signals = signals.copy()
    
    for i in range(len(signals)):
        if current_news[i] == "positive":
            adjusted_signals[i] = max(1, signals[i])
        elif current_news[i] == "negative":
            adjusted_signals[i] = min(-1, signals[i])
    return adjusted_signals

def main():
    exchange_id = 'binance'  # Example: Binance
    symbol = 'BTC/USDT'      # Bitcoin against USDT
    timeframe = '1d'         # Daily candles
    
    # Create an exchange instance
    exchange_class = getattr(ccxt, exchange_id)
    exchange = exchange_class({
        'enableRateLimit': True,
    })

    # Get current milliseconds and calculate the 'since' for 100 days ago
    current_time = exchange.milliseconds()
    since = current_time - 86400000 * 100  # 100 days ago in milliseconds
    
    limit = 100
    price_data = fetch_crypto_data(exchange_id, symbol, timeframe, since, limit)
    
    signals = generate_trading_signals(price_data)
    current_news = ["neutral"] * (len(signals) - 5) + ["positive"] * 5
    adjusted_signals = qualitative_adjustment(signals, current_news)
    
    print("Original Signals:", signals[-10:])
    print("Adjusted Signals:", adjusted_signals[-10:])
    
if __name__ == '__main__':
    main()

上記の修正版コードでは、仮想通貨市場から特定の通貨ペアの価格データを取得し、そのデータに基づいてトレーディングシグナルを生成し、さらにニュースデータに基づいてシグナルを調整するプロセスが含まれています。各ステップの詳細な解説は以下の通りです。

1. ライブラリのインポート

import numpy as np
import pandas as pd
import ccxt
from datetime import datetime

• numpy: 数値計算を効率的に行うためのライブラリです。
• pandas: データ分析を支援する機能を提供するライブラリで、データの操作や分析に広く用いられます。
• ccxt: さまざまな仮想通貨取引所のAPIにアクセスするためのライブラリです。
• datetime: 日付や時刻を扱うための標準ライブラリです。

2. fetch_crypto_data 関数

def fetch_crypto_data(exchange_id, symbol, timeframe, since, limit):
    exchange_class = getattr(ccxt, exchange_id)
    exchange = exchange_class({
        'enableRateLimit': True,
    })
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, since, 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)
    return df['close']

この関数は指定された取引所から指定された通貨ペアのOHLCVデータ（Open, High, Low, Close, Volume）を取得します。データは pandas の DataFrame 形式で整理され、タイムスタンプは datetime 形式に変換されます。

3. calculate_moving_averages 関数

def calculate_moving_averages(price_data, window=5):
    return price_data.rolling(window=window).mean()

この関数は価格データの移動平均を計算します。window パラメータで平均を取る期間を指定します。

4. generate_trading_signals 関数

def generate_trading_signals(price_data):
    short_ma = calculate_moving_averages(price_data, window=5)
    long_ma = calculate_moving_averages(price_data, window=20)
    signals = np.where(short_ma > long_ma, 1, 0) - np.where(short_ma < long_ma, 1, 0)
    return signals

5. qualitative_adjustment 関数

def qualitative_adjustment(signals, current_news):
    adjusted_signals = signals.copy()
    for i in range(len(signals)):
        if current_news[i] == "positive":
            adjusted_signals[i] = max(1, signals[i])
        elif current_news[i] == "negative":
            adjusted_signals[i] = min(-1, signals[i])
    return adjusted_signals

生成されたトレーディングシグナルをニュース情報に基づいて調整します。ポジティブなニュースはシグナルを強化し、ネガティブなニュースはシグナルを弱めます。

6. main 関数

def main():
    ...
    print("Original Signals:", signals[-10:])
    print("Adjusted Signals:", adjusted_signals[-10:])

メイン関数では、上述の関数を使用してデータを取得し、シグナルを生成して、それを調整します。最後に、最新の10個のシグナル（元のものと調整後のもの）を出力します。

4.最適ポジショニング

最適ポジショニング戦略では、投資家のリスク許容度に基づいてポジションサイズを計算し、調整することを目的とします。この戦略は、特に変動の大きい市場や多様な投資ポートフォリオにおいて、リスクを適切に管理するために重要です。以下のPythonコードは、投資家のリスク許容度に基づいてポジションサイズを計算する方法を示します。

必要なライブラリのインポート

import numpy as np
import pandas as pd

リスク許容度に基づくポジションサイズの計算

def calculate_position_size(account_balance, risk_per_trade, stop_loss_percent):
    """
    ポジションサイズを計算する。
    :param account_balance: 口座の残高
    :param risk_per_trade: 一回の取引に許容するリスクの割合（例: 0.01 は1%）
    :param stop_loss_percent: ストップロスの設定割合（例: 0.02 は2%）
    :return: 最適なポジションサイズ
    """
    risk_amount = account_balance * risk_per_trade  # トレードあたりのリスク額
    position_size = risk_amount / stop_loss_percent  # 最適なポジションサイズ
    return position_size

ポジションサイズの計算と結果の表示

def main():
    account_balance = 10000  # 口座の残高（USD）
    risk_per_trade = 0.01  # トレードあたりのリスク許容度（1%）
    stop_loss_percent = 0.02  # ストップロス設定（2%）

    position_size = calculate_position_size(account_balance, risk_per_trade, stop_loss_percent)
    print(f"Recommended position size: ${position_size:,.2f}")

if __name__ == '__main__':
    main()

注意点と実装上の考慮事項

• リスク管理の重要性: ポジションサイズの計算は、リスクを適切に管理するための最も基本的な要素の一つです。計算されたポジションサイズに従うことで、不測の市場の動きに対しても資金を守ることができます。
• 市場条件の変化への対応: 市場条件が変化した場合には、リスク許容度やストップロスのパーセンテージを調整する必要があります。
• レバレッジの影響: 使用するレバレッジによってリスクが増加するため、レバレッジを活用する際は特に慎重なリスク管理が求められます。

このコードは、リスク許容度に応じてポジションサイズを計算する基本的なフレームワークを提供します。投資戦略やポートフォリオ管理においてリスク管理は非常に重要であり、この方法は資金を効率的に管理し、市場での不確実性に対処するための一助となります。

5.ガンジーポートフォリオ構築

ガンジーポートフォリオ構築戦略は、異なる金融商品間で生じる無裁定機会を利用してリスクを最小化しつつ利益を最大化することを目的としています。この戦略では、価格の不一致や相関関係を利用して、裁定取引を行います。以下のコードは、シンプルなガンジーポートフォリオを構築するためのステップを示し、異なる金融商品間での裁定機会を計算する方法を実装します。

必要なライブラリのインポート

import pandas as pd
import numpy as np

データ取得と前処理

この例では、異なる資産の価格データをシミュレートします。実際のアプリケーションでは、適切なデータソースから実際の市場データを取得します。

# ダミーデータの生成
dates = pd.date_range('2021-01-01', periods=100)
data1 = np.random.normal(loc=100, scale=10, size=(100,))  # 資産1の価格
data2 = np.random.normal(loc=100, scale=15, size=(100,))  # 資産2の価格
df = pd.DataFrame({'Asset1': data1, 'Asset2': data2}, index=dates)

# 簡単な統計分析で相関を確認
correlation = df.corr()
print("Correlation between assets:", correlation)

裁定機会の識別

無裁定機会を識別するために、価格の差（スプレッド）が特定の閾値を超えた場合にポジションを取ることを検討します。

def identify_arbitrage_opportunities(df, threshold=20):
    # スプレッドの計算
    df['Spread'] = df['Asset1'] - df['Asset2']
    df['Mean_Spread'] = df['Spread'].mean()
    df['Spread_Deviation'] = df['Spread'] - df['Mean_Spread']

    # 裁定機会の識別
    df['Arbitrage_Opportunity'] = np.where(df['Spread_Deviation'].abs() > threshold, 1, 0)

    return df[df['Arbitrage_Opportunity'] == 1]

arbitrage_opportunities = identify_arbitrage_opportunities(df)
print("Arbitrage Opportunities:\n", arbitrage_opportunities[['Asset1', 'Asset2', 'Spread', 'Spread_Deviation']])

ポートフォリオ構築と最適化

ここで裁定ポートフォリオを構築し、リスクを管理しながら利益を最大化するための取引を行います。

def build_portfolio(df, opportunities):
    # 実際のポートフォリオ構築ロジックは、裁定機会に基づいて取引を行います
    # 例えば、Asset1を売ってAsset2を買うなど
    for index, row in opportunities.iterrows():
        if row['Spread_Deviation'] > 0:
            print(f"Buy Asset2 and sell Asset1 on {index.date()}")
        else:
            print(f"Sell Asset2 and buy Asset1 on {index.date()}")

build_portfolio(df, arbitrage_opportunities)

注意点と実装上の考慮事項

• 市場の流動性と取引コスト: 実際の取引では、市場の流動性と取引コストがポートフォリオのパフォーマンスに大きく影響します。
• データの更新と監視: 市場データは継続的に更新される必要があります。リアルタイムのデータフィードと自動化されたトレーディングシステムが推奨されます。
• 法規制とコンプライアンス: 特定の裁定戦略は規制の対象となることがあります。関連する法規制を遵守してください。

まとめ

今回も5種類の発展的なロジックの雛形をまとめました。

基本的なロジックに工夫を取り入れることで競合のbotに差をつけることができます。

また、小さな工夫の積み重ねが強いbot作りにつながります。