上場戦botの軌跡
おはこんばんにちは。
今回はBybitでの上場戦botを作っていた記録を時系列順に書いていきます。久しぶりの更新なので、ふざけた書き方を忘れました🥹
(深夜に書いたので日本語バグってますが、ご了承ください)
序章の序章
事の発端としては5月。
Pythonを用いたBot開発に注力していた時期で、CEXは上場戦bot, 裁量bot、そしてDEXはアビトラbot, 清算botなど作っていました。
一夜で大半のロジは完成させて、チューニングに結構な時間を費やしていました。
そして、CEXbotとしてエッジが効きそうだと思った分野として、上場戦でした。
その時はプレマが結構騒がれていたので、エアドロ系のトークンの先行取引で取得して、上場時の上ヒゲで売るという鞘の取り方が出来ていたらしいです。
(今はプレマ価格が上場価格より上が多くなってきたので、もう壊滅的です)
そして、研究を始めました。
序章: Rustで書き始める
Pythonでは言語処理的に遅いと言われていた為、何も考えずにRustを採用。
ただ、Rustは完全初学者だったので、Claudeを駆使しまくって、なんとか箱自体は作れました。
その時の結果やストーリーに関しては過去に書いたnoteをご参照ください
そして、結果的には儲からなかったです。
記事には100USD儲かったと言ってますが、結局は自分が望むようなロジとは程遠い動きでした。
そして、僕はPythonで何とかする事に決めました。
第一章: pybitでロジ実装
まず、僕はこの時点で間違っていました。
自分は裁量botを作っていた経験から、Bybit pythonの公式ライブラリであるpybitを使用してGET/POSTを叩いて実装しました。
そして、ロジの流れ的には下記の通りです。
configで指定した時刻日の5分前にPRICE GET連打
↓
retCode0吐いたら指定パラメータで成功するまでBUY POST
↓
PRICE GET連打でconfig指定の指定倍数分まで待機
↓
SELL POST
また、botterの皆様ならinstrument-infoで注文形式の端数考慮はしないのかと疑問に思ったでしょう。
恥ずかしながら、自分は当初脳筋思考だったので、外部ファイルから自力実装して調整していました。
ただ、これは後々に生かされていました。
そして、結果ですが、、当然儲かりませんでした。
敗北原因は明確でした。
価格取得が遅すぎて注文工程まで入らん。
APIのせいなのか俺のせいなのか良くわかりませんが、新規ペアがAPIへ落とされるまでが遅く、無効処理としてパスされます。
これじゃ価格取得の意味がないので、スクレピングで脳筋実装を試みます。
第二章: 脱API (GET)
SeleniumというスクレピングやWeb操作に特化したライブラリを使用しました。
とりあえずmarkPriceの取得をブラウザ上からできた為、成功です。
また、それと同時に GET系に関してはAPIを使いたくなかったので、端数調整を別ファイルとして持たせて、NumPyで高速演算させることにしました。
そして、注文ロジと組み合わせ、テストしました。
なにか嫌な予感がしますが、まあいいでしょう。
そして、本番。
まあ、この時の俺は気付いてませんが、単に指値が通ってないだけでなく、POSTスピードが遅かったでした。
つまり、現状の注文構造だと到底無理。（msレベル）
どうにかして注文スピードを早めたい。。
ということで、WebSoscketの存在を今更ながら知ります。
第三章: WSの偉大さ。
WSとREST APIの違いをとりあえず簡単に教えてもらいました
そして、それを知ったワイはすぐに実装しました。
そして、結果的に1/100レベルで短縮することに成功。
ただ、DEBUGデータを吐き出させると極端に遅くなるので、力技で連投させることにしました。
第四章: 初勝利
ドキドキしながら本番に挑みました。
そしたら、なんと。
注文が通り、上場価格に近い価格で購入することに成功しました。
小さな金額ですが、この瞬間に70USD取ることができました。
自分の戦略が効果的なエッジとして輝いた瞬間でした。
第五章: 結局。。
次回戦もウキウキで試みましたが、、
自分は大事な部分を見落としていました。
はい。
なぜ、指値でmarkPriceより高い価格で注文したら即約定されるのか。
それは、orderBookなので、当然ask板がないとfillされない = 約定されない
ということを、ここで初めて知りました。
(普通にアホ)
そして、この時点でやる気がなくなって、辞退しました。
結局はペアによっての運だからですね。
なので、ここまで見てくださった皆様には、コードを貼ります。
自己責任でお願いいたします。
main.py
import json
import time
import asyncio
import websockets
import hmac
import hashlib
from datetime import datetime
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from webdriver_manager.chrome import ChromeDriverManager
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import numpy as np
import concurrent.futures
from adjust_values import adjust_price, adjust_qty
DEBUG = True
def log(message, level='INFO', color=None):
if level == 'DEBUG' and not DEBUG:
return
timestamp = datetime.now().strftime("%H:%M:%S.%f")[:-3]
if color == 'green':
print(f"\033[92m{timestamp} - {level} - {message}\033[0m")
elif color == 'red':
print(f"\033[91m{timestamp} - {level} - {message}\033[0m")
else:
print(f"{timestamp} - {level} - {message}")
def load_config():
with open('settings/config.json') as f:
return json.load(f)
def setup_driver():
options = Options()
options.add_argument("--disable-gpu")
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
options.add_argument("--log-level=3")
service = Service(ChromeDriverManager().install())
driver = webdriver.Chrome(service=service, options=options)
driver.set_page_load_timeout(30)
return driver
async def authenticate(ws, api_key, api_secret):
expires = int((time.time_ns() // 1_000_000) + 10000)
signature = hmac.new(api_secret.encode(), f'GET/realtime{expires}'.encode(), hashlib.sha256).hexdigest()
auth_msg = {
"reqId": "auth",
"op": "auth",
"args": [api_key, expires, signature]
}
await ws.send(json.dumps(auth_msg))
response = await ws.recv()
log(f"Auth response: {response}")
async def send_order(ws, order):
sub_msg = {
"reqId": str(time.time_ns() // 1_000_000),
"header": {
"X-BAPI-TIMESTAMP": str(time.time_ns() // 1_000_000),
"X-BAPI-RECV-WINDOW": "8000",
},
"op": "order.create",
"args": [order]
}
await ws.send(json.dumps(sub_msg))
def get_prices(driver):
script = """
var askElements = document.querySelectorAll('.handicap-list.is-sell li');
var bidElements = document.querySelectorAll('.handicap-list.is-buy li');
var askPrice = askElements.length > 0 ? askElements[askElements.length - 1].querySelector('em').textContent : null;
var bidPrice = bidElements.length > 0 ? bidElements[0].querySelector('em').textContent : null;
var markPrice = document.querySelector('li.last-price > p.num').textContent;
return [askPrice, bidPrice, markPrice];
"""
return driver.execute_script(script)
def parse_price(price_str):
if price_str is None or price_str == '--':
return None
try:
return float(price_str.replace(',', '').replace('≈', '').replace('USD', '').strip())
except ValueError:
return None
async def scrape_prices_task(driver):
loop = asyncio.get_running_loop()
with concurrent.futures.ThreadPoolExecutor() as pool:
while True:
prices = await loop.run_in_executor(pool, get_prices, driver)
ask_price, bid_price, mark_price = map(parse_price, prices)
if ask_price is not None and bid_price is not None and mark_price is not None:
return ask_price, bid_price, mark_price
log(f"Invalid prices received: {prices}. Retrying...", level='DEBUG')
await asyncio.sleep(0.001)
async def place_buy_order_task(ws, symbol, qty, price):
order = {
"category": "spot",
"orderType": "Limit",
"price": str(price),
"qty": str(qty),
"side": "Buy",
"symbol": symbol,
}
await send_order(ws, order)
async def place_sell_order_task(ws, symbol, qty, price):
order = {
"category": "spot",
"orderType": "Limit",
"price": str(price),
"qty": str(qty),
"side": "Sell",
"symbol": symbol,
}
await send_order(ws, order)
async def main():
config = load_config()
api_key = config['APIKEY']
api_secret = config['SECRETKEY']
usdt_amount = float(config['USDT_AMOUNT'])
listing_time_str = config['LISTING_TIME']
listing_time = datetime.strptime(listing_time_str, '%Y-%m-%d %H:%M:%S')
symbol = config['SYMBOL']
start_interval = int(config['START_INTERVAL'])
coin = symbol.replace('USDT', '')
url = f"https://bybit.com/en/trade/spot/{coin}/USDT"
driver = setup_driver()
driver.get(url)
uri = "wss://stream.bybit.com/v5/trade"
async with websockets.connect(uri) as ws:
await authenticate(ws, api_key, api_secret)
log(f"Browser opened. Waiting for listing time ({listing_time_str})...")
while True:
current_time = datetime.now()
if (listing_time - current_time).total_seconds() <= start_interval:
break
await asyncio.sleep(0)
log("Starting rapid buy orders...")
initial_ask, initial_bid, initial_mark = await scrape_prices_task(driver)
log(f"Initial prices - Ask: {initial_ask}, Bid: {initial_bid}, Mark: {initial_mark}", color='green')
if initial_ask >= 5 * initial_mark:
log("Ask price is 5 times or more than mark price. Stopping script.", color='red')
driver.quit()
return
buy_price = adjust_price(np.float64(initial_ask) * np.float64(1.03))
raw_qty = np.float64(usdt_amount) * np.float64(0.997) / buy_price
qty = adjust_qty(buy_price, raw_qty)
log(f"Raw quantity calculation: {raw_qty}")
log(f"Adjusted quantity: {qty}")
if qty <= 0:
log("Calculated quantity is 0 or negative. Cannot place order.", color='red')
driver.quit()
return
log(f"Placing buy order - Price: {buy_price}, Quantity: {qty}")
order_start_time = time.time_ns()
buy_task = asyncio.create_task(place_buy_order_task(ws, symbol, qty, buy_price))
await buy_task
order_end_time = time.time_ns()
log(f"Buy order placed. Time taken: {(order_end_time - order_start_time) / 1000:.3f} seconds", color='green')
log("Waiting for sell condition...")
while True:
current_ask, current_bid, current_mark = await scrape_prices_task(driver)
log(f"Current prices - Ask: {current_ask}, Bid: {current_bid}, Mark: {current_mark}")
if current_bid > buy_price:
sell_price = adjust_price(np.float64(current_bid) * np.float64(0.997)) # 0.3% discount
sell_qty = adjust_qty(sell_price, qty)
log(f"Placing sell order (profit) - Price: {sell_price}, Quantity: {sell_qty}")
sell_task = asyncio.create_task(place_sell_order_task(ws, symbol, sell_qty, sell_price))
await sell_task
log("Sell order placed (profit)", color='green')
break
elif current_bid < buy_price * 1.0: # 上場価格から10倍まで許容
sell_price = adjust_price(np.float64(current_bid))
sell_qty = adjust_qty(sell_price, qty)
log(f"Placing sell order (stop loss) - Price: {sell_price}, Quantity: {sell_qty}")
sell_task = asyncio.create_task(place_sell_order_task(ws, symbol, sell_qty, sell_price))
await sell_task
log("Sell order placed (stop loss)", color='red')
break
await asyncio.sleep(0.001)
driver.quit()
if __name__ == "__main__":
asyncio.run(main())adjust_values.py (脳筋実装)
import numpy as np
def adjust_price(price):
conditions = [
(0.000001 <= price) & (price < 0.00001),
(0.00001 <= price) & (price < 0.0001),
(0.0001 <= price) & (price < 0.001),
(0.001 <= price) & (price < 0.01),
(0.01 <= price) & (price < 0.1),
(0.1 <= price) & (price < 1),
(1 <= price) & (price < 10),
(10 <= price) & (price < 100),
(100 <= price) & (price < 1000),
(1000 <= price) & (price < 10000),
(10000 <= price) & (price < 100000),
(price >= 100000)
]
decimals = [7, 6, 5, 4, 4, 4, 3, 1, 1, 0, 0, 0]
return np.round(price, decimals[np.argmax(conditions)])
def adjust_qty(price, qty):
conditions = [
(0.000001 <= price) & (price < 0.001),
(0.01 <= price) & (price < 1),
(10 <= price) & (price < 100),
(1000 <= price) & (price < 10000)
]
decimals = [0, 2, 3, 4]
qty = np.where(conditions[0], np.floor(qty), qty)
qty = np.where(conditions[1], np.floor(qty * 100) / 100, qty)
qty = np.where(conditions[2], np.floor(qty * 1000) / 1000, qty)
qty = np.where(conditions[3], np.floor(qty * 10000) / 10000, qty)
return qty補足
実行スピードは速いはずなので、選定ペアとorder bookを自分で揃えるなどしたら、ワンチャン勝てるかもです。
例えば、
エアドロトークンがあり、Bybitに上場する。
エアドロトークンを先にclaimして、アカウントに着金させる。
スクリプト改良を行い、ask板を自分で揃え、それを自分で買う。
それと同時にfillさせて高値売却。
これができたら結構強い🦆です。
終章: botは儲からない
とにかく儲からないです。
しんどいです。
以上です。
あとがき
今年の誕生日ですが、とても嬉しかったです😸
renくんに初めてお会いして、今年初うなぎをご馳走させていただきました。
そして、DAO TOKYOに行きました
ただ、あんまピンと来なかったので、renくんのお知り合いがいるコワーキングへ移動。
そこで、ぶしおbotさんやwasabiさん等お方にお会いできました。
そして、夜はなおさんと焼肉をご馳走させていただきましたっ
周りには感謝しきれないほど、恵まれていて嬉しい限りですっ！
16の年も頑張っていこうと思います！！