実用性ゼロの新規上場銘柄即日買いbotを作ってみた件：Bitget編

趣旨

なぜ即日買いなのに実用性がゼロなのか？

上場直後カンマ0.1秒で買いに行くbotならまだしも上場直後の高騰している時間に注文を成行で行う最悪のbotです。

実装する必要性は全くありません。
はっきり言って「時間の無駄」です。

単に思いついたロジックをコードに落とし込めるかどうか確かめたかっただけです。
練習ですね。

ただし、新規上場銘柄すべてをホドルする目的があっていちいち新規銘柄をチェックするのが煩わしいというのならアリかも知れません。

ロジック

ざっくりいうと、こんな感じです。

①　1日前の上場銘柄リストがディレクトリにあるか確認する。
②　Bitgetの上場銘柄をAPIで取ってきてリスト化する。
③　①と②のデータを比較
④　もし新しい上場銘柄が存在するならその銘柄を成り行き注文する
⑤　購入枚数を記録する

単純なロジックですが、実際に実装するとなるといくつか障壁がありました。

部分的な動作確認はしたのですが、新規上場銘柄がないと全ての動作確認ができないのでその部分だけ脳内シミュレーションで補いました。実装してもその部分のせいで正常に動かないかも知れないですが、どうせ使えないbotなのでフルでテストするのは諦めました。

コード（コピペ可）

仮想環境設定

仮想環境のことをよく知らなくても念のためにやっておいた方が良いです。

python3 -m venv myenv
source myenv/bin/activate

専用のディレクトリを構築します。

mkdir jikannomuda
cd jikannomuda

成行注文のコード

注意書きを羅列します。

• 上場銘柄情報を取得するときは”BTCUSDT”のような形で取得するのですが、注文する際は”BTC/USDT”という形でしか受け付けてもらえないみたいでした。

• bitget.options["defaultType"] = "swap"と書くと現物を注文します。
  先物を注文したい場合は以下の記事が参考になります。

• 現物の成行注文の場合はcreateMarketBuyOrderRequiresPriceをFalseにしないとうまくいきませんでした。

• 指値注文の場合は"market"ではなく"limit"にして価格情報も引数にセットする必要があります。

• create_order関数は順序が定まってる引数と順序が定まっていない引数があるようでした。順番を変えるとうまくいかなかったりします。bitget.create_order(symbol, 'market', side, size, None, {'force': 'FOK'})

market_order.py

import ccxt
import config
import json
import time
from data_functions import update_symbol_data, load_previous_data, save_data_to_json

def place_market_order(symbol, side, size):
    try:
        # Initialize the Bitget exchange instance
        bitget = ccxt.bitget({
            "apiKey": config.bitget_api_key,
            "secret": config.bitget_secret_key,
            "password": config.bitget_password,
            "enableRateLimit": True,
            "options": {
                'createMarketBuyOrderRequiresPrice': False
                }
        })

        # Set the default type to "swap"
        bitget.options["defaultType"] = "swap"

        # Create a market order
        order = bitget.create_order(symbol, 'market', side, size, None, {'force': 'FOK'})

        # Print the order response or perform additional actions as needed
        print("Order successfully placed:")
        print(order)

        previous_data = load_previous_data()
        symbol = symbol.replace("/", "")

        # Wait for 30 seconds
        time.sleep(30)

        # Fetch balance for the newly bought coin
        coin_name = symbol.split("USDT")[0].upper()

        balance_data = bitget.fetch_balance({'type': 'spot'})
        print('printing balance data')

        available = None
        for coin in balance_data['info']:
            if coin['coinName'] == coin_name:
                available = coin['available']
                print("success")
                print(coin_name)
                print(coin['available'])

        update_symbol_data(symbol, available)

    except ccxt.BaseError as e:
        # Handle exceptions if the order placement fails
        print(f"Error placing order: {e}")
        return None

上場銘柄のリストを管理する関数

特筆すべきことはないですが、JSONファイルの扱いの備忘録程度には使えるコードかも知れません。

data_functions.py

import requests
import json
import os

# Define the Bitget API endpoint
api_url = "https://api.bitget.com/api/v2/spot/public/symbols"

def fetch_symbol_data():
    response = requests.get(api_url)
    if response.status_code == 200:
        data = response.json()
        if data.get("code") == "00000":
            symbols_data = data.get("data", [])
            # Filter symbols to include only those ending with "USDT"
            filtered_symbols_data = [symbol_info for symbol_info in symbols_data if symbol_info.get("symbol").endswith("USDT")]
            return filtered_symbols_data
        else:
            print(f"Error: {data.get('msg', 'Unknown Error')}")
    else:
        print(f"Failed to fetch data. HTTP Status Code: {response.status_code}")
    return []

def load_previous_data():
    if os.path.exists('symbol_data.json'):
        with open('symbol_data.json', 'r') as json_file:
            return json.load(json_file)
    else:
        return {}

def save_data_to_json(data):
    with open('symbol_data.json', 'w') as json_file:
        json.dump(data, json_file, indent=4)

def update_symbol_data(symbol, available):
    # Load the existing data
    if os.path.exists('symbol_data.json'):
        with open('symbol_data.json', 'r') as file:
            data = json.load(file)
    else:
        data = {}

    # Update the data with the new available balance
    if symbol in data:
        if len(data[symbol]) == 2: # originally 2
            data[symbol][1] = available  # originally 1
        else:
            data[symbol].append(available)  # Add balance if not already present
    else:
        data[symbol] = [available]  # Add new entry for the symbol

    # Save the updated data back to the JSON file
    with open('symbol_data.json', 'w') as file:
        json.dump(data, file, indent=4)

メインのコード

ここが動作確認できなかった部分を含んでいる箇所です。

実力不足でマジックナンバーを含んでしまいました。
どうやったらJSONに取り込んだminTradeUSDTを変数として扱えるのかわかりませんでした。

どうやって実装するのかご存じの方がいたら私のXか以下のコメント欄で指摘してもらえると助かります。

https://twitter.com/daniel_tanaka_

その他注意点

• place_market_order関数を呼ぶ際のsizeはUSDT建てです。minTradeUSDTが全てのUSDTペア銘柄で5だったので5超の数値で注文が成立します。５ぴったりではエラーになりました。

• sideは"buy"か"sell"で成立します。

main.py

from data_functions import fetch_symbol_data, load_previous_data, save_data_to_json, update_symbol_data
from datetime import datetime
from market_order import place_market_order

def main():
    # Load previously stored data
    previous_data = load_previous_data()
    # Fetch current data
    current_data = fetch_symbol_data()
    # Get the current date
    current_date = datetime.now().date()

    # Check if data needs to be updated (e.g., once a day)
    if previous_data and 'last_update_date' in previous_data:
        last_update_date = datetime.strptime(previous_data['last_update_date'], '%Y-%m-%d').date()
        if current_date == last_update_date:
            print("Data has already been updated today.")
            return
        else:
            # Compare the new data with the previous data to detect newly added symbols
            new_symbols = set()
            for symbol_info in current_data:
                symbol = symbol_info.get("symbol").replace("/", "") # this is probably unnecessary
                if symbol not in previous_data:
                    new_symbols.add(symbol)

            if new_symbols:
                print("Newly added symbols today:")
                for symbol in new_symbols:
                    # Length of 'USDT'
                    usdt_symbol_length = 4
                    # put / in between Coin name and USDT
                    formatted_symbol = symbol[:-usdt_symbol_length] + '/' + symbol[-usdt_symbol_length:]
                    min_trade_usdt = 5 + 1

                    # 5 is most likely minTradeUSDT but it will fail so it's better to add a random number here like +1
                    place_market_order(formatted_symbol, "buy", min_trade_usdt)

            else:
                print("No newly added symbols today.")
    else:
        print("Initializing data...")

    # Update the symbol data in the JSON file with the current data
    symbol_data_dict = {}
    for symbol_info in current_data:
        symbol = symbol_info.get("symbol").replace("/", "")
        min_trade_usdt = symbol_info.get("minTradeUSDT") # maybe it's better to delete this part?
        symbol_data_dict[symbol] = [min_trade_usdt] # maybe it's better to delete this part?

    # Update the last update date
    symbol_data_dict['last_update_date'] = current_date.strftime('%Y-%m-%d')
    # Save the data to the JSON file
    save_data_to_json(symbol_data_dict)

    # Add a log message
    print("Code executed successfully.")

if __name__ == "__main__":
    main()

ダブルクオーテーションの中にそれぞれ記入して保存します。

config.py

bitget_api_key = ""
bitget_secret_key = ""
bitget_password = ""

以下のコマンドでJSONファイルを初期化します。

python main.py 

翌日から新しいコインがあれば成行注文を行うので以下のコマンドをcronで毎日自動で起動するように設定すれば終わりです。

python main.py 

cronについての備忘録は以下の記事にあります。