トップ bitFlyer APIに関する質問 bitflyer Realtime APIに、websocketで接続して得たデータを変数に格納する

編集履歴

回答編集履歴

2022/08/22 12:00

投稿

スコア0

test CHANGED Viewed

@@ -104,3 +104,105 @@
 停止する場合はCTRL+Cを押してください。
 止めることができなくても構わない、というなら try ～ except で囲まなくてもいいですが・・・。
+-----------
+# 分割例
+仮に bitflyer.py と main.py を分割して、main.pyだけ実行して同じようなことがしたいなら、下記のようになります。
+ bitflyer.py
+```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
+```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()
+```

2022/08/22 11:42

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -79,7 +79,6 @@
         product_code = resp['product_code']
         volume = float(resp['volume'])
         ticker = [product_code, timestamp, volume]
-        # logger.info(f'ticker={ticker}')
         que.put(ticker) #キューに ticker を入れる。

2022/08/22 11:41

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -15,19 +15,19 @@
 bitflyer.py
 ```py
 import logging
+import json
+import sys
+import time
 import websocket
-import json
+from concurrent.futures import ThreadPoolExecutor
-import time
 from queue import Queue
-from concurrent.futures import ThreadPoolExecutor
-import sys
 logging.basicConfig(level=logging.INFO, stream=sys.stdout)
 logger = logging.getLogger(__name__)
 que = Queue()  # データ受け渡しのためのキュー
+stop = {"status":False}  # 停止するためのフラグ
-stop = {"status":False}  # 停止するためのフラグ
 class RealtimeAPI(object):
@@ -89,7 +89,6 @@
     RealtimeAPI(url=url, channel=channel)
 executor = ThreadPoolExecutor()
 executor.submit(task)
 try:

2022/08/22 11:40

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -21,6 +21,8 @@
 from queue import Queue
 from concurrent.futures import ThreadPoolExecutor
+import sys
+logging.basicConfig(level=logging.INFO, stream=sys.stdout)
 logger = logging.getLogger(__name__)
 que = Queue()  # データ受け渡しのためのキュー
@@ -87,6 +89,7 @@
     RealtimeAPI(url=url, channel=channel)
 executor = ThreadPoolExecutor()
 executor.submit(task)
 try:
@@ -97,9 +100,9 @@
         time.sleep(1)
 except KeyboardInterrupt:
     stop["status"] = True
-    executor.shutdown()
+    executor.shutdown()
 ```
 停止する場合はCTRL+Cを押してください。
-try: ～ exceptで囲んでいるのは安全に停止するためです。
 止めることができなくても構わない、というなら try ～ except で囲まなくてもいいですが・・・。

2022/08/22 11:36

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -6,7 +6,7 @@
 のような形で書いてticker_data の"内容を表示したい”」
 と理解しました。
-これは、下記のようにマルチスレッドを使用すればできます。
+これは、下記のようにマルチスレッドを使用し、キューを使って受け渡しすればできます。
 なお、下記の場合 **main.py は無関係になるので、main.py の実行は不要**であり、
 **bitflyer.py を実行**すればよいです。

2022/08/22 11:32

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -1,10 +1,9 @@
-コメントによるヒヤリングを経た結果
+コメントによるヒヤリングを経た結果、結局
-「print(ticker_data)」のticker_dataにデータを入れたい、
+「print(ticker_data)」のticker_dataにデータを入れたい」ということでした。
-すなわち　print と組み合わせていることから読み解くに
+print と組み合わせていることから読み解くに
-「bitflyer.py の中（下の方）に
+多分「bitflyer.py の中（下の方）に、print(ticker_data)
-print(ticker_data)
-のような形にして
-ticker_data の"内容を表示したい”」
+のような形で書いてticker_data の"内容を表示したい”」
 と理解しました。
 これは、下記のようにマルチスレッドを使用すればできます。

2022/08/22 11:31

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -100,5 +100,7 @@
     stop["status"] = True
     executor.shutdown()
 ```
+停止する場合はCTRL+Cを押してください。
+try: ～ exceptで囲んでいるのは安全に停止するためです。
-停止する場合はCTRL+Cを押してください。（try: ～ exceptで囲んでいるのはそのためです。安全に止めなくても構わない、というなら try ～ except で囲囲まなくてもいいですが・・・）
+止めることができなくても構わない、というなら try ～ except で囲まなくてもいいですが・・・。

2022/08/22 11:30

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -100,3 +100,5 @@
     stop["status"] = True
     executor.shutdown()
 ```
+停止する場合はCTRL+Cを押してください。（try: ～ exceptで囲んでいるのはそのためです。安全に止めなくても構わない、というなら try ～ except で囲囲まなくてもいいですが・・・）

2022/08/22 11:28

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -1,7 +1,7 @@
 コメントによるヒヤリングを経た結果
 「print(ticker_data)」のticker_dataにデータを入れたい、
 すなわち　print と組み合わせていることから読み解くに
-「bitflyer.py の中（最下行）に
+「bitflyer.py の中（下の方）に
 print(ticker_data)
 のような形にして
 ticker_data の"内容を表示したい”」
@@ -11,7 +11,7 @@
 なお、下記の場合 **main.py は無関係になるので、main.py の実行は不要**であり、
 **bitflyer.py を実行**すればよいです。
-（「bitflyer.py の中（最下行）に」print(ticker_data)と入れてticker_data の内容を表示したい、とのことですから当然といえば当然ですが）
+（「bitflyer.py の中（下の方）に」print(ticker_data)と入れてticker_data の内容を表示したい、とのことですから当然といえば当然ですが）
 bitflyer.py
 ```py

2022/08/22 11:27

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -18,13 +18,11 @@
 import logging
 import websocket
 import json
-import pybitflyer
 import time
 from queue import Queue
 from concurrent.futures import ThreadPoolExecutor
-logger = logging.getLogger(__name__)#logging
+logger = logging.getLogger(__name__)
 que = Queue()  # データ受け渡しのためのキュー
@@ -81,7 +79,7 @@
         volume = float(resp['volume'])
         ticker = [product_code, timestamp, volume]
         # logger.info(f'ticker={ticker}')
-        que.put(ticker) #= ticker#追記
+        que.put(ticker) #キューに ticker を入れる。
 def task():
@@ -89,15 +87,16 @@
     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から得たデータを表示
+        print(ticker_data)  # ここでAPIから入ってくるtickerのデータを表示
         time.sleep(1)
 except KeyboardInterrupt:
     stop["status"] = True
-    executor.shutdown()
+    executor.shutdown()
 ```

2022/08/22 11:25

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -3,15 +3,16 @@
 すなわち　print と組み合わせていることから読み解くに
 「bitflyer.py の中（最下行）に
 print(ticker_data)
-と入れて
+のような形にして
 ticker_data の"内容を表示したい”」
 と理解しました。
-これは、下記のようにすればできます。
+これは、下記のようにマルチスレッドを使用すればできます。
 なお、下記の場合 **main.py は無関係になるので、main.py の実行は不要**であり、
 **bitflyer.py を実行**すればよいです。
 （「bitflyer.py の中（最下行）に」print(ticker_data)と入れてticker_data の内容を表示したい、とのことですから当然といえば当然ですが）
 bitflyer.py
 ```py
 import logging
@@ -25,7 +26,7 @@
 logger = logging.getLogger(__name__)#logging
-ticker_data = Queue()  # データ受け渡しのためのキュー
+que = Queue()  # データ受け渡しのためのキュー
 stop = {"status":False}  # 停止するためのフラグ
@@ -80,22 +81,23 @@
         volume = float(resp['volume'])
         ticker = [product_code, timestamp, volume]
         # logger.info(f'ticker={ticker}')
-        ticker_data.put(ticker) #= ticker#追記
+        que.put(ticker) #= ticker#追記
 def task():
     url = "wss://ws.lightstream.bitflyer.com/json-rpc"
     channel = "lightning_ticker_FX_BTC_JPY"
-    json_rpc = RealtimeAPI(url=url, channel=channel)
+    RealtimeAPI(url=url, channel=channel)
 executor = ThreadPoolExecutor()
 executor.submit(task)
 try:
     while True:
+        # キューからデータを取り出す。
+        ticker_data = str(que.get())
-        print("--->"+str(ticker_data.get()))  # ここでAPIから得たデータを表示
+        print(ticker_data)  # ここでAPIから得たデータを表示
         time.sleep(1)
 except KeyboardInterrupt:
     stop["status"] = True
     executor.shutdown()
 ```

2022/08/22 11:21

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -19,9 +19,15 @@
 import json
 import pybitflyer
+import time
+from queue import Queue
+from concurrent.futures import ThreadPoolExecutor
 logger = logging.getLogger(__name__)#logging
-ticker_data = 0#追記
+ticker_data = Queue()  # データ受け渡しのためのキュー
+stop = {"status":False}  # 停止するためのフラグ
 class RealtimeAPI(object):
@@ -46,6 +52,10 @@
         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)
@@ -69,17 +79,23 @@
         product_code = resp['product_code']
         volume = float(resp['volume'])
         ticker = [product_code, timestamp, volume]
-        print(ticker)
-        logger.info(f'ticker={ticker}')
+        # logger.info(f'ticker={ticker}')
-        global ticker_data#追記
-        ticker_data = ticker#追記
+        ticker_data.put(ticker) #= ticker#追記
+def task():
-url = "wss://ws.lightstream.bitflyer.com/json-rpc"
+    url = "wss://ws.lightstream.bitflyer.com/json-rpc"
-channel = "lightning_ticker_FX_BTC_JPY"
+    channel = "lightning_ticker_FX_BTC_JPY"
-json_rpc = RealtimeAPI(url=url, channel=channel)
+    json_rpc = RealtimeAPI(url=url, channel=channel)
+executor = ThreadPoolExecutor()
+executor.submit(task)
+try:
-while True:
+    while True:
-    print(ticker_data)#APIから入ってくるtickerのデータが表示される
+        print("--->"+str(ticker_data.get()))  # ここでAPIから得たデータを表示
+        time.sleep(1)
+except KeyboardInterrupt:
+    stop["status"] = True
+    executor.shutdown()
 ```

2022/08/22 11:06

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -80,6 +80,6 @@
 channel = "lightning_ticker_FX_BTC_JPY"
 json_rpc = RealtimeAPI(url=url, channel=channel)
 while True:
-    print(ticker_data)#ここにAPIから入ってくるtickerのデータを入れたいです
+    print(ticker_data)#APIから入ってくるtickerのデータが表示される
 ```

2022/08/22 11:05

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -5,7 +5,7 @@
 print(ticker_data)
 と入れて
 ticker_data の"内容を表示したい”」
-と理解できます。
+と理解しました。
 これは、下記のようにすればできます。

2022/08/22 11:05

投稿

退会済みユーザー

スコア0

test CHANGED Viewed

@@ -1,15 +1,18 @@
 コメントによるヒヤリングを経た結果
+「print(ticker_data)」のticker_dataにデータを入れたい、
+すなわち　print と組み合わせていることから読み解くに
 「bitflyer.py の中（最下行）に
 print(ticker_data)
 と入れて
-ticker_data の内容を表示したい」
+ticker_data の"内容を表示したい”」
-とのこと。
+と理解できます。
-下記のようにすればできます。
+これは、下記のようにすればできます。
-なお、下記の場合main.pyは無関係になあるので、main.pyの実行は不要で
+なお、下記の場合 **main.py は無関係になるので、main.py の実行は不要**であり、
-bitflyer.py を実行すればよいです。
+**bitflyer.py を実行**すればよいです。
 （「bitflyer.py の中（最下行）に」print(ticker_data)と入れてticker_data の内容を表示したい、とのことですから当然といえば当然ですが）
+bitflyer.py
 ```py
 import logging
 import websocket