truefix_ig_client/
streaming.rs

1//! IG Lightstreamer integration built on the TLCP protocol client.
2
3use lightstreamer_rs::{
4    Client, ClientConfig, Credentials, FieldSchema, ItemGroup, ServerAddress, Snapshot,
5    Subscription, SubscriptionMode, Updates,
6};
7
8use crate::error::IgResult;
9
10// Keep the top-of-book subscription to IG's documented portable field set.
11// Optional fields differ by account and instrument and IG rejects the entire
12// subscription when even one requested field is unavailable.
13const MARKET_FIELDS: [&str; 5] = [
14    "BID",
15    "OFFER",
16    "UPDATE_TIME",
17    "MARKET_STATE",
18    "MARKET_DELAY",
19];
20const MARKET_LADDER_FIELDS: [&str; 20] = [
21    "BIDPRICE1",
22    "BIDPRICE2",
23    "BIDPRICE3",
24    "BIDPRICE4",
25    "BIDPRICE5",
26    "ASKPRICE1",
27    "ASKPRICE2",
28    "ASKPRICE3",
29    "ASKPRICE4",
30    "ASKPRICE5",
31    "BIDSIZE1",
32    "BIDSIZE2",
33    "BIDSIZE3",
34    "BIDSIZE4",
35    "BIDSIZE5",
36    "ASKSIZE1",
37    "ASKSIZE2",
38    "ASKSIZE3",
39    "ASKSIZE4",
40    "ASKSIZE5",
41];
42const ACCOUNT_FIELDS: [&str; 5] = ["PNL", "DEPOSIT", "USED_MARGIN", "AVAILABLE_CASH", "FUNDS"];
43const TRADE_FIELDS: [&str; 3] = ["CONFIRMS", "OPU", "WOU"];
44
45/// An authenticated IG Lightstreamer session.
46///
47/// Dropping this value disconnects the underlying session and closes its subscriptions.
48pub struct IgStreamingClient {
49    client: Client,
50    account_id: String,
51}
52
53impl std::fmt::Debug for IgStreamingClient {
54    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
55        formatter
56            .debug_struct("IgStreamingClient")
57            .field("account_id", &"<redacted>")
58            .finish_non_exhaustive()
59    }
60}
61
62impl IgStreamingClient {
63    pub(crate) async fn connect(
64        endpoint: &str,
65        account_id: String,
66        cst: String,
67        x_security_token: String,
68    ) -> IgResult<Self> {
69        let password = format!("CST-{cst}|XST-{x_security_token}");
70        let config = ClientConfig::builder(ServerAddress::try_new(endpoint)?)
71            .with_credentials(Credentials::new(&account_id, password))
72            .build()?;
73        let (client, session_events) = Client::connect(config).await?;
74        drop(session_events);
75        Ok(Self { client, account_id })
76    }
77
78    /// Subscribes to live prices for one or more IG epics.
79    pub async fn subscribe_markets<I, S>(&self, epics: I) -> IgResult<Updates>
80    where
81        I: IntoIterator<Item = S>,
82        S: AsRef<str>,
83    {
84        let items = epics
85            .into_iter()
86            .map(|epic| format!("MARKET:{}", epic.as_ref()));
87        self.subscribe(SubscriptionMode::Merge, items, MARKET_FIELDS, Snapshot::On)
88            .await
89    }
90
91    /// Subscribes to IG's optional native five-level price ladder.
92    ///
93    /// Ladder availability is account- and instrument-dependent. Keeping this
94    /// subscription separate prevents an unsupported ladder field set from
95    /// rejecting the ordinary top-of-book market subscription.
96    pub async fn subscribe_market_ladders<I, S>(&self, epics: I) -> IgResult<Updates>
97    where
98        I: IntoIterator<Item = S>,
99        S: AsRef<str>,
100    {
101        let items = epics
102            .into_iter()
103            .map(|epic| format!("MARKET:{}", epic.as_ref()));
104        self.subscribe(
105            SubscriptionMode::Merge,
106            items,
107            MARKET_LADDER_FIELDS,
108            Snapshot::On,
109        )
110        .await
111    }
112
113    /// Subscribes to account balance and margin updates.
114    pub async fn subscribe_account(&self) -> IgResult<Updates> {
115        self.subscribe(
116            SubscriptionMode::Merge,
117            [format!("ACCOUNT:{}", self.account_id)],
118            ACCOUNT_FIELDS,
119            Snapshot::On,
120        )
121        .await
122    }
123
124    /// Subscribes to deal confirmations, position updates, and working-order updates.
125    pub async fn subscribe_trades(&self) -> IgResult<Updates> {
126        self.subscribe(
127            SubscriptionMode::Distinct,
128            [format!("TRADE:{}", self.account_id)],
129            TRADE_FIELDS,
130            Snapshot::On,
131        )
132        .await
133    }
134
135    /// Gracefully closes the Lightstreamer session.
136    pub async fn disconnect(self) -> IgResult<()> {
137        self.client.disconnect().await?;
138        Ok(())
139    }
140
141    async fn subscribe<I, S, F>(
142        &self,
143        mode: SubscriptionMode,
144        items: I,
145        fields: F,
146        snapshot: Snapshot,
147    ) -> IgResult<Updates>
148    where
149        I: IntoIterator<Item = S>,
150        S: Into<String>,
151        F: IntoIterator,
152        F::Item: Into<String>,
153    {
154        let subscription = Subscription::new(
155            mode,
156            ItemGroup::from_items(items)?,
157            FieldSchema::from_fields(fields)?,
158        )
159        .with_snapshot(snapshot);
160        Ok(self.client.subscribe(subscription).await?)
161    }
162}
truefix_ig_client/streaming.rs

truefix_ig_client/
streaming.rs
