ig_client/application/
streaming_convert.rs

1/******************************************************************************
2   Author: Joaquín Béjar García
3   Email: jb@taunais.com
4   Date: 20/10/25
5******************************************************************************/
6
7//! Adapters from Lightstreamer `ItemUpdate` to the presentation-layer DTOs.
8//!
9//! The presentation DTOs stay transport-agnostic: they expose pure
10//! `from_fields(..)` constructors that take plain field maps. This module is the
11//! only place that depends on `lightstreamer_rs`; it reads an `ItemUpdate` and
12//! feeds the extracted metadata / field maps into those pure constructors.
13//!
14//! Both a `Result`-returning function (for callers that want to observe parse
15//! failures) and the `From<&ItemUpdate>` conversions (which degrade to a default
16//! on failure, preserving the previous streaming behaviour) are provided per
17//! type.
18
19use crate::error::AppError;
20use crate::presentation::account::AccountData;
21use crate::presentation::chart::ChartData;
22use crate::presentation::market::{MarketFields, PresentationMarketData};
23use crate::presentation::price::PriceData;
24use crate::presentation::trade::TradeData;
25use lightstreamer_rs::subscription::ItemUpdate;
26use std::collections::HashMap;
27
28/// Converts Lightstreamer's `changed_fields` (`HashMap<String, String>`) into the
29/// `HashMap<String, Option<String>>` shape the pure field parsers consume, so a
30/// changed value is treated identically to a present full-snapshot value.
31#[must_use]
32fn changed_fields_as_options(changed: &HashMap<String, String>) -> HashMap<String, Option<String>> {
33    changed
34        .iter()
35        .map(|(key, value)| (key.clone(), Some(value.clone())))
36        .collect()
37}
38
39/// Converts a Lightstreamer `ItemUpdate` into a [`PriceData`].
40///
41/// # Errors
42/// Returns [`AppError::Deserialization`] when a field fails to parse (e.g. a
43/// non-numeric price or an unknown dealing flag). The message names the
44/// offending field and value.
45#[must_use = "the parse result must be handled"]
46pub fn price_data_from_item_update(item_update: &ItemUpdate) -> Result<PriceData, AppError> {
47    let changed_fields = changed_fields_as_options(&item_update.changed_fields);
48    PriceData::from_fields(
49        item_update.item_name.as_deref(),
50        item_update.item_pos,
51        item_update.is_snapshot,
52        &item_update.fields,
53        &changed_fields,
54    )
55    .map_err(AppError::Deserialization)
56}
57
58impl From<&ItemUpdate> for PriceData {
59    fn from(item_update: &ItemUpdate) -> Self {
60        price_data_from_item_update(item_update).unwrap_or_else(|e| {
61            tracing::warn!(error = %e, "failed to convert ItemUpdate to PriceData, returning default");
62            PriceData::default()
63        })
64    }
65}
66
67/// Converts a Lightstreamer `ItemUpdate` into a [`PresentationMarketData`].
68///
69/// # Errors
70/// Returns [`AppError::Deserialization`] when a field fails to parse (e.g. an
71/// unknown market state or an invalid `MARKET_DELAY` value).
72#[must_use = "the parse result must be handled"]
73pub fn market_data_from_item_update(
74    item_update: &ItemUpdate,
75) -> Result<PresentationMarketData, AppError> {
76    let changed_fields = changed_fields_as_options(&item_update.changed_fields);
77    PresentationMarketData::from_fields(
78        item_update.item_name.as_deref(),
79        item_update.item_pos,
80        item_update.is_snapshot,
81        &item_update.fields,
82        &changed_fields,
83    )
84    .map_err(AppError::Deserialization)
85}
86
87impl From<&ItemUpdate> for PresentationMarketData {
88    fn from(item_update: &ItemUpdate) -> Self {
89        market_data_from_item_update(item_update).unwrap_or_else(|_| PresentationMarketData {
90            item_name: String::new(),
91            item_pos: 0,
92            fields: MarketFields::default(),
93            changed_fields: MarketFields::default(),
94            is_snapshot: false,
95        })
96    }
97}
98
99/// Converts a Lightstreamer `ItemUpdate` into a [`ChartData`].
100///
101/// # Errors
102/// Returns [`AppError::Deserialization`] when a field fails to parse (e.g. a
103/// non-numeric candle value).
104#[must_use = "the parse result must be handled"]
105pub fn chart_data_from_item_update(item_update: &ItemUpdate) -> Result<ChartData, AppError> {
106    let changed_fields = changed_fields_as_options(&item_update.changed_fields);
107    ChartData::from_fields(
108        item_update.item_name.as_deref(),
109        item_update.item_pos,
110        item_update.is_snapshot,
111        &item_update.fields,
112        &changed_fields,
113    )
114    .map_err(AppError::Deserialization)
115}
116
117impl From<&ItemUpdate> for ChartData {
118    fn from(item_update: &ItemUpdate) -> Self {
119        chart_data_from_item_update(item_update).unwrap_or_default()
120    }
121}
122
123/// Converts a Lightstreamer `ItemUpdate` into a [`TradeData`].
124///
125/// # Errors
126/// Returns [`AppError::Deserialization`] when the embedded OPU / WOU JSON
127/// payload fails to parse.
128#[must_use = "the parse result must be handled"]
129pub fn trade_data_from_item_update(item_update: &ItemUpdate) -> Result<TradeData, AppError> {
130    let changed_fields = changed_fields_as_options(&item_update.changed_fields);
131    TradeData::from_fields(
132        item_update.item_name.as_deref(),
133        item_update.item_pos,
134        item_update.is_snapshot,
135        &item_update.fields,
136        &changed_fields,
137    )
138    .map_err(AppError::Deserialization)
139}
140
141impl From<&ItemUpdate> for TradeData {
142    fn from(item_update: &ItemUpdate) -> Self {
143        trade_data_from_item_update(item_update).unwrap_or_default()
144    }
145}
146
147/// Converts a Lightstreamer `ItemUpdate` into an [`AccountData`].
148///
149/// # Errors
150/// Returns [`AppError::Deserialization`] when a field fails to parse (e.g. a
151/// non-numeric P&L or margin value).
152#[must_use = "the parse result must be handled"]
153pub fn account_data_from_item_update(item_update: &ItemUpdate) -> Result<AccountData, AppError> {
154    let changed_fields = changed_fields_as_options(&item_update.changed_fields);
155    AccountData::from_fields(
156        item_update.item_name.as_deref(),
157        item_update.item_pos,
158        item_update.is_snapshot,
159        &item_update.fields,
160        &changed_fields,
161    )
162    .map_err(AppError::Deserialization)
163}
164
165impl From<&ItemUpdate> for AccountData {
166    fn from(item_update: &ItemUpdate) -> Self {
167        account_data_from_item_update(item_update).unwrap_or_else(|_| AccountData::default())
168    }
169}
170
171#[cfg(test)]
172mod tests {
173    use super::*;
174    use crate::presentation::order::{Direction, OrderType, Status, TimeInForce};
175
176    /// Builds an `ItemUpdate` carrying a single TRADE field, mirroring how the
177    /// Lightstreamer TRADE subscription delivers OPU / WOU / CONFIRMS payloads.
178    ///
179    /// `item_name`, `item_pos` and `is_snapshot` are set to distinctive,
180    /// obviously-synthetic values so tests can assert whether they survive the
181    /// parse (they are dropped on the silent-default path — see the malformed
182    /// test).
183    fn trade_item_update(field_key: &str, field_value: &str) -> ItemUpdate {
184        let mut fields = HashMap::new();
185        fields.insert(field_key.to_string(), Some(field_value.to_string()));
186        ItemUpdate {
187            item_name: Some("TRADE:SYNTHETIC".to_string()),
188            item_pos: 2,
189            fields,
190            changed_fields: HashMap::new(),
191            is_snapshot: true,
192        }
193    }
194
195    // Realistic, sanitized OPU payload. Field names mirror IG's Open Position
196    // Update stream (camelCase); the deal ids are obviously synthetic and carry
197    // no secret material.
198    const SANITIZED_OPU_JSON: &str = r#"{
199        "dealReference": "REF_SYNTHETIC_0001",
200        "dealId": "DIAAAAF00SYNTH01",
201        "dealIdOrigin": "DIAAAAF00SYNTH01",
202        "direction": "BUY",
203        "epic": "IX.D.DAX.DAILY.IP",
204        "status": "OPEN",
205        "dealStatus": "ACCEPTED",
206        "level": "18000.5",
207        "size": "1.0",
208        "currency": "EUR",
209        "timestamp": "2024-01-15T10:30:00.000",
210        "channel": "PublicRestOTC",
211        "expiry": "DFB"
212    }"#;
213
214    // Realistic, sanitized WOU payload. Field names mirror IG's Working Order
215    // Update stream (camelCase); the deal ids are obviously synthetic.
216    const SANITIZED_WOU_JSON: &str = r#"{
217        "dealReference": "REF_SYNTHETIC_0002",
218        "dealId": "DIAAAAF00SYNTH02",
219        "direction": "SELL",
220        "epic": "CS.D.EURUSD.CFD.IP",
221        "status": "OPEN",
222        "dealStatus": "ACCEPTED",
223        "level": "1.1000",
224        "size": "10000",
225        "currency": "USD",
226        "timestamp": "2024-01-15T10:31:00.000",
227        "channel": "PublicRestOTC",
228        "expiry": "-",
229        "stopDistance": "10.5",
230        "limitDistance": "20.0",
231        "guaranteedStop": false,
232        "orderType": "LIMIT",
233        "timeInForce": "GOOD_TILL_CANCELLED",
234        "goodTillDate": ""
235    }"#;
236
237    #[test]
238    fn test_trade_data_from_item_update_parses_open_position_update() {
239        let item = trade_item_update("OPU", SANITIZED_OPU_JSON);
240
241        let result = trade_data_from_item_update(&item);
242        assert!(result.is_ok(), "OPU payload should parse: {result:?}");
243        let data = result.expect("OPU parse checked ok above");
244
245        // Update metadata is preserved on the success path.
246        assert_eq!(data.item_name, "TRADE:SYNTHETIC");
247        assert_eq!(data.item_pos, 2);
248        assert!(data.is_snapshot);
249
250        let opu = data.fields.opu.expect("OPU should be populated");
251        assert_eq!(opu.deal_reference.as_deref(), Some("REF_SYNTHETIC_0001"));
252        assert_eq!(opu.deal_id.as_deref(), Some("DIAAAAF00SYNTH01"));
253        assert_eq!(opu.deal_id_origin.as_deref(), Some("DIAAAAF00SYNTH01"));
254        assert_eq!(opu.direction, Some(Direction::Buy));
255        assert_eq!(opu.epic.as_deref(), Some("IX.D.DAX.DAILY.IP"));
256        assert_eq!(opu.status, Some(Status::Open));
257        assert_eq!(opu.deal_status, Some(Status::Accepted));
258        assert_eq!(opu.level, Some(18000.5));
259        assert_eq!(opu.size, Some(1.0));
260        assert_eq!(opu.currency.as_deref(), Some("EUR"));
261        assert_eq!(opu.expiry.as_deref(), Some("DFB"));
262
263        // No WOU or CONFIRMS present in this update.
264        assert!(data.fields.wou.is_none());
265        assert!(data.fields.confirms.is_none());
266    }
267
268    #[test]
269    fn test_trade_data_from_item_update_parses_working_order_update() {
270        let item = trade_item_update("WOU", SANITIZED_WOU_JSON);
271
272        let result = trade_data_from_item_update(&item);
273        assert!(result.is_ok(), "WOU payload should parse: {result:?}");
274        let data = result.expect("WOU parse checked ok above");
275
276        let wou = data.fields.wou.expect("WOU should be populated");
277        assert_eq!(wou.deal_reference.as_deref(), Some("REF_SYNTHETIC_0002"));
278        assert_eq!(wou.deal_id.as_deref(), Some("DIAAAAF00SYNTH02"));
279        assert_eq!(wou.direction, Some(Direction::Sell));
280        assert_eq!(wou.epic.as_deref(), Some("CS.D.EURUSD.CFD.IP"));
281        assert_eq!(wou.status, Some(Status::Open));
282        assert_eq!(wou.deal_status, Some(Status::Accepted));
283        assert_eq!(wou.level, Some(1.1000));
284        assert_eq!(wou.size, Some(10000.0));
285        assert_eq!(wou.currency.as_deref(), Some("USD"));
286        assert_eq!(wou.stop_distance, Some(10.5));
287        assert_eq!(wou.limit_distance, Some(20.0));
288        assert_eq!(wou.guaranteed_stop, Some(false));
289        assert_eq!(wou.order_type, Some(OrderType::Limit));
290        assert_eq!(wou.time_in_force, Some(TimeInForce::GoodTillCancelled));
291        // Empty string good-till-date degrades to None.
292        assert!(wou.good_till_date.is_none());
293
294        // No OPU or CONFIRMS present in this update.
295        assert!(data.fields.opu.is_none());
296        assert!(data.fields.confirms.is_none());
297    }
298
299    #[test]
300    fn test_trade_data_from_item_update_malformed_opu_is_error_on_direct_path() {
301        let item = trade_item_update("OPU", "{ this is not valid json ");
302
303        // The direct parse path surfaces the failure as an Err.
304        let result = trade_data_from_item_update(&item);
305        assert!(
306            result.is_err(),
307            "malformed OPU JSON must surface an error on the direct parse path"
308        );
309    }
310
311    #[test]
312    fn test_trade_data_from_impl_swallows_malformed_opu_into_default() {
313        // KNOWN GAP (pinned, do not "fix" here): `From<&ItemUpdate>` funnels any
314        // parse failure through `.unwrap_or_default()`, so a malformed TRADE
315        // payload is silently swallowed into `TradeData::default()` instead of
316        // being surfaced. This test PINS that current behavior so a future change
317        // to make the swallow deliberate (e.g. logging, or propagating the error)
318        // is a conscious, reviewed decision rather than an accidental regression.
319        //
320        // TODO(streaming): decide whether `From<&ItemUpdate>` should log the
321        // dropped payload at WARN or expose the parse error instead of defaulting.
322        let item = trade_item_update("OPU", "{ this is not valid json ");
323
324        let data = TradeData::from(&item);
325
326        // Everything degrades to the Default, including the update metadata that
327        // was present on the source ItemUpdate (item_name / item_pos / snapshot).
328        assert!(
329            data.item_name.is_empty(),
330            "silent-default path discards item_name"
331        );
332        assert_eq!(data.item_pos, 0, "silent-default path discards item_pos");
333        assert!(
334            !data.is_snapshot,
335            "silent-default path discards is_snapshot"
336        );
337        assert!(data.fields.opu.is_none());
338        assert!(data.fields.wou.is_none());
339        assert!(data.fields.confirms.is_none());
340    }
341
342    #[test]
343    fn test_price_data_from_item_update_defaults_on_unknown_dealing_flag() {
344        // An unknown dealing flag makes the pure parser fail; the `From` impl
345        // degrades to `PriceData::default()` while the direct fn surfaces the Err.
346        let mut fields = HashMap::new();
347        fields.insert("DLG_FLAG".to_string(), Some("NOT_A_FLAG".to_string()));
348        let item = ItemUpdate {
349            item_name: Some("PRICE:CS.D.EURUSD.MINI.IP".to_string()),
350            item_pos: 1,
351            fields,
352            changed_fields: HashMap::new(),
353            is_snapshot: true,
354        };
355
356        assert!(price_data_from_item_update(&item).is_err());
357        // The `From` impl degrades to the default: metadata from the source
358        // update is discarded (matching the pre-refactor silent-default path).
359        let data = PriceData::from(&item);
360        assert!(data.item_name.is_empty());
361        assert_eq!(data.item_pos, 0);
362        assert!(!data.is_snapshot);
363        assert!(data.fields.dealing_flag.is_none());
364    }
365}
ig_client/application/streaming_convert.rs

ig_client/application/
streaming_convert.rs
