停止しない自動売買ボットの作り方。python 3

自動売買ボットはいろいろな原因(バグ、API変更、サーバーリソースなど)で良く停止しますが、停止したまま気づかないと、大きく損失したりします。

python3で停止しないボットの作り方です。

方針

ヘルスチェックを使います。一定時間ごとにボットの状態が正常であることを報告し、一定時間報告がなければ再起動させます。正常かどうかは、データが最新か、最後にループが回った時刻が最近か、などで判断します。

python3 実装コード

import threading
import sys
import os
import traceback
import time
import psutil

class PanicManager:
   def __init__(self, logger=None):
       self.monitors = {}
       self.logger = logger
       self.lock = threading.Lock()
       self.thread = threading.Thread(target=self.run)
       self.thread.start()

   def register(self, tag=None, start_time=None, interval=None):
       self.logger.debug('panic_manager register tag {} start_time {} sec interval {} sec'.format(tag, start_time, interval))
       with self.lock:
           self.monitors[tag] = {
               'start_at': time.time(),
               'ping_at': None,
               'start_time': start_time,
               'interval': interval
           }

   def ping(self, tag=None):
       with self.lock:
           self.monitors[tag]['ping_at'] = time.time()

   def panic(self):
       restart_program(logger=self.logger)

   def run(self):
       while True:
           self.logger.debug('panic_manager loop')
           now = time.time()
           with self.lock:
               for tag in self.monitors:
                   monitor = self.monitors[tag]
                   if monitor['ping_at']:
                       if now - monitor['ping_at'] > monitor['interval']:
                           self.logger.error('{} ping delayed. restarting'.format(tag))
                           self.panic()
                   else:
                       if now - monitor['start_at'] > monitor['start_time']:
                           self.logger.error('{} start delayed. restarting'.format(tag))
                           self.panic()
           time.sleep(5)

# https://stackoverflow.com/questions/11329917/restart-python-script-from-within-itself
def restart_program(logger=None):
   try:
       p = psutil.Process(os.getpid())
       for f in p.open_files() + p.connections():
           os.close(f.fd)
   except Exception as e:
       logger.error('exception ' + traceback.format_exc())
   python = sys.executable
   os.execl(python, python, *sys.argv)

使い方

# 初期化
# start_time: 最初の報告までの猶予(秒)
# interval: 2回目以降報告までの猶予(秒)
# tag: 監視対象を表す任意のtag

self.panic_manager = PanicManager(logger=logger)
self.panic_manager.register(tag='trader', start_time=5 * 60, interval=60)

# 定期的に生存報告
# ok: 正常かどうか。ボットごとにカスタマイズ

if ok:
    self.panic_manager.ping('trader')

注意

restart_programはなんかハンドルが閉じていないっぽいです。数ヶ月動かして、再起動を何度も繰り返すと、linuxのハンドル制限にひっかかり、再起動できなくなります。

ライセンス

記事中ソースコードのライセンスはCC0