Photo by
yharuki
pybottersとasyncioでinago flyer!
前回の記事
https://note.com/clever_dahlia308/n/n2fdb1ca0c62e
まだまだpybottersはこんなもんじゃないっ・・・！もっと楽しんでほしいっ・・・！
という気持ちがあったので今度はinago flyerっぽいものを書いてみました。
pybotters・asyncioのファンとして、それらの可能性・面白さを感じてもらえるような書き方を精一杯してみました。楽しんでいただけたら幸いです。
ただ個人的にはまだまだこんなもんじゃないと思っています。それくらいpybotters/asyncioは奥が深いっ・・・！
ロジック
イナゴ検知したら成行で入って成行で出ていく
イナゴ検知は２段階
（I）短期足（数秒）の BUY / SELL volume（log）が閾値を超えたら
（II）（I）を検知後、イナゴ方向に値動きがあったら
（I）と（II）を満たしたタイミングで新規成行注文
決済注文はトレイルストップ
エントリー値±トレイルマージンを初期ストップ値とする
長期足（１分）が確定するたびに最終CLOSE値±トレイルマージンでストップ値を上書き
こういうのを取りたい気持ち
結構雑なイナゴスキャです。自分で言うのもアレですがこのbotで勝てると思ってません。フォワードテストとかバックテストとかしてません。これで勝てるか全く分かりません。むしろちょっと回した感じ負けてます。もしこのコードあるいはこのコードを改良したもので勝てたら流石にDMください。
コード
このコード一つで動きます。
使い方は
- 依存ライブラリをインストール
- api keyの入ったjsonファイルを指定してmain.pyを走らせる
です。jsonファイルのフォーマットについてはhttps://pybotters.readthedocs.io/ja/latest/Advanced-Usage.html#apisを参考にしてください。
pip install pybotters loguru pandas numpypython main.py --api_key YOUR_API_KEY_JSON_FILE_PATHfrom __future__ import annotations
from typing import TYPE_CHECKING, Callable, Generator, Union
if TYPE_CHECKING:
from pybotters.store import DataStoreManager
import asyncio
from collections import deque
import loguru
import numpy as np
import pandas as pd
import pybotters
import pybotters.models.bitflyer
def check_side(side):
assert side in ["BUY", "SELL"]
# 注文ヘルパー
async def market_order(client, symbol, side, size):
res = await client.post(
"/v1/me/sendchildorder",
data={
"product_code": symbol,
"side": side,
"size": f"{size:.8f}",
"child_order_type": "MARKET",
},
)
data = await res.json()
if res.status != 200:
raise RuntimeError(f"Invalid request: {data}")
else:
return data["child_order_acceptance_id"]
async def watch_execution(execution: pybotters.models.bitflyer.ChildOrders, order_id):
with execution.watch() as stream:
async for msg in stream:
if (
msg.operation == "insert"
and msg.data["child_order_acceptance_id"] == order_id
and msg.data["event_type"] == "EXECUTION"
):
return msg.data
class AbstractPriceTrailer:
"""面倒になったのでここは箱だけ..."""
class BarBasedPriceTrailer(AbstractPriceTrailer):
"""足が確定するたびに最後の足のclose ± marginのところにstopを置き直す（ストップ値が悪化（？）する場合は更新なし）"""
def __init__(
self,
price: int,
side: str,
bar: AbstractTimeBar,
margin: int,
logger=loguru.logger,
):
check_side(side)
self._price = price
self._side = side
self._bar = bar
self._margin = margin
self._logger = logger
self._task = asyncio.create_task(self.auto_trail())
def __del__(self):
# gcの機嫌次第でいつ呼ばれるか（はたまた本当に呼ばれるのか）わからないが一応オブジェクト破棄時に
# Taskをキャンセルをするようにする。
self.cancel()
async def auto_trail(self):
while True:
# 最新足確定まで待機
bar = await self._bar.get_bar_at_settled()
last_close = bar[-1, 3]
if self._side == "BUY":
new_price = last_close - self._margin
if new_price > self._price:
self._price = new_price
else:
new_price = last_close + self._margin
if new_price < self._price:
self._price = new_price
self._logger.debug(
f"[TRAIL] {self._price:.0f} (last_close={last_close:.0f})"
)
def cancel(self, msg=None):
return self._task.cancel(msg)
@property
def price(self):
return self._price
class StreamArray:
"""queue-likeなnumpy array"""
def __init__(self, shape, array=None):
self._a = np.full(shape, np.nan)
if array is not None:
self.init(array)
def __repr__(self):
return self._a.__repr__()
def __getitem__(self, *args):
return self._a.__getitem__(*args)
def __len__(self):
return self._a.__len__()
def append(self, x):
def shift(arr: np.ndarray, num: int, fill_value=0) -> np.ndarray:
result = np.empty_like(arr)
if num > 0:
result[:num] = fill_value
result[num:] = arr[:-num]
elif num < 0:
result[num:] = fill_value
result[:num] = arr[-num:]
else:
result[:] = arr
return result
self._a = shift(self._a, -1)
self._a[-1] = x
def init(self, array):
if self._a.shape[0] <= array.shape[0]:
array = array[-self._a.shape[0] :]
self._a = array
else:
self._a[-array.shape[0] :] = array
@property
def a(self):
return self._a
class AbstractTimeBar:
"""Time-bar用の抽象クラス。
storeには約定情報を取得する``DataStore``を与える（例: bitFlyer -> ``pybotters.models.bitflyer.Executions``)
"""
def __init__(
self,
store: "pybotters.store.DataStore",
unit_seconds: int,
maxlen: int = 9999,
callbacks: list[Callable[[AbstractTimeBar], None]] = (),
):
self._store = store
self._seconds = unit_seconds
self._rule = f"{unit_seconds}S"
self._bar = StreamArray((maxlen, 7))
self._cur_bar = np.zeros(7)
self._timestamp = deque(maxlen=maxlen)
# callback
self._callbacks = callbacks or []
self._d = {}
# 確定足取得用
self._queue = asyncio.Queue(1)
self._task = asyncio.create_task(self.auto_update())
async def init(self, executions: list[dict]):
[self.update(e) for e in executions]
def new_bar(
self,
price: Union[int, float],
volume: float,
side: str,
timestamp: pd.Timestamp = None,
):
"""最新足の確定"""
check_side(side)
if side == "BUY":
buy_volume, sell_volume = volume, 0
else:
buy_volume, sell_volume = 0, volume
self._bar.append(self._cur_bar)
self._cur_bar = np.array(
[price, price, price, price, volume, buy_volume, sell_volume]
)
if timestamp is None:
timestamp = pd.Timestamp.utcnow().floor(self._rule)
self._timestamp.append(timestamp)
# 確定した最新barを格納
try:
self._queue.put_nowait(self._bar)
except asyncio.QueueFull:
self._queue.get_nowait()
self._queue.put_nowait(self._bar)
def update_cur_bar(self, price: Union[int, float], volume: float, side: str):
"""未確定足の更新"""
# high
self._cur_bar[1] = max(price, self._cur_bar[1])
# low
self._cur_bar[2] = min(price, self._cur_bar[2])
# close
self._cur_bar[3] = price
# volume
self._cur_bar[4] += volume
if side == "BUY":
self._cur_bar[5] += volume
elif side == "SELL":
self._cur_bar[6] += volume
else:
raise RuntimeError(f"Unsupported: {side}")
def update(self, e: pybotters.store.StoreChange):
"""要オーバーライド"""
raise NotImplementedError
async def auto_update(self):
"""約定情報の受信タスク"""
async for e in self.execution_stream():
self.update(e)
# 約定情報を受信するたびにcallback
for cb in self._callbacks:
self.d.update(cb(self))
async def execution_stream(self):
"""pybottersのwatchを使って約定情報を随時generateする。"""
with self._store.watch() as stream:
async for msg in stream:
if msg.operation == "insert":
yield msg.data
async def get_bar_at_settled(self) -> np.ndarray:
"""``await bar.get_bar_at_settled()``で最新足確定時に取得できる。"""
return await self._queue.get()
@property
def store(self):
return self._store
@property
def bar(self):
return self._bar
@property
def timestamp(self):
return self._timestamp
@property
def cur_bar(self):
return self._cur_bar
@property
def open(self):
return self._bar[:, 0]
@property
def o(self):
return self._bar[-1, 0]
@property
def high(self):
return self._bar[:, 1]
@property
def h(self):
return self._bar[-1, 1]
@property
def low(self):
return self._bar[:, 2]
@property
def l(self):
return self._bar[-1, 2]
@property
def close(self):
return self._bar[:, 3]
@property
def c(self):
return self._bar[-1, 3]
@property
def volume(self):
return self._bar[:, 4]
@property
def v(self):
return self._bar[-1, 4]
@property
def buy_volume(self):
return self._bar[:, 5]
@property
def bv(self):
return self._bar[-1, 5]
@property
def sell_volume(self):
return self._bar[:, 6]
@property
def sv(self):
return self._bar[-1, 6]
@property
def d(self):
return self._d
class BitflyerTimeBar(AbstractTimeBar):
"""bitFlyer用のTimeBar実装"""
def __init__(
self,
store: pybotters.models.bitflyer.Executions,
unit_seconds: int,
maxlen: int = 9999,
callbacks: list[Callable[[BitflyerTimeBar], None]] = (),
):
super(BitflyerTimeBar, self).__init__(store, unit_seconds, maxlen, callbacks)
def update(self, e: dict):
cur_ts = pd.to_datetime(e["exec_date"]).floor(self._rule)
if len(self._timestamp) == 0:
last_ts = None
else:
last_ts = self._timestamp[-1].floor(f"{self._rule}")
if last_ts is None or last_ts != cur_ts:
self.new_bar(e["price"], e["size"], e["side"], cur_ts)
else:
self.update_cur_bar(e["price"], e["size"], e["side"])
class AbstractInagoBot:
"""イナゴBOT用の抽象クラス（仮）"""
def __init__(
self,
client: pybotters.Client,
is_inago_start_fn: Callable[[AbstractInagoBot], tuple[bool, str]] = None,
is_inago_end_fn: Callable[[AbstractInagoBot], bool] = None,
off_loop_interval: float = 0.5,
logger=loguru.logger,
):
self._client = client
self._is_inago_start_fn = is_inago_start_fn
self._is_inago_end_fn = is_inago_end_fn
self._off_loop_interval = off_loop_interval
self._logger = logger
self._side = None
# 要オーバーライド
async def inago_stream(self) -> Generator[dict]:
""" "イナゴ"（主に約定情報になると思う）を流す"""
raise NotImplementedError
async def on_inago(self, inago):
"""各イナゴに対する処理"""
pass
async def is_inago_start(self) -> tuple[bool, str]:
"""イナゴ到来判定"""
if self._is_inago_start_fn is None:
raise NotImplementedError
if asyncio.iscoroutinefunction(self._is_inago_start_fn):
return await self._is_inago_start_fn(self)
else:
return self._is_inago_start_fn(self)
async def is_inago_end(self) -> bool:
"""イナゴ終了判定"""
if self._is_inago_end_fn is None:
raise NotImplementedError
if asyncio.iscoroutinefunction(self._is_inago_end_fn):
return await self._is_inago_end_fn(self)
else:
return self._is_inago_end_fn(self)
async def loop(self):
"""イナゴBOTのメインループ
- off_loop(): イナゴ待機ループ（検知）
- on_loop(): イナゴ処理ループ（トレード）
この二つを繰り返す。
- *_begin(), *_end(): 各関数の前後で呼ばれるhook
"""
while True:
self._logger.debug("BEGIN LOOP")
await self.on_loop_begin()
self._logger.debug("BEGIN OFF LOOP")
await self.off_loop()
self._logger.debug(f"END OFF LOOP")
self._logger.debug(f"BEGIN ON LOOP: {self._side}")
await self.on_loop()
self._logger.debug("END ON LOOP")
await self.on_loop_end()
self._logger.debug("END LOOP")
async def off_loop(self):
"""イナゴオフ時のループ＝イナゴ検知"""
await self.on_off_loop_begin()
while True:
is_start, side = await self.is_inago_start()
if is_start:
self._side = side
break
await asyncio.sleep(self._off_loop_interval)
await self.on_off_loop_end()
async def on_loop(self):
"""イナゴオン時のループ＝トレード"""
await self.on_on_loop_begin()
async for inago in self.inago_stream():
await self.on_inago_begin(inago)
await self.on_inago(inago)
is_end = await self.is_inago_end()
if is_end:
self._side = None
break
await self.on_inago_end(inago)
await self.on_on_loop_end()
async def on_loop_begin(self):
pass
async def on_loop_end(self):
pass
async def on_off_loop_begin(self):
pass
async def on_off_loop_end(self):
pass
async def on_on_loop_begin(self):
pass
async def on_on_loop_end(self):
pass
async def on_inago_begin(self, inago):
pass
async def on_inago_end(self, inago):
pass
@property
def client(self):
return self._client
@property
def side(self):
return self._side
class BitflyerInagoBot(AbstractInagoBot):
"""今回のイナゴBOTのメイン実装。単純な成行IN/成行OUTのイナゴトレード。
- bar_l: イナゴ終了用の長時間足（１分）
- bar_s: イナゴ検知用の短時間足（数秒）
- lower_threshold, upper_threshold: イナゴ検知用の閾値
- entry_patience_seconds, entry_price_change: 騙し回避用のパラメター
- trail_margin: トレイル決済のパラメター
"""
def __init__(
self,
client: pybotters.Client,
store: pybotters.bitFlyerDataStore,
bar_l: BitflyerTimeBar,
bar_s: BitflyerTimeBar,
*,
lower_threshold: float,
upper_threshold: float,
entry_patience_seconds: int,
entry_price_change: int,
trail_margin: int,
symbol: str = "FX_BTC_JPY",
size: float = 0.01,
side: str = "BOTH",
**kwargs,
):
super(BitflyerInagoBot, self).__init__(client, **kwargs)
self._store = store
self._bar_l = bar_l
self._bar_s = bar_s
self._lower_threshold = lower_threshold
self._upper_threshold = upper_threshold
self._entry_patience_seconds = entry_patience_seconds
self._entry_price_change = entry_price_change
self._trail_margin = trail_margin
self._symbol = symbol
self._size = size
self._entry_side = side
self._entry_order_info = None
self._exit_order_info = None
self._asks, self._bids = None, None
asyncio.create_task(self.auto_ask_bid_update())
async def auto_ask_bid_update(self):
"""板情報の自動更新タスク"""
while True:
await self._store.board.wait()
self._asks, self._bids = self._store.board.sorted().values()
async def on_loop_end(self):
"""トレードログ"""
assert self._entry_order_info is not None
assert self._exit_order_info is not None
pnl = self._exit_order_info["price"] - self._entry_order_info["price"]
if self._entry_order_info["side"] == "SELL":
pnl *= -1
pnl *= self._entry_order_info["size"]
self._logger.debug(f"[LOOP FINISH] pnl={pnl}")
self._entry_order_info = None
self._exit_order_info = None
async def on_inago(self, inago):
self._logger.debug(f"[ON INAGO] {self._bar_s.d} {inago}")
async def inago_stream(self):
with self._store.executions.watch() as stream:
async for msg in stream:
yield msg.data
async def is_inago_start(self) -> tuple[bool, str]:
"""イナゴ検知ロジック。
２段階で検知する。
（１）閾値判定：短期足（秒足）でのボリュームが閾値をクリア
（２）経過判定：n秒（``self._entry_patience_seconds``）後にイナゴ方向に値動き（``self._entry_price_change``）があるか否か
"""
d = self._bar_s.d
if len(d) == 0:
self._logger.warning(f"[INFORMATION IS EMPTY] {d}")
return False, None
self._logger.debug(f"[WAITING INAGO] {d}")
async def _primary_check():
"""閾値判定"""
if (
self._entry_side in ("BUY", "BOTH")
and d["sv_log"]
< self._lower_threshold
< d["bv_log"]
< self._upper_threshold
):
self._logger.debug("[PRIMARY CHECK] YES BUY")
return "BUY"
elif (
self._entry_side in ("SELL", "BOTH")
and d["bv_log"]
< self._lower_threshold
< d["sv_log"]
< self._upper_threshold
):
self._logger.debug("[PRIMARY CHECK] YES SELL")
return "SELL"
else:
return None
async def _secondary_check(s):
"""時間経過判定"""
# 仲値を値動きの参照値に使う
mark_price_start = int(self.mid)
self._logger.debug(f"[SECONDARY CHECK] mark_price={mark_price_start}")
while True:
mark_price = int(self.mid)
price_change = mark_price - mark_price_start
if s == "SELL":
price_change *= -1
print(
f"\r\033[31m>>> [SECONDARY CHECK] {mark_price_start}/{mark_price}/{price_change:+.0f}\033[0m",
end="",
)
if price_change > self._entry_price_change:
# イナゴ方向への値動きがあった
break
await asyncio.sleep(0.1)
# 閾値判定
side = await _primary_check()
if side:
try:
# 経過判定
await asyncio.wait_for(
_secondary_check(side), timeout=self._entry_patience_seconds
)
# carriage return調整してるだけ
print()
# イナゴ検知
return True, side
except asyncio.TimeoutError as e:
# carriage return調整してるだけ
print()
# 指定秒数以内に値動きがみられなかったのでスルー
self._logger.debug(f"[CANCEL] mark_price={self.mid}")
return False, None
else:
return False, None
async def on_loop(self):
"""ロジック
- 色々とhookを用意したものの、ロジック的に当てはめられなかったのでon_loop丸ごとオーバーライドしている（爆）
- 「約定情報を参照して決済注文を出す」といったロジックであれば以下のように分けて実装できると思う（元々はそう考えていた）
- ``on_on_loop_begin``で新規注文
- ``is_inago_endo``で終了判定
- ``on_on_loop_end``で決済注文
"""
# 新規注文
order_id = await market_order(self.client, self._symbol, self.side, self._size)
self._entry_order_info = await watch_execution(
self.store.childorderevents, order_id
)
self._logger.debug(f"[ENTRY ORDER] {self._entry_order_info}")
entry_price = self._entry_order_info["price"]
# 建値±``_trail_margin`` を初期ストップ値としてトレイルスタート
if self.side == "BUY":
stop_price = entry_price - self._trail_margin
else:
stop_price = entry_price + self._trail_margin
trailer = BarBasedPriceTrailer(
stop_price, self.side, self._bar_l, self._trail_margin, self._logger
)
self._logger.debug(f"[TRAIL START] entry={entry_price} stop={stop_price}")
while True:
await asyncio.sleep(0.1)
# 最良気配値がストップ値を割ったら決済
# ストップ値はtrailerが長期足（e.g., １分足）の確定毎に更新
if self.side == "BUY":
mark_price = self.best_bid
pnl = mark_price - entry_price
if mark_price <= trailer.price:
break
else:
mark_price = self.best_ask
pnl = entry_price - mark_price
if mark_price >= trailer.price:
break
print(
f"\r\033[31m>>> [TRAILING] entry={entry_price:.0f} stop={trailer.price:.0f} mark={mark_price:.0f} pnl={pnl:+.0f}\033[0m",
end="",
)
# trailタスクが回り続けてしまうので明示的にキャンセルする
# asyncio.Taskはオブジェクトがスコープを外れて破壊されてもキャンセルされない
trailer.cancel()
side = "SELL" if self._entry_order_info["side"] == "BUY" else "BUY"
order_id = await market_order(self.client, self._symbol, side, self._size)
self._exit_order_info = await watch_execution(
self.store.childorderevents, order_id
)
self._logger.debug(f"[EXIT ORDER] {self._exit_order_info}")
@property
def store(self):
return self._store
@property
def best_ask(self):
if self._asks is None:
return -1
return self._asks[0]["price"]
@property
def best_bid(self):
if self._bids is None:
return -1
return self._bids[0]["price"]
@property
def mid(self):
return (self.best_ask + self.best_bid) / 2
async def main(args):
logger = loguru.logger
logger.add("log.txt", retention=3, rotation="10MB")
async with pybotters.Client(
args.api_key, base_url="https://api.bitflyer.com"
) as client:
store = pybotters.bitFlyerDataStore()
# time bar
def log_volume(bar: AbstractTimeBar):
"""最新足のbuy/sell volumeのログを計算するcallback"""
d = dict()
d["bv_log"] = np.log1p(bar.bv)
d["sv_log"] = np.log1p(bar.sv)
return d
bar_l = BitflyerTimeBar(
unit_seconds=args.bar_unit_seconds_long,
store=store.executions,
maxlen=args.bar_maxlen,
callbacks=[log_volume],
)
bar_s = BitflyerTimeBar(
unit_seconds=args.bar_unit_seconds_short,
store=store.executions,
maxlen=args.bar_maxlen,
callbacks=[log_volume],
)
# time barを約定履歴で初期化
resp = await client.get(
"/v1/getexecutions", params={"producet_code": args.symbol, "count": 500}
)
data = await resp.json()
await bar_l.init(data[::-1])
await bar_s.init(data[::-1])
# web socketに接続
await client.ws_connect(
"wss://ws.lightstream.bitflyer.com/json-rpc",
send_json=[
{
"method": "subscribe",
"params": {"channel": f"lightning_board_snapshot_{args.symbol}"},
"id": 1,
},
{
"method": "subscribe",
"params": {"channel": f"lightning_board_{args.symbol}"},
"id": 2,
},
{
"method": "subscribe",
"params": {"channel": f"lightning_executions_{args.symbol}"},
"id": 3,
},
{
"method": "subscribe",
"params": {"channel": "child_order_events"},
"id": 4,
},
],
hdlr_json=store.onmessage,
)
while not all([len(w) for w in [store.board, store.executions]]):
logger.debug("[WAITING SOCKET RESPONSE]")
await store.wait()
await BitflyerInagoBot(
client,
store,
bar_l,
bar_s,
lower_threshold=args.lower_threshold,
upper_threshold=args.upper_threshold,
entry_patience_seconds=args.entry_patience_seconds,
entry_price_change=args.entry_price_change,
trail_margin=args.trail_margin,
symbol=args.symbol,
size=args.size,
side=args.side,
logger=logger,
).loop()
if __name__ == "__main__":
from argparse import ArgumentParser
parser = ArgumentParser()
parser.add_argument("--api_key", required=True, help="apiキーが入ったJSONファイル")
parser.add_argument("--symbol", default="FX_BTC_JPY", help="取引通過")
parser.add_argument(
"--side", default="BOTH", choices=["BUY", "SELL", "BOTH"], help="エントリーサイド"
)
parser.add_argument("--size", default=0.01, type=float, help="注文サイズ")
parser.add_argument("--bar_unit_seconds_long", default=60, type=int, help="長期足")
parser.add_argument("--bar_unit_seconds_short", default=5, type=int, help="短期足")
parser.add_argument("--bar_maxlen", default=999, type=int, help="足の最大履歴")
parser.add_argument(
"--lower_threshold",
default=1,
type=float,
help="短期足のボリューム（log）がこの閾値以上であればエントリー待機",
)
parser.add_argument(
"--upper_threshold",
default=float("inf"),
type=float,
help="短期足のボリューム（log）がこの閾値以下であればエントリー待機",
)
parser.add_argument(
"--entry_patience_seconds", default=3, type=int, help="閾値クリア後の経過観察時間"
)
parser.add_argument(
"--entry_price_change", default=500, type=int, help="経過観察後の価格変動がこの閾値以上であればエントリー"
)
parser.add_argument("--trail_margin", default=2000, type=int, help="トレイル値幅")
args = parser.parse_args()
try:
asyncio.run(main(args))
except KeyboardInterrupt:
pass
解説などしたかったんですが、また面倒になってきたのでコード・コメント嫁でお願いしますすみません・・・。
注意
コードライセンス：MIT
ご自身の責任においてご使用ください。
リンク
pybotters
https://github.com/MtkN1/pybotters
asyncio
https://docs.python.org/ja/3/library/asyncio.html
git