nautilus_betfair/stream/
messages.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Betfair Exchange Stream API message definitions.
17//!
18//! The stream protocol uses newline-delimited JSON with an `op` field to
19//! discriminate message types. Field names are abbreviated for bandwidth
20//! efficiency (e.g. `pt` for publish time, `mc` for market changes).
21//!
22//! # References
23//!
24//! <https://docs.developer.betfair.com/display/1smk3cen4v3lu3yomq5qye0ni/Exchange+Stream+API>
25
26use std::str::FromStr;
27
28use ahash::AHashMap;
29use nautilus_core::{
30    serialization::{deserialize_decimal, deserialize_optional_decimal},
31    string::secret::SecretString,
32};
33use rust_decimal::Decimal;
34use serde::{
35    Deserialize, Deserializer, Serialize,
36    de::{MapAccess, Visitor, value::MapAccessDeserializer},
37};
38use ustr::Ustr;
39use zeroize::Zeroize;
40
41use crate::common::{
42    consts::{
43        STREAM_OP_AUTHENTICATION, STREAM_OP_CRICKET_SUBSCRIPTION, STREAM_OP_HEARTBEAT,
44        STREAM_OP_RACE_SUBSCRIPTION,
45    },
46    enums::{
47        ChangeType, LapseStatusReasonCode, MarketBettingType, MarketDataFilterField, MarketStatus,
48        PriceLadderType, RunnerStatus, SegmentType, StatusErrorCode, StreamingOrderStatus,
49        StreamingOrderType, StreamingPersistenceType, StreamingSide,
50    },
51    types::{
52        Handicap, JsonDecimal, MarketId, SelectionId, deserialize_optional_decimal_native,
53        deserialize_optional_string_lenient, deserialize_selection_id,
54    },
55};
56
57/// Top-level streaming message, discriminated by the `op` field.
58///
59/// Deserializing a raw JSON line into this enum replaces the Python
60/// `stream_decode()` function from `betfair_parser`.
61#[derive(Debug, Clone, Deserialize)]
62#[serde(tag = "op")]
63pub enum StreamMessage {
64    #[serde(rename = "connection")]
65    Connection(Connection),
66    #[serde(rename = "status")]
67    Status(Status),
68    #[serde(rename = "mcm")]
69    MarketChange(MCM),
70    #[serde(rename = "ocm")]
71    OrderChange(OCM),
72    #[serde(rename = "rcm")]
73    RaceChange(RCM),
74    #[serde(rename = "ccm")]
75    CricketChange(CCM),
76}
77
78/// Connection confirmation sent on stream connect.
79#[derive(Debug, Clone, Deserialize)]
80#[serde(rename_all = "camelCase")]
81pub struct Connection {
82    pub id: Option<u64>,
83    pub connection_id: String,
84}
85
86/// Status response for errors or informational messages.
87#[derive(Debug, Clone, Deserialize)]
88#[serde(rename_all = "camelCase")]
89pub struct Status {
90    pub id: Option<u64>,
91    pub connection_closed: bool,
92    pub connection_id: Option<String>,
93    pub connections_available: Option<u32>,
94    pub error_code: Option<StatusErrorCode>,
95    pub error_message: Option<String>,
96    pub status_code: Option<String>,
97}
98
99/// Market Change Message (MCM) - price/market data updates.
100#[derive(Debug, Clone, Deserialize)]
101pub struct MCM {
102    pub id: Option<u64>,
103    /// Publish time (epoch millis).
104    pub pt: u64,
105    /// Token used for resubscription.
106    pub clk: Option<String>,
107    /// Initial clock token (sent on first image).
108    #[serde(rename = "initialClk")]
109    pub initial_clk: Option<String>,
110    /// Market changes (None on heartbeat).
111    pub mc: Option<Vec<MarketChange>>,
112    /// Change type.
113    pub ct: Option<ChangeType>,
114    /// Conflation interval in milliseconds.
115    #[serde(rename = "conflateMs")]
116    pub conflate_ms: Option<u64>,
117    /// Heartbeat interval in milliseconds.
118    #[serde(rename = "heartbeatMs")]
119    pub heartbeat_ms: Option<u64>,
120    /// Segment type for large messages.
121    #[serde(rename = "segmentType")]
122    pub segment_type: Option<SegmentType>,
123    pub status: Option<i32>,
124}
125
126impl MCM {
127    #[must_use]
128    pub fn is_heartbeat(&self) -> bool {
129        self.ct == Some(ChangeType::Heartbeat)
130    }
131}
132
133/// Order Change Message (OCM) - order/position updates.
134#[derive(Debug, Clone, Deserialize)]
135pub struct OCM {
136    pub id: Option<u64>,
137    /// Publish time (epoch millis).
138    pub pt: u64,
139    pub clk: Option<String>,
140    #[serde(rename = "initialClk")]
141    pub initial_clk: Option<String>,
142    /// Order market changes (None on heartbeat).
143    pub oc: Option<Vec<OrderMarketChange>>,
144    pub ct: Option<ChangeType>,
145    #[serde(rename = "conflateMs")]
146    pub conflate_ms: Option<u64>,
147    #[serde(rename = "heartbeatMs")]
148    pub heartbeat_ms: Option<u64>,
149    #[serde(rename = "segmentType")]
150    pub segment_type: Option<SegmentType>,
151    pub status: Option<i32>,
152}
153
154impl OCM {
155    #[must_use]
156    pub fn is_heartbeat(&self) -> bool {
157        self.ct == Some(ChangeType::Heartbeat)
158    }
159}
160
161/// Delta update for a single market.
162#[derive(Debug, Clone, Deserialize)]
163pub struct MarketChange {
164    /// Market identifier.
165    pub id: MarketId,
166    /// Runner changes.
167    pub rc: Option<Vec<RunnerChange>>,
168    /// Whether there was a conflation.
169    pub con: Option<bool>,
170    /// Whether this is a full image (vs delta).
171    #[serde(default)]
172    pub img: bool,
173    /// Full market definition (sent on subscription or change).
174    #[serde(rename = "marketDefinition")]
175    pub market_definition: Option<MarketDefinition>,
176    /// Total volume matched on this market.
177    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
178    pub tv: Option<Decimal>,
179}
180
181/// Delta update for a single runner (selection).
182#[derive(Debug, Clone, Deserialize)]
183pub struct RunnerChange {
184    /// Selection identifier.
185    #[serde(deserialize_with = "deserialize_selection_id")]
186    pub id: SelectionId,
187    /// Handicap value.
188    #[serde(default, deserialize_with = "deserialize_optional_decimal_native")]
189    pub hc: Option<Handicap>,
190    /// Available to back.
191    pub atb: Option<Vec<PV>>,
192    /// Available to lay.
193    pub atl: Option<Vec<PV>>,
194    /// Best available to back (depth).
195    pub batb: Option<Vec<LPV>>,
196    /// Best available to lay (depth).
197    pub batl: Option<Vec<LPV>>,
198    /// Best display available to back.
199    pub bdatb: Option<Vec<LPV>>,
200    /// Best display available to lay.
201    pub bdatl: Option<Vec<LPV>>,
202    /// Starting price back.
203    pub spb: Option<Vec<PV>>,
204    /// Starting price lay.
205    pub spl: Option<Vec<PV>>,
206    /// Starting price near (projected SP).
207    #[serde(default, deserialize_with = "deserialize_optional_decimal_lenient")]
208    pub spn: Option<Decimal>,
209    /// Starting price far (actual BSP).
210    #[serde(default, deserialize_with = "deserialize_optional_decimal_lenient")]
211    pub spf: Option<Decimal>,
212    /// Traded volume by price level.
213    pub trd: Option<Vec<PV>>,
214    /// Last traded price.
215    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
216    pub ltp: Option<Decimal>,
217    /// Total volume matched on this runner.
218    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
219    pub tv: Option<Decimal>,
220}
221
222fn deserialize_optional_decimal_lenient<'de, D>(
223    deserializer: D,
224) -> Result<Option<Decimal>, D::Error>
225where
226    D: Deserializer<'de>,
227{
228    struct LenientOptionalDecimalVisitor;
229
230    impl<'de> Visitor<'de> for LenientOptionalDecimalVisitor {
231        type Value = Option<Decimal>;
232
233        fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
234            formatter.write_str("null or a decimal number as string, integer, or float")
235        }
236
237        fn visit_str<E: serde::de::Error>(self, value: &str) -> Result<Self::Value, E> {
238            Ok(parse_optional_decimal_lenient(value))
239        }
240
241        fn visit_string<E: serde::de::Error>(self, value: String) -> Result<Self::Value, E> {
242            self.visit_str(&value)
243        }
244
245        fn visit_i64<E: serde::de::Error>(self, value: i64) -> Result<Self::Value, E> {
246            Ok(Some(Decimal::from(value)))
247        }
248
249        fn visit_u64<E: serde::de::Error>(self, value: u64) -> Result<Self::Value, E> {
250            Ok(Some(Decimal::from(value)))
251        }
252
253        fn visit_i128<E: serde::de::Error>(self, value: i128) -> Result<Self::Value, E> {
254            Ok(Some(Decimal::from(value)))
255        }
256
257        fn visit_u128<E: serde::de::Error>(self, value: u128) -> Result<Self::Value, E> {
258            Ok(Some(Decimal::from(value)))
259        }
260
261        fn visit_f64<E: serde::de::Error>(self, value: f64) -> Result<Self::Value, E> {
262            Ok(Decimal::try_from(value).ok())
263        }
264
265        fn visit_map<A: MapAccess<'de>>(self, map: A) -> Result<Self::Value, A::Error> {
266            let number = serde_json::Number::deserialize(MapAccessDeserializer::new(map))?;
267            self.visit_str(&number.to_string())
268        }
269
270        fn visit_unit<E: serde::de::Error>(self) -> Result<Self::Value, E> {
271            Ok(None)
272        }
273
274        fn visit_none<E: serde::de::Error>(self) -> Result<Self::Value, E> {
275            Ok(None)
276        }
277    }
278
279    deserializer.deserialize_any(LenientOptionalDecimalVisitor)
280}
281
282fn parse_optional_decimal_lenient(value: &str) -> Option<Decimal> {
283    let trimmed = value.trim();
284    if trimmed.is_empty() || is_non_finite_decimal(trimmed) {
285        return None;
286    }
287
288    if trimmed.contains('e') || trimmed.contains('E') {
289        Decimal::from_scientific(trimmed).ok()
290    } else {
291        Decimal::from_str(trimmed).ok()
292    }
293}
294
295fn is_non_finite_decimal(value: &str) -> bool {
296    value.eq_ignore_ascii_case("nan")
297        || value.eq_ignore_ascii_case("inf")
298        || value.eq_ignore_ascii_case("+inf")
299        || value.eq_ignore_ascii_case("-inf")
300        || value.eq_ignore_ascii_case("infinity")
301        || value.eq_ignore_ascii_case("+infinity")
302        || value.eq_ignore_ascii_case("-infinity")
303}
304
305/// Full market definition snapshot.
306#[derive(Debug, Clone, Deserialize)]
307#[serde(rename_all = "camelCase")]
308pub struct MarketDefinition {
309    pub bet_delay: Option<i32>,
310    pub betting_type: Option<MarketBettingType>,
311    pub bsp_market: Option<bool>,
312    pub bsp_reconciled: Option<bool>,
313    pub competition_id: Option<String>,
314    pub competition_name: Option<String>,
315    pub complete: Option<bool>,
316    pub country_code: Option<Ustr>,
317    pub cross_matching: Option<bool>,
318    pub discount_allowed: Option<bool>,
319    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
320    pub each_way_divisor: Option<Decimal>,
321    pub event_id: Option<String>,
322    pub event_name: Option<String>,
323    #[serde(default, deserialize_with = "deserialize_optional_string_lenient")]
324    pub event_type_id: Option<String>,
325    pub event_type_name: Option<Ustr>,
326    pub in_play: Option<bool>,
327    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
328    pub line_interval: Option<Decimal>,
329    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
330    pub line_max_unit: Option<Decimal>,
331    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
332    pub line_min_unit: Option<Decimal>,
333    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
334    pub market_base_rate: Option<Decimal>,
335    pub market_id: Option<MarketId>,
336    pub market_name: Option<String>,
337    pub market_time: Option<String>,
338    pub market_type: Option<Ustr>,
339    pub number_of_active_runners: Option<u32>,
340    pub number_of_winners: Option<u32>,
341    pub open_date: Option<String>,
342    pub persistence_enabled: Option<bool>,
343    pub price_ladder_definition: Option<PriceLadderDefinition>,
344    pub race_type: Option<Ustr>,
345    pub regulators: Option<Vec<Ustr>>,
346    pub runners: Option<Vec<RunnerDefinition>>,
347    pub runners_voidable: Option<bool>,
348    pub settled_time: Option<String>,
349    pub status: Option<MarketStatus>,
350    pub suspend_time: Option<String>,
351    pub timezone: Option<Ustr>,
352    pub turn_in_play_enabled: Option<bool>,
353    pub venue: Option<Ustr>,
354    pub version: Option<u64>,
355}
356
357/// Runner (selection) definition within a market definition.
358#[derive(Debug, Clone, Deserialize)]
359#[serde(rename_all = "camelCase")]
360pub struct RunnerDefinition {
361    #[serde(deserialize_with = "deserialize_selection_id")]
362    pub id: SelectionId,
363    #[serde(default, deserialize_with = "deserialize_optional_decimal_native")]
364    pub hc: Option<Handicap>,
365    pub sort_priority: Option<u32>,
366    pub name: Option<String>,
367    pub status: Option<RunnerStatus>,
368    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
369    pub adjustment_factor: Option<Decimal>,
370    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
371    pub bsp: Option<Decimal>,
372    pub removal_date: Option<String>,
373}
374
375/// Price ladder definition within a market definition.
376#[derive(Debug, Clone, Deserialize)]
377pub struct PriceLadderDefinition {
378    #[serde(rename = "type")]
379    pub ladder_type: Option<PriceLadderType>,
380}
381
382// Betfair encodes price-volume types as JSON arrays: [price, volume] and
383// [level, price, volume] respectively.
384
385/// Price-volume pair, serialized as a JSON array `[price, volume]`.
386#[derive(Debug, Clone, Copy, PartialEq)]
387pub struct PV {
388    pub price: Decimal,
389    pub volume: Decimal,
390}
391
392impl<'de> Deserialize<'de> for PV {
393    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
394    where
395        D: serde::Deserializer<'de>,
396    {
397        // Handles both `[price, volume]` and `[level, price, volume]` (RESUB_DELTA)
398        let arr: Vec<JsonDecimal> = Deserialize::deserialize(deserializer)?;
399        match arr.len() {
400            2 => Ok(Self {
401                price: arr[0].0,
402                volume: arr[1].0,
403            }),
404            3 => Ok(Self {
405                price: arr[1].0,
406                volume: arr[2].0,
407            }),
408            n => Err(serde::de::Error::invalid_length(n, &"2 or 3 elements")),
409        }
410    }
411}
412
413impl Serialize for PV {
414    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
415    where
416        S: serde::Serializer,
417    {
418        (self.price, self.volume).serialize(serializer)
419    }
420}
421
422/// Level-price-volume triple, serialized as a JSON array `[level, price, volume]`.
423#[derive(Debug, Clone, Copy, PartialEq)]
424pub struct LPV {
425    pub level: u32,
426    pub price: Decimal,
427    pub volume: Decimal,
428}
429
430impl<'de> Deserialize<'de> for LPV {
431    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
432    where
433        D: serde::Deserializer<'de>,
434    {
435        let arr: (u32, JsonDecimal, JsonDecimal) = Deserialize::deserialize(deserializer)?;
436        Ok(Self {
437            level: arr.0,
438            price: arr.1.0,
439            volume: arr.2.0,
440        })
441    }
442}
443
444impl Serialize for LPV {
445    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
446    where
447        S: serde::Serializer,
448    {
449        (self.level, self.price, self.volume).serialize(serializer)
450    }
451}
452
453/// Order changes for a single market.
454#[derive(Debug, Clone, Deserialize)]
455pub struct OrderMarketChange {
456    /// Market identifier.
457    pub id: MarketId,
458    #[serde(rename = "accountId")]
459    pub account_id: Option<u64>,
460    pub closed: Option<bool>,
461    #[serde(rename = "fullImage", default)]
462    pub full_image: bool,
463    /// Order runner changes.
464    pub orc: Option<Vec<OrderRunnerChange>>,
465}
466
467/// Order changes for a single runner within a market.
468#[derive(Debug, Clone, Deserialize)]
469pub struct OrderRunnerChange {
470    /// Selection identifier.
471    #[serde(deserialize_with = "deserialize_selection_id")]
472    pub id: SelectionId,
473    #[serde(rename = "fullImage", default)]
474    pub full_image: bool,
475    /// Handicap.
476    #[serde(default, deserialize_with = "deserialize_optional_decimal_native")]
477    pub hc: Option<Handicap>,
478    /// Matched backs.
479    pub mb: Option<Vec<MatchedOrder>>,
480    /// Matched lays.
481    pub ml: Option<Vec<MatchedOrder>>,
482    /// Strategy match changes, keyed by customer strategy ref.
483    pub smc: Option<AHashMap<String, StrategyMatchChange>>,
484    /// Unmatched orders.
485    pub uo: Option<Vec<UnmatchedOrder>>,
486}
487
488/// Matched order (price-size pair), serialized as `[price, size]`.
489#[derive(Debug, Clone, Copy, PartialEq)]
490pub struct MatchedOrder {
491    pub price: Decimal,
492    pub size: Decimal,
493}
494
495impl<'de> Deserialize<'de> for MatchedOrder {
496    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
497    where
498        D: serde::Deserializer<'de>,
499    {
500        let arr: (JsonDecimal, JsonDecimal) = Deserialize::deserialize(deserializer)?;
501        Ok(Self {
502            price: arr.0.0,
503            size: arr.1.0,
504        })
505    }
506}
507
508impl Serialize for MatchedOrder {
509    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
510    where
511        S: serde::Serializer,
512    {
513        (self.price, self.size).serialize(serializer)
514    }
515}
516
517/// Strategy-level match changes.
518#[derive(Debug, Clone, Deserialize)]
519pub struct StrategyMatchChange {
520    /// Matched backs.
521    pub mb: Option<Vec<MatchedOrder>>,
522    /// Matched lays.
523    pub ml: Option<Vec<MatchedOrder>>,
524}
525
526/// Unmatched order on the streaming API.
527#[derive(Debug, Clone, Deserialize)]
528pub struct UnmatchedOrder {
529    /// Bet identifier.
530    pub id: String,
531    /// Price.
532    #[serde(deserialize_with = "deserialize_decimal")]
533    pub p: Decimal,
534    /// Size.
535    #[serde(deserialize_with = "deserialize_decimal")]
536    pub s: Decimal,
537    /// Side (B=Back, L=Lay).
538    pub side: StreamingSide,
539    /// Order status (E=Executable, EC=ExecutionComplete).
540    pub status: StreamingOrderStatus,
541    /// Persistence type (L=Lapse, P=Persist, MOC=MarketOnClose).
542    ///
543    /// Betfair can omit this on some BSP market-on-close order updates.
544    #[serde(default)]
545    pub pt: Option<StreamingPersistenceType>,
546    /// Order type (L=Limit, LOC=LimitOnClose, MOC=MarketOnClose).
547    pub ot: StreamingOrderType,
548    /// Placed date (epoch millis).
549    pub pd: u64,
550    /// BSP liability.
551    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
552    pub bsp: Option<Decimal>,
553    /// Customer strategy reference.
554    pub rfo: Option<String>,
555    /// Regulator reference.
556    pub rfs: Option<String>,
557    /// Customer order reference.
558    pub rc: Option<String>,
559    /// Regulator auth code.
560    pub rac: Option<String>,
561    /// Matched date (epoch millis).
562    pub md: Option<u64>,
563    /// Cancelled date (epoch millis).
564    pub cd: Option<u64>,
565    /// Lapsed date (epoch millis).
566    pub ld: Option<u64>,
567    /// Average price matched.
568    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
569    pub avp: Option<Decimal>,
570    /// Size matched.
571    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
572    pub sm: Option<Decimal>,
573    /// Size remaining.
574    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
575    pub sr: Option<Decimal>,
576    /// Size lapsed.
577    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
578    pub sl: Option<Decimal>,
579    /// Size cancelled.
580    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
581    pub sc: Option<Decimal>,
582    /// Size voided.
583    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
584    pub sv: Option<Decimal>,
585    /// Lapse status reason code.
586    pub lsrc: Option<LapseStatusReasonCode>,
587}
588
589/// Authentication request sent on stream connect.
590#[derive(Debug, Clone, Serialize, Zeroize)]
591pub struct Authentication {
592    pub op: String,
593    pub id: Option<u64>,
594    #[serde(rename = "appKey")]
595    pub app_key: SecretString,
596    pub session: SecretString,
597}
598
599impl Authentication {
600    /// Creates a new authentication request.
601    #[must_use]
602    pub fn new(app_key: impl Into<SecretString>, session: impl Into<SecretString>) -> Self {
603        Self {
604            op: STREAM_OP_AUTHENTICATION.to_string(),
605            id: None,
606            app_key: app_key.into(),
607            session: session.into(),
608        }
609    }
610
611    /// Creates a correlated authentication request.
612    #[must_use]
613    pub fn with_id(
614        app_key: impl Into<SecretString>,
615        session: impl Into<SecretString>,
616        id: u64,
617    ) -> Self {
618        Self {
619            op: STREAM_OP_AUTHENTICATION.to_string(),
620            id: Some(id),
621            app_key: app_key.into(),
622            session: session.into(),
623        }
624    }
625}
626
627/// Market subscription request.
628#[derive(Debug, Clone, Serialize)]
629#[serde(rename_all = "camelCase")]
630pub struct MarketSubscription {
631    pub op: String,
632    pub id: Option<u64>,
633    pub market_filter: StreamMarketFilter,
634    pub market_data_filter: MarketDataFilter,
635    #[serde(skip_serializing_if = "Option::is_none")]
636    pub clk: Option<String>,
637    #[serde(skip_serializing_if = "Option::is_none")]
638    pub conflate_ms: Option<u64>,
639    #[serde(skip_serializing_if = "Option::is_none")]
640    pub heartbeat_ms: Option<u64>,
641    #[serde(skip_serializing_if = "Option::is_none")]
642    pub initial_clk: Option<String>,
643    #[serde(skip_serializing_if = "Option::is_none")]
644    pub segmentation_enabled: Option<bool>,
645}
646
647/// Order subscription request.
648#[derive(Debug, Clone, Serialize)]
649#[serde(rename_all = "camelCase")]
650pub struct OrderSubscription {
651    pub op: String,
652    pub id: Option<u64>,
653    #[serde(skip_serializing_if = "Option::is_none")]
654    pub order_filter: Option<OrderFilter>,
655    #[serde(skip_serializing_if = "Option::is_none")]
656    pub clk: Option<String>,
657    #[serde(skip_serializing_if = "Option::is_none")]
658    pub conflate_ms: Option<u64>,
659    #[serde(skip_serializing_if = "Option::is_none")]
660    pub heartbeat_ms: Option<u64>,
661    #[serde(skip_serializing_if = "Option::is_none")]
662    pub initial_clk: Option<String>,
663    #[serde(skip_serializing_if = "Option::is_none")]
664    pub segmentation_enabled: Option<bool>,
665}
666
667/// Race stream subscription request.
668#[derive(Debug, Clone, Serialize)]
669#[serde(rename_all = "camelCase")]
670pub struct RaceSubscription {
671    pub op: String,
672    pub id: Option<u64>,
673}
674
675impl RaceSubscription {
676    #[must_use]
677    pub fn new(id: u64) -> Self {
678        Self {
679            op: STREAM_OP_RACE_SUBSCRIPTION.to_string(),
680            id: Some(id),
681        }
682    }
683}
684
685/// Cricket stream subscription request.
686#[derive(Debug, Clone, Serialize)]
687#[serde(rename_all = "camelCase")]
688pub struct CricketSubscription {
689    pub op: String,
690    pub id: Option<u64>,
691}
692
693impl CricketSubscription {
694    #[must_use]
695    pub fn new(id: u64) -> Self {
696        Self {
697            op: STREAM_OP_CRICKET_SUBSCRIPTION.to_string(),
698            id: Some(id),
699        }
700    }
701}
702
703/// Heartbeat request to keep the connection alive.
704#[derive(Debug, Clone, Serialize)]
705pub struct StreamHeartbeat {
706    pub op: String,
707    pub id: Option<u64>,
708}
709
710impl StreamHeartbeat {
711    #[must_use]
712    pub fn new() -> Self {
713        Self {
714            op: STREAM_OP_HEARTBEAT.to_string(),
715            id: None,
716        }
717    }
718}
719
720impl Default for StreamHeartbeat {
721    fn default() -> Self {
722        Self::new()
723    }
724}
725
726/// Market filter for streaming subscriptions.
727#[derive(Debug, Clone, Default, Serialize, Deserialize)]
728#[serde(rename_all = "camelCase")]
729pub struct StreamMarketFilter {
730    #[serde(skip_serializing_if = "Option::is_none")]
731    pub betting_types: Option<Vec<MarketBettingType>>,
732    #[serde(skip_serializing_if = "Option::is_none")]
733    pub bsp_market: Option<bool>,
734    #[serde(skip_serializing_if = "Option::is_none")]
735    pub country_codes: Option<Vec<Ustr>>,
736    #[serde(skip_serializing_if = "Option::is_none")]
737    pub event_ids: Option<Vec<String>>,
738    #[serde(skip_serializing_if = "Option::is_none")]
739    pub event_type_ids: Option<Vec<String>>,
740    #[serde(skip_serializing_if = "Option::is_none")]
741    pub market_ids: Option<Vec<MarketId>>,
742    #[serde(skip_serializing_if = "Option::is_none")]
743    pub market_types: Option<Vec<Ustr>>,
744    #[serde(skip_serializing_if = "Option::is_none")]
745    pub race_types: Option<Vec<Ustr>>,
746    #[serde(skip_serializing_if = "Option::is_none")]
747    pub turn_in_play_enabled: Option<bool>,
748    #[serde(skip_serializing_if = "Option::is_none")]
749    pub venues: Option<Vec<Ustr>>,
750}
751
752/// Market data filter for streaming subscriptions.
753#[derive(Debug, Clone, Default, Serialize, Deserialize)]
754#[serde(rename_all = "camelCase")]
755pub struct MarketDataFilter {
756    #[serde(skip_serializing_if = "Option::is_none")]
757    pub fields: Option<Vec<MarketDataFilterField>>,
758    #[serde(skip_serializing_if = "Option::is_none")]
759    pub ladder_levels: Option<u32>,
760}
761
762/// Order filter for streaming subscriptions.
763#[derive(Debug, Clone, Serialize, Deserialize)]
764#[serde(rename_all = "camelCase")]
765pub struct OrderFilter {
766    #[serde(default = "default_true")]
767    pub include_overall_position: bool,
768    #[serde(skip_serializing_if = "Option::is_none")]
769    pub customer_strategy_refs: Option<Vec<String>>,
770    #[serde(default)]
771    pub partition_matched_by_strategy_ref: bool,
772    #[serde(skip_serializing_if = "Option::is_none")]
773    pub account_ids: Option<Vec<u64>>,
774}
775
776impl Default for OrderFilter {
777    fn default() -> Self {
778        Self {
779            include_overall_position: true,
780            customer_strategy_refs: None,
781            partition_matched_by_strategy_ref: false,
782            account_ids: None,
783        }
784    }
785}
786
787fn default_true() -> bool {
788    true
789}
790
791/// Race Change Message (RCM) - live GPS tracking data (Total Performance Data).
792#[derive(Debug, Clone, Deserialize)]
793pub struct RCM {
794    pub id: Option<u64>,
795    /// Publish time (epoch millis).
796    pub pt: u64,
797    /// Clock token (may be integer or string depending on feed state).
798    pub clk: Option<serde_json::Value>,
799    /// Race changes (None on heartbeat).
800    pub rc: Option<Vec<RaceChange>>,
801}
802
803/// Delta update for a single race within an RCM.
804#[derive(Debug, Clone, Deserialize)]
805pub struct RaceChange {
806    /// Race identifier (e.g. "28587288.1650").
807    pub id: Option<String>,
808    /// Betfair market identifier.
809    pub mid: Option<String>,
810    /// Individual runner GPS data changes.
811    pub rrc: Option<Vec<RaceRunnerChange>>,
812    /// Overall race progress summary.
813    pub rpc: Option<RaceProgressChange>,
814}
815
816/// GPS tracking data for a single runner.
817#[derive(Debug, Clone, Deserialize)]
818pub struct RaceRunnerChange {
819    /// Feed time (epoch millis).
820    pub ft: Option<u64>,
821    /// Selection identifier.
822    pub id: Option<i64>,
823    /// Latitude (GPS coordinate).
824    #[serde(default, deserialize_with = "deserialize_optional_f64")]
825    pub lat: Option<f64>,
826    /// Longitude (GPS coordinate).
827    #[serde(
828        rename = "long",
829        default,
830        deserialize_with = "deserialize_optional_f64"
831    )]
832    pub lng: Option<f64>,
833    /// Speed in m/s (Doppler-derived).
834    #[serde(default, deserialize_with = "deserialize_optional_f64")]
835    pub spd: Option<f64>,
836    /// Distance to finish in meters.
837    #[serde(default, deserialize_with = "deserialize_optional_f64")]
838    pub prg: Option<f64>,
839    /// Stride frequency in Hz.
840    #[serde(default, deserialize_with = "deserialize_optional_f64")]
841    pub sfq: Option<f64>,
842}
843
844/// Race-level progress summary.
845#[derive(Debug, Clone, Deserialize)]
846pub struct RaceProgressChange {
847    /// Feed time (epoch millis).
848    pub ft: Option<u64>,
849    /// Gate/sectional name (e.g. "1f", "2f", "Finish").
850    pub g: Option<String>,
851    /// Sectional time in seconds.
852    #[serde(default, deserialize_with = "deserialize_optional_f64")]
853    pub st: Option<f64>,
854    /// Running time since race start in seconds.
855    #[serde(default, deserialize_with = "deserialize_optional_f64")]
856    pub rt: Option<f64>,
857    /// Speed of lead horse in m/s.
858    #[serde(default, deserialize_with = "deserialize_optional_f64")]
859    pub spd: Option<f64>,
860    /// Distance to finish for leading horse in meters.
861    #[serde(default, deserialize_with = "deserialize_optional_f64")]
862    pub prg: Option<f64>,
863    /// Runner order by selection ID (current race position).
864    pub ord: Option<Vec<i64>>,
865    /// Obstacle data for jump races.
866    #[serde(rename = "J")]
867    pub jumps: Option<Vec<Jump>>,
868}
869
870/// Cricket Change Message (CCM) - live cricket match data.
871#[derive(Debug, Clone, Deserialize)]
872pub struct CCM {
873    /// Subscription identifier.
874    pub id: Option<u64>,
875    /// Publish time (epoch millis).
876    pub pt: u64,
877    /// Clock token (may be integer or string depending on feed state).
878    pub clk: Option<serde_json::Value>,
879    /// Cricket match changes (None on heartbeat).
880    pub cc: Option<Vec<CricketChange>>,
881}
882
883/// Delta update for a single cricket match within a CCM.
884#[derive(Debug, Clone, Deserialize)]
885#[serde(rename_all = "camelCase")]
886pub struct CricketChange {
887    /// Betfair event identifier.
888    #[serde(default, deserialize_with = "deserialize_optional_string_lenient")]
889    pub event_id: Option<String>,
890    /// Betfair market identifier.
891    pub market_id: Option<String>,
892    /// Fixture metadata.
893    pub fixture_info: Option<serde_json::Value>,
894    /// Home team metadata.
895    pub home_team: Option<serde_json::Value>,
896    /// Away team metadata.
897    pub away_team: Option<serde_json::Value>,
898    /// Match statistics.
899    pub match_stats: Option<serde_json::Value>,
900    /// Match incidents.
901    pub incident_list_wrapper: Option<serde_json::Value>,
902}
903
904/// Jump obstacle location data.
905#[derive(Debug, Clone, Serialize, Deserialize)]
906pub struct Jump {
907    /// Jump number.
908    #[serde(rename = "J")]
909    pub number: i32,
910    /// Distance from finish line in meters.
911    #[serde(rename = "L", deserialize_with = "deserialize_f64")]
912    pub distance: f64,
913}
914
915fn deserialize_f64<'de, D: Deserializer<'de>>(deserializer: D) -> Result<f64, D::Error> {
916    serde_json::Number::deserialize(deserializer)?
917        .as_f64()
918        .ok_or_else(|| serde::de::Error::custom("number out of range for f64"))
919}
920
921fn deserialize_optional_f64<'de, D: Deserializer<'de>>(
922    deserializer: D,
923) -> Result<Option<f64>, D::Error> {
924    Option::<serde_json::Number>::deserialize(deserializer)?
925        .map(|number| {
926            number
927                .as_f64()
928                .ok_or_else(|| serde::de::Error::custom("number out of range for f64"))
929        })
930        .transpose()
931}
932
933/// Decode a single JSON stream line into a [`StreamMessage`].
934///
935/// # Errors
936///
937/// Returns an error if the JSON is malformed or the `op` field is missing/unknown.
938pub fn stream_decode(data: &[u8]) -> Result<StreamMessage, serde_json::Error> {
939    serde_json::from_slice(data)
940}
941
942#[cfg(test)]
943mod tests {
944    use rstest::rstest;
945
946    use super::*;
947    use crate::common::testing::load_test_json;
948
949    #[rstest]
950    fn test_decimal_preserves_rounding() {
951        let text = include_str!("../../test_data/stream/decimal_compatibility.json");
952        let value: serde_json::Value = serde_json::from_str(text).unwrap();
953        let direct: RunnerChange = serde_json::from_str(text).unwrap();
954        let buffered: RunnerChange = serde_json::from_value(value.clone()).unwrap();
955        let tagged =
956            serde_json::json!({"op": "mcm", "pt": 123, "mc": [{"id": "1.2", "rc": [value]}]});
957
958        let StreamMessage::MarketChange(message) =
959            stream_decode(tagged.to_string().as_bytes()).unwrap()
960        else {
961            panic!("expected market change");
962        };
963
964        let tagged = message.mc.unwrap().remove(0).rc.unwrap().remove(0);
965
966        let numeric = if is_arbitrary_precision() {
967            Decimal::from_str_exact("13.223699999999997").unwrap()
968        } else {
969            Decimal::from_str_exact("13.2237").unwrap()
970        };
971
972        let rounded = Decimal::from_str_exact("0.1234567890123456789012345679").unwrap();
973
974        for (runner, expected_numeric) in [
975            (direct, numeric),
976            (buffered, Decimal::from_str_exact("13.2237").unwrap()),
977            (tagged, numeric),
978        ] {
979            assert_eq!(runner.spn, Some(expected_numeric));
980            assert_eq!(runner.spf, Some(rounded));
981            assert_eq!(runner.hc, Some(rounded));
982            assert_eq!(
983                runner.atb,
984                Some(vec![PV {
985                    price: rounded,
986                    volume: Decimal::new(12500, 4)
987                }])
988            );
989            assert_eq!(runner.atb.unwrap()[0].volume.scale(), 4);
990            assert_eq!(
991                runner.batb,
992                Some(vec![LPV {
993                    level: 3,
994                    price: rounded,
995                    volume: Decimal::new(23750, 4)
996                }])
997            );
998            assert_eq!(runner.batb.unwrap()[0].volume.scale(), 4);
999        }
1000    }
1001
1002    #[rstest]
1003    fn test_matched_order_preserves_rounding() {
1004        let value = serde_json::json!(["0.12345678901234567890123456789", "1.2500"]);
1005        let direct: MatchedOrder = serde_json::from_str(&value.to_string()).unwrap();
1006        let buffered: MatchedOrder = serde_json::from_value(value).unwrap();
1007        let expected = Decimal::from_str_exact("0.1234567890123456789012345679").unwrap();
1008
1009        for order in [direct, buffered] {
1010            assert_eq!(order.price, expected);
1011            assert_eq!(order.size, Decimal::new(12500, 4));
1012            assert_eq!(order.size.scale(), 4);
1013        }
1014    }
1015
1016    #[rstest]
1017    fn test_decimal_routes_preserve_available_precision() {
1018        let text = include_str!("../../test_data/stream/decimal_routes.json");
1019        let value: serde_json::Value = serde_json::from_str(text).unwrap();
1020
1021        let expected_price = if is_arbitrary_precision() {
1022            Decimal::from_str_exact("123456789.123456789").unwrap()
1023        } else {
1024            Decimal::from_str_exact("123456789.12345679").unwrap()
1025        };
1026
1027        let direct: MCM = serde_json::from_str(text).unwrap();
1028        let buffered: MCM = serde_json::from_value(value.clone()).unwrap();
1029
1030        let StreamMessage::MarketChange(tagged) = stream_decode(text.as_bytes()).unwrap() else {
1031            panic!("expected market change");
1032        };
1033
1034        let StreamMessage::MarketChange(buffered_tagged) = serde_json::from_value(value).unwrap()
1035        else {
1036            panic!("expected market change");
1037        };
1038
1039        let direct_scale = u32::from(is_arbitrary_precision());
1040        let buffered_scale = serde_json::from_value::<Decimal>(serde_json::json!(2.0))
1041            .unwrap()
1042            .scale();
1043
1044        for (message, numeric_scale) in [
1045            (direct, direct_scale),
1046            (buffered, buffered_scale),
1047            (tagged, direct_scale),
1048            (buffered_tagged, buffered_scale),
1049        ] {
1050            let markets = message.mc.unwrap();
1051            let runner = &markets[0].rc.as_ref().unwrap()[0];
1052            assert_eq!(message.pt, 123);
1053            assert_eq!(markets[0].id, "1.2");
1054            assert_eq!(runner.id, 7);
1055            assert_eq!(
1056                runner.hc,
1057                Some(Decimal::from_str_exact("-0.1234567890123456789012345678").unwrap())
1058            );
1059            assert_eq!(
1060                runner.atb,
1061                Some(vec![PV {
1062                    price: expected_price,
1063                    volume: Decimal::from(9_007_199_254_740_993u64)
1064                }])
1065            );
1066            assert_eq!(
1067                runner.atl,
1068                Some(vec![PV {
1069                    price: Decimal::new(2125, 3),
1070                    volume: Decimal::from(17)
1071                }])
1072            );
1073            assert_eq!(
1074                runner.batb,
1075                Some(vec![LPV {
1076                    level: 3,
1077                    price: Decimal::new(4875, 3),
1078                    volume: Decimal::from(29)
1079                }])
1080            );
1081            assert_eq!(runner.spn, None);
1082            assert_eq!(runner.spf, Some(Decimal::new(125, 1)));
1083            let other = &markets[0].rc.as_ref().unwrap()[1];
1084            assert_eq!(other.id, 8);
1085            assert_eq!(other.hc, Some(Decimal::from(2)));
1086            assert_eq!(other.hc.unwrap().scale(), numeric_scale);
1087            assert_eq!(
1088                crate::common::parse::make_symbol(&markets[0].id, other.id, other.hc.unwrap())
1089                    .as_str(),
1090                if numeric_scale == 0 {
1091                    "1.2-8-2"
1092                } else {
1093                    "1.2-8-2.0"
1094                }
1095            );
1096        }
1097    }
1098
1099    fn is_arbitrary_precision() -> bool {
1100        serde_json::from_str::<serde_json::Number>("1.2500")
1101            .unwrap()
1102            .to_string()
1103            == "1.2500"
1104    }
1105
1106    #[rstest]
1107    fn test_matched_order_decimal_routes() {
1108        let value = serde_json::json!(["0.1234567890123456789012345678", 9007199254740993u64]);
1109
1110        let expected = MatchedOrder {
1111            price: Decimal::from_str_exact("0.1234567890123456789012345678").unwrap(),
1112            size: Decimal::from(9_007_199_254_740_993u64),
1113        };
1114
1115        assert_eq!(
1116            serde_json::from_str::<MatchedOrder>(&value.to_string()).unwrap(),
1117            expected
1118        );
1119        assert_eq!(
1120            serde_json::from_value::<MatchedOrder>(value).unwrap(),
1121            expected
1122        );
1123
1124        for invalid in [
1125            serde_json::Value::Null,
1126            serde_json::json!(""),
1127            serde_json::json!(true),
1128        ] {
1129            assert!(
1130                serde_json::from_value::<MatchedOrder>(serde_json::json!([invalid, 1])).is_err()
1131            );
1132        }
1133    }
1134
1135    #[rstest]
1136    #[case("1e400")]
1137    #[case("-1e400")]
1138    fn test_deserialize_numeric_out_of_range(#[case] input: &str) {
1139        let required = deserialize_f64(&mut serde_json::Deserializer::from_str(input));
1140        let optional = deserialize_optional_f64(&mut serde_json::Deserializer::from_str(input));
1141
1142        assert!(required.is_err());
1143        assert!(optional.is_err());
1144    }
1145
1146    #[rstest]
1147    #[case(false)]
1148    #[case(true)]
1149    fn test_stream_decode_race_optional_floats(#[case] explicit_null: bool) {
1150        let runner = if explicit_null {
1151            serde_json::json!({"lat": null, "long": null, "spd": null, "prg": null, "sfq": null})
1152        } else {
1153            serde_json::json!({})
1154        };
1155        let progress = if explicit_null {
1156            serde_json::json!({"st": null, "rt": null, "spd": null, "prg": null})
1157        } else {
1158            serde_json::json!({})
1159        };
1160        let json = serde_json::json!({
1161            "op": "rcm", "pt": 123, "rc": [{"rrc": [runner], "rpc": progress}]
1162        });
1163        let StreamMessage::RaceChange(message) =
1164            stream_decode(json.to_string().as_bytes()).unwrap()
1165        else {
1166            panic!("Expected race change");
1167        };
1168        let races = message.rc.unwrap();
1169        let runner = &races[0].rrc.as_ref().unwrap()[0];
1170        let progress = races[0].rpc.as_ref().unwrap();
1171
1172        assert_eq!(
1173            (runner.lat, runner.lng, runner.spd, runner.prg, runner.sfq),
1174            (None, None, None, None, None)
1175        );
1176        assert_eq!(
1177            (progress.st, progress.rt, progress.spd, progress.prg),
1178            (None, None, None, None)
1179        );
1180    }
1181
1182    #[rstest]
1183    #[case(serde_json::json!({"J": 2}))]
1184    #[case(serde_json::json!({"J": 2, "L": null}))]
1185    #[case(serde_json::json!({"J": 2, "L": "370.1"}))]
1186    #[case(serde_json::json!({"J": 2, "L": {"number": "370.1"}}))]
1187    fn test_stream_decode_jump_requires_number(#[case] jump: serde_json::Value) {
1188        let json = serde_json::json!({
1189            "op": "rcm", "pt": 123, "rc": [{"rpc": {"J": [jump]}}]
1190        });
1191
1192        assert!(stream_decode(json.to_string().as_bytes()).is_err());
1193    }
1194
1195    #[rstest]
1196    fn test_jump_serialization() {
1197        let jump = Jump {
1198            number: 2,
1199            distance: 370.1,
1200        };
1201
1202        assert_eq!(
1203            serde_json::to_value(jump).unwrap(),
1204            serde_json::json!({"J": 2, "L": 370.1})
1205        );
1206    }
1207
1208    #[rstest]
1209    #[case("stream/ocm_NEW_FULL_IMAGE.json")]
1210    #[case("stream/ocm_FILLED.json")]
1211    #[case("stream/ocm_FULL_IMAGE.json")]
1212    #[case("stream/ocm_FULL_IMAGE_STRATEGY.json")]
1213    #[case("stream/ocm_CANCEL.json")]
1214    #[case("stream/ocm_UPDATE.json")]
1215    #[case("stream/ocm_SUB_IMAGE.json")]
1216    #[case("stream/ocm_MIXED.json")]
1217    #[case("stream/ocm_EMPTY_IMAGE.json")]
1218    #[case("stream/ocm_error_fill.json")]
1219    #[case("stream/ocm_filled_different_price.json")]
1220    #[case("stream/ocm_order_update.json")]
1221    fn test_stream_decode_ocm_fixtures(#[case] fixture: &str) {
1222        let data = load_test_json(fixture);
1223        let msg = stream_decode(data.as_bytes()).unwrap_or_else(|e| panic!("{fixture}: {e}"));
1224        assert!(matches!(msg, StreamMessage::OrderChange(_)), "{fixture}");
1225    }
1226
1227    #[rstest]
1228    #[case("stream/mcm_SUB_IMAGE.json")]
1229    #[case("stream/mcm_SUB_IMAGE_no_market_def.json")]
1230    #[case("stream/mcm_UPDATE.json")]
1231    #[case("stream/mcm_UPDATE_md.json")]
1232    #[case("stream/mcm_UPDATE_tv.json")]
1233    #[case("stream/mcm_HEARTBEAT.json")]
1234    #[case("stream/mcm_RESUB_DELTA.json")]
1235    #[case("stream/mcm_live_IMAGE.json")]
1236    #[case("stream/mcm_live_UPDATE.json")]
1237    #[case("stream/mcm_latency.json")]
1238    #[case("stream/market_definition_racing.json")]
1239    #[case("stream/market_definition_runner_removed.json")]
1240    fn test_stream_decode_mcm_fixtures(#[case] fixture: &str) {
1241        let data = load_test_json(fixture);
1242        let msg = stream_decode(data.as_bytes()).unwrap_or_else(|e| panic!("{fixture}: {e}"));
1243        assert!(matches!(msg, StreamMessage::MarketChange(_)), "{fixture}");
1244    }
1245
1246    /// Fixtures containing a JSON array of multiple MCM messages.
1247    #[rstest]
1248    #[case("stream/mcm_BSP.json")]
1249    #[case("stream/market_updates.json")]
1250    fn test_stream_decode_mcm_multi_fixtures(#[case] fixture: &str) {
1251        let data = load_test_json(fixture);
1252        let msgs: Vec<StreamMessage> =
1253            serde_json::from_str(&data).unwrap_or_else(|e| panic!("{fixture}: {e}"));
1254        assert!(!msgs.is_empty(), "{fixture}: empty array");
1255        for msg in &msgs {
1256            assert!(matches!(msg, StreamMessage::MarketChange(_)), "{fixture}");
1257        }
1258    }
1259
1260    /// Fixtures containing a JSON array of multiple OCM messages.
1261    #[rstest]
1262    #[case("stream/ocm_multiple_fills.json")]
1263    #[case("stream/ocm_DUPLICATE_EXECUTION.json")]
1264    fn test_stream_decode_ocm_multi_fixtures(#[case] fixture: &str) {
1265        let data = load_test_json(fixture);
1266        let msgs: Vec<StreamMessage> =
1267            serde_json::from_str(&data).unwrap_or_else(|e| panic!("{fixture}: {e}"));
1268        assert!(!msgs.is_empty(), "{fixture}: empty array");
1269        for msg in &msgs {
1270            assert!(matches!(msg, StreamMessage::OrderChange(_)), "{fixture}");
1271        }
1272    }
1273
1274    #[rstest]
1275    fn test_stream_decode_mcm_segments() {
1276        let data = load_test_json("stream/mcm_SEGMENTS.jsonl");
1277        let expected = [
1278            (SegmentType::SegStart, "1.100001", None),
1279            (SegmentType::Seg, "1.100002", None),
1280            (SegmentType::SegEnd, "1.100003", Some("mcm-segment-clk")),
1281        ];
1282
1283        let messages: Vec<StreamMessage> = data
1284            .lines()
1285            .map(|line| stream_decode(line.as_bytes()).unwrap())
1286            .collect();
1287
1288        assert_eq!(messages.len(), expected.len());
1289        for (message, (segment_type, market_id, clk)) in messages.into_iter().zip(expected) {
1290            let StreamMessage::MarketChange(mcm) = message else {
1291                panic!("Expected MarketChange");
1292            };
1293            assert_eq!(mcm.id, Some(1));
1294            assert_eq!(mcm.pt, 1_700_000_000_000);
1295            assert_eq!(mcm.clk.as_deref(), clk);
1296            assert_eq!(mcm.initial_clk, None);
1297            assert_eq!(mcm.ct, None);
1298            assert_eq!(mcm.conflate_ms, None);
1299            assert_eq!(mcm.heartbeat_ms, None);
1300            assert_eq!(mcm.segment_type, Some(segment_type));
1301            assert_eq!(mcm.status, None);
1302            let market_changes = mcm.mc.unwrap();
1303            assert_eq!(market_changes.len(), 1);
1304            assert_eq!(market_changes[0].id, market_id);
1305        }
1306    }
1307
1308    #[rstest]
1309    fn test_stream_decode_ocm_segments() {
1310        let data = load_test_json("stream/ocm_SEGMENTS.jsonl");
1311        let expected = [
1312            (SegmentType::SegStart, "1.100001", None),
1313            (SegmentType::Seg, "1.100002", None),
1314            (SegmentType::SegEnd, "1.100003", Some("ocm-segment-clk")),
1315        ];
1316
1317        let messages: Vec<StreamMessage> = data
1318            .lines()
1319            .map(|line| stream_decode(line.as_bytes()).unwrap())
1320            .collect();
1321
1322        assert_eq!(messages.len(), expected.len());
1323        for (message, (segment_type, market_id, clk)) in messages.into_iter().zip(expected) {
1324            let StreamMessage::OrderChange(ocm) = message else {
1325                panic!("Expected OrderChange");
1326            };
1327            assert_eq!(ocm.id, Some(1));
1328            assert_eq!(ocm.pt, 1_700_000_000_000);
1329            assert_eq!(ocm.clk.as_deref(), clk);
1330            assert_eq!(ocm.initial_clk, None);
1331            assert_eq!(ocm.ct, None);
1332            assert_eq!(ocm.conflate_ms, None);
1333            assert_eq!(ocm.heartbeat_ms, None);
1334            assert_eq!(ocm.segment_type, Some(segment_type));
1335            assert_eq!(ocm.status, None);
1336            let order_changes = ocm.oc.unwrap();
1337            assert_eq!(order_changes.len(), 1);
1338            assert_eq!(order_changes[0].id, market_id);
1339            assert_eq!(order_changes[0].orc.as_ref().unwrap().len(), 0);
1340        }
1341    }
1342
1343    #[rstest]
1344    fn test_stream_decode_connection() {
1345        let data = load_test_json("stream/connection.json");
1346        let msg = stream_decode(data.as_bytes()).unwrap();
1347        match msg {
1348            StreamMessage::Connection(conn) => {
1349                assert_eq!(conn.connection_id, "002-051134157842-432409");
1350            }
1351            other => panic!("Expected Connection, was {other:?}"),
1352        }
1353    }
1354
1355    #[rstest]
1356    fn test_stream_decode_status() {
1357        let data = load_test_json("stream/status.json");
1358        let msg = stream_decode(data.as_bytes()).unwrap();
1359        assert!(matches!(msg, StreamMessage::Status(_)));
1360    }
1361
1362    #[rstest]
1363    fn test_stream_decode_lenient_sp_fields() {
1364        let data = r#"{
1365            "op":"mcm",
1366            "pt":1773304044929,
1367            "mc":[{
1368                "id":"1.255095842",
1369                "rc":[{
1370                    "id":96146807,
1371                    "spn":"Infinity",
1372                    "spf":"NaN",
1373                    "ltp":5.0,
1374                    "tv":10.63
1375                }, {
1376                    "id":96146808,
1377                    "spn":6.75,
1378                    "spf":7.5e-1
1379                }]
1380            }]
1381        }"#;
1382
1383        let msg = stream_decode(data.as_bytes()).unwrap();
1384
1385        match msg {
1386            StreamMessage::MarketChange(mcm) => {
1387                let rc = &mcm.mc.as_ref().unwrap()[0].rc.as_ref().unwrap()[0];
1388                assert_eq!(rc.spn, None);
1389                assert_eq!(rc.spf, None);
1390                assert_eq!(rc.ltp, Some(Decimal::new(50, 1)));
1391                assert_eq!(rc.tv, Some(Decimal::new(1063, 2)));
1392                let rc = &mcm.mc.as_ref().unwrap()[0].rc.as_ref().unwrap()[1];
1393                assert_eq!(rc.spn, Some(Decimal::new(675, 2)));
1394                assert_eq!(rc.spf, Some(Decimal::new(75, 2)));
1395            }
1396            other => panic!("Expected MarketChange, was {other:?}"),
1397        }
1398    }
1399
1400    #[rstest]
1401    fn test_market_definition_standalone() {
1402        let data = load_test_json("stream/market_definition.json");
1403        let _def: MarketDefinition = serde_json::from_str(&data).unwrap();
1404    }
1405
1406    #[rstest]
1407    #[case("rest/market_definition_open.json")]
1408    #[case("rest/market_definition_closed.json")]
1409    #[case("rest/market_definition_runner_removed.json")]
1410    fn test_market_definition_response_fixtures(#[case] fixture: &str) {
1411        let data = load_test_json(fixture);
1412        let _def: MarketDefinition = serde_json::from_str(&data).unwrap();
1413    }
1414
1415    #[rstest]
1416    fn test_stream_decode_rcm_single() {
1417        let data = load_test_json("stream/rcm_single.json");
1418        let msg = stream_decode(data.as_bytes()).unwrap();
1419        match msg {
1420            StreamMessage::RaceChange(rcm) => {
1421                let rc = rcm.rc.as_ref().unwrap();
1422                assert_eq!(rc.len(), 1);
1423
1424                let race = &rc[0];
1425                assert_eq!(race.id.as_deref(), Some("28587288.1650"));
1426                assert_eq!(race.mid.as_deref(), Some("1.1234567"));
1427
1428                let runners = race.rrc.as_ref().unwrap();
1429                assert_eq!(runners.len(), 1);
1430                assert_eq!(runners[0].id, Some(7390417));
1431                assert!((runners[0].lat.unwrap() - 51.4189543).abs() < 1e-6);
1432                assert!((runners[0].spd.unwrap() - 17.8).abs() < 1e-6);
1433                assert!((runners[0].sfq.unwrap() - 2.07).abs() < 1e-6);
1434
1435                let progress = race.rpc.as_ref().unwrap();
1436                assert_eq!(progress.g.as_deref(), Some("1f"));
1437                assert!((progress.st.unwrap() - 10.6).abs() < 1e-6);
1438                assert!((progress.rt.unwrap() - 46.7).abs() < 1e-6);
1439
1440                let order = progress.ord.as_ref().unwrap();
1441                assert_eq!(order.len(), 5);
1442                assert_eq!(order[0], 7390417);
1443
1444                let jumps = progress.jumps.as_ref().unwrap();
1445                assert_eq!(jumps.len(), 2);
1446                assert_eq!(jumps[0].number, 2);
1447                assert!((jumps[0].distance - 370.1).abs() < 1e-6);
1448            }
1449            other => panic!("Expected RaceChange, was {other:?}"),
1450        }
1451    }
1452
1453    #[rstest]
1454    fn test_stream_decode_rcm_multi_runner() {
1455        let data = load_test_json("stream/rcm_multi_runner.json");
1456        let msg = stream_decode(data.as_bytes()).unwrap();
1457        match msg {
1458            StreamMessage::RaceChange(rcm) => {
1459                let rc = rcm.rc.as_ref().unwrap();
1460                let runners = rc[0].rrc.as_ref().unwrap();
1461                assert_eq!(runners.len(), 5);
1462
1463                let ids: Vec<i64> = runners.iter().filter_map(|r| r.id).collect();
1464                assert_eq!(ids, vec![35467839, 24947967, 299569, 31422647, 41694785]);
1465            }
1466            other => panic!("Expected RaceChange, was {other:?}"),
1467        }
1468    }
1469
1470    #[rstest]
1471    fn test_stream_decode_ccm_single() {
1472        let data = load_test_json("stream/ccm_single.json");
1473        let msg = stream_decode(data.as_bytes()).unwrap();
1474        match msg {
1475            StreamMessage::CricketChange(ccm) => {
1476                let cc = ccm.cc.as_ref().unwrap();
1477                assert_eq!(cc.len(), 1);
1478                assert_eq!(cc[0].event_id.as_deref(), Some("35741575"));
1479                assert_eq!(cc[0].market_id.as_deref(), Some("1.259334639"));
1480                assert!(cc[0].match_stats.is_some());
1481            }
1482            other => panic!("Expected CricketChange, was {other:?}"),
1483        }
1484    }
1485
1486    #[rstest]
1487    fn test_stream_decode_ocm_voided() {
1488        let data = load_test_json("stream/ocm_VOIDED.json");
1489        let msg = stream_decode(data.as_bytes()).unwrap();
1490        match msg {
1491            StreamMessage::OrderChange(ocm) => {
1492                let oc = ocm.oc.as_ref().unwrap();
1493                let orc = oc[0].orc.as_ref().unwrap();
1494                let uo = &orc[0].uo.as_ref().unwrap()[0];
1495                assert_eq!(uo.sv.unwrap(), rust_decimal::Decimal::from(50));
1496                assert_eq!(uo.sm.unwrap(), rust_decimal::Decimal::from(50));
1497                assert_eq!(uo.s, rust_decimal::Decimal::from(100));
1498            }
1499            other => panic!("Expected OrderChange, was {other:?}"),
1500        }
1501    }
1502
1503    #[rstest]
1504    fn test_stream_decode_ocm_missing_persistence_type_for_market_on_close() {
1505        let data = r#"{
1506            "op":"ocm",
1507            "id":1,
1508            "pt":1775175455685,
1509            "clk":"clk-1",
1510            "oc":[{
1511                "id":"1.256134154",
1512                "orc":[{
1513                    "id":77465280,
1514                    "uo":[{
1515                        "id":"424009603606",
1516                        "p":1.01,
1517                        "s":2.00,
1518                        "side":"B",
1519                        "status":"E",
1520                        "ot":"MOC",
1521                        "pd":1775175455000,
1522                        "sr":2.00
1523                    }]
1524                }]
1525            }]
1526        }"#;
1527
1528        let msg = stream_decode(data.as_bytes()).unwrap();
1529
1530        match msg {
1531            StreamMessage::OrderChange(ocm) => {
1532                let oc = ocm.oc.as_ref().unwrap();
1533                let orc = oc[0].orc.as_ref().unwrap();
1534                let uo = &orc[0].uo.as_ref().unwrap()[0];
1535                assert_eq!(uo.pt, None);
1536                assert_eq!(
1537                    uo.ot,
1538                    crate::common::enums::StreamingOrderType::MarketOnClose
1539                );
1540            }
1541            other => panic!("Expected OrderChange, was {other:?}"),
1542        }
1543    }
1544
1545    #[rstest]
1546    fn test_authentication_serializes_and_redacts_debug() {
1547        let authentication = Authentication::with_id(
1548            "application-key-token".to_string(),
1549            "session-token-value".to_string(),
1550            42,
1551        );
1552
1553        let json = serde_json::to_value(&authentication).unwrap();
1554        let formatted = format!("{authentication:?}");
1555
1556        assert_eq!(json["op"], STREAM_OP_AUTHENTICATION);
1557        assert_eq!(json["id"], 42);
1558        assert_eq!(json["appKey"], "application-key-token");
1559        assert_eq!(json["session"], "session-token-value");
1560        assert_eq!(
1561            formatted,
1562            "Authentication { op: \"authentication\", id: Some(42), app_key: <redacted>, session: <redacted> }",
1563        );
1564        assert!(!formatted.contains("application-key-token"));
1565        assert!(!formatted.contains("session-token-value"));
1566    }
1567}
nautilus_betfair/stream/messages.rs

nautilus_betfair/stream/
messages.rs
