仮想通貨botの開発記録#104(2024/9/14)「上場戦botの雛形」

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

今回は「新規トークンの上場戦bot」の雛形を紹介します。

よし、全部読めた。まずは「二次トークン市場」の章から学んだことを実践する。やっぱりDeFiって独特の文化があるな。https://t.co/Ci7Xb7ejkE

— よだか(夜鷹/yodaka) (@yodakablog) September 14, 2024

上場戦って何？

「上場戦」とは、特に金融市場において新しい株式や仮想通貨が公式に取引所に上場される際の、その直後の取引活動を指します。この時期には通常、価格の大きな変動が見られることが多いです。これは新規上場の資産に対する投資家の期待や市場の反応によって価格が急激に上下するためです。

上場戦の特徴

1. 高いボラティリティ:
   • 新規上場の資産は、その供給と需要のバランスが初めて市場によって設定されるため、非常に高いボラティリティ（価格変動）を経験します。
2. 大量の取引:
   • 上場初日には多くの投資家がその資産を取引しようとするため、通常よりもはるかに多い取引量が見られます。
3. 投機的取引:
   • 短期間で大きな利益を狙う投機的なトレーダーが多く参加するため、感情的な取引が多くなりがちです。

上場戦での戦略

投資家やトレーダーは、このような状況を利用して短期間で利益を出すための戦略を立てることが一般的です。たとえば、事前にその資産に関する情報を集め、市場の期待値を読み解き、上場直後に適切なタイミングで売買を行うという戦略があります。また、自動取引botを使用して、市場の動きをリアルタイムで分析し、迅速に反応する方法もあります。

新規上場の機会は大きな利益をもたらす可能性がありますが、同時に大きなリスクも伴います。価格の急激な変動により、大損する可能性もあるため、リスク管理と十分な市場分析が重要です。

botの雛形

import requests
import time
import os
from requests.exceptions import RequestException

class CryptoTradingBot:
    def __init__(self, base_url):
        self.api_key = os.getenv('API_KEY')
        self.api_secret = os.getenv('API_SECRET')
        self.base_url = base_url
        self.headers = {'X-MBX-APIKEY': self.api_key}
    
    def get_market_data(self, symbol):
        """市場データを取得する"""
        url = f"{self.base_url}/api/v3/ticker/price?symbol={symbol}"
        try:
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()  # エラーがあれば例外を発生させる
            data = response.json()
            return float(data['price'])
        except RequestException as e:
            print(f"データ取得中にエラーが発生しました: {e}")
            return None
    
    def place_order(self, symbol, quantity, price, order_type):
        """注文を出す"""
        # 注文のシミュレーションです。実際のAPIを使った注文処理には変更が必要です。
        order_data = {
            'symbol': symbol,
            'quantity': quantity,
            'price': price,
            'type': order_type
        }
        try:
            # この部分は具体的なAPIに合わせて適切に実装してください。
            print("注文を出しました:", order_data)
            return True
        except RequestException as e:
            print(f"注文時にエラーが発生しました: {e}")
            return False
    
    def strategy(self):
        """取引戦略を実行する"""
        symbol = os.getenv('TRADE_PAIR', 'NEWCOINBTC')
        buy_price = self.get_market_data(symbol)
        if buy_price is None:
            return
        
        buy_threshold = float(os.getenv('BUY_THRESHOLD', '0.00001000'))
        
        if buy_price < buy_threshold:
            self.place_order(symbol, 100, buy_price, 'buy')
        
        time.sleep(int(os.getenv('SLEEP_INTERVAL', '300')))  # 環境変数からスリープ時間を設定
        sell_price = self.get_market_data(symbol)
        if sell_price and sell_price > buy_price * 1.05:
            self.place_order(symbol, 100, sell_price, 'sell')

if __name__ == "__main__":
    base_url = 'https://api.exchange.com'  # 使う取引所のAPIのベースURL
    bot = CryptoTradingBot(base_url)
    bot.strategy()

説明:

1. API認証の改善: APIキーを環境変数から読み込み、リクエストのヘッダーに追加しています。
2. エラーハンドリングの強化: requestsのraise_for_status()を使用してHTTPエラーを捕捉し、例外を処理しています。
3. リアルタイム性とパラメータの柔軟性: time.sleep()の間隔や取引する通貨ペアなどのパラメータを環境変数で設定できるようにしました。
4. セキュリティ対策: APIキーとシークレットをソースコードから分離し、環境変数を通じて安全に管理しています。

このボットのロジックは、単純な「買い安売り高」の戦略に基づいて利益を出すことを目指しています。以下の流れで動作しています：

1. 市場データ取得: ボットはget_market_dataメソッドを使い、指定された通貨ペア（デフォルトではNEWCOINBTC）の現在の価格を取得します。
2. 買い注文の実行:
   • ボットはまず、環境変数BUY_THRESHOLDに設定された買い閾値（例: 0.00001000）と現在の市場価格を比較します。
   • 市場価格がこの買い閾値を下回った場合、ボットは指定された数量（デフォルトでは100単位）をその価格で買う注文を実行します。この注文はplace_orderメソッドで行われます。
3. 売り注文の実行:
   • 一度買い注文が成功すると、ボットは指定された時間（環境変数SLEEP_INTERVALで設定、デフォルトでは300秒）待機します。
   • その後、再度市場価格を取得し、買値の5%（buy_price * 1.05）以上の価格であれば、売り注文を実行して利益を確定します。

利益の出し方

このボットの利益の出し方は以下のとおりです：

• 安値で買い、高値で売る: 市場価格が一定の閾値（BUY_THRESHOLD）を下回ったときに購入し、その後価格が少なくとも5%上昇した時点で売ることで利益を確定します。
• 売買のタイミングをシンプルに設定: 基本的には買値から5%の値上がりが確認できたら売る、というシンプルなロジックに基づいています。

この戦略の成功は、価格が短期的に変動する市場であれば、ボットが頻繁に取引を行い、小さな利益を積み重ねることができますが、価格が一方向に急激に動いた場合（急落など）には利益を上げるのが難しくなります。

改善点

• この戦略では価格の変動率や他の指標を考慮していないため、損失リスクを減らすために損切り機能や追加のテクニカル指標を用いた高度な戦略を実装することも考慮できます。

さらなる改善点

上記の自動取引botのコードは基本的な枠組みとしては良くなっていますが、実際の取引環境での運用を考慮すると、まだいくつか検討すべき問題点があります。ここで挙げられるいくつかの追加の検討ポイントを確認しましょう：

1. 注文の詳細なエラーハンドリングが不足:
   • 注文処理部分には具体的なAPI呼び出しコードがなく、エラーハンドリングも十分ではありません。実際のAPIを使う場合、失敗した注文に対する適切な対応が必要です。例えば、注文が失敗した場合に再試行するロジックや、注文の失敗理由をログに記録する処理が考えられます。
2. マーケットデータのリアルタイム性の不足:
   • 現在の実装では、固定時間ごとにのみ市場データを取得しています。高速で変動する市場状況に対応するためには、より頻繁にまたはイベント駆動型で市場データを取得する仕組みが望ましいです。
3. ロジックの硬直性:
   • 取引ロジックが非常にシンプルであり、市場状況の変化に対応するための動的な調整が困難です。例えば、異なる市場条件に基づいて動的に売買しきい値を調整する機能が必要かもしれません。
4. 同時多発的な注文管理の不足:
   • 複数の注文を同時に管理する機能がありません。複数の通貨ペアを扱う場合や、複数の注文を同時に追跡する必要がある場合、注文管理のロジックを強化する必要があります。
5. APIレートリミットの考慮不足:
   • 多くの取引所APIにはレートリミットがありますが、このコードではその対策が見られません。API呼び出しの頻度を制限したり、レートリミットに達した際の適切な処理を実装する必要があります。
6. ロギングとモニタリングの不足:
   • 実際の運用では、ロギングとモニタリングが非常に重要です。システムの状態や取引の履歴を適切に記録し、問題発生時に迅速に対応できるようにする必要があります。

これらの問題点を検討し、それぞれに対して具体的な改善策を検討・実装することで、より実用的で信頼性の高い自動取引botを開発することができます。

戦略のスケーリング

新規上場時の取引戦略をスケールアップするためには、より洗練された取引アルゴリズムと、市場データの詳細な分析が必要になります。ここでは、スケーリングのための高度な戦略と、その戦略を実行するために必要な学習領域をいくつか紹介します。

スケールアップされた戦略

1. 量的分析と機械学習の活用:
   • 大量の市場データを収集し、過去の上場データからパターンを学習するために機械学習モデルを訓練します。これには、価格予測、ボラティリティ予測、および取引量の変動を予測するモデルが含まれます。
2. 高頻度取引（HFT）の技術:
   • ミリ秒単位で取引を行う高頻度取引技術を使用して、市場の小さな価格変動から利益を得る。この技術は非常に高速な取引実行能力と、低遅延のトレードインフラを必要とします。
3. センチメント分析の導入:
   • SNSやニュース記事からのデータを分析して市場の感情を評価し、その情報を取引戦略に組み込む。市場の期待や恐怖が価格にどのように影響するかを理解することが重要です。
4. 多変量統計モデルの適用:
   • 複数の市場指標を同時に分析し、それらの相互作用が価格にどのように影響するかを評価するための統計モデルを使用します。

必要な学習領域

1. データサイエンスと統計学:
   • 市場データを解析し、予測モデルを構築するためには、統計学とデータサイエンスの強い知識が必要です。特に回帰分析、時系列分析、確率論が重要になります。
2. 機械学習と人工知能:
   • 教師あり学習、教師なし学習、強化学習などの機械学習技術を理解し、実際の取引データに適用する能力が必要です。
3. コンピュータサイエンスとプログラミング:
   • 高速な取引システムを構築するためには、効率的なコーディングスキルと、システムアーキテクチャに関する知識が求められます。Python、C++、Javaなどの言語が一般的に使われます。
4. ファイナンスと市場理論:
   • 市場の基本的な動作理論や、金融商品の詳細な知識が必要です。経済指標やファイナンス理論を理解することで、より良い取引判断が可能になります。
5. 情報技術インフラ:
   • 低遅延取引を実現するためのハードウェアとソフトウェアの設計についての深い知識が求められます。ネットワークエンジニアリングやクラウドコンピューティングのスキルも役立ちます。

スケールアップを考慮した雛形

import os
import requests
import time
import pandas as pd
from sklearn.ensemble import RandomForestRegressor
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
from requests.exceptions import RequestException

class CryptoTradingBot:
    def __init__(self, base_url):
        self.api_key = os.getenv('API_KEY')
        self.api_secret = os.getenv('API_SECRET')
        self.base_url = base_url
        self.headers = {'X-MBX-APIKEY': self.api_key}
        self.model = self.train_model()

    def get_market_data(self, symbol):
        """リアルタイムで市場データを取得する"""
        url = f"{self.base_url}/api/v3/ticker/price?symbol={symbol}"
        try:
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            data = response.json()
            return float(data['price'])
        except RequestException as e:
            print(f"データ取得中にエラーが発生しました: {e}")
            return None

    def train_model(self):
        """機械学習モデルを訓練する（ダミーデータを使う）"""
        data = pd.read_csv('historical_prices.csv')  # 過去の価格データ
        X = data[['volume', 'open', 'high', 'low']]
        y = data['close']
        model = RandomForestRegressor(n_estimators=100)
        model.fit(X, y)
        return model
    
    def predict_price(self, current_data):
        """価格予測を行う"""
        return self.model.predict([current_data])[0]

    def analyze_sentiment(self, news):
        """市場のセンチメントを分析する"""
        analyzer = SentimentIntensityAnalyzer()
        sentiment = analyzer.polarity_scores(news)
        return sentiment['compound']

    def place_order(self, symbol, quantity, price, order_type):
        """注文を出す（注文処理はシミュレーション）"""
        order_data = {
            'symbol': symbol,
            'quantity': quantity,
            'price': price,
            'type': order_type
        }
        print("注文を出しました:", order_data)
        return True
    
    def strategy(self):
        """取引戦略を実行する"""
        symbol = os.getenv('TRADE_PAIR', 'NEWCOINBTC')
        current_price = self.get_market_data(symbol)
        if current_price is None:
            return
        
        # ニュース記事からセンチメントを分析
        news = "Bitcoin surges with new market entrant"
        sentiment_score = self.analyze_sentiment(news)
        
        # モデルに基づいて価格を予測
        predicted_price = self.predict_price([10000, current_price, current_price*1.01, current_price*0.99])
        
        # 戦略の決定
        if sentiment_score > 0.1 and predicted_price > current_price * 1.02:
            self.place_order(symbol, 100, current_price, 'buy')
        elif sentiment_score < -0.1 and predicted_price < current_price * 0.98:
            self.place_order(symbol, 100, current_price, 'sell')

if __name__ == "__main__":
    base_url = 'https://api.exchange.com'
    bot = CryptoTradingBot(base_url)
    bot.strategy()

Cursorで更に修正したコード

import logging
from dotenv import load_dotenv
import os
import time
import ccxt
from decimal import Decimal

    # ログの設定
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

    # 環境変数の読み込み
load_dotenv()

class ImprovedCryptoTradingBot:
    def __init__(self, exchange_id):
        try:
            self.exchange = getattr(ccxt, exchange_id)({
                'apiKey': os.getenv('API_KEY'),
                'secret': os.getenv('API_SECRET'),
            })
        except AttributeError:
            logger.error(f"Invalid exchange ID: {exchange_id}")
            raise
        except ccxt.AuthenticationError:
            logger.error("Authentication failed. Check your API key and secret.")
            raise
        except Exception as e:
            logger.error(f"Failed to initialize exchange: {str(e)}")
            raise
        self.base_asset = os.getenv('BASE_ASSET', 'BTC')
        self.quote_asset = os.getenv('QUOTE_ASSET', 'USDT')
        self.symbol = f'{self.base_asset}/{self.quote_asset}'
        self.max_position = Decimal(os.getenv('MAX_POSITION', '1000'))
        self.stop_loss_percent = Decimal(os.getenv('STOP_LOSS_PERCENT', '0.02'))
        self.take_profit_percent = Decimal(os.getenv('TAKE_PROFIT_PERCENT', '0.05'))
        self.model = self.train_model()

    def get_market_data(self, symbol):
        try:
            ticker = self.exchange.fetch_ticker(symbol)
            return Decimal(str(ticker['last']))
        except Exception as e:
            logger.error(f"市場データの取得に失敗しました: {str(e)}")
            return None

    def train_model(self):
        # モデルのトレーニングロジックをここに実装
        # この例では簡単のため、ダミーのモデルを返します
        return lambda x: sum(x) / len(x)

    def predict_price(self, current_data):
        return Decimal(str(self.model(current_data)))

    def analyze_sentiment(self, news):
        # センチメント分析のロジックをここに実装
        # この例では簡単のため、ランダムなスコアを返します
        return Decimal(str(np.random.uniform(-1, 1)))

    def get_account_balance(self):
        try:
            balance = self.exchange.fetch_balance()
            return Decimal(str(balance[self.quote_asset]['free']))
        except Exception as e:
            logger.error(f"口座残高の取得に失敗しました: {str(e)}")
            return None

    def calculate_position_size(self, price):
        balance = self.get_account_balance()
        if balance is None:
            return Decimal('0')
        return min(balance * Decimal('0.01'), self.max_position) / price

    def place_order(self, symbol, quantity, price, order_type):
        try:
            order = self.exchange.create_market_order(symbol, order_type, quantity)
            logger.info(f"{order_type.capitalize()}注文を出しました: {symbol}, 数量: {quantity}, 価格: {price}")
            return order
        except Exception as e:
            logger.error(f"注文の実行に失敗しました: {str(e)}")
            return None

    def place_order_with_risk_management(self, symbol, quantity, price, order_type):
        try:
            order = self.place_order(symbol, quantity, price, order_type)
            if order and order_type == 'buy':
                stop_loss_price = price * (Decimal('1') - self.stop_loss_percent)
                take_profit_price = price * (Decimal('1') + self.take_profit_percent)
                self.place_order(symbol, quantity, stop_loss_price, 'sell')
                self.place_order(symbol, quantity, take_profit_price, 'sell')
            return order
        except Exception as e:
            logger.error(f"リスク管理付き注文の実行に失敗しました: {str(e)}")
            return None

    def strategy(self):
        try:
            current_price = self.get_market_data(self.symbol)
            if current_price is None:
                logger.warning("市場データの取得に失敗しました")
                return

            news = "Bitcoin surges with new market entrant"
            sentiment_score = self.analyze_sentiment(news)

            predicted_price = self.predict_price([10000, float(current_price), float(current_price*Decimal('1.01')), float(current_price*Decimal('0.99'))])

            quantity = self.calculate_position_size(current_price)

            if sentiment_score > Decimal('0.1') and predicted_price > current_price * Decimal('1.02'):
                self.place_order_with_risk_management(self.symbol, quantity, current_price, 'buy')
            elif sentiment_score < Decimal('-0.1') and predicted_price < current_price * Decimal('0.98'):
                self.place_order_with_risk_management(self.symbol, quantity, current_price, 'sell')
            else:
                logger.info("取引条件を満たしていません。待機します。")

        except Exception as e:
            logger.error(f"戦略の実行中にエラーが発生しました: {str(e)}")

if __name__ == "__main__":
    exchange_id = os.getenv('EXCHANGE_ID', 'binance')
    bot = ImprovedCryptoTradingBot(exchange_id)
    
    while True:
        try:
            bot.strategy()
            time.sleep(60)  # 1分ごとに戦略を実行
        except Exception as e:
            logger.error(f"メインループでエラーが発生しました: {str(e)}")
            logger.error(traceback.format_exc())
            time.sleep(60)  # エラー発生時も1分待機

このコードは、暗号通貨の取引ボットを実装したものです。以下に各部分の役割を説明します。

インポートと設定

import logging
from dotenv import load_dotenv
import os
import time
import ccxt
from decimal import Decimal

 必要なライブラリをインポートしています。

loggingはログを記録するために使用。dotenvは環境変数を読み込むために使用。osは環境変数の取得に使用。timeは時間の操作に使用。ccxtは暗号通貨取引所のAPIを扱うためのライブラリ。Decimalは高精度の数値計算に使用。

ログの設定と環境変数の読み込み

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
load_dotenv()

 ログの設定を行い、ログのフォーマットを指定。

環境変数を読み込む。

ImprovedCryptoTradingBot クラス

class ImprovedCryptoTradingBot:
    def __init__(self, exchange_id):
        # ... 初期化コード ...

- ボットのクラスを定義。

__init__メソッドで取引所の初期化、環境変数の読み込み、モデルのトレーニングを行う。

市場データの取得

    def get_market_data(self, symbol):
        # ... 市場データの取得コード ...

- 指定されたシンボルの市場データを取得。

モデルのトレーニング

    def train_model(self):
        # ... モデルのトレーニングコード ...

簡単なダミーモデルをトレーニング。

価格予測

def predict_price(self, current_data):
        # ... 価格予測コード ...

- トレーニングしたモデルを使って価格を予測。

センチメント分析

    def analyze_sentiment(self, news):
        # ... センチメント分析コード ...

 ニュースのセンチメントを分析（ダミーのスコアを返す）。

口座残高の取得

    def get_account_balance(self):
        # ... 口座残高の取得コード ...

取引所から口座残高を取得。

ポジションサイズの計算

    def calculate_position_size(self, price):
        # ... ポジションサイズの計算コード ...

市場注文を実行。

リスク管理付き注文の実行

    def place_order_with_risk_management(self, symbol, quantity, price, order_type):
        # ... リスク管理付き注文の実行コード ...

ストップロスとテイクプロフィットを設定した注文を実行。

戦略の実行

    def strategy(self):
        # ... 戦略の実行コード ...

市場データ、ニュースのセンチメント、予測価格を基に取引戦略を実行。

メインループ

if __name__ == "__main__":
    exchange_id = os.getenv('EXCHANGE_ID', 'binance')
    bot = ImprovedCryptoTradingBot(exchange_id)
    
    while True:
        try:
            bot.strategy()
            time.sleep(60)  # 1分ごとに戦略を実行
        except Exception as e:
            logger.error(f"メインループでエラーが発生しました: {str(e)}")
            logger.error(traceback.format_exc())
            time.sleep(60)  # エラー発生時も1分待機

ボットを初期化し、1分ごとに戦略を実行するメインループ。

まとめ

今回は「上場戦bot」の雛形とCursorでの「改善型」のコードをまとめました。

今後もこの調子で仮想通貨botの開発状況を発信していきます。

Cursorでの開発、、「問題点を指摘して改良して」くらいの超ざっくりした指示だと「特製ラーメントッピング全部のせ」みたいな改善案を出してくるのが面白い。

シンプルさがなくなってbotの動きが悪くなりそうなので特定の範囲を指定をして「この部分の処理を高速化して」などの指示が良さげ。

— よだか(夜鷹/yodaka) (@yodakablog) September 14, 2024