text: from befh.restful_api_socket import RESTfulApiSocket
  from befh.exchanges.gateway import ExchangeGateway
  from befh.market_data import L2Depth, Trade
  from befh.util import Logger
  from befh.instrument import Instrument
  from befh.clients.sql_template import SqlClientTemplate
  from functools import partial
  from datetime import datetime
  import threading
  import time
  
  
  class ExchGwApiBitflyer(RESTfulApiSocket):
      """
      Exchange gateway RESTfulApi
      """
      def __init__(self):
          super(ExchGwApiBitflyer, self).__init__()
  
      @classmethod
      def get_timestamp_offset(cls):
          return 1000
  
      @classmethod
      def get_order_book_timestamp_field_name(cls):
          return 'date'
  
      @classmethod
      def get_trades_timestamp_field_name(cls):
          return 'exec_date'
  
      @classmethod
      def get_bids_field_name(cls):
          return 'bids'
  
      @classmethod
      def get_asks_field_name(cls):
          return 'asks'
  
      @classmethod
      def get_trade_side_field_name(cls):
          return 'side'
  
      @classmethod
      def get_trade_id_field_name(cls):
          return 'id'
  
      @classmethod
      def get_trade_price_field_name(cls):
          return 'price'
  
      @classmethod
      def get_trade_volume_field_name(cls):
          return 'size'
  
      @classmethod
      def get_order_book_link(cls, instmt):
          return 'https://api.bitflyer.jp/v1/getboard?product_code={}'.format(instmt.instmt_code)
  
      @classmethod
      def get_trades_link(cls, instmt):
          return 'https://api.bitflyer.jp/v1/getexecutions?product_cdoe={}'.format(instmt.instmt_code)
  
      @classmethod
      def parse_l2_depth(cls, instmt, raw):
          """
          Parse raw data to L2 depth
          :param instmt: Instrument
          :param raw: Raw data in JSON
          """
          l2_depth = L2Depth()
          keys = list(raw.keys())
          if cls.get_bids_field_name() in keys and \
             cls.get_asks_field_name() in keys:
  
              # No Date time information, has update id only
              l2_depth.date_time = datetime.utcnow().strftime("%Y%m%d %H:%M:%S.%f")
  
              # Bids
              bids = raw[cls.get_bids_field_name()]
              bids = sorted(bids, key=lambda x: x['price'], reverse=True)
              for i in range(0, 5):
                  l2_depth.bids[i].price = float(bids[i]['price']) if type(bids[i]['price']) != float else bids[i]['price']
                  l2_depth.bids[i].volume = float(bids[i]['size']) if type(bids[i]['size']) != float else bids[i]['size']
  
              # Asks
              asks = raw[cls.get_asks_field_name()]
              asks = sorted(asks, key=lambda x: x['price'])
              for i in range(0, 5):
                  l2_depth.asks[i].price = float(asks[i]['price']) if type(asks[i]['price']) != float else asks[i]['price']
                  l2_depth.asks[i].volume = float(asks[i]['size']) if type(asks[i]['size']) != float else asks[i]['size']
          else:
              raise Exception('Does not contain order book keys in instmt %s-%s.\nOriginal:\n%s' % \
                  (instmt.get_exchange_name(), instmt.get_instmt_name(), \
                   raw))
  
          return l2_depth
  
      @classmethod
      def parse_trade(cls, instmt, raw):
          """
          :param instmt: Instrument
          :param raw: Raw data in JSON
          :return:
          """
          trade = Trade()
          keys = list(raw.keys())
  
          if cls.get_trades_timestamp_field_name() in keys and \
             cls.get_trade_id_field_name() in keys and \
             cls.get_trade_price_field_name() in keys and \
             cls.get_trade_volume_field_name() in keys:
  
              # Date time
              date_time = raw[cls.get_trades_timestamp_field_name()]
              try:
                  trade.date_time = datetime.strptime(date_time, '%Y-%m-%dT%H:%M:%S.%f').strftime('%Y%m%d %H:%M:%S.%f')
              except Exception as e:
                  trade.date_time = datetime.strptime(date_time, '%Y-%m-%dT%H:%M:%S').strftime('%Y%m%d %H:%M:%S.%f')
  
              # Trade side
              trade.trade_side = Trade.parse_side(raw[cls.get_trade_side_field_name()])
              # Trade id
              trade.trade_id = str(raw[cls.get_trade_id_field_name()])
  
              # Trade price
              trade.trade_price = float(str(raw[cls.get_trade_price_field_name()]))
  
              # Trade volume
              trade.trade_volume = float(str(raw[cls.get_trade_volume_field_name()]))
          else:
              raise Exception('Does not contain trade keys in instmt %s-%s.\nOriginal:\n%s' % \
                  (instmt.get_exchange_name(), instmt.get_instmt_name(), \
                   raw))
  
          return trade
  
      @classmethod
      def get_order_book(cls, instmt):
          """
          Get order book
          :param instmt: Instrument
          :return: Object L2Depth
          """
          # If verify cert, got <urlopen error [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed (_ssl.c:749)>
          res = cls.request(cls.get_order_book_link(instmt), verify_cert=False)
          if len(res) > 0:
              return cls.parse_l2_depth(instmt=instmt,
                                         raw=res)
          else:
              return None
  
      @classmethod
      def get_trades(cls, instmt):
          """
          Get trades
          :param instmt: Instrument
          :param trade_id: Trade id
          :return: List of trades
          """
          link = cls.get_trades_link(instmt)
          # If verify cert, got <urlopen error [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed (_ssl.c:749)>
          res = cls.request(link, verify_cert=False)
          trades = []
          if len(res) > 0:
              for t in res:
                  trade = cls.parse_trade(instmt=instmt,
                                           raw=t)
                  trades.append(trade)
  
          return trades
  
  
  class ExchGwBitflyer(ExchangeGateway):
      """
      Exchange gateway
      """
      def __init__(self, db_clients):
          """
          Constructor
          :param db_client: Database client
          """
          ExchangeGateway.__init__(self, ExchGwApiBitflyer(), db_clients)
  
      @classmethod
      def get_exchange_name(cls):
          """
          Get exchange name
          :return: Exchange name string
          """
          return 'Bitflyer'
  
      def get_order_book_worker(self, instmt):
          """
          Get order book worker
          :param instmt: Instrument
          """
          while True:
              try:
                  l2_depth = self.api_socket.get_order_book(instmt)
                  if l2_depth is not None and l2_depth.is_diff(instmt.get_l2_depth()):
                      instmt.set_prev_l2_depth(instmt.get_l2_depth())
                      instmt.set_l2_depth(l2_depth)
                      instmt.incr_order_book_id()
                      self.insert_order_book(instmt)
              except Exception as e:
                  Logger.error(self.__class__.__name__, "Error in order book: %s" % e)
              time.sleep(1)
  
      def get_trades_worker(self, instmt):
          """
          Get order book worker thread
          :param instmt: Instrument name
          """
          while True:
              try:
                  ret = self.api_socket.get_trades(instmt)
                  if ret is None or len(ret) == 0:
                      time.sleep(1)
                      continue
              except Exception as e:
                  Logger.error(self.__class__.__name__, "Error in trades: %s" % e)
                  time.sleep(1)
                  continue
  
              for trade in ret:
                  assert isinstance(trade.trade_id, str), "trade.trade_id(%s) = %s" % (type(trade.trade_id), trade.trade_id)
                  assert isinstance(instmt.get_exch_trade_id(), str), \
                         "instmt.get_exch_trade_id()(%s) = %s" % (type(instmt.get_exch_trade_id()), instmt.get_exch_trade_id())
                  if int(trade.trade_id) > int(instmt.get_exch_trade_id()):
                      instmt.set_exch_trade_id(trade.trade_id)
                      instmt.incr_trade_id()
                      self.insert_trade(instmt, trade)
  
              # After the first time of getting the trade, indicate the instrument
              # is recovered
              if not instmt.get_recovered():
                  instmt.set_recovered(True)
  
              time.sleep(1)
  
      def start(self, instmt):
          """
          Start the exchange gateway
          :param instmt: Instrument
          :return List of threads
          """
          instmt.set_l2_depth(L2Depth(5))
          instmt.set_prev_l2_depth(L2Depth(5))
          instmt.set_instmt_snapshot_table_name(self.get_instmt_snapshot_table_name(instmt.get_exchange_name(),
                                                                                    instmt.get_instmt_name()))
          self.init_instmt_snapshot_table(instmt)
          instmt.set_recovered(False)
          t1 = threading.Thread(target=partial(self.get_order_book_worker, instmt))
          t2 = threading.Thread(target=partial(self.get_trades_worker, instmt))
          t1.start()
          t2.start()
          return [t1, t2]
  
  
  if __name__ == '__main__':
      exchange_name = 'Bitflyer'
      instmt_name = 'BTC_JPY'
      instmt_code = 'BTC_JPY'
      instmt = Instrument(exchange_name, instmt_name, instmt_code)
      Logger.init_log()
      db_client = SqlClientTemplate()
      exch = ExchGwBitflyer([db_client])
      instmt.set_l2_depth(L2Depth(5))
      instmt.set_prev_l2_depth(L2Depth(5))
      instmt.set_recovered(False)
      exch.start(instmt)
      #exch.get_order_book_worker(instmt)
      #exch.get_trades_worker(instmt)
  