"""Module for IQ option websocket."""
import json
import logging
import websocket
[docs]class WebsocketClient(object):
"""Class for work with IQ option websocket."""
def __init__(self, api):
"""
:param api: The instance of :class:`IQOptionAPI
<iqoptionapi.api.IQOptionAPI>`.
"""
self.api = api
self.wss = websocket.WebSocketApp(
self.api.wss_url, on_message=self.on_message,
on_error=self.on_error, on_close=self.on_close,
on_open=self.on_open)
[docs] def on_message(self, wss, message): # pylint: disable=unused-argument
"""Method to process websocket messages."""
logger = logging.getLogger(__name__)
logger.debug(message)
message = json.loads(str(message))
if message["name"] == "timeSync":
self.api.timesync.server_timestamp = message["msg"]
elif message["name"] == "profile":
self.api.profile.balance = message["msg"]["balance"]
elif message["name"] == "candles":
self.api.candles.candles_data = message["msg"]["data"]
elif message["name"] == "buyComplete":
self.api.buySuccessful = message["msg"]["isSuccessful"]
elif message["name"] == "listInfoData":
listinfodata = lambda: None
listinfodata.__dict__ = message["msg"][0]
self.api.listinfodata.add_listinfodata(listinfodata)
[docs] @staticmethod
def on_error(wss, error): # pylint: disable=unused-argument
"""Method to process websocket errors."""
logger = logging.getLogger(__name__)
logger.error(error)
[docs] @staticmethod
def on_open(wss): # pylint: disable=unused-argument
"""Method to process websocket open."""
logger = logging.getLogger(__name__)
logger.debug("Websocket client connected.")
[docs] @staticmethod
def on_close(wss): # pylint: disable=unused-argument
"""Method to process websocket close."""
logger = logging.getLogger(__name__)
logger.debug("Websocket connection closed.")