仮想通貨のbotを作りたいが、みんなどう書いてるの？
こんにちは
仮想通貨botter Advent Calendar 2022 の11日目の記事のつもりで書いています。
これからBot作りたいけど、みんなはどう設計しているのだろうという人向け。
loop頻度が1分以上のMLbotを前提としてますが、高頻度でも基本は変わらないと考えています。
クラスの責務の範囲や、起動周りの機構などに悩んでいたら参考になれば、、
結論
メインループ
下記のノリで各クラスの責務を分離をすると苦しくなりにくいです。
while True:
# 新規データ取得
data = stream.get_marcket_data(target_start_unix)
# データ追加
data_manager.add(data)
# データ => 指標への変換
metrics = model.translate(data_manager.tail())
# ステータス取得
status = status.get(symbol)
# アクション生成
actions = trader.cal(metrics, status)
# 執行
result = executor.request(actions)
登場人物
※ 関数などは全部書いても仕方がないので省略してます。
ExchangeClient
取引所のAPIを叩くクラス。
こいつを後続のクラスに差して使います。
実行速度を気にしないのであればnetoutなどの共通化はこのクラスでやって置くのが良いです。
またbacktestという架空の取引所を用意しましょう。
Stream
取引所から取れる情報をまとめて、bot側が扱えるように整理する。
（例えばOIやLongShortRatioなどを合わせたり）
exchange_client を差し込んで初期化します。
get_marcket_data はデータの準備が完了次第値を返すので（15分足が埋まるなど）
メインループ の周期は Stream に依存しています。
DataManager
streamから取れるdataを保管して置く場所。
次回起動がサクッとできるようにピッケルなどに状態を保存すると便利。
tailでケツからいくつかとって使う。
Model
datamanagerケツからとった値を指標に変換するモデル。
どんなModelを使ったかがアクションの生成に関わってくるのでTraderの中に突っ込んでしまうのもアリな気がする。
Status
ポジション等を管理するクラス、
15分インターバルのbotではポジションの自炊とかは必要なく、毎度APIを叩いて確認すればいいと思います。
candleなどのマーケット情報もstatusじゃないかという気持ちも分かりつつ、自身のアクションによってのみ発生する情報のことをstatusとしています。
Trader
指標とポジションなどから次の行動を作るクラス。
actionsの内容は、{orders: [], cancels: []} みたいなノリ。
Executor
actionを受け取って取引所にオーダーを出す。
ポイントはどこ
本番でもバックテストでも同じロジックを通せると安心です。
そのために下記の2点はあると良いかなと考えています。
取引所は共通化してbacktest用の取引所を作る
これをするとbotを起動するときに使う取引所を差し替えるだけでバックテストができるので、バックテストで走ったロジックをそのまま本番で使えるようになります。
下記のようにloopで用いる取引所を初期化して、作ったclientを必要なクラスに差し込んであげて初期化すると同じループで複数の取引所を扱えます。
新しい取引所に対応したいときはラッパーを作ってExchangeClientに対応させれば良いです。
# 取引所を決定
exchange_client = ExchangeClient(enum.GMO)
# 各クラスを作成
stream = Stream(exchange_client)
data_manager = DataManager()
model = Model()
status = Status(exchange_client)
trader = Trader()
executor = Executor(exchange_client)
データの取得を起点にして駆動する
loop側でインターバルを管理するのではなく、Streamの都合に合わせると、結果的にStreamに指定された取引所の都合に合わせてloopのインターバルが変わります。
backtestで起動したのに実時間で動いてしまうのでloopに一時的にコメントアウトを入れる、などを避けられるのでおすすめです。
重要な考え方として、botをdataを放り込んだら動く関数と捉えると、dataの取得元がcsvでもscoketでもapiをポーリングした結果でもbot側としては気にする必要がなくなります。
# APIから取るとき
while True:
data = get_data_from_market()
bot.act(data)
# 保存したCSVを利用してバックテストするとき
datas = load_from_csv()
for data in datas:
bot.act(data)
しかし、loopの記述を変えるのも面倒なので、Streamに要求して、準備ができていれば取れる、という形をとって隠蔽することで、
while True:
data = stream.get_marcket_data()
bot.act(data)
のようにすることでメインループをいじらずに済むようになります。
以上のことに気をつければ、突然取引所が増えても半日もあればロジックを流用できるかと思います。
以上です。
質問や指摘は気軽にお願いします🙏
付録
Backtestの取引所を作る
backtestの取引所を作る上でのポイントは3つです。
- 初期化時にmarket_dataをcsvから取り込んでcandlesなどであたかも取れているように振る舞う。
- 架空のcreate_orderの際にorder_idを発行することで、普通の取引所と同様にorder_idを指定してcancelができるようにする。
- この時
map[order_id] = orderとして保存map型で保存することで削除時にorderを探す際に、O(logN)で見つかるのでバックテストが早くなります。
- この時
- 新しいcandleのdataを排出する際に、現在の指値orderが約定したかを確認する関数を作成する。
- 下記の例だとapply_candleがそれにあたります。現在は外から叩いていますが、candlesが叩かれたときに勝手に叩いてもいいかも。
具体的には下記のようなものを利用しています。
class BackTestFiatBase(Client):
def __init__(self, api_key: str, sec_key: str) -> None:
self.long_size = 0
self.short_size = 0
self.cash = 100000.0
self.order_id_sequence = 100000
self.maker_fee = 0.0001
self.ltp = 0
self.orders = {}
self.profit_list = []
pass
def create_order(
self,
symbol: exchange.Symbol,
side: exchange.Side,
order_type: exchange.OrderType,
size: float,
price: Optional[float] = None,
stop_loss: Optional[float] = None,
close_pos: bool = False,
) -> Tuple[Optional[str], Optional[str]]:
self.order_id_sequence += 1
self.orders[str(self.order_id_sequence)] = {
"side": side,
"order_type": order_type,
"size": size,
"price": price,
}
return str(self.order_id_sequence), None
def cancel_order(self, symbol: exchange.Symbol, order_id: str) -> Optional[str]:
self.orders.pop(order_id, None)
self.orders = {}
return None
def _cancel_order_with_id(
self, symbol: exchange.Symbol, order_id: str
) -> Optional[str]:
self.orders.pop(order_id, None)
return None
def order_list(
self, symbol: exchange.Symbol
) -> Tuple[Optional[List[exchange.Order]], Optional[str]]:
ret = []
for key in self.orders.keys():
obj = self.orders[key]
ret.append(
exchange.Order(str(key), obj["price"], obj["size"], obj["side"], 0)
)
return ret, None
def position(
self, symbol: exchange.Symbol
) -> Tuple[Optional[exchange.Position], Optional[str]]:
return (
exchange.Position(
self.long_size, self.short_size, self.long_size - self.short_size
),
None,
)
def candles(
self,
symbol: exchange.Symbol,
interval_min: int,
from_unix: int,
limit: int = None,
) -> Tuple[Optional[List[exchange.OHLCV]], Optional[str]]:
if (
from_unix in self.candle_map
and from_unix + constants.interval_sec in self.candle_map
):
return [
self.candle_map[from_unix],
self.candle_map[from_unix + constants.interval_sec],
], None
return None, "no data"
def balance_of(self, coin: exchange.Coin) -> Tuple[Optional[float], Optional[str]]:
# 損益情報の蓄積
pos = self.long_size - self.short_size
return self.cash + self.ltp * pos, None
def enque_candles(self, candles: List[exchange.OHLCV]) -> Optional[str]:
self.candle_map = {}
for c in candles:
self.candle_map[c.start_at] = c
return None
def apply_candle(self, candle: exchange.OHLCV) -> Optional[str]:
self.ltp = candle.close
# 約定確認
delete_ids = []
for order_id in self.orders.keys():
order = self.orders[order_id]
if order["side"] == exchange.Side.BUY:
if order["price"] > candle.low:
self.long_size += order["size"]
self.cash -= order["size"] * order["price"] * (1 - self.maker_fee)
delete_ids.append(order_id)
else:
if order["price"] < candle.high:
self.short_size += order["size"]
self.cash += order["size"] * order["price"] * (1 + self.maker_fee)
delete_ids.append(order_id)
base_size = min(self.long_size, self.short_size)
self.long_size -= base_size
self.short_size -= base_size
# 約定したオーダーの削除
for order_id in delete_ids:
self._cancel_order_with_id(constants.symbol, order_id)
# 損益情報の蓄積
balance, _ = self.balance_of(exchange.Coin.BTC)
self.profit_list.append(
exchange.Profit(
candle.start_at + constants.interval_sec,
balance,
ltp=candle.close,
)
)
return None
def profits(self) -> List[exchange.Profit]:
return self.profit_list
AutoML良さそう
最近AI熱がまた盛り上がっているので、MLbotをもう少しちゃんとしようと思ってAutoMLを使いました。1回のtrainに3000円かかって仰天したのですが、modelを検討するのは職人芸なのである程度お金を払ってでも避けたい、、我こそは質の良いデータを持っている、という人は試してみるといいかも。
Discussion