何が実現できるか

証拠金情報をチェックして、定期的に時給や分給を LINE/Discord に通知。こんな感じ。

言葉の定義
・トータル損益(時給, 分給)：Bot(or 証拠金監視)の開始から現在時刻までの情報
・日次損益(時給, 分給)：日付変更から現在時刻までの情報
(※13時に監視開始した場合、それまでの日次損益は無視して13時を基点としてます。24時になったらリセット)
・区間損益(時給, 分給)：監視周期レベルでの情報
(※監視周期10分で、区間損益が100円なら、時給600円、分給10円)

注意事項

書くの忘れてた。

・あくまでも「ある時点(起動時とか日付変更時とか)の証拠金」を基準にしているので、動かしてる最中に証拠金を抜いたり足したりすると計算が狂います
・評価損益は計算対象に含まれていません

コード

上の方にあるAPI KeyとかSecretを変更すればOK。botに組み込むなら、thread作って叩けば良いと思う。Line通知とか自前の実装がある場合は、各自で良い感じに統合してください。
例によって特に解説してないので、詳細はコード読んでください。自分のbotコードから適当に抜き出してきてるので、コードの内容が統一されていないかも、気持ち悪い部分は各自で直してください。

import ccxt
import requests
import concurrent.futures
from time import sleep
from datetime import datetime, timedelta
from logging import getLogger, INFO, StreamHandler
logger = getLogger(__name__)
handler = StreamHandler()
handler.setLevel(INFO)
logger.setLevel(INFO)
logger.addHandler(handler)

API_KEY = 'YourApiKey'
API_SECRET = 'YourApiSecret'
ITV_NOTIFY = 60  # 通知間隔[min]

DISCORD_USER = 'YourDiscordBotUser'
DISCORD_URL = 'YourDiscordUrl'
LINE_TOKEN = 'YourLineNotifyToken'

def basedate(year=-1, month=-1, day=-1, hour=-1, minute=-1, second=-1):
  now = datetime.now()
  y  = int(year) if year != -1 else now.year
  mo = int(month) if month != -1 else now.month
  d  = int(day) if day != -1 else now.day
  h  = int(hour) if hour != -1 else now.hour
  m  = int(minute) if minute != -1 else now.minute
  s  = int(second) if second != -1 else now.second
  return datetime(y, mo, d, h, m, s)

def getNotifyTime(itvNotify):
  itvSec = itvNotify * 60
  nextNotifyTs = (int(datetime.now().timestamp() / itvSec) + 1) * itvSec
  return datetime.fromtimestamp(nextNotifyTs)

class Discord:
  def __init__(self, user, url):
    self.user = user
    self.url = url

  def send(self, message, fileName=None, isMention=False):
    msg = "@everyone\n" if isMention else ""
    msg += message
    data = {"content": " " + msg + " ", "username": self.user}
    file, f = {}, None
    if fileName:
      f = open(fileName, "rb")
      file = {"imageFile": f}
    try:
      r = requests.post(self.url, data=data, files = file)
    except:
      r = requests.post(self.url, data=data)
      if r.status_code == 204:
        logger.debug("正常に送信しました")
      elif r.status_code == 404:
        if f:  f.close()
        raise RuntimeError("指定URLは存在しません")
    if f:  f.close()
    return r

class Line:
  def __init__(self, token):
    self.token = token
    self.url ='https://notify-api.line.me/api/notify'

  def send(self, message, fileName=None):
    payload={'message': message}
    headers={'Authorization':'Bearer '+ self.token}
    file = {}
    if fileName:
      file["imageFile"] = open(fileName, "rb")
    try:
      r = requests.post(self.url, data=payload, headers=headers, files = file)
    except:
      r = requests.post(self.url, data=payload, headers=headers)
      if r.status_code == requests.codes.ok:
        logger.debug("正常に送信しました")
      elif r.status_code == 400:
        raise RuntimeError("リクエスト内容が間違っています")
      elif r.status_code == 401:
        raise RuntimeError("指定トークンは存在しません")
      elif r.status_code == 500:
        raise RuntimeError("Lineサーバーエラーにより失敗")
      else:
        logger.error("予期せぬレスポンス :" + r.status_code)
    if "imageFile" in file:  file["imageFile"].close()
    return r

class PnlNotifier:
  def __init__(self, itv, key=API_KEY, secret=API_SECRET):
    self.bf = ccxt.bitflyer({"apiKey": key, "secret": secret})
    self.itv = itv
    self.isActive = True
    self.notifyTime = 0
    res = self.getColl()
    self.coll = {
      "start": datetime.now(),
      "init": res["collateral"],
      "dayInit": res["collateral"],
      "day": basedate(hour=0, minute=0, second=0),
      "cur": res["collateral"],
      "notified": res["collateral"],
    }
    self.nextNotify = getNotifyTime(ITV_NOTIFY)
    self.line = None
    self.discord = None
    logger.info("Create PNL notifier. Initial collateral {}. First notification at {}".format(self.coll["init"], self.nextNotify.strftime("%Y/%d/%m %H:%M")))

  def initLine(self, token):
    self.line = Line(token)

  def initDiscord(self, user, url):
    self.discord = Discord(user, url)

  def run(self):
    while self.isActive:
      try:
        sleep((self.nextNotify - datetime.now()).total_seconds())
        res = self.getColl()
        self.coll["cur"] = res["collateral"]
        self.reportProfit()
        self.nextNotify = getNotifyTime(ITV_NOTIFY)
      except Exception as e:
        logger.error(e)
        raise e

  def stop(self):
    self.isActive = False

  def getColl(self):
    while True:
      try:
        res = self.bf.private_get_getcollateral()
        break
      except Exception as e:
        logger.error(e)
        sleep(5)
        continue
    return res

  # 利益情報の定期配信
  def reportProfit(self):
    elapsed = datetime.now() - self.coll["start"]
    elapsedStr = str(elapsed)
    if elapsedStr.find(".") > -1:
      elapsedStr = elapsedStr[:elapsedStr.find(".")]

    # 時間
    msg  = "開始時刻\n{}".format(self.coll["start"].strftime("%Y/%m/%d %H:%M:%S"))
    msg += "\n経過時間\n{}".format(elapsedStr)
    msg += "\n" + "-"*10

    # 証拠金
    msg += "\n初期証拠金\n{}".format(int(self.coll["init"]))
    msg += "\n現在証拠金\n{}".format(int(self.coll["cur"]))
    msg += "\n" + "-"*10

    # 稼働トータル
    profitTotal = self.coll["cur"] - self.coll["init"]
    profitPerHourTotal = round(profitTotal / (elapsed.total_seconds() / 60 / 60), 1)
    profitPerMinuteTotal = round(profitPerHourTotal / 60, 1)
    msg += "\nトータル損益\n{}".format(int(profitTotal))
    msg += "\n時給(トータル)\n{}".format(profitPerHourTotal)
    msg += "\n分給(トータル)\n{}".format(profitPerMinuteTotal)
    msg += "\n" + "-"*10

    # 日次
    profitDaily = self.coll["cur"] - self.coll["dayInit"]
    profitPerHourDaily = round(profitDaily / ((datetime.now() - self.coll["day"]).total_seconds() / 60 / 60), 1)
    profitPerMinuteDaily = round(profitPerHourDaily / 60, 1)
    msg += "\n日次損益\n{}".format(int(profitDaily))
    msg += "\n時給(日次)\n{}".format(profitPerHourDaily)
    msg += "\n分給(日次)\n{}".format(profitPerMinuteDaily)
    msg += "\n" + "-"*10

    # 区間
    profitItv = self.coll["cur"] - self.coll["notified"]
    profitPerHourItv = round(profitItv / (self.itv / 60), 1)
    profitPerMinuteItv = round(profitPerHourItv / 60, 1)
    msg += "\n区間損益\n{}".format(int(profitItv))
    msg += "\n時給(区間)\n{}".format(profitPerHourItv)
    msg += "\n分給(区間)\n{}".format(profitPerMinuteItv)

    if self.line: self.line.send("\n" + msg)
    if self.discord:  self.discord.send(datetime.now().strftime("%Y/%m/%d %H:%M:%S\n") + '```' + msg + '```', isMention=True)

    logger.info("{}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}".format(elapsedStr, int(self.coll["init"]), int(self.coll["cur"]), int(profitTotal), profitPerHourTotal, profitPerMinuteTotal, int(profitDaily), profitPerHourDaily, profitPerMinuteDaily, int(profitItv), profitPerHourItv, profitPerMinuteItv))

    self.coll["notified"] = self.coll["cur"]
    if self.coll["day"] + timedelta(days=1) < datetime.now():
      self.coll["day"] = basedate(hour=0, minute=0, second=0)
      self.coll["dayInit"] = self.coll["cur"]

if __name__ == '__main__':
  pnlNotifier = PnlNotifier(ITV_NOTIFY)
  pnlNotifier.initLine(LINE_TOKEN)
  pnlNotifier.initDiscord(DISCORD_USER, DISCORD_URL)
  executor = concurrent.futures.ThreadPoolExecutor(max_workers=1)
  executor.submit(pnlNotifier.run)

  try:
    sleep(1)
  except (KeyboardInterrupt, Exception) as e:
    logger.error(e)
    pnlNotifier.stop()
    executor.shutdown(wait=True)

Special thanks

LINE/Discord 通知は、れぷすさんのコードをベースにさせてもらってます。

おわりに

有料(¥100)にしてるけど、これで内容は全部です。募金してくれる人がいれば、ジュース代としていただけると嬉しい。コードは、インデントくずれが起きたりするようなので、コピペ時には注意してください。

マガジン

コメント用note(未購入者向け)

干し芋