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::{Deserialize, Deserializer, Serialize, de::Visitor};
35use ustr::Ustr;
36use zeroize::Zeroize;
37
38use crate::common::{
39    consts::{
40        STREAM_OP_AUTHENTICATION, STREAM_OP_CRICKET_SUBSCRIPTION, STREAM_OP_HEARTBEAT,
41        STREAM_OP_RACE_SUBSCRIPTION,
42    },
43    enums::{
44        ChangeType, LapseStatusReasonCode, MarketBettingType, MarketDataFilterField, MarketStatus,
45        PriceLadderType, RunnerStatus, SegmentType, StatusErrorCode, StreamingOrderStatus,
46        StreamingOrderType, StreamingPersistenceType, StreamingSide,
47    },
48    types::{
49        Handicap, MarketId, SelectionId, deserialize_optional_string_lenient,
50        deserialize_selection_id,
51    },
52};
53
54/// Top-level streaming message, discriminated by the `op` field.
55///
56/// Deserializing a raw JSON line into this enum replaces the Python
57/// `stream_decode()` function from `betfair_parser`.
58#[derive(Debug, Clone, Deserialize)]
59#[serde(tag = "op")]
60pub enum StreamMessage {
61    #[serde(rename = "connection")]
62    Connection(Connection),
63    #[serde(rename = "status")]
64    Status(Status),
65    #[serde(rename = "mcm")]
66    MarketChange(MCM),
67    #[serde(rename = "ocm")]
68    OrderChange(OCM),
69    #[serde(rename = "rcm")]
70    RaceChange(RCM),
71    #[serde(rename = "ccm")]
72    CricketChange(CCM),
73}
74
75/// Connection confirmation sent on stream connect.
76#[derive(Debug, Clone, Deserialize)]
77#[serde(rename_all = "camelCase")]
78pub struct Connection {
79    pub id: Option<u64>,
80    pub connection_id: String,
81}
82
83/// Status response for errors or informational messages.
84#[derive(Debug, Clone, Deserialize)]
85#[serde(rename_all = "camelCase")]
86pub struct Status {
87    pub id: Option<u64>,
88    pub connection_closed: bool,
89    pub connection_id: Option<String>,
90    pub connections_available: Option<u32>,
91    pub error_code: Option<StatusErrorCode>,
92    pub error_message: Option<String>,
93    pub status_code: Option<String>,
94}
95
96/// Market Change Message (MCM) - price/market data updates.
97#[derive(Debug, Clone, Deserialize)]
98pub struct MCM {
99    pub id: Option<u64>,
100    /// Publish time (epoch millis).
101    pub pt: u64,
102    /// Token used for resubscription.
103    pub clk: Option<String>,
104    /// Initial clock token (sent on first image).
105    #[serde(rename = "initialClk")]
106    pub initial_clk: Option<String>,
107    /// Market changes (None on heartbeat).
108    pub mc: Option<Vec<MarketChange>>,
109    /// Change type.
110    pub ct: Option<ChangeType>,
111    /// Conflation interval in milliseconds.
112    #[serde(rename = "conflateMs")]
113    pub conflate_ms: Option<u64>,
114    /// Heartbeat interval in milliseconds.
115    #[serde(rename = "heartbeatMs")]
116    pub heartbeat_ms: Option<u64>,
117    /// Segment type for large messages.
118    #[serde(rename = "segmentType")]
119    pub segment_type: Option<SegmentType>,
120    pub status: Option<i32>,
121}
122
123impl MCM {
124    #[must_use]
125    pub fn is_heartbeat(&self) -> bool {
126        self.ct == Some(ChangeType::Heartbeat)
127    }
128}
129
130/// Order Change Message (OCM) - order/position updates.
131#[derive(Debug, Clone, Deserialize)]
132pub struct OCM {
133    pub id: Option<u64>,
134    /// Publish time (epoch millis).
135    pub pt: u64,
136    pub clk: Option<String>,
137    #[serde(rename = "initialClk")]
138    pub initial_clk: Option<String>,
139    /// Order market changes (None on heartbeat).
140    pub oc: Option<Vec<OrderMarketChange>>,
141    pub ct: Option<ChangeType>,
142    #[serde(rename = "conflateMs")]
143    pub conflate_ms: Option<u64>,
144    #[serde(rename = "heartbeatMs")]
145    pub heartbeat_ms: Option<u64>,
146    #[serde(rename = "segmentType")]
147    pub segment_type: Option<SegmentType>,
148    pub status: Option<i32>,
149}
150
151impl OCM {
152    #[must_use]
153    pub fn is_heartbeat(&self) -> bool {
154        self.ct == Some(ChangeType::Heartbeat)
155    }
156}
157
158/// Delta update for a single market.
159#[derive(Debug, Clone, Deserialize)]
160pub struct MarketChange {
161    /// Market identifier.
162    pub id: MarketId,
163    /// Runner changes.
164    pub rc: Option<Vec<RunnerChange>>,
165    /// Whether there was a conflation.
166    pub con: Option<bool>,
167    /// Whether this is a full image (vs delta).
168    #[serde(default)]
169    pub img: bool,
170    /// Full market definition (sent on subscription or change).
171    #[serde(rename = "marketDefinition")]
172    pub market_definition: Option<MarketDefinition>,
173    /// Total volume matched on this market.
174    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
175    pub tv: Option<Decimal>,
176}
177
178/// Delta update for a single runner (selection).
179#[derive(Debug, Clone, Deserialize)]
180pub struct RunnerChange {
181    /// Selection identifier.
182    #[serde(deserialize_with = "deserialize_selection_id")]
183    pub id: SelectionId,
184    /// Handicap value.
185    pub hc: Option<Handicap>,
186    /// Available to back.
187    pub atb: Option<Vec<PV>>,
188    /// Available to lay.
189    pub atl: Option<Vec<PV>>,
190    /// Best available to back (depth).
191    pub batb: Option<Vec<LPV>>,
192    /// Best available to lay (depth).
193    pub batl: Option<Vec<LPV>>,
194    /// Best display available to back.
195    pub bdatb: Option<Vec<LPV>>,
196    /// Best display available to lay.
197    pub bdatl: Option<Vec<LPV>>,
198    /// Starting price back.
199    pub spb: Option<Vec<PV>>,
200    /// Starting price lay.
201    pub spl: Option<Vec<PV>>,
202    /// Starting price near (projected SP).
203    #[serde(default, deserialize_with = "deserialize_optional_decimal_lenient")]
204    pub spn: Option<Decimal>,
205    /// Starting price far (actual BSP).
206    #[serde(default, deserialize_with = "deserialize_optional_decimal_lenient")]
207    pub spf: Option<Decimal>,
208    /// Traded volume by price level.
209    pub trd: Option<Vec<PV>>,
210    /// Last traded price.
211    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
212    pub ltp: Option<Decimal>,
213    /// Total volume matched on this runner.
214    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
215    pub tv: Option<Decimal>,
216}
217
218fn deserialize_optional_decimal_lenient<'de, D>(
219    deserializer: D,
220) -> Result<Option<Decimal>, D::Error>
221where
222    D: Deserializer<'de>,
223{
224    struct LenientOptionalDecimalVisitor;
225
226    impl Visitor<'_> for LenientOptionalDecimalVisitor {
227        type Value = Option<Decimal>;
228
229        fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
230            formatter.write_str("null or a decimal number as string, integer, or float")
231        }
232
233        fn visit_str<E: serde::de::Error>(self, value: &str) -> Result<Self::Value, E> {
234            Ok(parse_optional_decimal_lenient(value))
235        }
236
237        fn visit_string<E: serde::de::Error>(self, value: String) -> Result<Self::Value, E> {
238            self.visit_str(&value)
239        }
240
241        fn visit_i64<E: serde::de::Error>(self, value: i64) -> Result<Self::Value, E> {
242            Ok(Some(Decimal::from(value)))
243        }
244
245        fn visit_u64<E: serde::de::Error>(self, value: u64) -> Result<Self::Value, E> {
246            Ok(Some(Decimal::from(value)))
247        }
248
249        fn visit_i128<E: serde::de::Error>(self, value: i128) -> Result<Self::Value, E> {
250            Ok(Some(Decimal::from(value)))
251        }
252
253        fn visit_u128<E: serde::de::Error>(self, value: u128) -> Result<Self::Value, E> {
254            Ok(Some(Decimal::from(value)))
255        }
256
257        fn visit_f64<E: serde::de::Error>(self, value: f64) -> Result<Self::Value, E> {
258            Ok(Decimal::try_from(value).ok())
259        }
260
261        fn visit_unit<E: serde::de::Error>(self) -> Result<Self::Value, E> {
262            Ok(None)
263        }
264
265        fn visit_none<E: serde::de::Error>(self) -> Result<Self::Value, E> {
266            Ok(None)
267        }
268    }
269
270    deserializer.deserialize_any(LenientOptionalDecimalVisitor)
271}
272
273fn parse_optional_decimal_lenient(value: &str) -> Option<Decimal> {
274    let trimmed = value.trim();
275    if trimmed.is_empty() || is_non_finite_decimal(trimmed) {
276        return None;
277    }
278
279    if trimmed.contains('e') || trimmed.contains('E') {
280        Decimal::from_scientific(trimmed).ok()
281    } else {
282        Decimal::from_str(trimmed).ok()
283    }
284}
285
286fn is_non_finite_decimal(value: &str) -> bool {
287    value.eq_ignore_ascii_case("nan")
288        || value.eq_ignore_ascii_case("inf")
289        || value.eq_ignore_ascii_case("+inf")
290        || value.eq_ignore_ascii_case("-inf")
291        || value.eq_ignore_ascii_case("infinity")
292        || value.eq_ignore_ascii_case("+infinity")
293        || value.eq_ignore_ascii_case("-infinity")
294}
295
296/// Full market definition snapshot.
297#[derive(Debug, Clone, Deserialize)]
298#[serde(rename_all = "camelCase")]
299pub struct MarketDefinition {
300    pub bet_delay: Option<i32>,
301    pub betting_type: Option<MarketBettingType>,
302    pub bsp_market: Option<bool>,
303    pub bsp_reconciled: Option<bool>,
304    pub competition_id: Option<String>,
305    pub competition_name: Option<String>,
306    pub complete: Option<bool>,
307    pub country_code: Option<Ustr>,
308    pub cross_matching: Option<bool>,
309    pub discount_allowed: Option<bool>,
310    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
311    pub each_way_divisor: Option<Decimal>,
312    pub event_id: Option<String>,
313    pub event_name: Option<String>,
314    #[serde(default, deserialize_with = "deserialize_optional_string_lenient")]
315    pub event_type_id: Option<String>,
316    pub event_type_name: Option<Ustr>,
317    pub in_play: Option<bool>,
318    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
319    pub line_interval: Option<Decimal>,
320    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
321    pub line_max_unit: Option<Decimal>,
322    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
323    pub line_min_unit: Option<Decimal>,
324    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
325    pub market_base_rate: Option<Decimal>,
326    pub market_id: Option<MarketId>,
327    pub market_name: Option<String>,
328    pub market_time: Option<String>,
329    pub market_type: Option<Ustr>,
330    pub number_of_active_runners: Option<u32>,
331    pub number_of_winners: Option<u32>,
332    pub open_date: Option<String>,
333    pub persistence_enabled: Option<bool>,
334    pub price_ladder_definition: Option<PriceLadderDefinition>,
335    pub race_type: Option<Ustr>,
336    pub regulators: Option<Vec<Ustr>>,
337    pub runners: Option<Vec<RunnerDefinition>>,
338    pub runners_voidable: Option<bool>,
339    pub settled_time: Option<String>,
340    pub status: Option<MarketStatus>,
341    pub suspend_time: Option<String>,
342    pub timezone: Option<Ustr>,
343    pub turn_in_play_enabled: Option<bool>,
344    pub venue: Option<Ustr>,
345    pub version: Option<u64>,
346}
347
348/// Runner (selection) definition within a market definition.
349#[derive(Debug, Clone, Deserialize)]
350#[serde(rename_all = "camelCase")]
351pub struct RunnerDefinition {
352    #[serde(deserialize_with = "deserialize_selection_id")]
353    pub id: SelectionId,
354    pub hc: Option<Handicap>,
355    pub sort_priority: Option<u32>,
356    pub name: Option<String>,
357    pub status: Option<RunnerStatus>,
358    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
359    pub adjustment_factor: Option<Decimal>,
360    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
361    pub bsp: Option<Decimal>,
362    pub removal_date: Option<String>,
363}
364
365/// Price ladder definition within a market definition.
366#[derive(Debug, Clone, Deserialize)]
367pub struct PriceLadderDefinition {
368    #[serde(rename = "type")]
369    pub ladder_type: Option<PriceLadderType>,
370}
371
372// Betfair encodes price-volume types as JSON arrays: [price, volume] and
373// [level, price, volume] respectively.
374
375/// Price-volume pair, serialized as a JSON array `[price, volume]`.
376#[derive(Debug, Clone, Copy, PartialEq)]
377pub struct PV {
378    pub price: Decimal,
379    pub volume: Decimal,
380}
381
382impl<'de> Deserialize<'de> for PV {
383    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
384    where
385        D: serde::Deserializer<'de>,
386    {
387        // Handles both `[price, volume]` and `[level, price, volume]` (RESUB_DELTA)
388        let arr: Vec<Decimal> = Deserialize::deserialize(deserializer)?;
389        match arr.len() {
390            2 => Ok(Self {
391                price: arr[0],
392                volume: arr[1],
393            }),
394            3 => Ok(Self {
395                price: arr[1],
396                volume: arr[2],
397            }),
398            n => Err(serde::de::Error::invalid_length(n, &"2 or 3 elements")),
399        }
400    }
401}
402
403impl Serialize for PV {
404    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
405    where
406        S: serde::Serializer,
407    {
408        (self.price, self.volume).serialize(serializer)
409    }
410}
411
412/// Level-price-volume triple, serialized as a JSON array `[level, price, volume]`.
413#[derive(Debug, Clone, Copy, PartialEq)]
414pub struct LPV {
415    pub level: u32,
416    pub price: Decimal,
417    pub volume: Decimal,
418}
419
420impl<'de> Deserialize<'de> for LPV {
421    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
422    where
423        D: serde::Deserializer<'de>,
424    {
425        let arr: (u32, Decimal, Decimal) = Deserialize::deserialize(deserializer)?;
426        Ok(Self {
427            level: arr.0,
428            price: arr.1,
429            volume: arr.2,
430        })
431    }
432}
433
434impl Serialize for LPV {
435    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
436    where
437        S: serde::Serializer,
438    {
439        (self.level, self.price, self.volume).serialize(serializer)
440    }
441}
442
443/// Order changes for a single market.
444#[derive(Debug, Clone, Deserialize)]
445pub struct OrderMarketChange {
446    /// Market identifier.
447    pub id: MarketId,
448    #[serde(rename = "accountId")]
449    pub account_id: Option<u64>,
450    pub closed: Option<bool>,
451    #[serde(rename = "fullImage", default)]
452    pub full_image: bool,
453    /// Order runner changes.
454    pub orc: Option<Vec<OrderRunnerChange>>,
455}
456
457/// Order changes for a single runner within a market.
458#[derive(Debug, Clone, Deserialize)]
459pub struct OrderRunnerChange {
460    /// Selection identifier.
461    #[serde(deserialize_with = "deserialize_selection_id")]
462    pub id: SelectionId,
463    #[serde(rename = "fullImage", default)]
464    pub full_image: bool,
465    /// Handicap.
466    pub hc: Option<Handicap>,
467    /// Matched backs.
468    pub mb: Option<Vec<MatchedOrder>>,
469    /// Matched lays.
470    pub ml: Option<Vec<MatchedOrder>>,
471    /// Strategy match changes, keyed by customer strategy ref.
472    pub smc: Option<AHashMap<String, StrategyMatchChange>>,
473    /// Unmatched orders.
474    pub uo: Option<Vec<UnmatchedOrder>>,
475}
476
477/// Matched order (price-size pair), serialized as `[price, size]`.
478#[derive(Debug, Clone, Copy, PartialEq)]
479pub struct MatchedOrder {
480    pub price: Decimal,
481    pub size: Decimal,
482}
483
484impl<'de> Deserialize<'de> for MatchedOrder {
485    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
486    where
487        D: serde::Deserializer<'de>,
488    {
489        let arr: (Decimal, Decimal) = Deserialize::deserialize(deserializer)?;
490        Ok(Self {
491            price: arr.0,
492            size: arr.1,
493        })
494    }
495}
496
497impl Serialize for MatchedOrder {
498    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
499    where
500        S: serde::Serializer,
501    {
502        (self.price, self.size).serialize(serializer)
503    }
504}
505
506/// Strategy-level match changes.
507#[derive(Debug, Clone, Deserialize)]
508pub struct StrategyMatchChange {
509    /// Matched backs.
510    pub mb: Option<Vec<MatchedOrder>>,
511    /// Matched lays.
512    pub ml: Option<Vec<MatchedOrder>>,
513}
514
515/// Unmatched order on the streaming API.
516#[derive(Debug, Clone, Deserialize)]
517pub struct UnmatchedOrder {
518    /// Bet identifier.
519    pub id: String,
520    /// Price.
521    #[serde(deserialize_with = "deserialize_decimal")]
522    pub p: Decimal,
523    /// Size.
524    #[serde(deserialize_with = "deserialize_decimal")]
525    pub s: Decimal,
526    /// Side (B=Back, L=Lay).
527    pub side: StreamingSide,
528    /// Order status (E=Executable, EC=ExecutionComplete).
529    pub status: StreamingOrderStatus,
530    /// Persistence type (L=Lapse, P=Persist, MOC=MarketOnClose).
531    ///
532    /// Betfair can omit this on some BSP market-on-close order updates.
533    #[serde(default)]
534    pub pt: Option<StreamingPersistenceType>,
535    /// Order type (L=Limit, LOC=LimitOnClose, MOC=MarketOnClose).
536    pub ot: StreamingOrderType,
537    /// Placed date (epoch millis).
538    pub pd: u64,
539    /// BSP liability.
540    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
541    pub bsp: Option<Decimal>,
542    /// Customer strategy reference.
543    pub rfo: Option<String>,
544    /// Regulator reference.
545    pub rfs: Option<String>,
546    /// Customer order reference.
547    pub rc: Option<String>,
548    /// Regulator auth code.
549    pub rac: Option<String>,
550    /// Matched date (epoch millis).
551    pub md: Option<u64>,
552    /// Cancelled date (epoch millis).
553    pub cd: Option<u64>,
554    /// Lapsed date (epoch millis).
555    pub ld: Option<u64>,
556    /// Average price matched.
557    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
558    pub avp: Option<Decimal>,
559    /// Size matched.
560    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
561    pub sm: Option<Decimal>,
562    /// Size remaining.
563    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
564    pub sr: Option<Decimal>,
565    /// Size lapsed.
566    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
567    pub sl: Option<Decimal>,
568    /// Size cancelled.
569    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
570    pub sc: Option<Decimal>,
571    /// Size voided.
572    #[serde(default, deserialize_with = "deserialize_optional_decimal")]
573    pub sv: Option<Decimal>,
574    /// Lapse status reason code.
575    pub lsrc: Option<LapseStatusReasonCode>,
576}
577
578/// Authentication request sent on stream connect.
579#[derive(Debug, Clone, Serialize, Zeroize)]
580pub struct Authentication {
581    pub op: String,
582    pub id: Option<u64>,
583    #[serde(rename = "appKey")]
584    pub app_key: SecretString,
585    pub session: SecretString,
586}
587
588impl Authentication {
589    /// Creates a new authentication request.
590    #[must_use]
591    pub fn new(app_key: impl Into<SecretString>, session: impl Into<SecretString>) -> Self {
592        Self {
593            op: STREAM_OP_AUTHENTICATION.to_string(),
594            id: None,
595            app_key: app_key.into(),
596            session: session.into(),
597        }
598    }
599
600    /// Creates a correlated authentication request.
601    #[must_use]
602    pub fn with_id(
603        app_key: impl Into<SecretString>,
604        session: impl Into<SecretString>,
605        id: u64,
606    ) -> Self {
607        Self {
608            op: STREAM_OP_AUTHENTICATION.to_string(),
609            id: Some(id),
610            app_key: app_key.into(),
611            session: session.into(),
612        }
613    }
614}
615
616/// Market subscription request.
617#[derive(Debug, Clone, Serialize)]
618#[serde(rename_all = "camelCase")]
619pub struct MarketSubscription {
620    pub op: String,
621    pub id: Option<u64>,
622    pub market_filter: StreamMarketFilter,
623    pub market_data_filter: MarketDataFilter,
624    #[serde(skip_serializing_if = "Option::is_none")]
625    pub clk: Option<String>,
626    #[serde(skip_serializing_if = "Option::is_none")]
627    pub conflate_ms: Option<u64>,
628    #[serde(skip_serializing_if = "Option::is_none")]
629    pub heartbeat_ms: Option<u64>,
630    #[serde(skip_serializing_if = "Option::is_none")]
631    pub initial_clk: Option<String>,
632    #[serde(skip_serializing_if = "Option::is_none")]
633    pub segmentation_enabled: Option<bool>,
634}
635
636/// Order subscription request.
637#[derive(Debug, Clone, Serialize)]
638#[serde(rename_all = "camelCase")]
639pub struct OrderSubscription {
640    pub op: String,
641    pub id: Option<u64>,
642    #[serde(skip_serializing_if = "Option::is_none")]
643    pub order_filter: Option<OrderFilter>,
644    #[serde(skip_serializing_if = "Option::is_none")]
645    pub clk: Option<String>,
646    #[serde(skip_serializing_if = "Option::is_none")]
647    pub conflate_ms: Option<u64>,
648    #[serde(skip_serializing_if = "Option::is_none")]
649    pub heartbeat_ms: Option<u64>,
650    #[serde(skip_serializing_if = "Option::is_none")]
651    pub initial_clk: Option<String>,
652    #[serde(skip_serializing_if = "Option::is_none")]
653    pub segmentation_enabled: Option<bool>,
654}
655
656/// Race stream subscription request.
657#[derive(Debug, Clone, Serialize)]
658#[serde(rename_all = "camelCase")]
659pub struct RaceSubscription {
660    pub op: String,
661    pub id: Option<u64>,
662}
663
664impl RaceSubscription {
665    #[must_use]
666    pub fn new(id: u64) -> Self {
667        Self {
668            op: STREAM_OP_RACE_SUBSCRIPTION.to_string(),
669            id: Some(id),
670        }
671    }
672}
673
674/// Cricket stream subscription request.
675#[derive(Debug, Clone, Serialize)]
676#[serde(rename_all = "camelCase")]
677pub struct CricketSubscription {
678    pub op: String,
679    pub id: Option<u64>,
680}
681
682impl CricketSubscription {
683    #[must_use]
684    pub fn new(id: u64) -> Self {
685        Self {
686            op: STREAM_OP_CRICKET_SUBSCRIPTION.to_string(),
687            id: Some(id),
688        }
689    }
690}
691
692/// Heartbeat request to keep the connection alive.
693#[derive(Debug, Clone, Serialize)]
694pub struct StreamHeartbeat {
695    pub op: String,
696    pub id: Option<u64>,
697}
698
699impl StreamHeartbeat {
700    #[must_use]
701    pub fn new() -> Self {
702        Self {
703            op: STREAM_OP_HEARTBEAT.to_string(),
704            id: None,
705        }
706    }
707}
708
709impl Default for StreamHeartbeat {
710    fn default() -> Self {
711        Self::new()
712    }
713}
714
715/// Market filter for streaming subscriptions.
716#[derive(Debug, Clone, Default, Serialize, Deserialize)]
717#[serde(rename_all = "camelCase")]
718pub struct StreamMarketFilter {
719    #[serde(skip_serializing_if = "Option::is_none")]
720    pub betting_types: Option<Vec<MarketBettingType>>,
721    #[serde(skip_serializing_if = "Option::is_none")]
722    pub bsp_market: Option<bool>,
723    #[serde(skip_serializing_if = "Option::is_none")]
724    pub country_codes: Option<Vec<Ustr>>,
725    #[serde(skip_serializing_if = "Option::is_none")]
726    pub event_ids: Option<Vec<String>>,
727    #[serde(skip_serializing_if = "Option::is_none")]
728    pub event_type_ids: Option<Vec<String>>,
729    #[serde(skip_serializing_if = "Option::is_none")]
730    pub market_ids: Option<Vec<MarketId>>,
731    #[serde(skip_serializing_if = "Option::is_none")]
732    pub market_types: Option<Vec<Ustr>>,
733    #[serde(skip_serializing_if = "Option::is_none")]
734    pub race_types: Option<Vec<Ustr>>,
735    #[serde(skip_serializing_if = "Option::is_none")]
736    pub turn_in_play_enabled: Option<bool>,
737    #[serde(skip_serializing_if = "Option::is_none")]
738    pub venues: Option<Vec<Ustr>>,
739}
740
741/// Market data filter for streaming subscriptions.
742#[derive(Debug, Clone, Default, Serialize, Deserialize)]
743#[serde(rename_all = "camelCase")]
744pub struct MarketDataFilter {
745    #[serde(skip_serializing_if = "Option::is_none")]
746    pub fields: Option<Vec<MarketDataFilterField>>,
747    #[serde(skip_serializing_if = "Option::is_none")]
748    pub ladder_levels: Option<u32>,
749}
750
751/// Order filter for streaming subscriptions.
752#[derive(Debug, Clone, Serialize, Deserialize)]
753#[serde(rename_all = "camelCase")]
754pub struct OrderFilter {
755    #[serde(default = "default_true")]
756    pub include_overall_position: bool,
757    #[serde(skip_serializing_if = "Option::is_none")]
758    pub customer_strategy_refs: Option<Vec<String>>,
759    #[serde(default)]
760    pub partition_matched_by_strategy_ref: bool,
761    #[serde(skip_serializing_if = "Option::is_none")]
762    pub account_ids: Option<Vec<u64>>,
763}
764
765impl Default for OrderFilter {
766    fn default() -> Self {
767        Self {
768            include_overall_position: true,
769            customer_strategy_refs: None,
770            partition_matched_by_strategy_ref: false,
771            account_ids: None,
772        }
773    }
774}
775
776fn default_true() -> bool {
777    true
778}
779
780/// Race Change Message (RCM) - live GPS tracking data (Total Performance Data).
781#[derive(Debug, Clone, Deserialize)]
782pub struct RCM {
783    pub id: Option<u64>,
784    /// Publish time (epoch millis).
785    pub pt: u64,
786    /// Clock token (may be integer or string depending on feed state).
787    pub clk: Option<serde_json::Value>,
788    /// Race changes (None on heartbeat).
789    pub rc: Option<Vec<RaceChange>>,
790}
791
792/// Delta update for a single race within an RCM.
793#[derive(Debug, Clone, Deserialize)]
794pub struct RaceChange {
795    /// Race identifier (e.g. "28587288.1650").
796    pub id: Option<String>,
797    /// Betfair market identifier.
798    pub mid: Option<String>,
799    /// Individual runner GPS data changes.
800    pub rrc: Option<Vec<RaceRunnerChange>>,
801    /// Overall race progress summary.
802    pub rpc: Option<RaceProgressChange>,
803}
804
805/// GPS tracking data for a single runner.
806#[derive(Debug, Clone, Deserialize)]
807pub struct RaceRunnerChange {
808    /// Feed time (epoch millis).
809    pub ft: Option<u64>,
810    /// Selection identifier.
811    pub id: Option<i64>,
812    /// Latitude (GPS coordinate).
813    pub lat: Option<f64>,
814    /// Longitude (GPS coordinate).
815    #[serde(rename = "long")]
816    pub lng: Option<f64>,
817    /// Speed in m/s (Doppler-derived).
818    pub spd: Option<f64>,
819    /// Distance to finish in meters.
820    pub prg: Option<f64>,
821    /// Stride frequency in Hz.
822    pub sfq: Option<f64>,
823}
824
825/// Race-level progress summary.
826#[derive(Debug, Clone, Deserialize)]
827pub struct RaceProgressChange {
828    /// Feed time (epoch millis).
829    pub ft: Option<u64>,
830    /// Gate/sectional name (e.g. "1f", "2f", "Finish").
831    pub g: Option<String>,
832    /// Sectional time in seconds.
833    pub st: Option<f64>,
834    /// Running time since race start in seconds.
835    pub rt: Option<f64>,
836    /// Speed of lead horse in m/s.
837    pub spd: Option<f64>,
838    /// Distance to finish for leading horse in meters.
839    pub prg: Option<f64>,
840    /// Runner order by selection ID (current race position).
841    pub ord: Option<Vec<i64>>,
842    /// Obstacle data for jump races.
843    #[serde(rename = "J")]
844    pub jumps: Option<Vec<Jump>>,
845}
846
847/// Cricket Change Message (CCM) - live cricket match data.
848#[derive(Debug, Clone, Deserialize)]
849pub struct CCM {
850    /// Subscription identifier.
851    pub id: Option<u64>,
852    /// Publish time (epoch millis).
853    pub pt: u64,
854    /// Clock token (may be integer or string depending on feed state).
855    pub clk: Option<serde_json::Value>,
856    /// Cricket match changes (None on heartbeat).
857    pub cc: Option<Vec<CricketChange>>,
858}
859
860/// Delta update for a single cricket match within a CCM.
861#[derive(Debug, Clone, Deserialize)]
862#[serde(rename_all = "camelCase")]
863pub struct CricketChange {
864    /// Betfair event identifier.
865    #[serde(default, deserialize_with = "deserialize_optional_string_lenient")]
866    pub event_id: Option<String>,
867    /// Betfair market identifier.
868    pub market_id: Option<String>,
869    /// Fixture metadata.
870    pub fixture_info: Option<serde_json::Value>,
871    /// Home team metadata.
872    pub home_team: Option<serde_json::Value>,
873    /// Away team metadata.
874    pub away_team: Option<serde_json::Value>,
875    /// Match statistics.
876    pub match_stats: Option<serde_json::Value>,
877    /// Match incidents.
878    pub incident_list_wrapper: Option<serde_json::Value>,
879}
880
881/// Jump obstacle location data.
882#[derive(Debug, Clone, Serialize, Deserialize)]
883pub struct Jump {
884    /// Jump number.
885    #[serde(rename = "J")]
886    pub number: i32,
887    /// Distance from finish line in meters.
888    #[serde(rename = "L")]
889    pub distance: f64,
890}
891
892/// Decode a single JSON stream line into a [`StreamMessage`].
893///
894/// # Errors
895///
896/// Returns an error if the JSON is malformed or the `op` field is missing/unknown.
897pub fn stream_decode(data: &[u8]) -> Result<StreamMessage, serde_json::Error> {
898    serde_json::from_slice(data)
899}
900
901#[cfg(test)]
902mod tests {
903    use rstest::rstest;
904
905    use super::*;
906    use crate::common::testing::load_test_json;
907
908    #[rstest]
909    #[case("stream/ocm_NEW_FULL_IMAGE.json")]
910    #[case("stream/ocm_FILLED.json")]
911    #[case("stream/ocm_FULL_IMAGE.json")]
912    #[case("stream/ocm_FULL_IMAGE_STRATEGY.json")]
913    #[case("stream/ocm_CANCEL.json")]
914    #[case("stream/ocm_UPDATE.json")]
915    #[case("stream/ocm_SUB_IMAGE.json")]
916    #[case("stream/ocm_MIXED.json")]
917    #[case("stream/ocm_EMPTY_IMAGE.json")]
918    #[case("stream/ocm_error_fill.json")]
919    #[case("stream/ocm_filled_different_price.json")]
920    #[case("stream/ocm_order_update.json")]
921    fn test_stream_decode_ocm_fixtures(#[case] fixture: &str) {
922        let data = load_test_json(fixture);
923        let msg = stream_decode(data.as_bytes()).unwrap_or_else(|e| panic!("{fixture}: {e}"));
924        assert!(matches!(msg, StreamMessage::OrderChange(_)), "{fixture}");
925    }
926
927    #[rstest]
928    #[case("stream/mcm_SUB_IMAGE.json")]
929    #[case("stream/mcm_SUB_IMAGE_no_market_def.json")]
930    #[case("stream/mcm_UPDATE.json")]
931    #[case("stream/mcm_UPDATE_md.json")]
932    #[case("stream/mcm_UPDATE_tv.json")]
933    #[case("stream/mcm_HEARTBEAT.json")]
934    #[case("stream/mcm_RESUB_DELTA.json")]
935    #[case("stream/mcm_live_IMAGE.json")]
936    #[case("stream/mcm_live_UPDATE.json")]
937    #[case("stream/mcm_latency.json")]
938    #[case("stream/market_definition_racing.json")]
939    #[case("stream/market_definition_runner_removed.json")]
940    fn test_stream_decode_mcm_fixtures(#[case] fixture: &str) {
941        let data = load_test_json(fixture);
942        let msg = stream_decode(data.as_bytes()).unwrap_or_else(|e| panic!("{fixture}: {e}"));
943        assert!(matches!(msg, StreamMessage::MarketChange(_)), "{fixture}");
944    }
945
946    /// Fixtures containing a JSON array of multiple MCM messages.
947    #[rstest]
948    #[case("stream/mcm_BSP.json")]
949    #[case("stream/market_updates.json")]
950    fn test_stream_decode_mcm_multi_fixtures(#[case] fixture: &str) {
951        let data = load_test_json(fixture);
952        let msgs: Vec<StreamMessage> =
953            serde_json::from_str(&data).unwrap_or_else(|e| panic!("{fixture}: {e}"));
954        assert!(!msgs.is_empty(), "{fixture}: empty array");
955        for msg in &msgs {
956            assert!(matches!(msg, StreamMessage::MarketChange(_)), "{fixture}");
957        }
958    }
959
960    /// Fixtures containing a JSON array of multiple OCM messages.
961    #[rstest]
962    #[case("stream/ocm_multiple_fills.json")]
963    #[case("stream/ocm_DUPLICATE_EXECUTION.json")]
964    fn test_stream_decode_ocm_multi_fixtures(#[case] fixture: &str) {
965        let data = load_test_json(fixture);
966        let msgs: Vec<StreamMessage> =
967            serde_json::from_str(&data).unwrap_or_else(|e| panic!("{fixture}: {e}"));
968        assert!(!msgs.is_empty(), "{fixture}: empty array");
969        for msg in &msgs {
970            assert!(matches!(msg, StreamMessage::OrderChange(_)), "{fixture}");
971        }
972    }
973
974    #[rstest]
975    fn test_stream_decode_mcm_segments() {
976        let data = load_test_json("stream/mcm_SEGMENTS.jsonl");
977        let expected = [
978            (SegmentType::SegStart, "1.100001", None),
979            (SegmentType::Seg, "1.100002", None),
980            (SegmentType::SegEnd, "1.100003", Some("mcm-segment-clk")),
981        ];
982
983        let messages: Vec<StreamMessage> = data
984            .lines()
985            .map(|line| stream_decode(line.as_bytes()).unwrap())
986            .collect();
987
988        assert_eq!(messages.len(), expected.len());
989        for (message, (segment_type, market_id, clk)) in messages.into_iter().zip(expected) {
990            let StreamMessage::MarketChange(mcm) = message else {
991                panic!("Expected MarketChange");
992            };
993            assert_eq!(mcm.id, Some(1));
994            assert_eq!(mcm.pt, 1_700_000_000_000);
995            assert_eq!(mcm.clk.as_deref(), clk);
996            assert_eq!(mcm.initial_clk, None);
997            assert_eq!(mcm.ct, None);
998            assert_eq!(mcm.conflate_ms, None);
999            assert_eq!(mcm.heartbeat_ms, None);
1000            assert_eq!(mcm.segment_type, Some(segment_type));
1001            assert_eq!(mcm.status, None);
1002            let market_changes = mcm.mc.unwrap();
1003            assert_eq!(market_changes.len(), 1);
1004            assert_eq!(market_changes[0].id, market_id);
1005        }
1006    }
1007
1008    #[rstest]
1009    fn test_stream_decode_ocm_segments() {
1010        let data = load_test_json("stream/ocm_SEGMENTS.jsonl");
1011        let expected = [
1012            (SegmentType::SegStart, "1.100001", None),
1013            (SegmentType::Seg, "1.100002", None),
1014            (SegmentType::SegEnd, "1.100003", Some("ocm-segment-clk")),
1015        ];
1016
1017        let messages: Vec<StreamMessage> = data
1018            .lines()
1019            .map(|line| stream_decode(line.as_bytes()).unwrap())
1020            .collect();
1021
1022        assert_eq!(messages.len(), expected.len());
1023        for (message, (segment_type, market_id, clk)) in messages.into_iter().zip(expected) {
1024            let StreamMessage::OrderChange(ocm) = message else {
1025                panic!("Expected OrderChange");
1026            };
1027            assert_eq!(ocm.id, Some(1));
1028            assert_eq!(ocm.pt, 1_700_000_000_000);
1029            assert_eq!(ocm.clk.as_deref(), clk);
1030            assert_eq!(ocm.initial_clk, None);
1031            assert_eq!(ocm.ct, None);
1032            assert_eq!(ocm.conflate_ms, None);
1033            assert_eq!(ocm.heartbeat_ms, None);
1034            assert_eq!(ocm.segment_type, Some(segment_type));
1035            assert_eq!(ocm.status, None);
1036            let order_changes = ocm.oc.unwrap();
1037            assert_eq!(order_changes.len(), 1);
1038            assert_eq!(order_changes[0].id, market_id);
1039            assert_eq!(order_changes[0].orc.as_ref().unwrap().len(), 0);
1040        }
1041    }
1042
1043    #[rstest]
1044    fn test_stream_decode_connection() {
1045        let data = load_test_json("stream/connection.json");
1046        let msg = stream_decode(data.as_bytes()).unwrap();
1047        match msg {
1048            StreamMessage::Connection(conn) => {
1049                assert_eq!(conn.connection_id, "002-051134157842-432409");
1050            }
1051            other => panic!("Expected Connection, was {other:?}"),
1052        }
1053    }
1054
1055    #[rstest]
1056    fn test_stream_decode_status() {
1057        let data = load_test_json("stream/status.json");
1058        let msg = stream_decode(data.as_bytes()).unwrap();
1059        assert!(matches!(msg, StreamMessage::Status(_)));
1060    }
1061
1062    #[rstest]
1063    fn test_stream_decode_lenient_sp_fields() {
1064        let data = r#"{
1065            "op":"mcm",
1066            "pt":1773304044929,
1067            "mc":[{
1068                "id":"1.255095842",
1069                "rc":[{
1070                    "id":96146807,
1071                    "spn":"Infinity",
1072                    "spf":"NaN",
1073                    "ltp":5.0,
1074                    "tv":10.63
1075                }]
1076            }]
1077        }"#;
1078
1079        let msg = stream_decode(data.as_bytes()).unwrap();
1080
1081        match msg {
1082            StreamMessage::MarketChange(mcm) => {
1083                let rc = &mcm.mc.as_ref().unwrap()[0].rc.as_ref().unwrap()[0];
1084                assert_eq!(rc.spn, None);
1085                assert_eq!(rc.spf, None);
1086                assert_eq!(rc.ltp, Some(Decimal::new(50, 1)));
1087                assert_eq!(rc.tv, Some(Decimal::new(1063, 2)));
1088            }
1089            other => panic!("Expected MarketChange, was {other:?}"),
1090        }
1091    }
1092
1093    #[rstest]
1094    fn test_market_definition_standalone() {
1095        let data = load_test_json("stream/market_definition.json");
1096        let _def: MarketDefinition = serde_json::from_str(&data).unwrap();
1097    }
1098
1099    #[rstest]
1100    #[case("rest/market_definition_open.json")]
1101    #[case("rest/market_definition_closed.json")]
1102    #[case("rest/market_definition_runner_removed.json")]
1103    fn test_market_definition_response_fixtures(#[case] fixture: &str) {
1104        let data = load_test_json(fixture);
1105        let _def: MarketDefinition = serde_json::from_str(&data).unwrap();
1106    }
1107
1108    #[rstest]
1109    fn test_stream_decode_rcm_single() {
1110        let data = load_test_json("stream/rcm_single.json");
1111        let msg = stream_decode(data.as_bytes()).unwrap();
1112        match msg {
1113            StreamMessage::RaceChange(rcm) => {
1114                let rc = rcm.rc.as_ref().unwrap();
1115                assert_eq!(rc.len(), 1);
1116
1117                let race = &rc[0];
1118                assert_eq!(race.id.as_deref(), Some("28587288.1650"));
1119                assert_eq!(race.mid.as_deref(), Some("1.1234567"));
1120
1121                let runners = race.rrc.as_ref().unwrap();
1122                assert_eq!(runners.len(), 1);
1123                assert_eq!(runners[0].id, Some(7390417));
1124                assert!((runners[0].lat.unwrap() - 51.4189543).abs() < 1e-6);
1125                assert!((runners[0].spd.unwrap() - 17.8).abs() < 1e-6);
1126                assert!((runners[0].sfq.unwrap() - 2.07).abs() < 1e-6);
1127
1128                let progress = race.rpc.as_ref().unwrap();
1129                assert_eq!(progress.g.as_deref(), Some("1f"));
1130                assert!((progress.st.unwrap() - 10.6).abs() < 1e-6);
1131                assert!((progress.rt.unwrap() - 46.7).abs() < 1e-6);
1132
1133                let order = progress.ord.as_ref().unwrap();
1134                assert_eq!(order.len(), 5);
1135                assert_eq!(order[0], 7390417);
1136
1137                let jumps = progress.jumps.as_ref().unwrap();
1138                assert_eq!(jumps.len(), 2);
1139                assert_eq!(jumps[0].number, 2);
1140                assert!((jumps[0].distance - 370.1).abs() < 1e-6);
1141            }
1142            other => panic!("Expected RaceChange, was {other:?}"),
1143        }
1144    }
1145
1146    #[rstest]
1147    fn test_stream_decode_rcm_multi_runner() {
1148        let data = load_test_json("stream/rcm_multi_runner.json");
1149        let msg = stream_decode(data.as_bytes()).unwrap();
1150        match msg {
1151            StreamMessage::RaceChange(rcm) => {
1152                let rc = rcm.rc.as_ref().unwrap();
1153                let runners = rc[0].rrc.as_ref().unwrap();
1154                assert_eq!(runners.len(), 5);
1155
1156                let ids: Vec<i64> = runners.iter().filter_map(|r| r.id).collect();
1157                assert_eq!(ids, vec![35467839, 24947967, 299569, 31422647, 41694785]);
1158            }
1159            other => panic!("Expected RaceChange, was {other:?}"),
1160        }
1161    }
1162
1163    #[rstest]
1164    fn test_stream_decode_ccm_single() {
1165        let data = load_test_json("stream/ccm_single.json");
1166        let msg = stream_decode(data.as_bytes()).unwrap();
1167        match msg {
1168            StreamMessage::CricketChange(ccm) => {
1169                let cc = ccm.cc.as_ref().unwrap();
1170                assert_eq!(cc.len(), 1);
1171                assert_eq!(cc[0].event_id.as_deref(), Some("35741575"));
1172                assert_eq!(cc[0].market_id.as_deref(), Some("1.259334639"));
1173                assert!(cc[0].match_stats.is_some());
1174            }
1175            other => panic!("Expected CricketChange, was {other:?}"),
1176        }
1177    }
1178
1179    #[rstest]
1180    fn test_stream_decode_ocm_voided() {
1181        let data = load_test_json("stream/ocm_VOIDED.json");
1182        let msg = stream_decode(data.as_bytes()).unwrap();
1183        match msg {
1184            StreamMessage::OrderChange(ocm) => {
1185                let oc = ocm.oc.as_ref().unwrap();
1186                let orc = oc[0].orc.as_ref().unwrap();
1187                let uo = &orc[0].uo.as_ref().unwrap()[0];
1188                assert_eq!(uo.sv.unwrap(), rust_decimal::Decimal::from(50));
1189                assert_eq!(uo.sm.unwrap(), rust_decimal::Decimal::from(50));
1190                assert_eq!(uo.s, rust_decimal::Decimal::from(100));
1191            }
1192            other => panic!("Expected OrderChange, was {other:?}"),
1193        }
1194    }
1195
1196    #[rstest]
1197    fn test_stream_decode_ocm_missing_persistence_type_for_market_on_close() {
1198        let data = r#"{
1199            "op":"ocm",
1200            "id":1,
1201            "pt":1775175455685,
1202            "clk":"clk-1",
1203            "oc":[{
1204                "id":"1.256134154",
1205                "orc":[{
1206                    "id":77465280,
1207                    "uo":[{
1208                        "id":"424009603606",
1209                        "p":1.01,
1210                        "s":2.00,
1211                        "side":"B",
1212                        "status":"E",
1213                        "ot":"MOC",
1214                        "pd":1775175455000,
1215                        "sr":2.00
1216                    }]
1217                }]
1218            }]
1219        }"#;
1220
1221        let msg = stream_decode(data.as_bytes()).unwrap();
1222
1223        match msg {
1224            StreamMessage::OrderChange(ocm) => {
1225                let oc = ocm.oc.as_ref().unwrap();
1226                let orc = oc[0].orc.as_ref().unwrap();
1227                let uo = &orc[0].uo.as_ref().unwrap()[0];
1228                assert_eq!(uo.pt, None);
1229                assert_eq!(
1230                    uo.ot,
1231                    crate::common::enums::StreamingOrderType::MarketOnClose
1232                );
1233            }
1234            other => panic!("Expected OrderChange, was {other:?}"),
1235        }
1236    }
1237
1238    #[rstest]
1239    fn test_authentication_serializes_and_redacts_debug() {
1240        let authentication = Authentication::with_id(
1241            "application-key-token".to_string(),
1242            "session-token-value".to_string(),
1243            42,
1244        );
1245
1246        let json = serde_json::to_value(&authentication).unwrap();
1247        let formatted = format!("{authentication:?}");
1248
1249        assert_eq!(json["op"], STREAM_OP_AUTHENTICATION);
1250        assert_eq!(json["id"], 42);
1251        assert_eq!(json["appKey"], "application-key-token");
1252        assert_eq!(json["session"], "session-token-value");
1253        assert_eq!(
1254            formatted,
1255            "Authentication { op: \"authentication\", id: Some(42), app_key: <redacted>, session: <redacted> }",
1256        );
1257        assert!(!formatted.contains("application-key-token"));
1258        assert!(!formatted.contains("session-token-value"));
1259    }
1260}
nautilus_betfair/stream/messages.rs

nautilus_betfair/stream/
messages.rs
