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
597/// Market subscription request.
598#[derive(Debug, Clone, Serialize)]
599#[serde(rename_all = "camelCase")]
600pub struct MarketSubscription {
601    pub op: String,
602    pub id: Option<u64>,
603    pub market_filter: StreamMarketFilter,
604    pub market_data_filter: MarketDataFilter,
605    #[serde(skip_serializing_if = "Option::is_none")]
606    pub clk: Option<String>,
607    #[serde(skip_serializing_if = "Option::is_none")]
608    pub conflate_ms: Option<u64>,
609    #[serde(skip_serializing_if = "Option::is_none")]
610    pub heartbeat_ms: Option<u64>,
611    #[serde(skip_serializing_if = "Option::is_none")]
612    pub initial_clk: Option<String>,
613    #[serde(skip_serializing_if = "Option::is_none")]
614    pub segmentation_enabled: Option<bool>,
615}
616
617/// Order subscription request.
618#[derive(Debug, Clone, Serialize)]
619#[serde(rename_all = "camelCase")]
620pub struct OrderSubscription {
621    pub op: String,
622    pub id: Option<u64>,
623    #[serde(skip_serializing_if = "Option::is_none")]
624    pub order_filter: Option<OrderFilter>,
625    #[serde(skip_serializing_if = "Option::is_none")]
626    pub clk: Option<String>,
627    #[serde(skip_serializing_if = "Option::is_none")]
628    pub conflate_ms: Option<u64>,
629    #[serde(skip_serializing_if = "Option::is_none")]
630    pub heartbeat_ms: Option<u64>,
631    #[serde(skip_serializing_if = "Option::is_none")]
632    pub initial_clk: Option<String>,
633    #[serde(skip_serializing_if = "Option::is_none")]
634    pub segmentation_enabled: Option<bool>,
635}
636
637/// Race stream subscription request.
638#[derive(Debug, Clone, Serialize)]
639#[serde(rename_all = "camelCase")]
640pub struct RaceSubscription {
641    pub op: String,
642    pub id: Option<u64>,
643}
644
645impl RaceSubscription {
646    #[must_use]
647    pub fn new(id: u64) -> Self {
648        Self {
649            op: STREAM_OP_RACE_SUBSCRIPTION.to_string(),
650            id: Some(id),
651        }
652    }
653}
654
655/// Cricket stream subscription request.
656#[derive(Debug, Clone, Serialize)]
657#[serde(rename_all = "camelCase")]
658pub struct CricketSubscription {
659    pub op: String,
660    pub id: Option<u64>,
661}
662
663impl CricketSubscription {
664    #[must_use]
665    pub fn new(id: u64) -> Self {
666        Self {
667            op: STREAM_OP_CRICKET_SUBSCRIPTION.to_string(),
668            id: Some(id),
669        }
670    }
671}
672
673/// Heartbeat request to keep the connection alive.
674#[derive(Debug, Clone, Serialize)]
675pub struct StreamHeartbeat {
676    pub op: String,
677    pub id: Option<u64>,
678}
679
680impl StreamHeartbeat {
681    #[must_use]
682    pub fn new() -> Self {
683        Self {
684            op: STREAM_OP_HEARTBEAT.to_string(),
685            id: None,
686        }
687    }
688}
689
690impl Default for StreamHeartbeat {
691    fn default() -> Self {
692        Self::new()
693    }
694}
695
696/// Market filter for streaming subscriptions.
697#[derive(Debug, Clone, Default, Serialize, Deserialize)]
698#[serde(rename_all = "camelCase")]
699pub struct StreamMarketFilter {
700    #[serde(skip_serializing_if = "Option::is_none")]
701    pub betting_types: Option<Vec<MarketBettingType>>,
702    #[serde(skip_serializing_if = "Option::is_none")]
703    pub bsp_market: Option<bool>,
704    #[serde(skip_serializing_if = "Option::is_none")]
705    pub country_codes: Option<Vec<Ustr>>,
706    #[serde(skip_serializing_if = "Option::is_none")]
707    pub event_ids: Option<Vec<String>>,
708    #[serde(skip_serializing_if = "Option::is_none")]
709    pub event_type_ids: Option<Vec<String>>,
710    #[serde(skip_serializing_if = "Option::is_none")]
711    pub market_ids: Option<Vec<MarketId>>,
712    #[serde(skip_serializing_if = "Option::is_none")]
713    pub market_types: Option<Vec<Ustr>>,
714    #[serde(skip_serializing_if = "Option::is_none")]
715    pub race_types: Option<Vec<Ustr>>,
716    #[serde(skip_serializing_if = "Option::is_none")]
717    pub turn_in_play_enabled: Option<bool>,
718    #[serde(skip_serializing_if = "Option::is_none")]
719    pub venues: Option<Vec<Ustr>>,
720}
721
722/// Market data filter for streaming subscriptions.
723#[derive(Debug, Clone, Default, Serialize, Deserialize)]
724#[serde(rename_all = "camelCase")]
725pub struct MarketDataFilter {
726    #[serde(skip_serializing_if = "Option::is_none")]
727    pub fields: Option<Vec<MarketDataFilterField>>,
728    #[serde(skip_serializing_if = "Option::is_none")]
729    pub ladder_levels: Option<u32>,
730}
731
732/// Order filter for streaming subscriptions.
733#[derive(Debug, Clone, Serialize, Deserialize)]
734#[serde(rename_all = "camelCase")]
735pub struct OrderFilter {
736    #[serde(default = "default_true")]
737    pub include_overall_position: bool,
738    #[serde(skip_serializing_if = "Option::is_none")]
739    pub customer_strategy_refs: Option<Vec<String>>,
740    #[serde(default)]
741    pub partition_matched_by_strategy_ref: bool,
742    #[serde(skip_serializing_if = "Option::is_none")]
743    pub account_ids: Option<Vec<u64>>,
744}
745
746impl Default for OrderFilter {
747    fn default() -> Self {
748        Self {
749            include_overall_position: true,
750            customer_strategy_refs: None,
751            partition_matched_by_strategy_ref: false,
752            account_ids: None,
753        }
754    }
755}
756
757fn default_true() -> bool {
758    true
759}
760
761/// Race Change Message (RCM) - live GPS tracking data (Total Performance Data).
762#[derive(Debug, Clone, Deserialize)]
763pub struct RCM {
764    pub id: Option<u64>,
765    /// Publish time (epoch millis).
766    pub pt: u64,
767    /// Clock token (may be integer or string depending on feed state).
768    pub clk: Option<serde_json::Value>,
769    /// Race changes (None on heartbeat).
770    pub rc: Option<Vec<RaceChange>>,
771}
772
773/// Delta update for a single race within an RCM.
774#[derive(Debug, Clone, Deserialize)]
775pub struct RaceChange {
776    /// Race identifier (e.g. "28587288.1650").
777    pub id: Option<String>,
778    /// Betfair market identifier.
779    pub mid: Option<String>,
780    /// Individual runner GPS data changes.
781    pub rrc: Option<Vec<RaceRunnerChange>>,
782    /// Overall race progress summary.
783    pub rpc: Option<RaceProgressChange>,
784}
785
786/// GPS tracking data for a single runner.
787#[derive(Debug, Clone, Deserialize)]
788pub struct RaceRunnerChange {
789    /// Feed time (epoch millis).
790    pub ft: Option<u64>,
791    /// Selection identifier.
792    pub id: Option<i64>,
793    /// Latitude (GPS coordinate).
794    pub lat: Option<f64>,
795    /// Longitude (GPS coordinate).
796    #[serde(rename = "long")]
797    pub lng: Option<f64>,
798    /// Speed in m/s (Doppler-derived).
799    pub spd: Option<f64>,
800    /// Distance to finish in meters.
801    pub prg: Option<f64>,
802    /// Stride frequency in Hz.
803    pub sfq: Option<f64>,
804}
805
806/// Race-level progress summary.
807#[derive(Debug, Clone, Deserialize)]
808pub struct RaceProgressChange {
809    /// Feed time (epoch millis).
810    pub ft: Option<u64>,
811    /// Gate/sectional name (e.g. "1f", "2f", "Finish").
812    pub g: Option<String>,
813    /// Sectional time in seconds.
814    pub st: Option<f64>,
815    /// Running time since race start in seconds.
816    pub rt: Option<f64>,
817    /// Speed of lead horse in m/s.
818    pub spd: Option<f64>,
819    /// Distance to finish for leading horse in meters.
820    pub prg: Option<f64>,
821    /// Runner order by selection ID (current race position).
822    pub ord: Option<Vec<i64>>,
823    /// Obstacle data for jump races.
824    #[serde(rename = "J")]
825    pub jumps: Option<Vec<Jump>>,
826}
827
828/// Cricket Change Message (CCM) - live cricket match data.
829#[derive(Debug, Clone, Deserialize)]
830pub struct CCM {
831    /// Subscription identifier.
832    pub id: Option<u64>,
833    /// Publish time (epoch millis).
834    pub pt: u64,
835    /// Clock token (may be integer or string depending on feed state).
836    pub clk: Option<serde_json::Value>,
837    /// Cricket match changes (None on heartbeat).
838    pub cc: Option<Vec<CricketChange>>,
839}
840
841/// Delta update for a single cricket match within a CCM.
842#[derive(Debug, Clone, Deserialize)]
843#[serde(rename_all = "camelCase")]
844pub struct CricketChange {
845    /// Betfair event identifier.
846    #[serde(default, deserialize_with = "deserialize_optional_string_lenient")]
847    pub event_id: Option<String>,
848    /// Betfair market identifier.
849    pub market_id: Option<String>,
850    /// Fixture metadata.
851    pub fixture_info: Option<serde_json::Value>,
852    /// Home team metadata.
853    pub home_team: Option<serde_json::Value>,
854    /// Away team metadata.
855    pub away_team: Option<serde_json::Value>,
856    /// Match statistics.
857    pub match_stats: Option<serde_json::Value>,
858    /// Match incidents.
859    pub incident_list_wrapper: Option<serde_json::Value>,
860}
861
862/// Jump obstacle location data.
863#[derive(Debug, Clone, Serialize, Deserialize)]
864pub struct Jump {
865    /// Jump number.
866    #[serde(rename = "J")]
867    pub number: i32,
868    /// Distance from finish line in meters.
869    #[serde(rename = "L")]
870    pub distance: f64,
871}
872
873/// Decode a single JSON stream line into a [`StreamMessage`].
874///
875/// # Errors
876///
877/// Returns an error if the JSON is malformed or the `op` field is missing/unknown.
878pub fn stream_decode(data: &[u8]) -> Result<StreamMessage, serde_json::Error> {
879    serde_json::from_slice(data)
880}
881
882#[cfg(test)]
883mod tests {
884    use rstest::rstest;
885
886    use super::*;
887    use crate::common::testing::load_test_json;
888
889    #[rstest]
890    #[case("stream/ocm_NEW_FULL_IMAGE.json")]
891    #[case("stream/ocm_FILLED.json")]
892    #[case("stream/ocm_FULL_IMAGE.json")]
893    #[case("stream/ocm_FULL_IMAGE_STRATEGY.json")]
894    #[case("stream/ocm_CANCEL.json")]
895    #[case("stream/ocm_UPDATE.json")]
896    #[case("stream/ocm_SUB_IMAGE.json")]
897    #[case("stream/ocm_MIXED.json")]
898    #[case("stream/ocm_EMPTY_IMAGE.json")]
899    #[case("stream/ocm_error_fill.json")]
900    #[case("stream/ocm_filled_different_price.json")]
901    #[case("stream/ocm_order_update.json")]
902    fn test_stream_decode_ocm_fixtures(#[case] fixture: &str) {
903        let data = load_test_json(fixture);
904        let msg = stream_decode(data.as_bytes()).unwrap_or_else(|e| panic!("{fixture}: {e}"));
905        assert!(matches!(msg, StreamMessage::OrderChange(_)), "{fixture}");
906    }
907
908    #[rstest]
909    #[case("stream/mcm_SUB_IMAGE.json")]
910    #[case("stream/mcm_SUB_IMAGE_no_market_def.json")]
911    #[case("stream/mcm_UPDATE.json")]
912    #[case("stream/mcm_UPDATE_md.json")]
913    #[case("stream/mcm_UPDATE_tv.json")]
914    #[case("stream/mcm_HEARTBEAT.json")]
915    #[case("stream/mcm_RESUB_DELTA.json")]
916    #[case("stream/mcm_live_IMAGE.json")]
917    #[case("stream/mcm_live_UPDATE.json")]
918    #[case("stream/mcm_latency.json")]
919    #[case("stream/market_definition_racing.json")]
920    #[case("stream/market_definition_runner_removed.json")]
921    fn test_stream_decode_mcm_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::MarketChange(_)), "{fixture}");
925    }
926
927    /// Fixtures containing a JSON array of multiple MCM messages.
928    #[rstest]
929    #[case("stream/mcm_BSP.json")]
930    #[case("stream/market_updates.json")]
931    fn test_stream_decode_mcm_multi_fixtures(#[case] fixture: &str) {
932        let data = load_test_json(fixture);
933        let msgs: Vec<StreamMessage> =
934            serde_json::from_str(&data).unwrap_or_else(|e| panic!("{fixture}: {e}"));
935        assert!(!msgs.is_empty(), "{fixture}: empty array");
936        for msg in &msgs {
937            assert!(matches!(msg, StreamMessage::MarketChange(_)), "{fixture}");
938        }
939    }
940
941    /// Fixtures containing a JSON array of multiple OCM messages.
942    #[rstest]
943    #[case("stream/ocm_multiple_fills.json")]
944    #[case("stream/ocm_DUPLICATE_EXECUTION.json")]
945    fn test_stream_decode_ocm_multi_fixtures(#[case] fixture: &str) {
946        let data = load_test_json(fixture);
947        let msgs: Vec<StreamMessage> =
948            serde_json::from_str(&data).unwrap_or_else(|e| panic!("{fixture}: {e}"));
949        assert!(!msgs.is_empty(), "{fixture}: empty array");
950        for msg in &msgs {
951            assert!(matches!(msg, StreamMessage::OrderChange(_)), "{fixture}");
952        }
953    }
954
955    #[rstest]
956    fn test_stream_decode_connection() {
957        let data = load_test_json("stream/connection.json");
958        let msg = stream_decode(data.as_bytes()).unwrap();
959        match msg {
960            StreamMessage::Connection(conn) => {
961                assert_eq!(conn.connection_id, "002-051134157842-432409");
962            }
963            other => panic!("Expected Connection, was {other:?}"),
964        }
965    }
966
967    #[rstest]
968    fn test_stream_decode_status() {
969        let data = load_test_json("stream/status.json");
970        let msg = stream_decode(data.as_bytes()).unwrap();
971        assert!(matches!(msg, StreamMessage::Status(_)));
972    }
973
974    #[rstest]
975    fn test_stream_decode_lenient_sp_fields() {
976        let data = r#"{
977            "op":"mcm",
978            "pt":1773304044929,
979            "mc":[{
980                "id":"1.255095842",
981                "rc":[{
982                    "id":96146807,
983                    "spn":"Infinity",
984                    "spf":"NaN",
985                    "ltp":5.0,
986                    "tv":10.63
987                }]
988            }]
989        }"#;
990
991        let msg = stream_decode(data.as_bytes()).unwrap();
992
993        match msg {
994            StreamMessage::MarketChange(mcm) => {
995                let rc = &mcm.mc.as_ref().unwrap()[0].rc.as_ref().unwrap()[0];
996                assert_eq!(rc.spn, None);
997                assert_eq!(rc.spf, None);
998                assert_eq!(rc.ltp, Some(Decimal::new(50, 1)));
999                assert_eq!(rc.tv, Some(Decimal::new(1063, 2)));
1000            }
1001            other => panic!("Expected MarketChange, was {other:?}"),
1002        }
1003    }
1004
1005    #[rstest]
1006    fn test_market_definition_standalone() {
1007        let data = load_test_json("stream/market_definition.json");
1008        let _def: MarketDefinition = serde_json::from_str(&data).unwrap();
1009    }
1010
1011    #[rstest]
1012    #[case("rest/market_definition_open.json")]
1013    #[case("rest/market_definition_closed.json")]
1014    #[case("rest/market_definition_runner_removed.json")]
1015    fn test_market_definition_response_fixtures(#[case] fixture: &str) {
1016        let data = load_test_json(fixture);
1017        let _def: MarketDefinition = serde_json::from_str(&data).unwrap();
1018    }
1019
1020    #[rstest]
1021    fn test_stream_decode_rcm_single() {
1022        let data = load_test_json("stream/rcm_single.json");
1023        let msg = stream_decode(data.as_bytes()).unwrap();
1024        match msg {
1025            StreamMessage::RaceChange(rcm) => {
1026                let rc = rcm.rc.as_ref().unwrap();
1027                assert_eq!(rc.len(), 1);
1028
1029                let race = &rc[0];
1030                assert_eq!(race.id.as_deref(), Some("28587288.1650"));
1031                assert_eq!(race.mid.as_deref(), Some("1.1234567"));
1032
1033                let runners = race.rrc.as_ref().unwrap();
1034                assert_eq!(runners.len(), 1);
1035                assert_eq!(runners[0].id, Some(7390417));
1036                assert!((runners[0].lat.unwrap() - 51.4189543).abs() < 1e-6);
1037                assert!((runners[0].spd.unwrap() - 17.8).abs() < 1e-6);
1038                assert!((runners[0].sfq.unwrap() - 2.07).abs() < 1e-6);
1039
1040                let progress = race.rpc.as_ref().unwrap();
1041                assert_eq!(progress.g.as_deref(), Some("1f"));
1042                assert!((progress.st.unwrap() - 10.6).abs() < 1e-6);
1043                assert!((progress.rt.unwrap() - 46.7).abs() < 1e-6);
1044
1045                let order = progress.ord.as_ref().unwrap();
1046                assert_eq!(order.len(), 5);
1047                assert_eq!(order[0], 7390417);
1048
1049                let jumps = progress.jumps.as_ref().unwrap();
1050                assert_eq!(jumps.len(), 2);
1051                assert_eq!(jumps[0].number, 2);
1052                assert!((jumps[0].distance - 370.1).abs() < 1e-6);
1053            }
1054            other => panic!("Expected RaceChange, was {other:?}"),
1055        }
1056    }
1057
1058    #[rstest]
1059    fn test_stream_decode_rcm_multi_runner() {
1060        let data = load_test_json("stream/rcm_multi_runner.json");
1061        let msg = stream_decode(data.as_bytes()).unwrap();
1062        match msg {
1063            StreamMessage::RaceChange(rcm) => {
1064                let rc = rcm.rc.as_ref().unwrap();
1065                let runners = rc[0].rrc.as_ref().unwrap();
1066                assert_eq!(runners.len(), 5);
1067
1068                let ids: Vec<i64> = runners.iter().filter_map(|r| r.id).collect();
1069                assert_eq!(ids, vec![35467839, 24947967, 299569, 31422647, 41694785]);
1070            }
1071            other => panic!("Expected RaceChange, was {other:?}"),
1072        }
1073    }
1074
1075    #[rstest]
1076    fn test_stream_decode_ccm_single() {
1077        let data = load_test_json("stream/ccm_single.json");
1078        let msg = stream_decode(data.as_bytes()).unwrap();
1079        match msg {
1080            StreamMessage::CricketChange(ccm) => {
1081                let cc = ccm.cc.as_ref().unwrap();
1082                assert_eq!(cc.len(), 1);
1083                assert_eq!(cc[0].event_id.as_deref(), Some("35741575"));
1084                assert_eq!(cc[0].market_id.as_deref(), Some("1.259334639"));
1085                assert!(cc[0].match_stats.is_some());
1086            }
1087            other => panic!("Expected CricketChange, was {other:?}"),
1088        }
1089    }
1090
1091    #[rstest]
1092    fn test_stream_decode_ocm_voided() {
1093        let data = load_test_json("stream/ocm_VOIDED.json");
1094        let msg = stream_decode(data.as_bytes()).unwrap();
1095        match msg {
1096            StreamMessage::OrderChange(ocm) => {
1097                let oc = ocm.oc.as_ref().unwrap();
1098                let orc = oc[0].orc.as_ref().unwrap();
1099                let uo = &orc[0].uo.as_ref().unwrap()[0];
1100                assert_eq!(uo.sv.unwrap(), rust_decimal::Decimal::from(50));
1101                assert_eq!(uo.sm.unwrap(), rust_decimal::Decimal::from(50));
1102                assert_eq!(uo.s, rust_decimal::Decimal::from(100));
1103            }
1104            other => panic!("Expected OrderChange, was {other:?}"),
1105        }
1106    }
1107
1108    #[rstest]
1109    fn test_stream_decode_ocm_missing_persistence_type_for_market_on_close() {
1110        let data = r#"{
1111            "op":"ocm",
1112            "id":1,
1113            "pt":1775175455685,
1114            "clk":"clk-1",
1115            "oc":[{
1116                "id":"1.256134154",
1117                "orc":[{
1118                    "id":77465280,
1119                    "uo":[{
1120                        "id":"424009603606",
1121                        "p":1.01,
1122                        "s":2.00,
1123                        "side":"B",
1124                        "status":"E",
1125                        "ot":"MOC",
1126                        "pd":1775175455000,
1127                        "sr":2.00
1128                    }]
1129                }]
1130            }]
1131        }"#;
1132
1133        let msg = stream_decode(data.as_bytes()).unwrap();
1134
1135        match msg {
1136            StreamMessage::OrderChange(ocm) => {
1137                let oc = ocm.oc.as_ref().unwrap();
1138                let orc = oc[0].orc.as_ref().unwrap();
1139                let uo = &orc[0].uo.as_ref().unwrap()[0];
1140                assert_eq!(uo.pt, None);
1141                assert_eq!(
1142                    uo.ot,
1143                    crate::common::enums::StreamingOrderType::MarketOnClose
1144                );
1145            }
1146            other => panic!("Expected OrderChange, was {other:?}"),
1147        }
1148    }
1149}
nautilus_betfair/stream/messages.rs

nautilus_betfair/stream/
messages.rs
