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

nautilus_betfair/stream/
messages.rs
