bitflyer Realtime APIに、websocketで接続して得たデータを変数に格納する

Question

前提

bitflyer Realtime APIに、websocketで接続して得たデータを変数に格納するために
試行錯誤しましたが、やり方がわかりませんでした

実現したいこと

bitflyer Realtime APIに、websocketで接続して得たデータを変数に格納する

発生している問題・エラーメッセージ

--- Logging error ---
Traceback (most recent call last):
File "/Users/a1/opt/anaconda3/envs/m/lib/python3.9/site-packages/websocket/_app.py", line 383, in run_forever
dispatcher.read(self.sock.sock, read, check)
File "/Users/a1/opt/anaconda3/envs/m/lib/python3.9/site-packages/websocket/_app.py", line 64, in read
r = self.select()
File "/Users/a1/opt/anaconda3/envs/m/lib/python3.9/site-packages/websocket/_app.py", line 78, in select
r = sel.select(self.ping_timeout)
File "/Users/a1/opt/anaconda3/envs/m/lib/python3.9/selectors.py", line 562, in select
kev_list = self._selector.control(None, max_ev, timeout)
KeyboardInterrupt

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/Users/a1/opt/anaconda3/envs/m/lib/python3.9/site-packages/websocket/_app.py", line 424, in _callback
callback(self, *args)
TypeError: on_close() takes 2 positional arguments but 4 were given

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
File "/Users/a1/opt/anaconda3/envs/m/lib/python3.9/logging/init.py", line 1086, in emit
stream.write(msg + self.terminator)
AttributeError: 'int' object has no attribute 'write'
Call stack:
File "/Users/a1/Desktop/ml/main.py", line 14, in <module>
json_rpc = RealtimeAPI(url=url, channel=channel)
File "/Users/a1/Desktop/ml/bitflyer.py", line 15, in init
self.connect()
File "/Users/a1/Desktop/ml/bitflyer.py", line 22, in connect
self.ws.run_forever()
File "/Users/a1/opt/anaconda3/envs/m/lib/python3.9/site-packages/websocket/_app.py", line 390, in run_forever
teardown()
File "/Users/a1/opt/anaconda3/envs/m/lib/python3.9/site-packages/websocket/_app.py", line 318, in teardown
self._callback(self.on_close, close_status_code, close_reason)
File "/Users/a1/opt/anaconda3/envs/m/lib/python3.9/site-packages/websocket/_app.py", line 429, in _callback
self.on_error(self, e)
File "/Users/a1/Desktop/ml/bitflyer.py", line 35, in on_error
logger.error(error)
Message: TypeError('on_close() takes 2 positional arguments but 4 were given')
Arguments: ()

該当のソースコード

使用言語　python3.9

コード　ファイル名　bitflyer.py

import logging
import websocket
import json

import pybitflyer

logger = logging.getLogger(__name__)#logging
ticker_data = 0#追記

class RealtimeAPI(object):

    def __init__(self, url, channel):
        self.url = url
        self.channel = channel
        self.connect()

    def connect(self):
        self.ws = websocket.WebSocketApp\
            (self.url,header=None,on_open=self.on_open,\
                 on_message=self.on_message, on_error=self.on_error,\
                     on_close=self.on_close)
        self.ws.run_forever()
        logger.info('Web Socket process ended.')

    """
    Below are callback functions of websocket.
    """
    # when we get message
    def on_message(self, ws, message):
        resp = json.loads(message)['params']['message']
        self.set_realtime_ticker(resp)

    # when error occurs
    def on_error(self, ws, error):
        logger.error(error)

    # when websocket closed.
    def on_close(self, ws):
        logger.info('disconnected streaming server')

    # when websocket opened.
    def on_open(self, ws):
        logger.info('connected streaming server')
        input_data = json.dumps(
            {'method' : 'subscribe',
            'params' : {'channel' : self.channel}
            }
        )
        ws.send(input_data)

    def set_realtime_ticker(self, resp):
        timestamp = resp['timestamp']
        product_code = resp['product_code']
        volume = float(resp['volume'])
        ticker = [product_code, timestamp, volume]
        print(ticker)
        logger.info(f'ticker={ticker}')

        global ticker_data#追記
        ticker_data = ticker#追記

print(ticker_data)#ここにAPIから入ってくるtickerのデータを入れたいです

---------------------------------ここからもう一つのファイル---------------------------------------
#ファイル名　main.py
import logging
import sys

from bitflyer import RealtimeAPI

logging.basicConfig(level=logging.INFO, stream=sys.stdout)
# logging.basicConfig(level=logging.INFO, stream=a)
#ログレベル INFO　出力先　stream=sys.stdout

if __name__ == "__main__":
    url = "wss://ws.lightstream.bitflyer.com/json-rpc"
    channel = "lightning_ticker_FX_BTC_JPY"
    json_rpc = RealtimeAPI(url=url, channel=channel)


------------------------------コードはここまで---------------------------------------

### 試したこと

①ファイル名bitflyer.pyのdef set_realtime_tickerにあるtickerをグローバル変数にした
②ファイル名　main.pyのlogging.basicConfigにあるstream=変数にした

### 補足情報（FW/ツールのバージョンなど）
Visual Studio Codeを使用しています

ここにより詳細な情報を記載してください。

よろしくお願いいたします

Comments

[退会済みユーザー]

「変数に格納する」とは、具体的にいうと、質問に記載されているソースコードの中のどの変数を、どのようにしたいのでしょうか？
手元でbitflyer.pyとmain.pyを同じディレクトリに配置してmain.pyを実行したところ、tickerの内容は正常にターミナルに表示されました。

[161]

追記依頼いただき、ありがとうございます
仰る通りターミナルに表示されるところまではできています
そのターミナルに表示されているデータを、コード中には記入していないのですが
ファイル名　bitflyer.py の def set_realtime_ticker内に
tick_data=ticker として変数 tick_data 内に格納し
それを関数の外でも使えるグローバル変数として使用したいと考えています

[退会済みユーザー]

「それを関数の外でも使えるグローバル変数として使用したいと考えています」
→これは、何も情報を持っていない人から見ると、手掛かりがなく答えようがありません。
というか範囲が広すぎます。
エスパーを駆使して具体的なシチュエーションを勝手に作って例示することはできなくもないですが、仮にそれが質問者さんの解決したいことではなかった場合、無駄打ちになってしまいます。

（よくある「XY問題」というものに近いです）

「（グローバル変数などにすることによって）どのように使いたいのか」という「使い方」を具体的なコードも含めて、質問文に書いてください。
つまり「このようなコードで、こういう風にtickerを取り出して、このように加工したいと考えているが、現状できない」というのがわかる内容（コード）、ということです。

動かないコードでも構いません。そのコードを動かしてエラーが出ているならばそのエラーも省略せずに記載してください。

[161]

内容を修正しました、コードの変更（ファイル名　bitflyer.pyの#追記となっている部分と、最終行）
とコード変更に伴って生じたエラーを記載しました
よろしくお願いいたします

Answer 1

コメントによるヒヤリングを経た結果、結局
「print(ticker_data)」のticker_dataにデータを入れたい」ということでした。

print と組み合わせていることから読み解くに
多分「bitflyer.py の中（下の方）に、print(ticker_data)
のような形で書いてticker_data の"内容を表示したい”」
と理解しました。

これは、下記のようにマルチスレッドを使用し、キューを使って受け渡しすればできます。

なお、下記の場合 main.py は無関係になるので、main.py の実行は不要であり、
bitflyer.py を実行すればよいです。
（「bitflyer.py の中（下の方）に」print(ticker_data)と入れてticker_data の内容を表示したい、とのことですから当然といえば当然ですが）

bitflyer.py

import logging
import json
import sys
import time
import websocket
from concurrent.futures import ThreadPoolExecutor
from queue import Queue

logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger(__name__)

que = Queue()  # データ受け渡しのためのキュー
stop = {"status":False}  # 停止するためのフラグ


class RealtimeAPI(object):

    def __init__(self, url, channel):
        self.url = url
        self.channel = channel
        self.connect()

    def connect(self):
        self.ws = websocket.WebSocketApp\
            (self.url,header=None,on_open=self.on_open,\
                 on_message=self.on_message, on_error=self.on_error,\
                     on_close=self.on_close)
        self.ws.run_forever()
        logger.info('Web Socket process ended.')

    """
    Below are callback functions of websocket.
    """
    # when we get message
    def on_message(self, ws, message):
        resp = json.loads(message)['params']['message']
        self.set_realtime_ticker(resp)

        if stop["status"]:
            # 停止フラグがTrueならwebsocketをクローズする。
            self.ws.close()

    # when error occurs
    def on_error(self, ws, error):
        logger.error(error)

    # when websocket closed.
    def on_close(self, ws):
        logger.info('disconnected streaming server')

    # when websocket opened.
    def on_open(self, ws):
        logger.info('connected streaming server')
        input_data = json.dumps(
            {'method' : 'subscribe',
            'params' : {'channel' : self.channel}
            }
        )
        ws.send(input_data)

    def set_realtime_ticker(self, resp):
        timestamp = resp['timestamp']
        product_code = resp['product_code']
        volume = float(resp['volume'])
        ticker = [product_code, timestamp, volume]
        que.put(ticker) #キューに ticker を入れる。


def task():
    url = "wss://ws.lightstream.bitflyer.com/json-rpc"
    channel = "lightning_ticker_FX_BTC_JPY"
    RealtimeAPI(url=url, channel=channel)


executor = ThreadPoolExecutor()
executor.submit(task)
try:
    while True:
        # キューから ticker を取り出す。
        ticker_data = str(que.get())
        print(ticker_data)  # ここでAPIから入ってくるtickerのデータを表示
        time.sleep(1)
except KeyboardInterrupt:
    stop["status"] = True
    executor.shutdown()    

停止する場合はCTRL+Cを押してください。
止めることができなくても構わない、というなら try ～ except で囲まなくてもいいですが・・・。

---

分割例

仮に bitflyer.py と main.py を分割して、main.pyだけ実行して同じようなことがしたいなら、下記のようになります。

bitflyer.py

import json
import logging
import sys
import websocket
from queue import Queue

logging.basicConfig(level=logging.INFO, stream=sys.stdout)
logger = logging.getLogger(__name__)

que = Queue()  # データ受け渡しのためのキュー

stop = {"status":False}  # 停止するためのフラグ


class RealtimeAPI(object):

    def __init__(self, url, channel):
        self.url = url
        self.channel = channel
        self.connect()

    def connect(self):
        self.ws = websocket.WebSocketApp\
            (self.url,header=None,on_open=self.on_open,\
                 on_message=self.on_message, on_error=self.on_error,\
                     on_close=self.on_close)
        self.ws.run_forever()
        logger.info('Web Socket process ended.')

    """
    Below are callback functions of websocket.
    """
    # when we get message
    def on_message(self, ws, message):
        resp = json.loads(message)['params']['message']
        self.set_realtime_ticker(resp)

        if stop["status"]:
            # 停止フラグがTrueならwebsocketをクローズする。
            self.ws.close()

    # when error occurs
    def on_error(self, ws, error):
        logger.error(error)

    # when websocket closed.
    def on_close(self, ws):
        logger.info('disconnected streaming server')

    # when websocket opened.
    def on_open(self, ws):
        logger.info('connected streaming server')
        input_data = json.dumps(
            {'method' : 'subscribe',
            'params' : {'channel' : self.channel}
            }
        )
        ws.send(input_data)

    def set_realtime_ticker(self, resp):
        timestamp = resp['timestamp']
        product_code = resp['product_code']
        volume = float(resp['volume'])
        ticker = [product_code, timestamp, volume]
        que.put(ticker) #キューに ticker を入れる。

main.py

import time
from concurrent.futures import ThreadPoolExecutor
from bitflyer import RealtimeAPI, que, stop


def task():
    url = "wss://ws.lightstream.bitflyer.com/json-rpc"
    channel = "lightning_ticker_FX_BTC_JPY"
    RealtimeAPI(url=url, channel=channel)

if __name__ == "__main__":
    executor = ThreadPoolExecutor()
    executor.submit(task)
    try:
        while True:
            # キューから ticker を取り出す。
            ticker_data = str(que.get())
            print(ticker_data)  # ここでAPIから入ってくるtickerのデータを表示
            time.sleep(1)
    except KeyboardInterrupt:
        stop["status"] = True
        executor.shutdown()   

Comments

[161]

ご回答をいただき感謝しています
ありがとうございました