truefix_ig_client/
client.rs

1use reqwest::{Method, RequestBuilder, StatusCode};
2use serde::{Serialize, de::DeserializeOwned};
3use std::sync::RwLock;
4use std::time::{Duration, Instant};
5use url::Url;
6
7use crate::{
8    config::{AuthenticationVersion, ClientConfig, Credentials},
9    error::{IgError, IgResult},
10    streaming::IgStreamingClient,
11    types::{
12        AccountsResponse, CreatePositionRequest, CreateWorkingOrderRequest, DealConfirmation,
13        DealReferenceResponse, HistoricalPricesQuery, HistoricalPricesResponse, LoginResponse,
14        MarketDetails, MarketsResponse, OAuthToken, PositionsResponse, SwitchAccountResponse,
15        UpdateWorkingOrderRequest, V3LoginResponse, WorkingOrdersResponse,
16    },
17};
18
19#[derive(Debug, Clone)]
20struct SessionTokens {
21    cst: String,
22    x_security_token: String,
23    account_id: String,
24    lightstreamer_endpoint: String,
25}
26
27#[derive(Debug, Clone)]
28struct OAuthSession {
29    access_token: String,
30    refresh_token: String,
31    account_id: String,
32    lightstreamer_endpoint: String,
33    expires_at: Instant,
34}
35
36#[derive(Debug, Clone)]
37enum Session {
38    V2(SessionTokens),
39    V3(OAuthSession),
40}
41
42const TOKEN_REFRESH_THRESHOLD: Duration = Duration::from_secs(10);
43
44/// IG REST trading client.
45///
46/// Session credentials are kept behind a lock so that one client can be shared among
47/// concurrent read-only tasks after [`Self::login`] succeeds. Calls which change server state are
48/// deliberately made once and are never automatically replayed.
49#[derive(Debug)]
50pub struct IgClient {
51    http: reqwest::Client,
52    config: ClientConfig,
53    base_url: Url,
54    session: RwLock<Option<Session>>,
55}
56
57impl IgClient {
58    /// Constructs a client without contacting IG.
59    pub fn new(config: ClientConfig) -> IgResult<Self> {
60        let base_url = Url::parse(config.environment.rest_base()).map_err(|error| {
61            IgError::InvalidConfiguration(format!("invalid REST base URL: {error}"))
62        })?;
63        let mut http = reqwest::Client::builder().timeout(config.timeout);
64        if let Some(proxy) = &config.proxy {
65            http = http.proxy(reqwest::Proxy::all(proxy).map_err(|error| {
66                IgError::InvalidConfiguration(format!("invalid proxy URL: {error}"))
67            })?);
68        }
69        Ok(Self {
70            http: http.build()?,
71            config,
72            base_url,
73            session: RwLock::new(None),
74        })
75    }
76
77    /// Authenticates using the protocol selected in [`ClientConfig::authentication`].
78    pub async fn login(&self) -> IgResult<LoginResponse> {
79        match &self.config.authentication {
80            AuthenticationVersion::V2 => self.login_v2().await,
81            AuthenticationVersion::V3 { account_id } => self.login_v3(account_id).await,
82        }
83    }
84
85    /// Authenticates with v2 and stores the CST/X-SECURITY-TOKEN pair.
86    pub async fn login_v2(&self) -> IgResult<LoginResponse> {
87        #[derive(Serialize)]
88        struct LoginRequest<'a> {
89            identifier: &'a str,
90            password: &'a str,
91        }
92
93        let credentials = self.credentials()?;
94        let response = self
95            .request(Method::POST, "session", 2, false)?
96            .json(&LoginRequest {
97                identifier: credentials.identifier(),
98                password: credentials.password(),
99            })
100            .send()
101            .await?;
102        let (tokens, login) = self.decode_login(response).await?;
103        *self.session.write().map_err(|_| IgError::MissingSession)? = Some(Session::V2(tokens));
104        Ok(login)
105    }
106
107    /// Authenticates with v3 OAuth and stores refreshable bearer credentials.
108    pub async fn login_v3(&self, account_id: &str) -> IgResult<LoginResponse> {
109        #[derive(Serialize)]
110        #[serde(rename_all = "camelCase")]
111        struct LoginRequest<'a> {
112            identifier: &'a str,
113            password: &'a str,
114            account_id: &'a str,
115        }
116
117        if account_id.is_empty() {
118            return Err(IgError::InvalidConfiguration(
119                "v3 authentication requires a non-empty account ID".to_owned(),
120            ));
121        }
122        let credentials = self.credentials()?;
123        let response = self
124            .request(Method::POST, "session", 3, false)?
125            .header("IG-ACCOUNT-ID", account_id)
126            .json(&LoginRequest {
127                identifier: credentials.identifier(),
128                password: credentials.password(),
129                account_id,
130            })
131            .send()
132            .await?;
133        let response = Self::ensure_success(response).await?;
134        let login: V3LoginResponse = response.json().await?;
135        let oauth = login.oauth_token.ok_or_else(|| {
136            IgError::InvalidConfiguration(
137                "IG v3 login response did not contain an OAuth token".to_owned(),
138            )
139        })?;
140        let selected_account = login
141            .account_id
142            .clone()
143            .unwrap_or_else(|| account_id.to_owned());
144        let result = LoginResponse {
145            current_account_id: login.current_account_id.or(login.account_id),
146            lightstreamer_endpoint: login.lightstreamer_endpoint,
147            client_id: login.client_id,
148            currency_iso_code: login.currency_iso_code,
149            dealing_enabled: login.dealing_enabled,
150        };
151        *self.session.write().map_err(|_| IgError::MissingSession)? =
152            Some(Session::V3(oauth_session(
153                oauth,
154                selected_account,
155                result.lightstreamer_endpoint.clone(),
156            )));
157        Ok(result)
158    }
159
160    /// Deletes the current session and clears local tokens only after IG accepts the request.
161    pub async fn logout(&self) -> IgResult<()> {
162        self.refresh_oauth_if_needed().await?;
163        let response = self
164            .request(Method::DELETE, "session", 1, true)?
165            .send()
166            .await?;
167        Self::ensure_success(response).await?;
168        *self.session.write().map_err(|_| IgError::MissingSession)? = None;
169        Ok(())
170    }
171
172    /// Returns the accounts available to the authenticated client.
173    pub async fn accounts(&self) -> IgResult<AccountsResponse> {
174        self.get("accounts", 1).await
175    }
176
177    /// Switches the active V2 account without changing the user's default.
178    ///
179    /// IG rotates the X-SECURITY-TOKEN when the active account changes. The
180    /// returned header is committed atomically with the local account ID so
181    /// subsequent REST and Lightstreamer calls cannot retain the old account
182    /// token.
183    pub async fn switch_account(&self, account_id: &str) -> IgResult<SwitchAccountResponse> {
184        #[derive(Serialize)]
185        #[serde(rename_all = "camelCase")]
186        struct SwitchRequest<'a> {
187            account_id: &'a str,
188            default_account: bool,
189        }
190
191        if account_id.trim().is_empty() {
192            return Err(IgError::InvalidConfiguration(
193                "account switch requires a non-empty account ID".to_owned(),
194            ));
195        }
196        let response = self
197            .request(Method::PUT, "session", 1, true)?
198            .json(&SwitchRequest {
199                account_id,
200                default_account: false,
201            })
202            .send()
203            .await?;
204        let response = Self::ensure_success(response).await?;
205        let next_token = response
206            .headers()
207            .get("X-SECURITY-TOKEN")
208            .and_then(|value| value.to_str().ok())
209            .map(str::to_owned);
210        let result: SwitchAccountResponse = response.json().await?;
211        let mut session = self.session.write().map_err(|_| IgError::MissingSession)?;
212        let Some(Session::V2(tokens)) = session.as_mut() else {
213            return Err(IgError::InvalidConfiguration(
214                "account switching requires a V2 CST/XST session".to_owned(),
215            ));
216        };
217        tokens.account_id = account_id.to_owned();
218        if let Some(token) = next_token {
219            tokens.x_security_token = token;
220        }
221        Ok(result)
222    }
223
224    /// Returns all open positions for the active account.
225    pub async fn positions(&self) -> IgResult<PositionsResponse> {
226        self.get("positions", 2).await
227    }
228
229    /// Returns market metadata and a latest price snapshot for one IG epic.
230    pub async fn market(&self, epic: &str) -> IgResult<MarketDetails> {
231        self.get(&format!("markets/{}", encode_path_segment(epic)), 3)
232            .await
233    }
234
235    /// Searches the market catalogue using IG's `/markets` endpoint.
236    pub async fn search_markets(&self, query: &str) -> IgResult<MarketsResponse> {
237        self.refresh_oauth_if_needed().await?;
238        let mut url = self.url("markets")?;
239        url.query_pairs_mut().append_pair("searchTerm", query);
240        self.send_json(self.request_url(Method::GET, url, 1, true)?)
241            .await
242    }
243
244    /// Returns historical prices for an epic at the requested IG resolution.
245    pub async fn historical_prices(
246        &self,
247        epic: &str,
248        query: HistoricalPricesQuery<'_>,
249    ) -> IgResult<HistoricalPricesResponse> {
250        self.refresh_oauth_if_needed().await?;
251        let mut url = self.url(&format!("prices/{}", encode_path_segment(epic)))?;
252        {
253            let mut pairs = url.query_pairs_mut();
254            pairs.append_pair("resolution", query.resolution);
255            if let Some(value) = query.from {
256                pairs.append_pair("from", value);
257            }
258            if let Some(value) = query.to {
259                pairs.append_pair("to", value);
260            }
261            if let Some(value) = query.max {
262                pairs.append_pair("max", &value.to_string());
263            }
264        }
265        self.send_json(self.request_url(Method::GET, url, 3, true)?)
266            .await
267    }
268
269    /// Creates a position. This operation is never automatically retried.
270    pub async fn create_position(
271        &self,
272        request: &CreatePositionRequest,
273    ) -> IgResult<DealReferenceResponse> {
274        self.refresh_oauth_if_needed().await?;
275        let response = self
276            .request(Method::POST, "positions/otc", 2, true)?
277            .json(request)
278            .send()
279            .await?;
280        Self::decode_json(response).await
281    }
282
283    /// Resolves a submission acknowledgement into IG's authoritative deal
284    /// confirmation. Callers must not treat the initial deal reference as an
285    /// accepted order.
286    pub async fn deal_confirmation(&self, deal_reference: &str) -> IgResult<DealConfirmation> {
287        self.get(
288            &format!("confirms/{}", encode_path_segment(deal_reference)),
289            1,
290        )
291        .await
292    }
293
294    /// Returns every working order for the active account.
295    pub async fn working_orders(&self) -> IgResult<WorkingOrdersResponse> {
296        self.get("workingorders", 2).await
297    }
298
299    /// Creates a working order. Resolve the returned reference with
300    /// [`Self::deal_confirmation`] before treating it as accepted.
301    pub async fn create_working_order(
302        &self,
303        request: &CreateWorkingOrderRequest,
304    ) -> IgResult<DealReferenceResponse> {
305        self.mutate_json(Method::POST, "workingorders/otc", 2, request)
306            .await
307    }
308
309    /// Updates an existing working order. Writes are never automatically replayed.
310    pub async fn update_working_order(
311        &self,
312        deal_id: &str,
313        request: &UpdateWorkingOrderRequest,
314    ) -> IgResult<DealReferenceResponse> {
315        self.mutate_json(
316            Method::PUT,
317            &format!("workingorders/otc/{}", encode_path_segment(deal_id)),
318            2,
319            request,
320        )
321        .await
322    }
323
324    /// Deletes an existing working order. Writes are never automatically replayed.
325    pub async fn delete_working_order(&self, deal_id: &str) -> IgResult<DealReferenceResponse> {
326        self.refresh_oauth_if_needed().await?;
327        let response = self
328            .request(
329                Method::DELETE,
330                &format!("workingorders/otc/{}", encode_path_segment(deal_id)),
331                2,
332                true,
333            )?
334            .send()
335            .await?;
336        Self::decode_json(response).await
337    }
338
339    /// Opens IG's Lightstreamer endpoint using credentials issued by v2 login.
340    pub async fn connect_streaming(&self) -> IgResult<IgStreamingClient> {
341        enum StreamingCredentials {
342            Ready(SessionTokens),
343            Fetch {
344                account_id: String,
345                endpoint: String,
346            },
347        }
348        let credentials = {
349            let session = self.session.read().map_err(|_| IgError::MissingSession)?;
350            match session.as_ref() {
351                Some(Session::V2(tokens)) => StreamingCredentials::Ready(tokens.clone()),
352                Some(Session::V3(oauth)) => StreamingCredentials::Fetch {
353                    account_id: oauth.account_id.clone(),
354                    endpoint: oauth.lightstreamer_endpoint.clone(),
355                },
356                None => return Err(IgError::MissingSession),
357            }
358        };
359        let tokens = match credentials {
360            StreamingCredentials::Ready(tokens) => tokens,
361            StreamingCredentials::Fetch {
362                account_id,
363                endpoint,
364            } => self.fetch_streaming_tokens(account_id, endpoint).await?,
365        };
366        IgStreamingClient::connect(
367            &tokens.lightstreamer_endpoint,
368            tokens.account_id,
369            tokens.cst,
370            tokens.x_security_token,
371        )
372        .await
373    }
374
375    async fn fetch_streaming_tokens(
376        &self,
377        account_id: String,
378        lightstreamer_endpoint: String,
379    ) -> IgResult<SessionTokens> {
380        self.refresh_oauth_if_needed().await?;
381        let mut url = self.url("session")?;
382        url.query_pairs_mut()
383            .append_pair("fetchSessionTokens", "true");
384        let response = self.request_url(Method::GET, url, 1, true)?.send().await?;
385        let response = Self::ensure_success(response).await?;
386        let cst = response
387            .headers()
388            .get("CST")
389            .and_then(|value| value.to_str().ok())
390            .ok_or(IgError::MissingToken("CST"))?
391            .to_owned();
392        let x_security_token = response
393            .headers()
394            .get("X-SECURITY-TOKEN")
395            .and_then(|value| value.to_str().ok())
396            .ok_or(IgError::MissingToken("X-SECURITY-TOKEN"))?
397            .to_owned();
398        Ok(SessionTokens {
399            cst,
400            x_security_token,
401            account_id,
402            lightstreamer_endpoint,
403        })
404    }
405
406    async fn mutate_json<B: Serialize + ?Sized, T: DeserializeOwned>(
407        &self,
408        method: Method,
409        path: &str,
410        version: u8,
411        body: &B,
412    ) -> IgResult<T> {
413        self.refresh_oauth_if_needed().await?;
414        let response = self
415            .request(method, path, version, true)?
416            .json(body)
417            .send()
418            .await?;
419        Self::decode_json(response).await
420    }
421
422    fn credentials(&self) -> IgResult<&Credentials> {
423        self.config
424            .credentials
425            .as_ref()
426            .ok_or(IgError::MissingCredentials)
427    }
428
429    fn url(&self, path: &str) -> IgResult<Url> {
430        let mut url = self.base_url.clone();
431        url.set_path(&format!(
432            "{}/{}",
433            self.base_url.path().trim_end_matches('/'),
434            path.trim_start_matches('/')
435        ));
436        url.set_query(None);
437        Ok(url)
438    }
439
440    fn request(
441        &self,
442        method: Method,
443        path: &str,
444        version: u8,
445        include_session: bool,
446    ) -> IgResult<RequestBuilder> {
447        self.request_url(method, self.url(path)?, version, include_session)
448    }
449
450    fn request_url(
451        &self,
452        method: Method,
453        url: Url,
454        version: u8,
455        include_session: bool,
456    ) -> IgResult<RequestBuilder> {
457        let credentials = self.credentials()?;
458        let mut request = self
459            .http
460            .request(method, url)
461            .header("X-IG-API-KEY", credentials.api_key())
462            .header("Accept", "application/json")
463            .header("Content-Type", "application/json")
464            .header("Version", version.to_string());
465        if include_session {
466            let session = self.session.read().map_err(|_| IgError::MissingSession)?;
467            let session = session.as_ref().ok_or(IgError::MissingSession)?;
468            request = match session {
469                Session::V2(tokens) => request
470                    .header("CST", &tokens.cst)
471                    .header("X-SECURITY-TOKEN", &tokens.x_security_token),
472                Session::V3(oauth) => request
473                    .header("Authorization", format!("Bearer {}", oauth.access_token))
474                    .header("IG-ACCOUNT-ID", &oauth.account_id),
475            };
476        }
477        Ok(request)
478    }
479
480    async fn get<T: DeserializeOwned>(&self, path: &str, version: u8) -> IgResult<T> {
481        self.refresh_oauth_if_needed().await?;
482        self.send_json(self.request(Method::GET, path, version, true)?)
483            .await
484    }
485
486    async fn send_json<T: DeserializeOwned>(&self, request: RequestBuilder) -> IgResult<T> {
487        Self::decode_json(request.send().await?).await
488    }
489
490    async fn decode_login(
491        &self,
492        response: reqwest::Response,
493    ) -> IgResult<(SessionTokens, LoginResponse)> {
494        let response = Self::ensure_success(response).await?;
495        let cst = response
496            .headers()
497            .get("CST")
498            .and_then(|value| value.to_str().ok())
499            .ok_or(IgError::MissingToken("CST"))?
500            .to_owned();
501        let x_security_token = response
502            .headers()
503            .get("X-SECURITY-TOKEN")
504            .and_then(|value| value.to_str().ok())
505            .ok_or(IgError::MissingToken("X-SECURITY-TOKEN"))?
506            .to_owned();
507        let login: LoginResponse = response.json().await?;
508        let account_id = login.current_account_id.clone().ok_or_else(|| {
509            IgError::InvalidConfiguration(
510                "IG v2 login response did not contain currentAccountId".to_owned(),
511            )
512        })?;
513        let lightstreamer_endpoint = login.lightstreamer_endpoint.clone();
514        Ok((
515            SessionTokens {
516                cst,
517                x_security_token,
518                account_id,
519                lightstreamer_endpoint,
520            },
521            login,
522        ))
523    }
524
525    async fn refresh_oauth_if_needed(&self) -> IgResult<()> {
526        let refresh_token = {
527            let session = self.session.read().map_err(|_| IgError::MissingSession)?;
528            match session.as_ref() {
529                Some(Session::V3(oauth))
530                    if oauth.expires_at.saturating_duration_since(Instant::now())
531                        <= TOKEN_REFRESH_THRESHOLD =>
532                {
533                    Some(oauth.refresh_token.clone())
534                }
535                _ => None,
536            }
537        };
538        let Some(refresh_token) = refresh_token else {
539            return Ok(());
540        };
541
542        #[derive(Serialize)]
543        #[serde(rename_all = "camelCase")]
544        struct RefreshTokenRequest<'a> {
545            refresh_token: &'a str,
546        }
547
548        let response = self
549            .request(Method::POST, "session/refresh-token", 1, false)?
550            .json(&RefreshTokenRequest {
551                refresh_token: &refresh_token,
552            })
553            .send()
554            .await?;
555        let oauth: OAuthToken = Self::decode_json(response).await?;
556        let mut session = self.session.write().map_err(|_| IgError::MissingSession)?;
557        let Some(Session::V3(current)) = session.as_mut() else {
558            return Ok(());
559        };
560        current.access_token = oauth.access_token;
561        current.refresh_token = oauth.refresh_token;
562        current.expires_at = expiry_from(oauth.expires_in);
563        Ok(())
564    }
565
566    async fn decode_json<T: DeserializeOwned>(response: reqwest::Response) -> IgResult<T> {
567        Self::ensure_success(response)
568            .await?
569            .json()
570            .await
571            .map_err(IgError::from)
572    }
573
574    async fn ensure_success(response: reqwest::Response) -> IgResult<reqwest::Response> {
575        if response.status().is_success() {
576            return Ok(response);
577        }
578        let status = response.status();
579        let body = response.text().await.unwrap_or_default();
580        Err(exchange_error(status, &body))
581    }
582}
583
584fn exchange_error(status: StatusCode, body: &str) -> IgError {
585    #[derive(serde::Deserialize)]
586    #[serde(rename_all = "camelCase")]
587    struct ErrorBody {
588        error_code: Option<String>,
589        message: Option<String>,
590    }
591    let parsed = serde_json::from_str::<ErrorBody>(body).ok();
592    let code = parsed
593        .as_ref()
594        .and_then(|value| value.error_code.clone())
595        .unwrap_or_else(|| "unknown".to_owned());
596    let message = parsed
597        .and_then(|value| value.message)
598        .unwrap_or_else(|| body.to_owned());
599    IgError::Exchange {
600        status: status.as_u16(),
601        code,
602        message,
603    }
604}
605
606fn oauth_session(
607    token: OAuthToken,
608    account_id: String,
609    lightstreamer_endpoint: String,
610) -> OAuthSession {
611    OAuthSession {
612        access_token: token.access_token,
613        refresh_token: token.refresh_token,
614        account_id,
615        lightstreamer_endpoint,
616        expires_at: expiry_from(token.expires_in),
617    }
618}
619
620fn expiry_from(expires_in: u64) -> Instant {
621    Instant::now() + Duration::from_secs(expires_in.max(1))
622}
623
624fn encode_path_segment(value: &str) -> String {
625    let mut encoded = String::with_capacity(value.len());
626    for byte in value.bytes() {
627        if byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b'~') {
628            encoded.push(char::from(byte));
629        } else {
630            encoded.push_str(&format!("%{byte:02X}"));
631        }
632    }
633    encoded
634}
635
636#[cfg(test)]
637mod tests {
638    use super::*;
639    use crate::config::Environment;
640
641    #[test]
642    fn custom_base_preserves_gateway_path() {
643        let config = ClientConfig {
644            environment: Environment::Custom {
645                rest_base: "http://127.0.0.1:8080/gateway/deal".to_owned(),
646            },
647            ..ClientConfig::default()
648        };
649        let client = IgClient::new(config).unwrap();
650        assert_eq!(
651            client.url("session").unwrap().as_str(),
652            "http://127.0.0.1:8080/gateway/deal/session"
653        );
654    }
655
656    #[test]
657    fn epics_are_encoded_as_one_path_segment() {
658        assert_eq!(encode_path_segment("CS.D/EUR USD"), "CS.D%2FEUR%20USD");
659    }
660
661    #[test]
662    fn encoded_epic_is_not_encoded_twice_in_a_url() {
663        let config = ClientConfig {
664            environment: Environment::Custom {
665                rest_base: "http://127.0.0.1:8080/gateway/deal".to_owned(),
666            },
667            ..ClientConfig::default()
668        };
669        let client = IgClient::new(config).unwrap();
670        assert_eq!(
671            client.url("markets/CS.D%2FEUR").unwrap().path(),
672            "/gateway/deal/markets/CS.D%2FEUR"
673        );
674    }
675}
truefix_ig_client/client.rs

truefix_ig_client/
client.rs
