Jump to content
  • 0

Subscription through API



I am trying to subscribe to my first streaming data. I basically copy pasted code from the IG Labs forum (what's the difference between here and there?), added my own api key, and it works. Well, almost. It connected well but I can't subscribe. What am I doing wrong?
Subscription returns error 17 Data Adapter not found
This is the subscription code:
    fields = ["offer"]
    fields = [field.upper() for field in fields]
    # Making a new Subscription in MERGE mode
    subscription = Subscription(
        # items=["EURJPY", "USDJPY"],
        # items=["CS.D.USDJPY.CFD.IP", "CS.D.EURJPY.CFD.IP"],
and this is the rest of the code (mostly copy pasted from the IG LAB's forum) 
# when this site used as main (for debugging), we need to add its parent
# folder to the PYTHONPATH, so we could import correctly.
if __name__ == "__main__":
    import os
    import sys
    import inspect
    currentdir = os.path.dirname(os.path.abspath(
    parentdir = os.path.dirname(currentdir)
    sys.path.insert(0, parentdir)

import utils.log_utils as log_utils
import sys
import logging as lg
import threading
import time
import traceback
import requests
import configs.load_config as load_config

from urllib.request import urlopen as _urlopen
from urllib.parse import (urlparse as parse_url, urljoin, urlencode)

def _url_encode(params):
    return urlencode(params).encode("utf-8")

def _iteritems(d):
    return iter(d.items())

def wait_for_input():

CONNECTION_URL_PATH = "lightstreamer/create_session.txt"
BIND_URL_PATH = "lightstreamer/bind_session.txt"
CONTROL_URL_PATH = "lightstreamer/control.txt"
# Request parameter to create and activate a new Table.
OP_ADD = "add"
# Request parameter to delete a previously created Table.
OP_DELETE = "delete"
# Request parameter to force closure of an existing session.
OP_DESTROY = "destroy"
# List of possible server responses

class Subscription(object):
    """Represents a Subscription to be submitted to a Lightstreamer Server."""

    def __init__(self, mode, items, fields, adapter=""):
        class_hierarchy = __name__ + "Subscription"
        self.logger = lg.getLogger(class_hierarchy)
        self.item_names = items
        self._items_map = {}
        self.field_names = fields
        self.adapter = adapter
        self.mode = mode
        self.snapshot = "true"
        self._listeners = []

    def _decode(self, value, last):
        """Decode the field value according to
        Lightstreamer Text Protocol specifications.
        if value == "$":
            return u''
        elif value == "#":
            return None
        elif not value:
            return last
        elif value[0] in "#$":
            value = value[1:]

        return value

    def addlistener(self, listener):

    def notifyupdate(self, item_line):
        """Invoked by LSClient each time Lightstreamer Server pushes
        a new item event.
        # Tokenize the item line as sent by Lightstreamer
        toks = item_line.rstrip('\r\n').split('|')
        undecoded_item = dict(list(zip(self.field_names, toks[1:])))

        # Retrieve the previous item stored into the map, if present.
        # Otherwise create a new empty dict.
        item_pos = int(toks[0])
        curr_item = self._items_map.get(item_pos, {})
        # Update the map with new values, merging with the
        # previous ones if any.
        self._items_map[item_pos] = dict([
            (k, self._decode(v, curr_item.get(k))) for k, v
            in list(undecoded_item.items())
        # Make an item info as a new event to be passed to listeners
        item_info = {
            'pos': item_pos,
            'name': self.item_names[item_pos - 1],
            'values': self._items_map[item_pos]

        # Update each registered listener with new event
        for on_item_update in self._listeners:

class LSClient(object):
    """Manages the communication with Lightstreamer Server"""

    def __init__(self, base_url, adapter_set="", user="", password=""):
        class_hierarchy = __name__ + "LSClient"
        self.logger = lg.getLogger(class_hierarchy)
        self._base_url = parse_url(base_url)
        self._adapter_set = adapter_set
        self._user = user
        self._password = password
        self._session = {}
        self._subscriptions = {}
        self._current_subscription_key = 0
        self._stream_connection = None
        self._stream_connection_thread = None
        self._bind_counter = 0

    def _encode_params(self, params):
        """Encode the parameter for HTTP POST submissions, but
        only for non empty values..."""
        return _url_encode(
            dict([(k, v) for (k, v) in _iteritems(params) if v])

    def _call(self, base_url, url, params):
        """Open a network connection and performs HTTP Post
        with provided params.
        # Combines the "base_url" with the
        # required "url" to be used for the specific request.
        url = urljoin(base_url.geturl(), url)
        body = self._encode_params(params)
        self.logger.debug("Making a request to <%s> with body <%s>", url, body)
        return _urlopen(url, data=body)

    def _set_control_link_url(self, custom_address=None):
        """Set the address to use for the Control Connection
        in such cases where Lightstreamer is behind a Load Balancer.
        if custom_address is None:
            self._control_url = self._base_url
            parsed_custom_address = parse_url("//" + custom_address)
            self._control_url = parsed_custom_address._replace(

    def _control(self, params):
        """Create a Control Connection to send control commands
        that manage the content of Stream Connection.
        params["LS_session"] = self._session["SessionId"]
        response = self._call(self._control_url, CONTROL_URL_PATH, params)
        decoded_response = ""
        for i in range(max_reads):
            response_line = response.readline().decode("utf-8").rstrip()
            if not response_line:
                decoded_response += response_line + "\n"
        self.logger.debug("Server response: <%s>", decoded_response)
        return decoded_response

    def _read_from_stream(self):
        """Read a single line of content of the Stream Connection."""
        line = self._stream_connection.readline().decode("utf-8").rstrip()
        return line

    def connect(self):
        """Establish a connection to Lightstreamer Server to create a new
        self.logger.debug("Opening a new session to <%s>",
        self._stream_connection = self._call(
                "LS_op2": 'create',
                "LS_cid": 'mgQkwtwdysogQz2BJ4Ji kOj2Bg',
                "LS_adapter_set": self._adapter_set,
                "LS_user": self._user,
                "LS_password": self._password}
        stream_line = self._read_from_stream()

    def bind(self):
        """Replace a completely consumed connection in listening for an active
        self.logger.debug("Binding to <%s>", self._control_url.geturl())
        self._stream_connection = self._call(
                "LS_session": self._session["SessionId"]

        self._bind_counter += 1
        stream_line = self._read_from_stream()
        self.logger.info("Bound to <%s>", self._control_url.geturl())

    def _handle_stream(self, stream_line):
        if stream_line == OK_CMD:
            self.logger.info("Successfully connected to <%s>",
            self.logger.debug("Starting to handling real-time stream")
            # Parsing session inkion
            while 1:
                next_stream_line = self._read_from_stream()
                if next_stream_line:
                    session_key, session_value = next_stream_line.split(":", 1)
                    self._session[session_key] = session_value

            # Setup of the control link url

            # Start a new thread to handle real time updates sent
            # by Lightstreamer Server on the stream connection.
            self._stream_connection_thread = threading.Thread(
            self.logger.info("Started handling of real-time stream")
            lines = self._stream_connection.readlines()
            lines.insert(0, stream_line)
            self.logger.error("Server response error: \n%s",
                              " ".join((str(x) for x in lines)))
            raise IOError()

    def _join(self):
        """Await the natural StreamThread termination."""
        if self._stream_connection_thread:
            self.logger.debug("Waiting for thread to terminate")
            self._stream_connection_thread = None
            self.logger.debug("Thread terminated")

    def disconnect(self):
        """Request to close the session previously opened with the connect()
        if self._stream_connection is not None:
            self.logger.debug("Closing session to <%s>",
            server_response = self._control({"LS_op": OP_DESTROY})
            # There is no need to explicitly close the connection, since it is
            # handled by thread completion.
            self.logger.info("Closed session to <%s>", self._base_url.geturl())
            self.logger.warning("No connection to Lightstreamer")

    def subscribe(self, subscription):
        """"Perform a subscription request to Lightstreamer Server."""
        # Register the Subscription with a new subscription key
        self._current_subscription_key += 1
        self._subscriptions[self._current_subscription_key] = subscription

        # Send the control request to perform the subscription
        self.logger.debug("Making a new subscription request")
        server_response = self._control({
            "LS_Table": self._current_subscription_key,
            "LS_op": OP_ADD,
            "LS_data_adapter": subscription.adapter,
            "LS_mode": subscription.mode,
            "LS_schema": " ".join(subscription.field_names),
            "LS_id": " ".join(subscription.item_names),
        if server_response.startswith(OK_CMD):
            self.logger.info("Successfully subscribed ")
            self.logger.warning("Subscription error:"+ str(server_response))
        return self._current_subscription_key

    def unsubscribe(self, subcription_key):
        """Unregister the Subscription associated to the
        specified subscription_key.
        self.logger.debug("Making an unsubscription request")
        if subcription_key in self._subscriptions:
            server_response = self._control({
                "LS_Table": subcription_key,
                "LS_op": OP_DELETE

            if server_response == OK_CMD:
                del self._subscriptions[subcription_key]
                self.logger.info("Successfully unsubscribed")
                self.logger.warning("Unsubscription error")
                "No subscription key %s found!", subcription_key)

    def _forward_update_message(self, update_message):
        """Forwards the real time update to the relative
        Subscription instance for further dispatching to its listeners.
        self.logger.debug("Received update message: <%s>", update_message)
            tok = update_message.split(',', 1)
            table, item = int(tok[0]), tok[1]
            if table in self._subscriptions:
                self.logger.warning("No subscription found!")
        except Exception:

    def _receive(self):
        rebind = False
        receive = True
        while receive:
            self.logger.debug("Waiting for a new message")
                message = self._read_from_stream()
                self.logger.debug("Received message: <%s>", message)
                if not message.strip():
                    message = None
            except Exception:
                self.logger.error("Communication error")
                message = None

            if message is None:
                receive = False
                self.logger.warning("No new message received")
            elif message == PROBE_CMD:
                # Skipping the PROBE message, keep on receiving messages.
                self.logger.debug("PROBE message")
            elif message.startswith(ERROR_CMD):
                # Terminate the receiving loop on ERROR message
                receive = False
            elif message.startswith(LOOP_CMD):
                # Terminate the the receiving loop on LOOP message.
                # A complete implementation should proceed with
                # a rebind of the session.
                receive = False
                rebind = True
            elif message.startswith(SYNC_ERROR_CMD):
                # Terminate the receiving loop on SYNC ERROR message.
                # A complete implementation should create a new session
                # and re-subscribe to all the old items and relative fields.
                self.logger.error("SYNC ERROR")
                receive = False
            elif message.startswith(END_CMD):
                # Terminate the receiving loop on END message.
                # The session has been forcibly closed on the server side.
                # A complete implementation should handle the
                # "cause_code" if present.
                self.logger.info("Connection closed by the server")
                receive = False
            elif message.startswith("Preamble"):
                # Skipping Preamble message, keep on receiving messages.

        if not rebind:
            self.logger.debug("No rebind to <%s>, clearing internal session data",
            # Clear internal data structures for session
            # and subscriptions management.
            self._stream_connection = None
            self._current_subscription_key = 0
            self.logger.debug("Binding to this active session")

def authenticate(configs, demo):
    # IG rest API parameters
    if demo:
        rest_api_key = configs.ig_demo.access
        rest_identifier = configs.ig_demo["rest_user"]
        rest_password = configs.ig_demo["rest_pass"]
        rest_api_key = configs.ig.access
        rest_identifier = configs.ig["rest_user"]
        rest_password = configs.ig["rest_pass"]

    # IG rest login request
    ig_domain = "https://demo-api.ig.com/" if demo else "https://api.ig.com/"
    rest_url = ig_domain + "gateway/deal/session"

    headers = {}
    headers["Content-Type"] = "application/json; charset=UTF-8"
    headers["Accept"] = "application/json; charset=UTF-8"
    headers["Version"] = "2"
    headers["X-IG-API-KEY"] = rest_api_key

    request_json = {}
    request_json["identifier"] = rest_identifier
    request_json["password"] = rest_password

    rest_response = requests.request(
        "POST", rest_url, headers=headers, json=request_json)
    if rest_response.status_code != 200:
        print("error", rest_response.status_code, rest_url, rest_response.text)

    # collect params from IG rest login response

    xst = rest_response.headers["X-SECURITY-TOKEN"]
    cst = rest_response.headers["CST"]
    pwd = 'CST-' + cst + '|XST-' + xst
    return pwd

if __name__ == '__main__':
    demo = False
    # Establishing a new connection to Lightstreamer Server
    print("Starting connection")
    # lightstreamer_client = LSClient("http://localhost:8080", "DEMO")
    # lightstreamer_client = LSClient("http://push.lightstreamer.com", "DEMO")
    # Establishing a new connection to Lightstreamer Server

    IG_url = "https://demo-apd.marketdatasystems.com" if demo else "https://apd.marketdatasystems.com"

    configs = load_config.load_config()
    usr = configs.ig_demo["rest_user"] if demo else configs.ig["rest_user"]
    pwd = authenticate(configs, demo)

    lightstreamer_client = LSClient(base_url=IG_url,

    except Exception as e:
        print("Unable to connect to Lightstreamer Server")
    fields = ["stock_name", "last_price", "time", "bid", "ask"]
    fields = ["UPDATE_TIME", "bid", "offer"]
    fields = ["offer"]
    fields = [field.upper() for field in fields]
    # Making a new Subscription in MERGE mode
    subscription = Subscription(
        # items=["EURJPY", "USDJPY"],
        # items=["CS.D.USDJPY.CFD.IP", "CS.D.EURJPY.CFD.IP"],

    # A simple function acting as a Subscription listener

    def on_item_update(item_update):
        print("{stock_name:<19}: Last{last_price:>6} - Time {time:<8} - "
              "Bid {bid:>5} - Ask {ask:>5}".format(**item_update["values"]))

    # Adding the "on_item_update" function to Subscription

    # Registering the Subscription
    sub_key = lightstreamer_client.subscribe(subscription)


    # Unsubscribing from Lightstreamer by using the subscription key

    # Disconnecting
Link to comment

0 answers to this question

Recommended Posts

There have been no answers to this question yet

Create an account or sign in to comment

You need to be a member in order to leave a comment

Create an account

Sign up for a new account in our community. It's easy!

Register a new account

Sign in

Already have an account? Sign in here.

Sign In Now
  • Posts

    • Sainsburys full year earnings and Unilever’s first quarter trading update both say the same thing, UK consumers are in for higher prices. The war in Ukraine, supply chain issues and the effects of ongoing Covid all to blame.      
    • US Dollar (DXY) Daily Price and Analysis US Q1 GDP may stall the greenback’s advance. A 20-year high nears for the US dollar. The multi-month US dollar rally continues with the greenback printing a fresh high today ahead of the first look at US Q1 GDP at 12.30 GMT. The US dollar basket (DXY) has been boosted by renewed weakness in the Euro and the Japanese Yen, as investors move from lower-yielding to higher-yielding currencies, while safe-haven flows continue to benefit the greenback. The US growth release later in the session is expected to show a sharp slowdown from the robust Q4 figure of 6.9%. The markets are currently pricing in growth of just 1% for the first three months of this year, with the slowdown mainly due to a reduction in inventory accrual over the quarter. This release is unlikely to move the greenback, unless there is a large miss or beat, as the Fed believe that 2022 US growth will be robust enough to let them tighten monetary policy sharply without damaging the economy. The latest US Core PCE data – the Fed’s preferred inflation reading – is released on Friday and this may have more effect on the US dollar than today’s GDP data. For all market moving economic data and events, see the DailyFX Calendar. The ongoing US dollar rally has been aided by weakness across a range of G7 currencies including the Euro, the Japanese Yen, and the British Pound. The Euro continues to battle with lowly growth expectations, exacerbated by energy concerns, the British Pound is mired by weak economic data, while the Japanese Yen is in freefall as the BoJ continues with its ultra-loose monetary policy.   The US dollar continues to press higher and looks set to break above 103.96, the March 2020 high. Above here the US dollar would be back at levels last seen nearly two decades ago. The March resistance will likely hold in the short-term, especially with month-end portfolio rebalancing at the end of the week, but US dollar strength is set to continue in the months ahead. USDOLLAR (DXY) WEEKLY PRICE CHART – APRIL 28, 2022 {{THE_FUNDAMENTALS_OF_BREAKOUT_TRADING}} What is your view on the US Dollar – bullish or bearish?   Apr 28, 2022 | DailyFX Nick Cawley, Strategist
    • While Tesla has nothing directly to do with Elon Musk buying Twitter - TSLA stock closed down 12% on news that Musk may have to sell stock and use other holdings to stand against the loan to finalise the purchase of the social media giant.        
  • Create New...