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
10const MARKET_FIELDS: [&str; 11] = [
11    "BID",
12    "OFFER",
13    "HIGH",
14    "LOW",
15    "MID_OPEN",
16    "CHANGE",
17    "CHANGE_PCT",
18    "UPDATE_TIME",
19    "MARKET_STATE",
20    "MARKET_DELAY",
21    "LTV",
22];
23const ACCOUNT_FIELDS: [&str; 5] = ["PNL", "DEPOSIT", "USED_MARGIN", "AVAILABLE_CASH", "FUNDS"];
24const TRADE_FIELDS: [&str; 3] = ["CONFIRMS", "OPU", "WOU"];
25
26/// An authenticated IG Lightstreamer session.
27///
28/// Dropping this value disconnects the underlying session and closes its subscriptions.
29pub struct IgStreamingClient {
30    client: Client,
31    account_id: String,
32}
33
34impl std::fmt::Debug for IgStreamingClient {
35    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
36        formatter
37            .debug_struct("IgStreamingClient")
38            .field("account_id", &"<redacted>")
39            .finish_non_exhaustive()
40    }
41}
42
43impl IgStreamingClient {
44    pub(crate) async fn connect(
45        endpoint: &str,
46        account_id: String,
47        cst: String,
48        x_security_token: String,
49    ) -> IgResult<Self> {
50        let password = format!("CST-{cst}|XST-{x_security_token}");
51        let config = ClientConfig::builder(ServerAddress::try_new(endpoint)?)
52            .with_credentials(Credentials::new(&account_id, password))
53            .build()?;
54        let (client, session_events) = Client::connect(config).await?;
55        drop(session_events);
56        Ok(Self { client, account_id })
57    }
58
59    /// Subscribes to live prices for one or more IG epics.
60    pub async fn subscribe_markets<I, S>(&self, epics: I) -> IgResult<Updates>
61    where
62        I: IntoIterator<Item = S>,
63        S: AsRef<str>,
64    {
65        let items = epics
66            .into_iter()
67            .map(|epic| format!("MARKET:{}", epic.as_ref()));
68        self.subscribe(SubscriptionMode::Merge, items, MARKET_FIELDS, Snapshot::On)
69            .await
70    }
71
72    /// Subscribes to account balance and margin updates.
73    pub async fn subscribe_account(&self) -> IgResult<Updates> {
74        self.subscribe(
75            SubscriptionMode::Merge,
76            [format!("ACCOUNT:{}", self.account_id)],
77            ACCOUNT_FIELDS,
78            Snapshot::On,
79        )
80        .await
81    }
82
83    /// Subscribes to deal confirmations, position updates, and working-order updates.
84    pub async fn subscribe_trades(&self) -> IgResult<Updates> {
85        self.subscribe(
86            SubscriptionMode::Distinct,
87            [format!("TRADE:{}", self.account_id)],
88            TRADE_FIELDS,
89            Snapshot::On,
90        )
91        .await
92    }
93
94    /// Gracefully closes the Lightstreamer session.
95    pub async fn disconnect(self) -> IgResult<()> {
96        self.client.disconnect().await?;
97        Ok(())
98    }
99
100    async fn subscribe<I, S, F>(
101        &self,
102        mode: SubscriptionMode,
103        items: I,
104        fields: F,
105        snapshot: Snapshot,
106    ) -> IgResult<Updates>
107    where
108        I: IntoIterator<Item = S>,
109        S: Into<String>,
110        F: IntoIterator,
111        F::Item: Into<String>,
112    {
113        let subscription = Subscription::new(
114            mode,
115            ItemGroup::from_items(items)?,
116            FieldSchema::from_fields(fields)?,
117        )
118        .with_snapshot(snapshot);
119        Ok(self.client.subscribe(subscription).await?)
120    }
121}
truefix_ig_client/streaming.rs

truefix_ig_client/
streaming.rs
