I’m trying to make a connection with lightstreamer using python. While I am able to connect to IG and retrieve account details, I am unable to connect to Lightstreamer and the live prices
My inline debug tells me that it might have connected but the live prices are not streamed and connection details are not available
I have tried the ig_trading simple example but was unable to make that run ither
Is anyone able to have a look at the code and help out pls
#Using Lightstreamer SDK Version: 2.2.0
import requests
import time
import logging
from lightstreamer.client import LightstreamerClient, Subscription, SubscriptionListener
from requests import session
logging.basicConfig(level=logging.INFO)
IG_API_URL = "https://demo-api.ig.com/gateway/deal" # Use "https://api.ig.com/gateway/deal" for live trading
API_KEY = ""
USERNAME = ""
PASSWORD = ""
def ig_auth():
headers = {
"X-IG-API-KEY": API_KEY,
"Content-Type": "application/json",
"Version": "2"
}
data = {
"identifier": USERNAME,
"password": PASSWORD
}
response = requests.post(f"{IG_API_URL}/session", headers=headers, json=data)
if response.status_code != 200:
print("❌ Authentication Failed:", response.json())
return None
session_data = response.json()
auth_data = {
"CST": response.headers.get("CST"),
"X-SECURITY-TOKEN": response.headers.get("X-SECURITY-TOKEN"),
"accountId": session_data.get("currentAccountId"),
"lightstreamerEndpoint": session_data.get("lightstreamerEndpoint", "N/A")
}
if not auth_data["CST"] or not auth_data["X-SECURITY-TOKEN"]:
print("❌ ERROR: CST or X-SECURITY-TOKEN is missing! Authentication failed.")
return None
print(f"🔑 CST: {auth_data['CST']}")
print(f"🔑 X-SECURITY-TOKEN: {auth_data['X-SECURITY-TOKEN']}")
print(f"🆔 Account ID: {auth_data['accountId']}")
print(f"🆔 LS End Point: {auth_data['lightstreamerEndpoint']}")
return auth_data
class PriceUpdateListener(SubscriptionListener):
def onListenStart(self):
print("🔍 Listener started, waiting for updates...")
def onSubscriptionError(self, code, message):
print(f"❌ Subscription Error: Code {code}, Message: {message}")
def onItemUpdate(self, item_update):
print("Item Updated")
""" ✅ This method should be called on every price update """
print(
f"⏱ Time: {item_update.getValue('UPDATE_TIME')} | "
f"📉 Bid: {item_update.getValue('BID')} | "
f"📈 Ask: {item_update.getValue('OFFER')}"
)
class IGStreamClient:
def init(self, auth_headers, epics):
self.auth_headers = auth_headers
self.epics = [f"MARKET:{epic}" for epic in epics]
self.client = None
self.subscription = None
def connect(self):
print("🚀 Connecting to Lightstreamer...")
self.client = LightstreamerClient(
"https://demo-apd.marketdatasystems.com",
"DEFAULT"
)
self.client.connectionDetails.setUser(
f"CST-{self.auth_headers['CST']}|XST-{self.auth_headers['X-SECURITY-TOKEN']}")
self.client.connectionDetails.setPassword(PASSWORD)
try:
self.client.connect()
mystr = self.client.getStatus()
print(f"Connection Status ", mystr)
print("✅ Connected to Lightstreamer")
except Exception as e:
print(f"❌ Connection error: {e}")
return
except Exception as e:
print(f"❌ Lightstreamer Connection Error: {e}")
def disconnect(self):
""" Disconnect from Lightstreamer """
if self.client:
print("🔴 Disconnecting from Lightstreamer...")
try:
self.client.unsubscribe(self.subscription)
self.client.disconnect()
print("✅ Disconnected")
except Exception as e:
print(f"❌ Error while disconnecting: {e}")
if name == "main":
auth_headers = ig_auth()
if not auth_headers:
print("❌ Authentication failed. Exiting...")
exit()
epics = ["IX.D.NASDAQ.CASH.IP", "IX.D.DOW.DAILY.IP:TICK", "CS.D.BITCOIN.CASH.IP"]
print(f"📡 Streaming live prices for: {', '.join(epics)}...")
stream_client = IGStreamClient(auth_headers, epics)
stream_client.connect()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
print("\n🛑 Stopping price streaming...")
stream_client.disconnect()
