前回の記事でBitMexの資産を監視して変化があったタイミングでLINEに通知するbotを開発しました。
実際に運用していると資産の変化だけでなく、どんな取引をして資産が変更したかを知りたいという要望がチラホラありました。
そこで、前回開発したbotに取引通知をLINEに通知する機能を実装したいと思います。
GitHubはこちら
https://github.com/cryptocat-miner/BitMexLineNotifyBot
目次
ロジック・仕様
- 取引履歴情報を定期的に監視(資産情報の監視と同じタイミング)
- 取引履歴情報に変化があったタイミングで更新された取引履歴を通知
- 成行注文などで取引が複数回になると通知が鬱陶しいので、同じ注文IDの取引は集計して1回で通知する
- Funding情報も通知する
ロジック自体はシンプルです。
同じIDの注文を集計して平均価格を算出するのが若干面倒でしたが。
コード
最初に取引履歴のリストを取得し、後はその先頭の取引情報に変化が無いかを監視しています。
変化があった場合はif trade["id"] == self.latestTradeHistory["id"]:で前回の先頭(最新)取引のIDまでの取引内容をチェックします。
import ccxtWrapper
import LineNotify
import io
class tradeHistory(ccxtWrapper.ccxtWrapper):
exchange = None
tradeHistory = None
latestTradeHistory = None
tradeHistoryList = list()
def __init__(self, instance:ccxtWrapper.ccxtWrapper):
self.exchange = instance
self.tradeHistory = self.exchange.fetchMyTrades(since=None,limit=50)
def checkTradeHistory(self):
tradeHistory = self.exchange.fetchMyTrades(since=None, limit=50)
if tradeHistory != None:
if self.latestTradeHistory != None:
if self.latestTradeHistory["id"] != tradeHistory[0]["id"]:
tradeHistoryItem = self.tradeHistoryItem()
for trade in tradeHistory:
print(trade)
print("\r\n\r\n")
if trade["id"] == self.latestTradeHistory["id"]:
if tradeHistoryItem.amount != 0:
tradeHistoryItem.averagePrice = tradeHistoryItem.averagePrice / tradeHistoryItem.amount
self.tradeHistoryList.append(tradeHistoryItem)
break
if tradeHistoryItem.orderId == trade["info"]["orderID"]:
tradeHistoryItem.amount += trade["amount"]
tradeHistoryItem.averagePrice = tradeHistoryItem.averagePrice + (trade["info"]["lastPx"] * trade["amount"])
tradeHistoryItem.feeCost += trade["fee"]["cost"]
else:
if tradeHistoryItem.orderId != None:
if tradeHistoryItem.amount != 0:
tradeHistoryItem.averagePrice = tradeHistoryItem.averagePrice / tradeHistoryItem.amount
self.tradeHistoryList.append(tradeHistoryItem)
tradeHistoryItem = self.tradeHistoryItem()
tradeHistoryItem.execType = trade["info"]["execType"]
tradeHistoryItem.orderId = trade["info"]["orderID"]
tradeHistoryItem.text = trade["info"]["text"]
tradeHistoryItem.amount = trade["amount"]
if tradeHistoryItem.text != "Liquidation":
tradeHistoryItem.averagePrice = trade["info"]["lastPx"] * trade["amount"]
else:
tradeHistoryItem.averagePrice = trade["info"]["stopPx"] * trade["amount"]
tradeHistoryItem.feeCost = trade["fee"]["cost"]
tradeHistoryItem.feeRate = trade["fee"]["rate"]
tradeHistoryItem.side = trade["side"]
tradeHistoryItem.orderType = trade["type"]
for trade in self.tradeHistoryList:
if trade.execType == "Trade":
self.sendTradeMessage(amount=trade.amount, price=trade.averagePrice, feeCost=trade.feeCost, side=trade.side, orderType=trade.orderType, text=trade.text)
elif trade.execType == "Funding":
self.sendFundingMessage(amount=trade.amount, price=trade.averagePrice, feeCost=trade.feeCost, feeRate=trade.feeRate)
self.tradeHistory = tradeHistory
self.latestTradeHistory = self.tradeHistory[0]
self.tradeHistoryList.clear()
def sendTradeMessage(self, amount:int, price:float, feeCost:float, side:str, orderType:str, text:str):
message = "取引情報を通知します\r\n"
if side == "buy":
message += "方向:ロング\r\n"
elif side == "sell":
message += "方向:ショート\r\n"
if orderType == "limit":
message += "タイプ:指値\r\n"
elif orderType == "market":
message += "タイプ:成行\r\n"
elif orderType == "stop":
message += "タイプ:ストップ成行\r\n"
elif orderType == "stoplimit" and text == "Liquidation":
message += "タイプ:精算\r\n"
elif orderType == "marketiftouched":
message += "タイプ:利食い成行\r\n"
elif orderType == "limitiftouched":
message += "タイプ:利食い指値\r\n"
else:
message += "タイプ:不明\r\n"
message += "取引量:" + str(int(amount)) + "\r\n"
message += "価格:" + "{:.1f}".format(price) + "\r\n"
if (feeCost >= 0):
message += "取引手数料:-" + "{:.4f}".format(abs(feeCost)) + " XBT"
else:
message += "取引手数料:+" + "{:.4f}".format(abs(feeCost)) + " XBT"
LineNotify.PostMessage(message)
def sendFundingMessage(self, amount:int, price:float, feeCost:float, feeRate:float):
message = "Funding情報を通知します\r\n"
message += "数量:" + str(amount) + "\r\n"
message += "執行価格:" + "{:.1f}".format(price) + "\r\n"
if (feeCost >= 0):
message += "Funding手数料:-" + "{:.4f}".format(abs(feeCost)) + " XBT(" + "{:.4f}".format(feeRate * 100) + "%)"
else:
message += "Funding手数料:+" + "{:.4f}".format(abs(feeCost)) + " XBT(" + "{:.4f}".format(feeRate * 100) + "%)"
LineNotify.PostMessage(message)
class tradeHistoryItem:
execType = None
orderType = None
orderId = None
text = None
id = None
side = None
takerOrMaker = None
amount = 0.0
averagePrice = 0.0
feeCost = 0.0
feeRate = 0.0
取引情報を取得するにはfetch_my_trade()を使用します。
引数のsinceを指定しない場合は最新/最古から、limitを指定しない場合はBitMexの場合はDefaultで100件取得します。
paramを指定しないとデフォルトでは古い方から取得されてindexの0から詰められるようなので、params={"reverse": True}を指定して最新情報から取得し、その後sortメソッドで"datetime"をキーにindexの0が最新になるようにソートしています。
def fetchMyTrades(self, since, limit):
trades = None
symbol = "BTC/USD"
print("fetchTarde()")
try:
#"reverse":Trueで新しい順に配列の最後から詰められる
trades = self.__bitmex.fetch_my_trades(symbol, since=since, limit=limit, params={"reverse": True})
#listの"datetime"をキーにreverseする事で配列の最初(index=0)からソートする
trades.sort(key=lambda x: x["datetime"],reverse=True)
except Exception as error:
print(error)
trades = None
return trades
メインプログラムに取引履歴を監視するメソッドを追加しました。
最初はorderManager.py内で資産に変更があったタイミングで取引履歴を取得していましたが、資産監視と取引履歴は別機能だし一緒にするとコードがグチャグチャになると思い(なった)、tradeHistory.pyとして分離させました。
#!/usr/bin/python3
import time
from datetime import datetime
import calendar
import ccxt
import ccxtWrapper
import math
import io
import orderManager
import tradeHistory
import LineNotify
# ------------------------------------------------------------------------------------------------
# メインプログラム
# 初期化
# bitmex = ccxtWrapper.ccxtWrapper(ccxtWrapper.ccxtWrapper.EXCHANGE_TYPE_TESTNET) #テストネット
bitmex = ccxtWrapper.ccxtWrapper(ccxtWrapper.ccxtWrapper.EXCHANGE_TYPE_MAINNET)
mTradeHistory = tradeHistory.tradeHistory(bitmex)
mOrderManager = orderManager.orderManager(bitmex)
openOrders = bitmex.fetchOpenOrders()
while True:
currentTime = datetime.utcnow() # 現在時刻(UTC)
mTradeHistory.checkTradeHistory()
mOrderManager.switchState()
time.sleep(6)
まとめ
既に前回の記事のコードを使用されている方も
- tradeHistory.pyをプロジェクトに追加
- ccxtWrapper.pyを上書き
- LineNotifyBot.pyを上書き
すれば取引履歴の通知機能を追加できます。
逆に資産監視か取引履歴どちらかの通知だけでよい、という方はLineNotifyBot.pyのmTradeHistory.checkTradeHistory()かmOrderManager.switchState()を任意でコメントアウトすればOKです。
