nautilus_betfair/
execution.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//! Live execution client for the Betfair adapter.
17//!
18//! # Stream reconnect lifecycle
19//!
20//! On every reconnect (second `Connection` message), the OCM handler raises
21//! both `pending_resync` (drains buffered OCMs on the next command) and
22//! `is_reconciling` (halts exposure-increasing commands). The reconnect
23//! background task then refreshes the session, fetches account state, and
24//! pulls a `list_current_orders` mass status to recover any fills that
25//! completed and rolled off the unmatched book during the gap. Once the
26//! mass status is dispatched the halt clears.
27//!
28//! While `is_reconciling` is set, `submit_order` and `submit_order_list`
29//! emit `OrderDenied` with `STREAM_RECONCILING`; cancels and modifies
30//! pass through unchanged. The halt is fail-open: a transient reconnect
31//! failure clears the flag rather than locking trading indefinitely,
32//! consistent with the rest of Nautilus.
33
34use std::{
35    fmt,
36    future::Future,
37    sync::{
38        Arc, Mutex,
39        atomic::{AtomicBool, Ordering},
40    },
41};
42
43use ahash::AHashSet;
44use async_trait::async_trait;
45use nautilus_common::{
46    clients::ExecutionClient,
47    live::{
48        get_runtime,
49        runner::{get_data_event_sender, get_exec_event_sender},
50    },
51    messages::{
52        DataEvent, ExecutionReport,
53        execution::{
54            BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
55            GenerateFillReportsBuilder, GenerateOrderStatusReports,
56            GenerateOrderStatusReportsBuilder, ModifyOrder, QueryOrder, SubmitOrder,
57            SubmitOrderList,
58        },
59    },
60};
61use nautilus_core::{
62    MUTEX_POISONED, UUID4, UnixNanos,
63    datetime::NANOSECONDS_IN_SECOND,
64    time::{AtomicTime, get_atomic_clock_realtime},
65};
66use nautilus_live::{ExecutionClientCore, ExecutionEventEmitter};
67use nautilus_model::{
68    accounts::AccountAny,
69    data::Data,
70    enums::{AccountType, OmsType, OrderStatus, OrderType, TimeInForce},
71    events::{
72        OrderAccepted, OrderCanceled, OrderDeniedReason, OrderEventAny, OrderFilled, OrderUpdated,
73    },
74    identifiers::{AccountId, ClientId, ClientOrderId, InstrumentId, Venue, VenueOrderId},
75    instruments::InstrumentAny,
76    orders::Order,
77    reports::{ExecutionMassStatus, FillReport, OrderStatusReport},
78    types::{AccountBalance, Currency, MarginBalance},
79};
80use nautilus_network::socket::TcpMessageHandler;
81use rust_decimal::Decimal;
82use tokio::task::JoinHandle;
83
84use crate::{
85    common::{
86        consts::{
87            BETFAIR_VENUE, METHOD_CANCEL_ORDERS, METHOD_GET_ACCOUNT_FUNDS,
88            METHOD_LIST_CURRENT_ORDERS, METHOD_PLACE_ORDERS, METHOD_REPLACE_ORDERS,
89        },
90        credential::BetfairCredential,
91        enums::{
92            BetfairOrderStatus, BetfairOrderType, BetfairSide, BetfairTimeInForce,
93            ExecutionReportErrorCode, ExecutionReportStatus, InstructionReportErrorCode,
94            InstructionReportStatus, OrderProjection, PersistenceType, StreamingOrderStatus,
95            StreamingSide,
96        },
97        parse::{
98            extract_market_id, extract_selection_id, make_customer_order_ref,
99            make_customer_order_ref_legacy, make_instrument_id, parse_account_state,
100            parse_betfair_quantity, parse_millis_timestamp,
101        },
102        types::{BetId, OrderSyncEntry},
103    },
104    config::BetfairExecConfig,
105    data::custom_data_with_instrument,
106    data_types::{BetfairOrderVoided, register_betfair_custom_data},
107    http::{
108        client::BetfairHttpClient,
109        models::{
110            AccountFundsResponse, CancelExecutionReport, CancelInstruction, CancelOrdersParams,
111            CurrentOrderSummary, CurrentOrderSummaryReport, LimitOnCloseOrder, LimitOrder,
112            ListCurrentOrdersParams, MarketOnCloseOrder, MarketVersion, PlaceExecutionReport,
113            PlaceInstruction, PlaceInstructionReport, PlaceOrdersParams, ReplaceExecutionReport,
114            ReplaceInstruction, ReplaceInstructionReport, ReplaceOrdersParams, TimeRange,
115        },
116        parse::{parse_current_order_fill_report, parse_current_order_report},
117    },
118    stream::{
119        client::BetfairStreamClient,
120        config::BetfairStreamConfig,
121        messages::{OCM, StreamMessage, stream_decode},
122        ocm::OcmState,
123        parse::{has_cancel_quantity, parse_order_status_report},
124    },
125};
126
127/// Betfair live execution client.
128#[derive(Debug)]
129pub struct BetfairExecutionClient {
130    core: ExecutionClientCore,
131    clock: &'static AtomicTime,
132    emitter: ExecutionEventEmitter,
133    http_client: Arc<BetfairHttpClient>,
134    stream_client: Option<Arc<BetfairStreamClient>>,
135    credential: BetfairCredential,
136    stream_config: BetfairStreamConfig,
137    config: BetfairExecConfig,
138    currency: Currency,
139    ocm_state: Arc<Mutex<OcmState>>,
140    pending_resync: Arc<AtomicBool>,
141    is_reconciling: Arc<AtomicBool>,
142    replay_buffer: Arc<Mutex<Vec<OCM>>>,
143    pending_tasks: Mutex<Vec<JoinHandle<()>>>,
144    keep_alive_handle: Option<JoinHandle<()>>,
145    account_state_handle: Option<JoinHandle<()>>,
146    reconnect_handle: Option<JoinHandle<()>>,
147}
148
149impl BetfairExecutionClient {
150    /// Creates a new [`BetfairExecutionClient`] instance.
151    #[must_use]
152    pub fn new(
153        core: ExecutionClientCore,
154        http_client: BetfairHttpClient,
155        credential: BetfairCredential,
156        stream_config: BetfairStreamConfig,
157        config: BetfairExecConfig,
158        currency: Currency,
159    ) -> Self {
160        let clock = get_atomic_clock_realtime();
161        let emitter = ExecutionEventEmitter::new(
162            clock,
163            core.trader_id,
164            core.account_id,
165            AccountType::Betting,
166            None,
167        );
168
169        Self {
170            core,
171            clock,
172            emitter,
173            http_client: Arc::new(http_client),
174            stream_client: None,
175            credential,
176            stream_config,
177            config,
178            currency,
179            ocm_state: Arc::new(Mutex::new(OcmState::default())),
180            pending_resync: Arc::new(AtomicBool::new(false)),
181            is_reconciling: Arc::new(AtomicBool::new(false)),
182            replay_buffer: Arc::new(Mutex::new(Vec::new())),
183            pending_tasks: Mutex::new(Vec::new()),
184            keep_alive_handle: None,
185            account_state_handle: None,
186            reconnect_handle: None,
187        }
188    }
189
190    /// Returns true while post-reconnect reconciliation is in flight.
191    #[must_use]
192    pub fn is_reconciling(&self) -> bool {
193        self.is_reconciling.load(Ordering::Acquire)
194    }
195
196    fn spawn_task<F>(&self, description: &'static str, fut: F)
197    where
198        F: Future<Output = anyhow::Result<()>> + Send + 'static,
199    {
200        let runtime = get_runtime();
201        let handle = runtime.spawn(async move {
202            if let Err(e) = fut.await {
203                log::warn!("{description} failed: {e:?}");
204            }
205        });
206
207        let mut tasks = self.pending_tasks.lock().expect(MUTEX_POISONED);
208        tasks.retain(|handle| !handle.is_finished());
209        tasks.push(handle);
210    }
211
212    fn reconcile_market_ids(&self) -> Option<Vec<String>> {
213        if self.config.reconcile_market_ids_only
214            && let Some(ids) = &self.config.reconcile_market_ids
215        {
216            return Some(ids.clone());
217        }
218        self.config.stream_market_ids_filter.clone()
219    }
220
221    /// Returns the market version for price protection on order placement.
222    ///
223    /// When `use_market_version` is enabled, reads the `version` field from
224    /// the instrument's `info` metadata. Betfair lapses orders submitted with
225    /// a stale version rather than matching against a moved book.
226    fn get_market_version(&self, instrument_id: &InstrumentId) -> Option<MarketVersion> {
227        if !self.config.use_market_version {
228            return None;
229        }
230
231        let cache = self.core.cache();
232        let instrument = cache.instrument(instrument_id)?;
233
234        if let InstrumentAny::Betting(betting) = instrument {
235            let version = betting.info.as_ref()?.get_i64("version")?;
236            return Some(MarketVersion {
237                version: Some(version),
238            });
239        }
240
241        None
242    }
243
244    /// Pre-populates OCM state from cached orders to prevent duplicate fills
245    /// and terminal events after reconnect.
246    fn sync_ocm_state_from_cache(&self) {
247        let cache = self.core.cache();
248        let venue = *BETFAIR_VENUE;
249        let orders = cache.orders(Some(&venue), None, None, None, None);
250
251        let order_data: Vec<OrderSyncEntry> = orders
252            .iter()
253            .filter_map(|order| {
254                let venue_order_id = order.venue_order_id()?;
255                let bet_id = venue_order_id.to_string();
256                let filled_qty = order.filled_qty().as_decimal();
257                let avg_px = order.avg_px().map_or(Decimal::ZERO, |px| {
258                    Decimal::try_from(px).unwrap_or(Decimal::ZERO)
259                });
260                let trade_ids = order
261                    .trade_ids()
262                    .iter()
263                    .map(|t| t.to_string())
264                    .collect::<Vec<_>>();
265                Some(OrderSyncEntry {
266                    bet_id,
267                    client_order_id: order.client_order_id(),
268                    filled_qty,
269                    avg_px,
270                    is_closed: order.is_closed(),
271                    trade_ids,
272                })
273            })
274            .collect();
275
276        let mut state = self.ocm_state.lock().expect(MUTEX_POISONED);
277        state.sync_from_orders(&order_data);
278
279        log::debug!("Synced OCM state from {} cached orders", order_data.len());
280    }
281
282    /// Resyncs OCM state from cache and drains any OCMs the network handler
283    /// buffered while waiting (cache is `!Send` so this must run on the
284    /// engine thread).
285    fn process_pending_resync(&self) {
286        if !self.core.is_connected() {
287            return;
288        }
289
290        if !self.pending_resync.load(Ordering::Acquire) {
291            return;
292        }
293
294        let data_sender = get_data_event_sender();
295        let market_ids_filter = self
296            .config
297            .stream_market_ids_filter
298            .as_ref()
299            .map(|ids| ids.iter().cloned().collect::<ahash::AHashSet<String>>());
300
301        // Sync only at the start: per-iteration re-sync would clobber
302        // tracker updates `process_ocm` just made with a staler cache view.
303        self.sync_ocm_state_from_cache();
304
305        loop {
306            let mut buf = self.replay_buffer.lock().expect(MUTEX_POISONED);
307            if buf.is_empty() {
308                self.pending_resync.store(false, Ordering::Release);
309                return;
310            }
311            let drained: Vec<OCM> = std::mem::take(&mut *buf);
312            drop(buf);
313
314            for ocm in &drained {
315                Self::process_ocm(
316                    ocm,
317                    self.core.account_id,
318                    self.currency,
319                    &self.emitter,
320                    &self.ocm_state,
321                    &data_sender,
322                    market_ids_filter.as_ref(),
323                    self.config.ignore_external_orders,
324                );
325            }
326        }
327    }
328
329    /// Clears the resync flag and replay buffer so post-shutdown
330    /// `is_connected()` polls don't dispatch buffered OCMs.
331    fn clear_resync_state(&self) {
332        let mut buf = self.replay_buffer.lock().expect(MUTEX_POISONED);
333        buf.clear();
334        self.pending_resync.store(false, Ordering::Release);
335
336        // An aborted reconnect must not leave submits permanently halted
337        self.is_reconciling.store(false, Ordering::Release);
338    }
339
340    fn abort_pending_tasks(&self) {
341        let mut tasks = self.pending_tasks.lock().expect(MUTEX_POISONED);
342        for handle in tasks.drain(..) {
343            handle.abort();
344        }
345    }
346
347    fn abort_background_tasks(&mut self) {
348        if let Some(handle) = self.keep_alive_handle.take() {
349            handle.abort();
350        }
351
352        if let Some(handle) = self.account_state_handle.take() {
353            handle.abort();
354        }
355
356        if let Some(handle) = self.reconnect_handle.take() {
357            handle.abort();
358        }
359    }
360
361    #[expect(clippy::too_many_arguments)]
362    fn create_ocm_handler(
363        emitter: ExecutionEventEmitter,
364        account_id: AccountId,
365        currency: Currency,
366        ocm_state: Arc<Mutex<OcmState>>,
367        data_sender: tokio::sync::mpsc::UnboundedSender<DataEvent>,
368        market_ids_filter: Option<ahash::AHashSet<String>>,
369        ignore_external_orders: bool,
370        reconnect_tx: tokio::sync::mpsc::UnboundedSender<()>,
371        pending_resync: Arc<AtomicBool>,
372        is_reconciling: Arc<AtomicBool>,
373        replay_buffer: Arc<Mutex<Vec<OCM>>>,
374    ) -> TcpMessageHandler {
375        let has_initial_connection = Arc::new(AtomicBool::new(false));
376
377        Arc::new(move |data: &[u8]| {
378            let msg = match stream_decode(data) {
379                Ok(msg) => msg,
380                Err(e) => {
381                    log::warn!("Failed to decode stream message: {e}");
382                    return;
383                }
384            };
385
386            match msg {
387                StreamMessage::OrderChange(ocm) => {
388                    if ocm.is_heartbeat() {
389                        return;
390                    }
391
392                    // Lock spans the flag check so the drainer's clear-flag
393                    // step cannot race a producer push
394                    let mut buf = replay_buffer.lock().expect(MUTEX_POISONED);
395                    if pending_resync.load(Ordering::Acquire) {
396                        buf.push(ocm);
397                        return;
398                    }
399                    drop(buf);
400
401                    Self::process_ocm(
402                        &ocm,
403                        account_id,
404                        currency,
405                        &emitter,
406                        &ocm_state,
407                        &data_sender,
408                        market_ids_filter.as_ref(),
409                        ignore_external_orders,
410                    );
411                }
412                StreamMessage::Connection(_) => {
413                    if has_initial_connection.swap(true, Ordering::SeqCst) {
414                        log::info!("Betfair execution stream reconnected");
415                        pending_resync.store(true, Ordering::Release);
416                        is_reconciling.store(true, Ordering::Release);
417                        let _ = reconnect_tx.send(());
418                    } else {
419                        log::debug!("Betfair execution stream connected");
420                    }
421                }
422                StreamMessage::Status(status) => {
423                    if status.connection_closed {
424                        log::warn!(
425                            "Betfair execution stream closed: {:?} - {:?}",
426                            status.error_code,
427                            status.error_message,
428                        );
429                    }
430                }
431                StreamMessage::MarketChange(_)
432                | StreamMessage::RaceChange(_)
433                | StreamMessage::CricketChange(_) => {}
434            }
435        })
436    }
437
438    #[expect(clippy::too_many_arguments)]
439    fn process_ocm(
440        ocm: &OCM,
441        account_id: AccountId,
442        currency: Currency,
443        emitter: &ExecutionEventEmitter,
444        ocm_state: &Arc<Mutex<OcmState>>,
445        data_sender: &tokio::sync::mpsc::UnboundedSender<DataEvent>,
446        market_ids_filter: Option<&ahash::AHashSet<String>>,
447        ignore_external_orders: bool,
448    ) {
449        let Some(order_changes) = &ocm.oc else {
450            return;
451        };
452
453        let ts_event = parse_millis_timestamp(ocm.pt);
454        let ts_init = ts_event;
455
456        for omc in order_changes {
457            if let Some(filter) = market_ids_filter
458                && !filter.contains(&omc.id)
459            {
460                continue;
461            }
462            let Some(orc_list) = &omc.orc else {
463                continue;
464            };
465
466            for orc in orc_list {
467                let handicap = orc.hc.unwrap_or(Decimal::ZERO);
468                let instrument_id = make_instrument_id(&omc.id, orc.id, handicap);
469
470                let Some(unmatched_orders) = &orc.uo else {
471                    continue;
472                };
473
474                for uo in unmatched_orders {
475                    // Treat an empty `rfo` string the same as a missing one: parsers
476                    // elsewhere normalise it to `None`, so the skip must too.
477                    let has_customer_ref = uo.rfo.as_deref().is_some_and(|s| !s.is_empty());
478                    if ignore_external_orders && !has_customer_ref {
479                        continue;
480                    }
481
482                    Self::process_unmatched_order(
483                        uo,
484                        instrument_id,
485                        account_id,
486                        currency,
487                        emitter,
488                        ocm_state,
489                        ts_event,
490                        ts_init,
491                    );
492
493                    if uo.status == StreamingOrderStatus::ExecutionComplete
494                        && uo.sv.is_some_and(|sv| sv > Decimal::ZERO)
495                    {
496                        let sv = uo.sv.unwrap();
497                        let side_str = match uo.side {
498                            StreamingSide::Back => "BACK",
499                            StreamingSide::Lay => "LAY",
500                        };
501                        let voided = BetfairOrderVoided::new(
502                            instrument_id,
503                            uo.rfo.as_deref().unwrap_or("").to_string(),
504                            uo.id.clone(),
505                            sv,
506                            uo.p,
507                            uo.s,
508                            side_str.to_string(),
509                            uo.avp,
510                            uo.sm,
511                            String::new(),
512                            ts_event,
513                            ts_init,
514                        );
515                        log::debug!("Order voided: bet_id={}, size_voided={sv}", uo.id);
516                        let custom = custom_data_with_instrument(Arc::new(voided), instrument_id);
517
518                        if let Err(e) = data_sender.send(DataEvent::Data(Data::Custom(custom))) {
519                            log::warn!("Failed to send voided event: {e}");
520                        }
521                    }
522                }
523            }
524        }
525    }
526
527    #[expect(clippy::too_many_arguments)]
528    fn process_unmatched_order(
529        uo: &crate::stream::messages::UnmatchedOrder,
530        instrument_id: InstrumentId,
531        account_id: AccountId,
532        currency: Currency,
533        emitter: &ExecutionEventEmitter,
534        ocm_state: &Arc<Mutex<OcmState>>,
535        ts_event: UnixNanos,
536        ts_init: UnixNanos,
537    ) -> bool {
538        let mut report =
539            match parse_order_status_report(uo, instrument_id, account_id, ts_event, ts_init) {
540                Ok(report) => report,
541                Err(e) => {
542                    log::warn!("Failed to parse order status report for {instrument_id}: {e}");
543                    return false;
544                }
545            };
546
547        let Ok(mut state) = ocm_state.lock() else {
548            log::error!("OcmState mutex poisoned");
549            return false;
550        };
551
552        if state.terminal_orders.contains(&uo.id) {
553            return false;
554        }
555
556        let resolved_client_order_id = state.resolve_client_order_id(uo.rfo.as_deref());
557
558        // Patch the truncated rfo-derived client_order_id with the full
559        // resolved value so downstream reconciliation matches correctly.
560        if resolved_client_order_id.is_some() {
561            report.client_order_id = resolved_client_order_id;
562        }
563
564        if uo.status == StreamingOrderStatus::ExecutionComplete
565            && has_cancel_quantity(uo)
566            && let Some(ref client_oid) = resolved_client_order_id
567            && state.should_suppress_cancel(client_oid, &uo.id)
568        {
569            log::debug!(
570                "Suppressing cancel for bet_id={} (pending replace or already replaced)",
571                uo.id,
572            );
573            return false;
574        }
575
576        if let Some(client_oid) = resolved_client_order_id {
577            state.stream_reported_client_orders.insert(client_oid);
578        }
579
580        // Tracked orders take the direct-event path, not reports (routing contract,
581        // developer_guide/adapters.md): direct terminal events stay authoritative instead
582        // of being deferred by reconciliation while the order is locally `PendingCancel`.
583        let tracked = resolved_client_order_id.and_then(|client_oid| {
584            state
585                .order_strategy_id(&client_oid)
586                .map(|strategy_id| (client_oid, strategy_id))
587        });
588
589        // Advance the fill tracker for both routes so cumulative-size bookkeeping stays correct.
590        let fill = state.fill_tracker.maybe_fill_report(
591            uo,
592            uo.s,
593            instrument_id,
594            account_id,
595            currency,
596            ts_event,
597            ts_init,
598        );
599
600        if report.order_status == OrderStatus::Canceled
601            && let Some(reason) = report.cancel_reason.as_deref()
602        {
603            log::debug!(
604                "Betfair order {} ({}) canceled: reason={}, matched={}, canceled={}, lapsed={}, voided={}",
605                report
606                    .client_order_id
607                    .unwrap_or_else(|| ClientOrderId::from(uo.id.as_str())),
608                uo.id,
609                reason,
610                uo.sm.unwrap_or(Decimal::ZERO),
611                uo.sc.unwrap_or(Decimal::ZERO),
612                uo.sl.unwrap_or(Decimal::ZERO),
613                uo.sv.unwrap_or(Decimal::ZERO),
614            );
615        }
616
617        if let Some((client_oid, strategy_id)) = tracked {
618            // Synthesize OrderAccepted on first sight: a stream-first order, or a fill ahead
619            // of the HTTP ack, needs acceptance applied before any fill or terminal event.
620            if state.mark_accepted(client_oid) {
621                let accepted = OrderAccepted::new(
622                    emitter.trader_id(),
623                    strategy_id,
624                    instrument_id,
625                    client_oid,
626                    report.venue_order_id,
627                    account_id,
628                    UUID4::new(),
629                    report.ts_accepted,
630                    ts_init,
631                    false,
632                );
633                emitter.send_order_event(OrderEventAny::Accepted(accepted));
634            }
635
636            if let Some(fill_report) = fill {
637                log::debug!(
638                    "Fill: bet_id={}, last_qty={}, last_px={}",
639                    uo.id,
640                    fill_report.last_qty,
641                    fill_report.last_px,
642                );
643                let filled = OrderFilled::new(
644                    emitter.trader_id(),
645                    strategy_id,
646                    instrument_id,
647                    client_oid,
648                    fill_report.venue_order_id,
649                    account_id,
650                    fill_report.trade_id,
651                    fill_report.order_side,
652                    report.order_type,
653                    fill_report.last_qty,
654                    fill_report.last_px,
655                    currency,
656                    fill_report.liquidity_side,
657                    UUID4::new(),
658                    fill_report.ts_event,
659                    ts_init,
660                    false,
661                    fill_report.venue_position_id,
662                    Some(fill_report.commission),
663                );
664                emitter.send_order_event(OrderEventAny::Filled(filled));
665            }
666
667            if report.order_status == OrderStatus::Canceled {
668                let canceled = OrderCanceled::new(
669                    emitter.trader_id(),
670                    strategy_id,
671                    instrument_id,
672                    client_oid,
673                    UUID4::new(),
674                    report.ts_last,
675                    ts_init,
676                    false,
677                    Some(report.venue_order_id),
678                    Some(account_id),
679                );
680                emitter.send_order_event(OrderEventAny::Canceled(canceled));
681            }
682        } else {
683            // External/unknown order: send the fill before the status report so reconciliation
684            // does not infer a duplicate fill from the cumulative filled_qty on the report.
685            if let Some(mut fill_report) = fill {
686                if resolved_client_order_id.is_some() {
687                    fill_report.client_order_id = resolved_client_order_id;
688                }
689                log::debug!(
690                    "Fill: bet_id={}, last_qty={}, last_px={}",
691                    uo.id,
692                    fill_report.last_qty,
693                    fill_report.last_px,
694                );
695                emitter.send_fill_report(fill_report);
696            }
697
698            emitter.send_order_status_report(report);
699        }
700
701        if uo.status == StreamingOrderStatus::ExecutionComplete {
702            state.terminal_orders.insert(uo.id.clone());
703            state.fill_tracker.prune(&uo.id);
704
705            if let Some(ref client_oid) = resolved_client_order_id {
706                state.cleanup_terminal_order(client_oid);
707            }
708        }
709
710        true
711    }
712}
713
714#[async_trait(?Send)]
715impl ExecutionClient for BetfairExecutionClient {
716    fn is_connected(&self) -> bool {
717        // Drain any OCMs the network handler buffered during reconnect
718        self.process_pending_resync();
719        self.core.is_connected()
720    }
721
722    fn client_id(&self) -> ClientId {
723        self.core.client_id
724    }
725
726    fn account_id(&self) -> AccountId {
727        self.core.account_id
728    }
729
730    fn venue(&self) -> Venue {
731        *BETFAIR_VENUE
732    }
733
734    fn oms_type(&self) -> OmsType {
735        self.core.oms_type
736    }
737
738    fn get_account(&self) -> Option<AccountAny> {
739        self.core.cache().account_owned(&self.core.account_id)
740    }
741
742    fn generate_account_state(
743        &self,
744        balances: Vec<AccountBalance>,
745        margins: Vec<MarginBalance>,
746        reported: bool,
747        ts_event: UnixNanos,
748    ) -> anyhow::Result<()> {
749        self.emitter
750            .emit_account_state(balances, margins, reported, ts_event);
751        Ok(())
752    }
753
754    fn start(&mut self) -> anyhow::Result<()> {
755        if self.core.is_started() {
756            return Ok(());
757        }
758
759        let sender = get_exec_event_sender();
760        self.emitter.set_sender(sender);
761        self.core.set_started();
762
763        log::info!(
764            "Started: client_id={}, account_id={}",
765            self.core.client_id,
766            self.core.account_id,
767        );
768        Ok(())
769    }
770
771    fn stop(&mut self) -> anyhow::Result<()> {
772        if self.core.is_stopped() {
773            return Ok(());
774        }
775
776        self.core.set_stopped();
777        self.core.set_disconnected();
778        self.abort_background_tasks();
779        self.abort_pending_tasks();
780        self.clear_resync_state();
781        log::info!("Stopped: client_id={}", self.core.client_id);
782        Ok(())
783    }
784
785    async fn connect(&mut self) -> anyhow::Result<()> {
786        if self.core.is_connected() {
787            return Ok(());
788        }
789
790        register_betfair_custom_data();
791
792        self.http_client
793            .connect()
794            .await
795            .map_err(|e| anyhow::anyhow!("{e}"))?;
796
797        let funds: AccountFundsResponse = self
798            .http_client
799            .send_accounts(METHOD_GET_ACCOUNT_FUNDS, serde_json::json!({}))
800            .await
801            .map_err(|e| anyhow::anyhow!("{e}"))?;
802
803        let ts_init = self.clock.get_time_ns();
804        let account_state = parse_account_state(
805            &funds,
806            self.core.account_id,
807            self.currency,
808            ts_init,
809            ts_init,
810        )?;
811        self.emitter.send_account_state(account_state);
812
813        let session_token = self
814            .http_client
815            .session_token()
816            .await
817            .ok_or_else(|| anyhow::anyhow!("No session token after login"))?;
818
819        // Sync OCM state from cached orders before stream connects
820        self.sync_ocm_state_from_cache();
821
822        let market_ids_filter = self
823            .config
824            .stream_market_ids_filter
825            .as_ref()
826            .map(|ids| ids.iter().cloned().collect::<ahash::AHashSet<String>>());
827
828        let (reconnect_tx, mut reconnect_rx) = tokio::sync::mpsc::unbounded_channel();
829
830        let handler = Self::create_ocm_handler(
831            self.emitter.clone(),
832            self.core.account_id,
833            self.currency,
834            Arc::clone(&self.ocm_state),
835            get_data_event_sender(),
836            market_ids_filter,
837            self.config.ignore_external_orders,
838            reconnect_tx,
839            Arc::clone(&self.pending_resync),
840            Arc::clone(&self.is_reconciling),
841            Arc::clone(&self.replay_buffer),
842        );
843
844        let stream_client = BetfairStreamClient::connect(
845            &self.credential,
846            session_token,
847            handler,
848            self.stream_config.clone(),
849        )
850        .await
851        .map_err(|e| anyhow::anyhow!("{e}"))?;
852
853        let stream_client = Arc::new(stream_client);
854
855        stream_client
856            .subscribe_orders(None, None)
857            .await
858            .map_err(|e| anyhow::anyhow!("{e}"))?;
859
860        self.stream_client = Some(stream_client);
861
862        // Spawn periodic keep-alive to prevent session expiry
863        let keep_alive_client = Arc::clone(&self.http_client);
864        let keep_alive_stream = Arc::clone(self.stream_client.as_ref().unwrap());
865        let keep_alive_app_key = self.credential.app_key().to_string();
866
867        self.keep_alive_handle = Some(get_runtime().spawn(async move {
868            const KEEP_ALIVE_INTERVAL_SECS: u64 = 36_000;
869            let interval = tokio::time::Duration::from_secs(KEEP_ALIVE_INTERVAL_SECS);
870            loop {
871                tokio::time::sleep(interval).await;
872
873                match keep_alive_client.keep_alive().await {
874                    Ok(()) => {}
875                    Err(ref e) if e.is_login_failed() => {
876                        log::warn!("Betfair execution session expired, attempting re-login: {e}");
877                        if let Err(e) = keep_alive_client.reconnect().await {
878                            log::warn!("Betfair execution re-login failed: {e}");
879                            continue;
880                        }
881                    }
882                    Err(e) => {
883                        log::warn!("Betfair execution keep-alive failed (transient): {e}");
884                        continue;
885                    }
886                }
887
888                if let Some(token) = keep_alive_client.session_token().await {
889                    keep_alive_stream.update_auth(&keep_alive_app_key, token);
890                }
891                log::debug!("Betfair execution session keep-alive sent");
892            }
893        }));
894
895        if self.config.calculate_account_state && self.config.request_account_state_secs > 0 {
896            let acct_client = Arc::clone(&self.http_client);
897            let acct_emitter = self.emitter.clone();
898            let acct_id = self.core.account_id;
899            let acct_currency = self.currency;
900            let acct_clock = self.clock;
901            let interval_secs = self.config.request_account_state_secs;
902            self.account_state_handle = Some(get_runtime().spawn(async move {
903                let interval = tokio::time::Duration::from_secs(interval_secs);
904                loop {
905                    tokio::time::sleep(interval).await;
906
907                    match acct_client
908                        .send_accounts::<AccountFundsResponse, _>(
909                            METHOD_GET_ACCOUNT_FUNDS,
910                            serde_json::json!({}),
911                        )
912                        .await
913                    {
914                        Ok(funds) => {
915                            let ts_init = acct_clock.get_time_ns();
916
917                            match parse_account_state(
918                                &funds,
919                                acct_id,
920                                acct_currency,
921                                ts_init,
922                                ts_init,
923                            ) {
924                                Ok(state) => acct_emitter.send_account_state(state),
925                                Err(e) => log::warn!("Failed to parse account state: {e}"),
926                            }
927                        }
928                        Err(e) => log::warn!("Failed to fetch account state: {e}"),
929                    }
930                }
931            }));
932        }
933
934        let reconnect_http = Arc::clone(&self.http_client);
935        let reconnect_stream = Arc::clone(self.stream_client.as_ref().unwrap());
936        let reconnect_emitter = self.emitter.clone();
937        let reconnect_app_key = self.credential.app_key().to_string();
938        let reconnect_clock = self.clock;
939        let reconnect_client_id = self.core.client_id;
940        let reconnect_acct_id = self.core.account_id;
941        let reconnect_currency = self.currency;
942        let reconnect_market_ids = self.reconcile_market_ids();
943        let reconnect_lookback_mins = self.config.stream_gap_recovery_lookback_mins;
944        let reconnect_ocm_state = Arc::clone(&self.ocm_state);
945        let reconnect_is_reconciling = Arc::clone(&self.is_reconciling);
946
947        self.reconnect_handle = Some(get_runtime().spawn(async move {
948            while reconnect_rx.recv().await.is_some() {
949                log::info!("Handling execution stream reconnection");
950
951                // Re-assert so a queued reconnect doesn't run with the halt cleared
952                // by the previous iteration.
953                reconnect_is_reconciling.store(true, Ordering::Release);
954
955                // Inner async block so early returns still hit the clear below.
956                let () = async {
957                    match reconnect_http.keep_alive().await {
958                        Ok(()) => {}
959                        Err(ref e) if e.is_login_failed() => {
960                            log::warn!("Session expired on reconnect, attempting re-login: {e}",);
961                            if let Err(e) = reconnect_http.reconnect().await {
962                                log::warn!("Re-login failed on reconnect: {e}");
963                                return;
964                            }
965                        }
966                        Err(e) => {
967                            log::warn!("Keep-alive failed on reconnect (transient): {e}");
968                            return;
969                        }
970                    }
971
972                    if let Some(token) = reconnect_http.session_token().await {
973                        reconnect_stream.update_auth(&reconnect_app_key, token);
974                    }
975
976                    match reconnect_http
977                        .send_accounts::<AccountFundsResponse, _>(
978                            METHOD_GET_ACCOUNT_FUNDS,
979                            serde_json::json!({}),
980                        )
981                        .await
982                    {
983                        Ok(funds) => {
984                            let ts_init = reconnect_clock.get_time_ns();
985
986                            match parse_account_state(
987                                &funds,
988                                reconnect_acct_id,
989                                reconnect_currency,
990                                ts_init,
991                                ts_init,
992                            ) {
993                                Ok(state) => reconnect_emitter.send_account_state(state),
994                                Err(e) => {
995                                    log::warn!("Failed to parse account state on reconnect: {e}");
996                                }
997                            }
998                        }
999                        Err(e) => {
1000                            log::warn!("Failed to fetch account state on reconnect: {e}");
1001                        }
1002                    }
1003
1004                    match fetch_post_reconnect_mass_status(
1005                        &reconnect_http,
1006                        reconnect_client_id,
1007                        reconnect_acct_id,
1008                        reconnect_currency,
1009                        reconnect_clock,
1010                        reconnect_market_ids.clone(),
1011                        reconnect_lookback_mins,
1012                        &reconnect_ocm_state,
1013                    )
1014                    .await
1015                    {
1016                        Ok(mass_status) => {
1017                            let order_count = mass_status.order_reports().len();
1018                            let fill_count: usize = mass_status
1019                                .fill_reports()
1020                                .values()
1021                                .map(|fills| fills.len())
1022                                .sum();
1023                            reconnect_emitter.send_execution_report(ExecutionReport::MassStatus(
1024                                Box::new(mass_status),
1025                            ));
1026                            log::info!(
1027                                "Post-reconnect reconciliation submitted: \
1028                                 orders={order_count}, fills={fill_count}",
1029                            );
1030                        }
1031                        Err(e) => {
1032                            log::warn!("Post-reconnect reconciliation failed: {e}");
1033                        }
1034                    }
1035                }
1036                .await;
1037
1038                // Fail-open: a failed iteration must not deny submits indefinitely.
1039                reconnect_is_reconciling.store(false, Ordering::Release);
1040            }
1041        }));
1042
1043        self.core.set_connected();
1044
1045        log::info!("Connected: client_id={}", self.core.client_id);
1046        Ok(())
1047    }
1048
1049    async fn disconnect(&mut self) -> anyhow::Result<()> {
1050        if self.core.is_disconnected() {
1051            return Ok(());
1052        }
1053
1054        self.abort_background_tasks();
1055        self.abort_pending_tasks();
1056
1057        if let Some(client) = &self.stream_client {
1058            client.close().await;
1059        }
1060
1061        self.http_client.disconnect().await;
1062        self.core.set_disconnected();
1063        self.clear_resync_state();
1064
1065        log::info!("Disconnected: client_id={}", self.core.client_id);
1066        Ok(())
1067    }
1068
1069    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
1070        self.process_pending_resync();
1071
1072        let http_client = Arc::clone(&self.http_client);
1073        let emitter = self.emitter.clone();
1074        let account_id = self.core.account_id;
1075        let ocm_state = Arc::clone(&self.ocm_state);
1076        let clock = self.clock;
1077        let client_order_id = cmd.client_order_id;
1078        let venue_order_id = cmd.venue_order_id;
1079        let instrument_id = cmd.instrument_id;
1080
1081        self.spawn_task("query_order", async move {
1082            let mut candidates: Vec<CurrentOrderSummary> = Vec::new();
1083            let mut seen_bet_ids: AHashSet<String> = AHashSet::new();
1084
1085            // Customer_order_ref lookup: Betfair reuses the ref across a
1086            // replace (old bet cancelled + new bet live), so this returns the
1087            // live replacement even when the cached bet_id is stale.
1088            let rfo = make_customer_order_ref(client_order_id.as_str());
1089            let rfo_params = list_current_orders_filter_ref(rfo.clone());
1090            match list_current_orders_with_retry(&http_client, &rfo_params).await {
1091                Ok(r) => extend_unique(&mut candidates, &mut seen_bet_ids, r.current_orders),
1092                Err(e) => log::warn!("Betfair query_order ref lookup failed: {e}"),
1093            }
1094
1095            if candidates.is_empty() {
1096                let rfo_legacy = make_customer_order_ref_legacy(client_order_id.as_str());
1097                if rfo_legacy != rfo {
1098                    let legacy_params = list_current_orders_filter_ref(rfo_legacy);
1099                    match list_current_orders_with_retry(&http_client, &legacy_params).await {
1100                        Ok(r) => {
1101                            extend_unique(&mut candidates, &mut seen_bet_ids, r.current_orders);
1102                        }
1103                        Err(e) => log::warn!("Betfair query_order legacy lookup failed: {e}"),
1104                    }
1105                }
1106            }
1107
1108            // Always also query by bet_id when known. This rescues
1109            // pre-existing orders without a recognizable ref and orders whose
1110            // ref-based results came back as foreign-market collisions only.
1111            if let Some(ref bet_id) = venue_order_id {
1112                let params = list_current_orders_filter_bet_id(bet_id.to_string());
1113                match list_current_orders_with_retry(&http_client, &params).await {
1114                    Ok(r) => extend_unique(&mut candidates, &mut seen_bet_ids, r.current_orders),
1115                    Err(e) => log::warn!("Betfair query_order bet_id lookup failed: {e}"),
1116                }
1117            }
1118
1119            if candidates.is_empty() {
1120                log::warn!(
1121                    "Betfair query_order found no order for client_order_id={client_order_id}, venue_order_id={venue_order_id:?}",
1122                );
1123                return Ok(());
1124            }
1125
1126            let Some(order) = select_order_for_query(
1127                &candidates,
1128                instrument_id,
1129                client_order_id,
1130                venue_order_id,
1131            ) else {
1132                return Ok(());
1133            };
1134
1135            let ts_init = clock.get_time_ns();
1136            let mut report = match parse_current_order_report(order, account_id, ts_init) {
1137                Ok(r) => r,
1138                Err(e) => {
1139                    log::warn!("Failed to parse order report for {}: {e}", order.bet_id);
1140                    return Ok(());
1141                }
1142            };
1143
1144            if report.client_order_id.is_none()
1145                && let Some(rfo) = order.customer_order_ref.as_deref()
1146                && let Ok(state) = ocm_state.lock()
1147                && let Some(full_id) = state.resolve_client_order_id(Some(rfo))
1148            {
1149                report.client_order_id = Some(full_id);
1150            }
1151
1152            if report.client_order_id.is_none() {
1153                report.client_order_id = Some(client_order_id);
1154            }
1155
1156            emitter.send_order_status_report(report);
1157            Ok(())
1158        });
1159
1160        Ok(())
1161    }
1162
1163    async fn generate_mass_status(
1164        &self,
1165        lookback_mins: Option<u64>,
1166    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
1167        self.process_pending_resync();
1168
1169        log::info!("Generating ExecutionMassStatus (lookback_mins={lookback_mins:?})");
1170
1171        let ts_now = self.clock.get_time_ns();
1172        let start = lookback_mins.map(|mins| {
1173            let lookback_ns = mins
1174                .saturating_mul(60)
1175                .saturating_mul(NANOSECONDS_IN_SECOND);
1176            UnixNanos::from(ts_now.as_u64().saturating_sub(lookback_ns))
1177        });
1178
1179        let order_cmd = GenerateOrderStatusReportsBuilder::default()
1180            .ts_init(ts_now)
1181            .open_only(false)
1182            .start(start)
1183            .build()
1184            .map_err(|e| anyhow::anyhow!("{e}"))?;
1185
1186        let fill_cmd = GenerateFillReportsBuilder::default()
1187            .ts_init(ts_now)
1188            .start(start)
1189            .build()
1190            .map_err(|e| anyhow::anyhow!("{e}"))?;
1191
1192        let (order_reports, fill_reports) = tokio::try_join!(
1193            self.generate_order_status_reports(&order_cmd),
1194            self.generate_fill_reports(fill_cmd),
1195        )?;
1196
1197        log::info!("Received {} OrderStatusReports", order_reports.len());
1198        log::info!("Received {} FillReports", fill_reports.len());
1199
1200        let mut mass_status = ExecutionMassStatus::new(
1201            self.core.client_id,
1202            self.core.account_id,
1203            *BETFAIR_VENUE,
1204            ts_now,
1205            None,
1206        );
1207
1208        mass_status.add_order_reports(order_reports);
1209        mass_status.add_fill_reports(fill_reports);
1210
1211        Ok(Some(mass_status))
1212    }
1213
1214    async fn generate_order_status_reports(
1215        &self,
1216        cmd: &GenerateOrderStatusReports,
1217    ) -> anyhow::Result<Vec<OrderStatusReport>> {
1218        self.process_pending_resync();
1219
1220        let reports = fetch_order_status_reports_via_http(
1221            &self.http_client,
1222            self.core.account_id,
1223            self.clock.get_time_ns(),
1224            self.reconcile_market_ids(),
1225            cmd.open_only,
1226            &self.ocm_state,
1227        )
1228        .await?;
1229
1230        log::debug!("Generated {} order status reports", reports.len());
1231        Ok(reports)
1232    }
1233
1234    async fn generate_fill_reports(
1235        &self,
1236        cmd: GenerateFillReports,
1237    ) -> anyhow::Result<Vec<FillReport>> {
1238        self.process_pending_resync();
1239
1240        let date_range = match (cmd.start, cmd.end) {
1241            (Some(start), Some(end)) => Some(TimeRange {
1242                from: Some(start.to_rfc3339()),
1243                to: Some(end.to_rfc3339()),
1244            }),
1245            (Some(start), None) => Some(TimeRange {
1246                from: Some(start.to_rfc3339()),
1247                to: None,
1248            }),
1249            (None, Some(end)) => Some(TimeRange {
1250                from: None,
1251                to: Some(end.to_rfc3339()),
1252            }),
1253            (None, None) => None,
1254        };
1255
1256        let reports = fetch_fill_reports_via_http(
1257            &self.http_client,
1258            self.core.account_id,
1259            self.currency,
1260            self.clock.get_time_ns(),
1261            self.reconcile_market_ids(),
1262            date_range,
1263            &self.ocm_state,
1264        )
1265        .await?;
1266
1267        log::debug!("Generated {} fill reports", reports.len());
1268        Ok(reports)
1269    }
1270
1271    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
1272        self.process_pending_resync();
1273
1274        let order = self.core.get_order(&cmd.client_order_id)?;
1275
1276        if self.is_reconciling.load(Ordering::Acquire) {
1277            log::warn!(
1278                "Halting submit for {} during post-reconnect reconciliation",
1279                order.client_order_id(),
1280            );
1281            self.emitter
1282                .emit_order_denied(&order, &OrderDeniedReason::StreamReconciling.to_string());
1283            return Ok(());
1284        }
1285
1286        if order.is_closed() {
1287            log::warn!("Cannot submit closed order {}", order.client_order_id());
1288            return Ok(());
1289        }
1290
1291        if let Ok(mut state) = self.ocm_state.lock() {
1292            state.register_customer_order_ref(order.client_order_id());
1293            state.register_order_identity(order.client_order_id(), order.strategy_id());
1294        }
1295
1296        let instrument_id = order.instrument_id();
1297        let market_id = extract_market_id(&instrument_id)?;
1298        let (selection_id, handicap) = extract_selection_id(&instrument_id)?;
1299
1300        let side = BetfairSide::from(order.order_side());
1301        let size = order.quantity().as_decimal();
1302        let handicap_opt = if handicap == Decimal::ZERO {
1303            None
1304        } else {
1305            Some(handicap)
1306        };
1307        let customer_order_ref = Some(make_customer_order_ref(order.client_order_id().as_str()));
1308
1309        let instruction = match order.order_type() {
1310            OrderType::Limit => {
1311                let price = order
1312                    .price()
1313                    .ok_or_else(|| anyhow::anyhow!("Limit order missing price"))?
1314                    .as_decimal();
1315
1316                // BSP LimitOnClose: participates in starting price calculation
1317                // with a price limit, using liability instead of size
1318                if matches!(
1319                    order.time_in_force(),
1320                    TimeInForce::AtTheClose | TimeInForce::AtTheOpen
1321                ) {
1322                    PlaceInstruction {
1323                        order_type: BetfairOrderType::LimitOnClose,
1324                        selection_id,
1325                        handicap: handicap_opt,
1326                        side,
1327                        limit_order: None,
1328                        limit_on_close_order: Some(LimitOnCloseOrder {
1329                            liability: size,
1330                            price,
1331                        }),
1332                        market_on_close_order: None,
1333                        customer_order_ref,
1334                    }
1335                } else {
1336                    let (persistence_type, time_in_force, min_fill_size) =
1337                        match order.time_in_force() {
1338                            TimeInForce::Ioc => (
1339                                None,
1340                                Some(BetfairTimeInForce::FillOrKill),
1341                                Some(Decimal::ZERO),
1342                            ),
1343                            TimeInForce::Fok => (None, Some(BetfairTimeInForce::FillOrKill), None),
1344                            TimeInForce::Gtc => (Some(PersistenceType::Persist), None, None),
1345                            _ => (Some(PersistenceType::Lapse), None, None),
1346                        };
1347
1348                    PlaceInstruction {
1349                        order_type: BetfairOrderType::Limit,
1350                        selection_id,
1351                        handicap: handicap_opt,
1352                        side,
1353                        limit_order: Some(LimitOrder {
1354                            size,
1355                            price,
1356                            persistence_type,
1357                            time_in_force,
1358                            min_fill_size,
1359                            bet_target_type: None,
1360                            bet_target_size: None,
1361                        }),
1362                        limit_on_close_order: None,
1363                        market_on_close_order: None,
1364                        customer_order_ref,
1365                    }
1366                }
1367            }
1368            OrderType::Market => {
1369                if order.time_in_force() != TimeInForce::AtTheClose {
1370                    anyhow::bail!(
1371                        "Market orders on Betfair are only supported with AtTheClose \
1372                         time in force (BSP MarketOnClose)"
1373                    );
1374                }
1375                PlaceInstruction {
1376                    order_type: BetfairOrderType::MarketOnClose,
1377                    selection_id,
1378                    handicap: handicap_opt,
1379                    side,
1380                    limit_order: None,
1381                    limit_on_close_order: None,
1382                    market_on_close_order: Some(MarketOnCloseOrder { liability: size }),
1383                    customer_order_ref,
1384                }
1385            }
1386            other => {
1387                anyhow::bail!("Unsupported order type for Betfair: {other:?}");
1388            }
1389        };
1390
1391        let market_version = self.get_market_version(&instrument_id);
1392
1393        let params = PlaceOrdersParams {
1394            market_id,
1395            instructions: vec![instruction],
1396            customer_ref: None,
1397            market_version,
1398            customer_strategy_ref: None,
1399        };
1400
1401        let client_order_id = order.client_order_id();
1402        let strategy_id = order.strategy_id();
1403
1404        log::debug!("OrderSubmitted client_order_id={client_order_id}");
1405        self.emitter.emit_order_submitted(&order);
1406
1407        let http_client = Arc::clone(&self.http_client);
1408        let emitter = self.emitter.clone();
1409        let clock = self.clock;
1410        let ocm_state = Arc::clone(&self.ocm_state);
1411
1412        self.spawn_task("submit-order", async move {
1413            let report: PlaceExecutionReport = match http_client
1414                .send_betting_order(METHOD_PLACE_ORDERS, &params)
1415                .await
1416            {
1417                Ok(r) => r,
1418                Err(e) => {
1419                    // Transport errors (502, timeout, network reset) may mean the
1420                    // order was placed but the response was lost. Do not reject
1421                    // because the OCM stream will reconcile via customerOrderRef.
1422                    if e.is_order_placement_ambiguous() {
1423                        log::warn!(
1424                            "Ambiguous submit response for {client_order_id}: {e}. \
1425                             Order may be live, awaiting OCM reconciliation",
1426                        );
1427                        return Ok(());
1428                    }
1429
1430                    let reason = format!("submit-order error: {e}");
1431                    if should_emit_http_reject(&ocm_state, &client_order_id, &reason) {
1432                        let ts_event = clock.get_time_ns();
1433                        emitter.emit_order_rejected_event(
1434                            strategy_id,
1435                            instrument_id,
1436                            client_order_id,
1437                            &reason,
1438                            ts_event,
1439                            false,
1440                        );
1441                    }
1442                    return Ok(());
1443                }
1444            };
1445
1446            if report.status == ExecutionReportStatus::Timeout {
1447                log::warn!(
1448                    "Betfair Timeout for {client_order_id}. \
1449                     Order may be live, awaiting OCM reconciliation",
1450                );
1451                return Ok(());
1452            }
1453
1454            if let Some(instruction_reports) = &report.instruction_reports {
1455                if let Some(ir) = instruction_reports.first() {
1456                    if ir.status == InstructionReportStatus::Failure {
1457                        let reason = format_place_instruction_reason(ir, &report);
1458                        if should_emit_http_reject(&ocm_state, &client_order_id, &reason) {
1459                            let ts_event = clock.get_time_ns();
1460                            emitter.emit_order_rejected_event(
1461                                strategy_id,
1462                                instrument_id,
1463                                client_order_id,
1464                                &reason,
1465                                ts_event,
1466                                false,
1467                            );
1468                        }
1469                        return Ok(());
1470                    }
1471
1472                    if let Some(bet_id) = &ir.bet_id {
1473                        let venue_order_id = VenueOrderId::from(bet_id.as_str());
1474                        let ts_event = clock.get_time_ns();
1475
1476                        emit_http_accept_if_claimed(&ocm_state, &client_order_id, || {
1477                            emitter.emit_order_accepted(&order, venue_order_id, ts_event);
1478                        });
1479                    }
1480                } else if report.status == ExecutionReportStatus::Failure
1481                    || report.status == ExecutionReportStatus::ProcessedWithErrors
1482                {
1483                    let reason = format_betfair_reason(
1484                        report.error_message.as_deref(),
1485                        report.error_code,
1486                        None,
1487                        "unknown error",
1488                    );
1489
1490                    if should_emit_http_reject(&ocm_state, &client_order_id, &reason) {
1491                        let ts_event = clock.get_time_ns();
1492                        emitter.emit_order_rejected_event(
1493                            strategy_id,
1494                            instrument_id,
1495                            client_order_id,
1496                            &reason,
1497                            ts_event,
1498                            false,
1499                        );
1500                    }
1501                }
1502            } else if report.status == ExecutionReportStatus::Failure
1503                || report.status == ExecutionReportStatus::ProcessedWithErrors
1504            {
1505                let reason = format_betfair_reason(
1506                    report.error_message.as_deref(),
1507                    report.error_code,
1508                    None,
1509                    "unknown error",
1510                );
1511
1512                if should_emit_http_reject(&ocm_state, &client_order_id, &reason) {
1513                    let ts_event = clock.get_time_ns();
1514                    emitter.emit_order_rejected_event(
1515                        strategy_id,
1516                        instrument_id,
1517                        client_order_id,
1518                        &reason,
1519                        ts_event,
1520                        false,
1521                    );
1522                }
1523            }
1524
1525            Ok(())
1526        });
1527
1528        Ok(())
1529    }
1530
1531    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
1532        self.process_pending_resync();
1533
1534        let instrument_id = cmd.instrument_id;
1535        let venue_order_id = match cmd.venue_order_id {
1536            Some(venue_order_id) => venue_order_id,
1537            None => {
1538                log::warn!(
1539                    "Cannot cancel order {}: no venue_order_id",
1540                    cmd.client_order_id
1541                );
1542                return Ok(());
1543            }
1544        };
1545        let market_id = match extract_market_id(&instrument_id) {
1546            Ok(market_id) => market_id,
1547            Err(e) => {
1548                log::warn!("Cannot cancel order {}: {e}", cmd.client_order_id);
1549                return Ok(());
1550            }
1551        };
1552        let bet_id: BetId = venue_order_id.to_string();
1553
1554        let params = CancelOrdersParams {
1555            market_id: Some(market_id),
1556            instructions: Some(vec![CancelInstruction {
1557                bet_id,
1558                size_reduction: None,
1559            }]),
1560            customer_ref: None,
1561        };
1562
1563        let client_order_id = cmd.client_order_id;
1564        let strategy_id = cmd.strategy_id;
1565        let http_client = Arc::clone(&self.http_client);
1566        let emitter = self.emitter.clone();
1567        let clock = self.clock;
1568
1569        self.spawn_task("cancel-order", async move {
1570            let result: Result<CancelExecutionReport, _> = http_client
1571                .send_betting_order(METHOD_CANCEL_ORDERS, &params)
1572                .await;
1573
1574            let report = match result {
1575                Ok(r) => r,
1576                Err(e) => {
1577                    log::warn!(
1578                        "Cancel request failed for {client_order_id}, awaiting OCM reconciliation: {e}",
1579                    );
1580                    return Ok(());
1581                }
1582            };
1583
1584            if report.status == ExecutionReportStatus::Timeout {
1585                log::warn!(
1586                    "Betfair Timeout for cancel {client_order_id}. \
1587                     Cancel may be delayed (in-play), awaiting OCM reconciliation",
1588                );
1589                return Ok(());
1590            }
1591
1592            if let Some(instruction_reports) = &report.instruction_reports
1593                && !instruction_reports.is_empty()
1594            {
1595                for ir in instruction_reports {
1596                    match ir.status {
1597                        InstructionReportStatus::Success => {}
1598                        InstructionReportStatus::Timeout => {
1599                            log::warn!(
1600                                "Cancel instruction timeout for {client_order_id}",
1601                            );
1602                        }
1603                        InstructionReportStatus::Failure => {
1604                            if ir.error_code
1605                                == Some(InstructionReportErrorCode::BetTakenOrLapsed)
1606                            {
1607                                log::debug!(
1608                                    "Cancel {client_order_id}: BetTakenOrLapsed, treating as success",
1609                                );
1610                                continue;
1611                            }
1612
1613                            let reason = format_cancel_instruction_reason(
1614                                ir.error_message.as_deref(),
1615                                ir.error_code,
1616                                report.error_message.as_deref(),
1617                                report.error_code,
1618                            );
1619                            let ts_event = clock.get_time_ns();
1620                            emitter.emit_order_cancel_rejected_event(
1621                                strategy_id,
1622                                instrument_id,
1623                                client_order_id,
1624                                Some(venue_order_id),
1625                                &reason,
1626                                ts_event,
1627                            );
1628                            return Ok(());
1629                        }
1630                    }
1631                }
1632            } else if report.status != ExecutionReportStatus::Success {
1633                let reason = format_betfair_reason(
1634                    report.error_message.as_deref(),
1635                    report.error_code,
1636                    None,
1637                    "unknown error",
1638                );
1639                log::warn!(
1640                    "Cancel {client_order_id} failed without per-order result, awaiting OCM reconciliation: {reason}",
1641                );
1642            }
1643
1644            Ok(())
1645        });
1646
1647        Ok(())
1648    }
1649
1650    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
1651        self.process_pending_resync();
1652
1653        let instrument_id = cmd.instrument_id;
1654        let market_id = extract_market_id(&instrument_id)?;
1655
1656        let venue_order_id = cmd
1657            .venue_order_id
1658            .ok_or_else(|| anyhow::anyhow!("Cannot modify order without venue_order_id"))?;
1659        let bet_id: BetId = venue_order_id.to_string();
1660
1661        // Compare against existing order to determine actual changes
1662        let existing_order = self.core.get_order(&cmd.client_order_id);
1663        let has_price_change = match (&cmd.price, &existing_order) {
1664            (Some(new_price), Ok(order)) => order.price() != Some(*new_price),
1665            (Some(_), Err(_)) => true,
1666            (None, _) => false,
1667        };
1668        let has_quantity_change = match (&cmd.quantity, &existing_order) {
1669            (Some(new_qty), Ok(order)) => order.quantity() != *new_qty,
1670            (Some(_), Err(_)) => true,
1671            (None, _) => false,
1672        };
1673
1674        // Betfair does not support atomic price+quantity modification
1675        if has_price_change && has_quantity_change {
1676            let ts_event = self.clock.get_time_ns();
1677            self.emitter.emit_order_modify_rejected_event(
1678                cmd.strategy_id,
1679                instrument_id,
1680                cmd.client_order_id,
1681                Some(venue_order_id),
1682                "cannot modify price and quantity simultaneously on Betfair",
1683                ts_event,
1684            );
1685            return Ok(());
1686        }
1687
1688        let client_order_id = cmd.client_order_id;
1689        let strategy_id = cmd.strategy_id;
1690        let http_client = Arc::clone(&self.http_client);
1691        let emitter = self.emitter.clone();
1692        let clock = self.clock;
1693
1694        if has_price_change {
1695            let new_price = cmd.price.unwrap().as_decimal();
1696            let old_bet_id = bet_id.clone();
1697            let update_price = cmd.price;
1698            let update_qty = existing_order.as_ref().ok().map(|order| order.quantity());
1699
1700            // Track pending replace so the OCM handler suppresses the
1701            // cancel event for the old bet that Betfair emits as part
1702            // of the replace operation.
1703            if let Ok(mut state) = self.ocm_state.lock() {
1704                state
1705                    .pending_update_keys
1706                    .insert((client_order_id, old_bet_id.clone()));
1707            }
1708
1709            let market_version = self.get_market_version(&instrument_id);
1710
1711            let params = ReplaceOrdersParams {
1712                market_id,
1713                instructions: vec![ReplaceInstruction { bet_id, new_price }],
1714                customer_ref: None,
1715                market_version,
1716            };
1717
1718            let ocm_state = Arc::clone(&self.ocm_state);
1719
1720            self.spawn_task("modify-order-price", async move {
1721                let result: Result<ReplaceExecutionReport, _> = http_client
1722                    .send_betting_order(METHOD_REPLACE_ORDERS, &params)
1723                    .await;
1724
1725                match result {
1726                    Ok(report) if report.status == ExecutionReportStatus::Success => {
1727                        if let Ok(mut state) = ocm_state.lock() {
1728                            state
1729                                .pending_update_keys
1730                                .remove(&(client_order_id, old_bet_id.clone()));
1731                            state.replaced_venue_order_ids.insert(old_bet_id);
1732                        }
1733
1734                        // Emit OrderUpdated from the replace confirmation, promoting the order
1735                        // to the new bet; the new bet's resting OCM then emits nothing for a
1736                        // tracked order.
1737                        if let (Some(new_bet_id), Some(quantity)) = (
1738                            report
1739                                .instruction_reports
1740                                .as_ref()
1741                                .and_then(|reports| reports.first())
1742                                .and_then(|ir| ir.place_instruction_report.as_ref())
1743                                .and_then(|pir| pir.bet_id.clone()),
1744                            update_qty,
1745                        ) {
1746                            let ts_event = clock.get_time_ns();
1747                            let updated = OrderUpdated::new(
1748                                emitter.trader_id(),
1749                                strategy_id,
1750                                instrument_id,
1751                                client_order_id,
1752                                quantity,
1753                                UUID4::new(),
1754                                ts_event,
1755                                ts_event,
1756                                false,
1757                                Some(VenueOrderId::from(new_bet_id.as_str())),
1758                                Some(emitter.account_id()),
1759                                update_price,
1760                                None,
1761                                None,
1762                                false,
1763                            );
1764                            emitter.send_order_event(OrderEventAny::Updated(updated));
1765                        }
1766                    }
1767                    Ok(report) if report.status == ExecutionReportStatus::Timeout => {
1768                        log::warn!(
1769                            "Betfair Timeout for modify {client_order_id}. \
1770                             Replace may be pending, awaiting reconciliation",
1771                        );
1772                    }
1773                    Ok(report) => {
1774                        if let Ok(mut state) = ocm_state.lock() {
1775                            state
1776                                .pending_update_keys
1777                                .remove(&(client_order_id, old_bet_id));
1778                        }
1779
1780                        if let Some(instruction_reports) = &report.instruction_reports
1781                            && !instruction_reports.is_empty()
1782                        {
1783                            for ir in instruction_reports {
1784                                match ir.status {
1785                                    InstructionReportStatus::Success => {}
1786                                    InstructionReportStatus::Timeout => {
1787                                        log::warn!(
1788                                            "Replace instruction timeout for {client_order_id}",
1789                                        );
1790                                    }
1791                                    InstructionReportStatus::Failure => {
1792                                        let reason = format_replace_instruction_reason(ir, &report);
1793                                        let ts_event = clock.get_time_ns();
1794                                        emitter.emit_order_modify_rejected_event(
1795                                            strategy_id,
1796                                            instrument_id,
1797                                            client_order_id,
1798                                            Some(venue_order_id),
1799                                            &reason,
1800                                            ts_event,
1801                                        );
1802                                        return Ok(());
1803                                    }
1804                                }
1805                            }
1806                        }
1807
1808                        let reason = format_betfair_reason(
1809                            report.error_message.as_deref(),
1810                            report.error_code,
1811                            None,
1812                            "unknown error",
1813                        );
1814                        let ts_event = clock.get_time_ns();
1815                        emitter.emit_order_modify_rejected_event(
1816                            strategy_id,
1817                            instrument_id,
1818                            client_order_id,
1819                            Some(venue_order_id),
1820                            &reason,
1821                            ts_event,
1822                        );
1823                    }
1824                    Err(e) => {
1825                        if let Ok(mut state) = ocm_state.lock() {
1826                            state
1827                                .pending_update_keys
1828                                .remove(&(client_order_id, old_bet_id));
1829                        }
1830                        let ts_event = clock.get_time_ns();
1831                        emitter.emit_order_modify_rejected_event(
1832                            strategy_id,
1833                            instrument_id,
1834                            client_order_id,
1835                            Some(venue_order_id),
1836                            &format!("modify-order error: {e}"),
1837                            ts_event,
1838                        );
1839                    }
1840                }
1841
1842                Ok(())
1843            });
1844        } else if has_quantity_change {
1845            // Quantity reduction via partial cancel
1846            let order = self.core.get_order(&client_order_id)?;
1847            let existing_qty = order.quantity().as_decimal();
1848            let new_qty = cmd.quantity.unwrap().as_decimal();
1849            let new_quantity = cmd.quantity.unwrap();
1850
1851            if new_qty >= existing_qty {
1852                let ts_event = self.clock.get_time_ns();
1853                self.emitter.emit_order_modify_rejected_event(
1854                    strategy_id,
1855                    instrument_id,
1856                    client_order_id,
1857                    Some(venue_order_id),
1858                    "can only reduce quantity on Betfair",
1859                    ts_event,
1860                );
1861                return Ok(());
1862            }
1863
1864            let size_reduction = existing_qty - new_qty;
1865            let params = CancelOrdersParams {
1866                market_id: Some(market_id),
1867                instructions: Some(vec![CancelInstruction {
1868                    bet_id,
1869                    size_reduction: Some(size_reduction),
1870                }]),
1871                customer_ref: None,
1872            };
1873
1874            self.spawn_task("modify-order-quantity", async move {
1875                let result: Result<CancelExecutionReport, _> = http_client
1876                    .send_betting_order(METHOD_CANCEL_ORDERS, &params)
1877                    .await;
1878
1879                match result {
1880                    Ok(report) if report.status != ExecutionReportStatus::Success => {
1881                        let reason = format_betfair_reason(
1882                            report.error_message.as_deref(),
1883                            report.error_code,
1884                            None,
1885                            "unknown error",
1886                        );
1887                        let ts_event = clock.get_time_ns();
1888                        emitter.emit_order_modify_rejected_event(
1889                            strategy_id,
1890                            instrument_id,
1891                            client_order_id,
1892                            Some(venue_order_id),
1893                            &reason,
1894                            ts_event,
1895                        );
1896                    }
1897                    Err(e) => {
1898                        let ts_event = clock.get_time_ns();
1899                        emitter.emit_order_modify_rejected_event(
1900                            strategy_id,
1901                            instrument_id,
1902                            client_order_id,
1903                            Some(venue_order_id),
1904                            &format!("modify-order error: {e}"),
1905                            ts_event,
1906                        );
1907                    }
1908                    Ok(report) => {
1909                        // Derive remaining quantity from the actual `size_cancelled`, not the
1910                        // requested target: a fill racing the reduction cancels less than
1911                        // requested, so the target would drop that fill as an overfill.
1912                        let updated_quantity = report
1913                            .instruction_reports
1914                            .as_ref()
1915                            .and_then(|reports| reports.first())
1916                            .and_then(|ir| ir.size_cancelled)
1917                            .and_then(|cancelled| {
1918                                parse_betfair_quantity(existing_qty - cancelled).ok()
1919                            })
1920                            .unwrap_or(new_quantity);
1921                        let ts_event = clock.get_time_ns();
1922                        let updated = OrderUpdated::new(
1923                            emitter.trader_id(),
1924                            strategy_id,
1925                            instrument_id,
1926                            client_order_id,
1927                            updated_quantity,
1928                            UUID4::new(),
1929                            ts_event,
1930                            ts_event,
1931                            false,
1932                            Some(venue_order_id),
1933                            Some(emitter.account_id()),
1934                            None,
1935                            None,
1936                            None,
1937                            false,
1938                        );
1939                        emitter.send_order_event(OrderEventAny::Updated(updated));
1940                    }
1941                }
1942
1943                Ok(())
1944            });
1945        } else {
1946            let ts_event = self.clock.get_time_ns();
1947            self.emitter.emit_order_modify_rejected_event(
1948                strategy_id,
1949                instrument_id,
1950                client_order_id,
1951                Some(venue_order_id),
1952                "no effective change in price or quantity",
1953                ts_event,
1954            );
1955        }
1956
1957        Ok(())
1958    }
1959
1960    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
1961        self.process_pending_resync();
1962
1963        let instrument_id = cmd.instrument_id;
1964        let market_id = match extract_market_id(&instrument_id) {
1965            Ok(market_id) => market_id,
1966            Err(e) => {
1967                log::warn!("Cannot cancel all orders for {instrument_id}: {e}");
1968                return Ok(());
1969            }
1970        };
1971
1972        let params = CancelOrdersParams {
1973            market_id: Some(market_id),
1974            instructions: None,
1975            customer_ref: None,
1976        };
1977
1978        let http_client = Arc::clone(&self.http_client);
1979
1980        self.spawn_task("cancel-all-orders", async move {
1981            let result = http_client
1982                .send_betting_order::<serde_json::Value, _>(METHOD_CANCEL_ORDERS, &params)
1983                .await;
1984
1985            if let Err(e) = result {
1986                log::warn!("Failed to cancel all orders: {e}");
1987            }
1988
1989            Ok(())
1990        });
1991
1992        Ok(())
1993    }
1994
1995    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
1996        self.process_pending_resync();
1997
1998        let instrument_id = cmd.instrument_id;
1999        let market_id = match extract_market_id(&instrument_id) {
2000            Ok(market_id) => market_id,
2001            Err(e) => {
2002                log::warn!("Cannot batch cancel orders for {instrument_id}: {e}");
2003                return Ok(());
2004            }
2005        };
2006
2007        let mut instructions = Vec::new();
2008        let mut valid_cancels = Vec::new();
2009
2010        for cancel in &cmd.cancels {
2011            match cancel.venue_order_id {
2012                Some(venue_order_id) => {
2013                    let bet_id: BetId = venue_order_id.to_string();
2014                    instructions.push(CancelInstruction {
2015                        bet_id,
2016                        size_reduction: None,
2017                    });
2018                    valid_cancels.push(cancel);
2019                }
2020                None => {
2021                    log::warn!(
2022                        "Cannot batch cancel order {}: no venue_order_id",
2023                        cancel.client_order_id,
2024                    );
2025                }
2026            }
2027        }
2028
2029        if valid_cancels.is_empty() {
2030            return Ok(());
2031        }
2032
2033        let params = CancelOrdersParams {
2034            market_id: Some(market_id),
2035            instructions: Some(instructions),
2036            customer_ref: None,
2037        };
2038
2039        let cancel_data: Vec<_> = valid_cancels
2040            .iter()
2041            .map(|c| {
2042                (
2043                    c.strategy_id,
2044                    c.instrument_id,
2045                    c.client_order_id,
2046                    c.venue_order_id,
2047                )
2048            })
2049            .collect();
2050
2051        let http_client = Arc::clone(&self.http_client);
2052        let emitter = self.emitter.clone();
2053        let clock = self.clock;
2054
2055        self.spawn_task("batch-cancel-orders", async move {
2056            let report: CancelExecutionReport = match http_client
2057                .send_betting_order(METHOD_CANCEL_ORDERS, &params)
2058                .await
2059            {
2060                Ok(r) => r,
2061                Err(e) => {
2062                    log::warn!(
2063                        "Batch cancel request failed for {} orders, awaiting OCM reconciliation: {e}",
2064                        cancel_data.len(),
2065                    );
2066                    return Ok(());
2067                }
2068            };
2069
2070            if report.status == ExecutionReportStatus::Timeout {
2071                log::warn!(
2072                    "Batch cancel request timed out for {} orders, awaiting OCM reconciliation",
2073                    cancel_data.len(),
2074                );
2075                return Ok(());
2076            }
2077
2078            if report.status == ExecutionReportStatus::Failure
2079                || report.status == ExecutionReportStatus::ProcessedWithErrors
2080            {
2081                let reason = format_betfair_reason(
2082                    report.error_message.as_deref(),
2083                    report.error_code,
2084                    None,
2085                    "unknown error",
2086                );
2087
2088                if report
2089                    .instruction_reports
2090                    .as_ref()
2091                    .is_none_or(Vec::is_empty)
2092                {
2093                    log::warn!(
2094                        "Batch cancel failed for {} orders without per-order results, awaiting OCM reconciliation: {reason}",
2095                        cancel_data.len(),
2096                    );
2097                    return Ok(());
2098                }
2099            }
2100
2101            if let Some(instruction_reports) = &report.instruction_reports {
2102                for (ir, (strategy_id, instr_id, client_oid, venue_oid)) in
2103                    instruction_reports.iter().zip(cancel_data.iter())
2104                {
2105                    match ir.status {
2106                        InstructionReportStatus::Success => {}
2107                        InstructionReportStatus::Timeout => {
2108                            log::warn!(
2109                                "Cancel timeout for {client_oid}: leaving order state unchanged",
2110                            );
2111                        }
2112                        InstructionReportStatus::Failure => {
2113                            // BetTakenOrLapsed means the bet already completed, treat as success
2114                            if ir.error_code == Some(InstructionReportErrorCode::BetTakenOrLapsed) {
2115                                continue;
2116                            }
2117
2118                            let reason = format_cancel_instruction_reason(
2119                                ir.error_message.as_deref(),
2120                                ir.error_code,
2121                                report.error_message.as_deref(),
2122                                report.error_code,
2123                            );
2124                            let ts_event = clock.get_time_ns();
2125                            emitter.emit_order_cancel_rejected_event(
2126                                *strategy_id,
2127                                *instr_id,
2128                                *client_oid,
2129                                *venue_oid,
2130                                &reason,
2131                                ts_event,
2132                            );
2133                        }
2134                    }
2135                }
2136            }
2137
2138            Ok(())
2139        });
2140
2141        Ok(())
2142    }
2143
2144    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
2145        self.process_pending_resync();
2146
2147        if self.is_reconciling.load(Ordering::Acquire) {
2148            log::warn!(
2149                "Halting submit_order_list ({} orders) during post-reconnect reconciliation",
2150                cmd.order_list.client_order_ids.len(),
2151            );
2152
2153            let denied = OrderDeniedReason::StreamReconciling.to_string();
2154
2155            for client_order_id in &cmd.order_list.client_order_ids {
2156                if let Ok(order) = self.core.get_order(client_order_id) {
2157                    self.emitter.emit_order_denied(&order, &denied);
2158                }
2159            }
2160            return Ok(());
2161        }
2162
2163        let instrument_id = cmd.instrument_id;
2164        let market_id = extract_market_id(&instrument_id)?;
2165        let (selection_id, handicap) = extract_selection_id(&instrument_id)?;
2166
2167        let handicap_opt = if handicap == Decimal::ZERO {
2168            None
2169        } else {
2170            Some(handicap)
2171        };
2172
2173        let mut instructions = Vec::new();
2174        let mut order_snapshots = Vec::new();
2175
2176        for client_order_id in &cmd.order_list.client_order_ids {
2177            let order = self.core.get_order(client_order_id)?;
2178
2179            if order.is_closed() {
2180                log::warn!("Skipping closed order {client_order_id}");
2181                continue;
2182            }
2183
2184            if let Ok(mut state) = self.ocm_state.lock() {
2185                state.register_customer_order_ref(order.client_order_id());
2186                state.register_order_identity(order.client_order_id(), order.strategy_id());
2187            }
2188
2189            let side = BetfairSide::from(order.order_side());
2190            let size = order.quantity().as_decimal();
2191            let customer_order_ref =
2192                Some(make_customer_order_ref(order.client_order_id().as_str()));
2193
2194            let instruction = match order.order_type() {
2195                OrderType::Limit => {
2196                    let price = order
2197                        .price()
2198                        .ok_or_else(|| anyhow::anyhow!("Limit order missing price"))?
2199                        .as_decimal();
2200
2201                    if matches!(
2202                        order.time_in_force(),
2203                        TimeInForce::AtTheClose | TimeInForce::AtTheOpen
2204                    ) {
2205                        PlaceInstruction {
2206                            order_type: BetfairOrderType::LimitOnClose,
2207                            selection_id,
2208                            handicap: handicap_opt,
2209                            side,
2210                            limit_order: None,
2211                            limit_on_close_order: Some(LimitOnCloseOrder {
2212                                liability: size,
2213                                price,
2214                            }),
2215                            market_on_close_order: None,
2216                            customer_order_ref,
2217                        }
2218                    } else {
2219                        let (persistence_type, time_in_force, min_fill_size) = match order
2220                            .time_in_force()
2221                        {
2222                            TimeInForce::Ioc => (
2223                                None,
2224                                Some(BetfairTimeInForce::FillOrKill),
2225                                Some(Decimal::ZERO),
2226                            ),
2227                            TimeInForce::Fok => (None, Some(BetfairTimeInForce::FillOrKill), None),
2228                            TimeInForce::Gtc => (Some(PersistenceType::Persist), None, None),
2229                            _ => (Some(PersistenceType::Lapse), None, None),
2230                        };
2231
2232                        PlaceInstruction {
2233                            order_type: BetfairOrderType::Limit,
2234                            selection_id,
2235                            handicap: handicap_opt,
2236                            side,
2237                            limit_order: Some(LimitOrder {
2238                                size,
2239                                price,
2240                                persistence_type,
2241                                time_in_force,
2242                                min_fill_size,
2243                                bet_target_type: None,
2244                                bet_target_size: None,
2245                            }),
2246                            limit_on_close_order: None,
2247                            market_on_close_order: None,
2248                            customer_order_ref,
2249                        }
2250                    }
2251                }
2252                OrderType::Market => {
2253                    if order.time_in_force() != TimeInForce::AtTheClose {
2254                        anyhow::bail!(
2255                            "Market orders on Betfair are only supported with AtTheClose \
2256                             time in force (BSP MarketOnClose)"
2257                        );
2258                    }
2259                    PlaceInstruction {
2260                        order_type: BetfairOrderType::MarketOnClose,
2261                        selection_id,
2262                        handicap: handicap_opt,
2263                        side,
2264                        limit_order: None,
2265                        limit_on_close_order: None,
2266                        market_on_close_order: Some(MarketOnCloseOrder { liability: size }),
2267                        customer_order_ref,
2268                    }
2269                }
2270                other => {
2271                    anyhow::bail!("Unsupported order type for Betfair: {other:?}");
2272                }
2273            };
2274
2275            instructions.push(instruction);
2276            order_snapshots.push((order.client_order_id(), order.strategy_id(), order.clone()));
2277        }
2278
2279        if instructions.is_empty() {
2280            return Ok(());
2281        }
2282
2283        let market_version = self.get_market_version(&instrument_id);
2284
2285        let params = PlaceOrdersParams {
2286            market_id,
2287            instructions,
2288            customer_ref: None,
2289            market_version,
2290            customer_strategy_ref: None,
2291        };
2292
2293        for (_, _, order) in &order_snapshots {
2294            log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
2295            self.emitter.emit_order_submitted(order);
2296        }
2297
2298        let http_client = Arc::clone(&self.http_client);
2299        let emitter = self.emitter.clone();
2300        let clock = self.clock;
2301        let ocm_state = Arc::clone(&self.ocm_state);
2302
2303        self.spawn_task("submit-order-list", async move {
2304            let report: PlaceExecutionReport = match http_client
2305                .send_betting_order(METHOD_PLACE_ORDERS, &params)
2306                .await
2307            {
2308                Ok(r) => r,
2309                Err(e) => {
2310                    if e.is_order_placement_ambiguous() {
2311                        log::warn!(
2312                            "Ambiguous submit response for order list: {e}. \
2313                             Orders may be live, awaiting OCM reconciliation",
2314                        );
2315                        return Ok(());
2316                    }
2317
2318                    let ts_event = clock.get_time_ns();
2319                    let reason = format!("submit-order-list error: {e}");
2320
2321                    for (client_oid, strategy_id, _) in &order_snapshots {
2322                        if should_emit_http_reject(&ocm_state, client_oid, &reason) {
2323                            emitter.emit_order_rejected_event(
2324                                *strategy_id,
2325                                instrument_id,
2326                                *client_oid,
2327                                &reason,
2328                                ts_event,
2329                                false,
2330                            );
2331                        }
2332                    }
2333                    return Ok(());
2334                }
2335            };
2336
2337            if report.status == ExecutionReportStatus::Failure {
2338                let reason = format_betfair_reason(
2339                    report.error_message.as_deref(),
2340                    report.error_code,
2341                    None,
2342                    "unknown error",
2343                );
2344
2345                if report.instruction_reports.is_none() {
2346                    let ts_event = clock.get_time_ns();
2347
2348                    for (client_oid, strategy_id, _) in &order_snapshots {
2349                        if should_emit_http_reject(&ocm_state, client_oid, &reason) {
2350                            emitter.emit_order_rejected_event(
2351                                *strategy_id,
2352                                instrument_id,
2353                                *client_oid,
2354                                &reason,
2355                                ts_event,
2356                                false,
2357                            );
2358                        }
2359                    }
2360                    return Ok(());
2361                }
2362            }
2363
2364            if report.status == ExecutionReportStatus::Timeout {
2365                log::warn!(
2366                    "Betfair Timeout for order list. \
2367                     Orders may be live, awaiting OCM reconciliation",
2368                );
2369                return Ok(());
2370            }
2371
2372            if let Some(instruction_reports) = &report.instruction_reports {
2373                for (ir, (client_oid, strategy_id, order)) in
2374                    instruction_reports.iter().zip(order_snapshots.iter())
2375                {
2376                    match ir.status {
2377                        InstructionReportStatus::Success => {
2378                            if let Some(bet_id) = &ir.bet_id {
2379                                let venue_order_id = VenueOrderId::from(bet_id.as_str());
2380                                let ts_event = clock.get_time_ns();
2381
2382                                emit_http_accept_if_claimed(&ocm_state, client_oid, || {
2383                                    emitter.emit_order_accepted(order, venue_order_id, ts_event);
2384                                });
2385                            }
2386                        }
2387                        InstructionReportStatus::Timeout => {
2388                            log::warn!(
2389                                "Submit timeout for {client_oid}: \
2390                                 leaving SUBMITTED for reconciliation",
2391                            );
2392                        }
2393                        InstructionReportStatus::Failure => {
2394                            let reason = format_place_instruction_reason(ir, &report);
2395                            if should_emit_http_reject(&ocm_state, client_oid, &reason) {
2396                                let ts_event = clock.get_time_ns();
2397                                emitter.emit_order_rejected_event(
2398                                    *strategy_id,
2399                                    instrument_id,
2400                                    *client_oid,
2401                                    &reason,
2402                                    ts_event,
2403                                    false,
2404                                );
2405                            }
2406                        }
2407                    }
2408                }
2409            }
2410
2411            Ok(())
2412        });
2413
2414        Ok(())
2415    }
2416}
2417
2418/// Paginates `list_current_orders` into `OrderStatusReport`s without touching
2419/// the engine cache, so it is callable from any tokio task.
2420async fn fetch_order_status_reports_via_http(
2421    http_client: &Arc<BetfairHttpClient>,
2422    account_id: AccountId,
2423    ts_init: UnixNanos,
2424    market_ids: Option<Vec<String>>,
2425    open_only: bool,
2426    ocm_state: &Arc<Mutex<OcmState>>,
2427) -> anyhow::Result<Vec<OrderStatusReport>> {
2428    let order_projection = if open_only {
2429        Some(OrderProjection::Executable)
2430    } else {
2431        Some(OrderProjection::All)
2432    };
2433
2434    let mut reports = Vec::new();
2435    let mut from_record: u32 = 0;
2436
2437    loop {
2438        let params = ListCurrentOrdersParams {
2439            bet_ids: None,
2440            market_ids: market_ids.clone(),
2441            order_projection,
2442            customer_order_refs: None,
2443            customer_strategy_refs: None,
2444            date_range: None,
2445            order_by: None,
2446            sort_dir: None,
2447            from_record: if from_record > 0 {
2448                Some(from_record)
2449            } else {
2450                None
2451            },
2452            record_count: None,
2453        };
2454
2455        let response = list_current_orders_with_retry(http_client, &params).await?;
2456        let page_size = response.current_orders.len() as u32;
2457
2458        for order in &response.current_orders {
2459            match parse_current_order_report(order, account_id, ts_init) {
2460                Ok(mut r) => {
2461                    if let Some(ref rfo) = order.customer_order_ref
2462                        && let Ok(state) = ocm_state.lock()
2463                        && let Some(full_id) = state.resolve_client_order_id(Some(rfo.as_str()))
2464                    {
2465                        r.client_order_id = Some(full_id);
2466                    }
2467                    reports.push(r);
2468                }
2469                Err(e) => log::warn!("Failed to parse order report for {}: {e}", order.bet_id),
2470            }
2471        }
2472
2473        if !response.more_available {
2474            break;
2475        }
2476
2477        from_record += page_size;
2478    }
2479
2480    Ok(reports)
2481}
2482
2483/// Paginates `list_current_orders` into `FillReport`s without touching the
2484/// engine cache, so it is callable from any tokio task.
2485async fn fetch_fill_reports_via_http(
2486    http_client: &Arc<BetfairHttpClient>,
2487    account_id: AccountId,
2488    currency: Currency,
2489    ts_init: UnixNanos,
2490    market_ids: Option<Vec<String>>,
2491    date_range: Option<TimeRange>,
2492    ocm_state: &Arc<Mutex<OcmState>>,
2493) -> anyhow::Result<Vec<FillReport>> {
2494    let mut reports = Vec::new();
2495    let mut from_record: u32 = 0;
2496
2497    loop {
2498        let params = ListCurrentOrdersParams {
2499            bet_ids: None,
2500            market_ids: market_ids.clone(),
2501            order_projection: Some(OrderProjection::All),
2502            customer_order_refs: None,
2503            customer_strategy_refs: None,
2504            date_range: date_range.clone(),
2505            order_by: None,
2506            sort_dir: None,
2507            from_record: if from_record > 0 {
2508                Some(from_record)
2509            } else {
2510                None
2511            },
2512            record_count: None,
2513        };
2514
2515        let response = list_current_orders_with_retry(http_client, &params).await?;
2516        let page_size = response.current_orders.len() as u32;
2517
2518        for order in &response.current_orders {
2519            let size_matched = order.size_matched.unwrap_or(Decimal::ZERO);
2520            if size_matched == Decimal::ZERO {
2521                continue;
2522            }
2523
2524            match parse_current_order_fill_report(order, account_id, currency, ts_init) {
2525                Ok(mut r) => {
2526                    if let Some(ref rfo) = order.customer_order_ref
2527                        && let Ok(state) = ocm_state.lock()
2528                        && let Some(full_id) = state.resolve_client_order_id(Some(rfo.as_str()))
2529                    {
2530                        r.client_order_id = Some(full_id);
2531                    }
2532                    reports.push(r);
2533                }
2534                Err(e) => log::warn!("Failed to parse fill report for {}: {e}", order.bet_id),
2535            }
2536        }
2537
2538        if !response.more_available {
2539            break;
2540        }
2541
2542        from_record += page_size;
2543    }
2544
2545    Ok(reports)
2546}
2547
2548/// Builds an [`ExecutionMassStatus`] over `lookback_mins` of REST history.
2549#[expect(clippy::too_many_arguments)]
2550async fn fetch_post_reconnect_mass_status(
2551    http_client: &Arc<BetfairHttpClient>,
2552    client_id: ClientId,
2553    account_id: AccountId,
2554    currency: Currency,
2555    clock: &'static AtomicTime,
2556    market_ids: Option<Vec<String>>,
2557    lookback_mins: u64,
2558    ocm_state: &Arc<Mutex<OcmState>>,
2559) -> anyhow::Result<ExecutionMassStatus> {
2560    let ts_now = clock.get_time_ns();
2561    let lookback_ns = lookback_mins
2562        .saturating_mul(60)
2563        .saturating_mul(NANOSECONDS_IN_SECOND);
2564    let start = UnixNanos::from(ts_now.as_u64().saturating_sub(lookback_ns));
2565
2566    let date_range = TimeRange {
2567        from: Some(start.to_rfc3339()),
2568        to: None,
2569    };
2570
2571    let (order_reports, fill_reports) = tokio::try_join!(
2572        fetch_order_status_reports_via_http(
2573            http_client,
2574            account_id,
2575            ts_now,
2576            market_ids.clone(),
2577            false,
2578            ocm_state,
2579        ),
2580        fetch_fill_reports_via_http(
2581            http_client,
2582            account_id,
2583            currency,
2584            ts_now,
2585            market_ids,
2586            Some(date_range),
2587            ocm_state,
2588        ),
2589    )?;
2590
2591    let mut mass_status =
2592        ExecutionMassStatus::new(client_id, account_id, *BETFAIR_VENUE, ts_now, None);
2593    mass_status.add_order_reports(order_reports);
2594    mass_status.add_fill_reports(fill_reports);
2595    Ok(mass_status)
2596}
2597
2598fn list_current_orders_filter_bet_id(bet_id: String) -> ListCurrentOrdersParams {
2599    ListCurrentOrdersParams {
2600        bet_ids: Some(vec![bet_id]),
2601        market_ids: None,
2602        order_projection: None,
2603        customer_order_refs: None,
2604        customer_strategy_refs: None,
2605        date_range: None,
2606        order_by: None,
2607        sort_dir: None,
2608        from_record: None,
2609        record_count: None,
2610    }
2611}
2612
2613fn list_current_orders_filter_ref(customer_order_ref: String) -> ListCurrentOrdersParams {
2614    ListCurrentOrdersParams {
2615        bet_ids: None,
2616        market_ids: None,
2617        order_projection: None,
2618        customer_order_refs: Some(vec![customer_order_ref]),
2619        customer_strategy_refs: None,
2620        date_range: None,
2621        order_by: None,
2622        sort_dir: None,
2623        from_record: None,
2624        record_count: None,
2625    }
2626}
2627
2628fn extend_unique(
2629    candidates: &mut Vec<CurrentOrderSummary>,
2630    seen: &mut AHashSet<String>,
2631    orders: Vec<CurrentOrderSummary>,
2632) {
2633    for order in orders {
2634        if seen.insert(order.bet_id.clone()) {
2635            candidates.push(order);
2636        }
2637    }
2638}
2639
2640fn select_order_for_query(
2641    orders: &[CurrentOrderSummary],
2642    expected_instrument_id: InstrumentId,
2643    expected_client_order_id: ClientOrderId,
2644    expected_venue_order_id: Option<VenueOrderId>,
2645) -> Option<&CurrentOrderSummary> {
2646    let matching: Vec<&CurrentOrderSummary> = orders
2647        .iter()
2648        .filter(|o| {
2649            make_instrument_id(&o.market_id, o.selection_id, o.handicap) == expected_instrument_id
2650        })
2651        .collect();
2652
2653    let candidates: Vec<&CurrentOrderSummary> = if matching.is_empty() {
2654        // No instrument match: accept only an exact venue_order_id hit
2655        // (pre-existing orders without a recognizable customer_order_ref).
2656        // A lone foreign-instrument candidate is not enough, since a 32-char
2657        // customer_order_ref collision can surface a single unrelated bet.
2658        if let Some(vid) = expected_venue_order_id
2659            && let Some(order) = orders.iter().find(|o| o.bet_id == vid.as_str())
2660        {
2661            return Some(order);
2662        }
2663        log::warn!(
2664            "Betfair query_order returned {} orders for client_order_id={expected_client_order_id}, none matching instrument {expected_instrument_id}; skipping to avoid cross-instrument reconciliation",
2665            orders.len(),
2666        );
2667        return None;
2668    } else {
2669        matching
2670    };
2671
2672    // Prefer EXECUTABLE so a live replacement wins over a cancelled
2673    // predecessor sharing the same customer_order_ref.
2674    let executable: Vec<&CurrentOrderSummary> = candidates
2675        .iter()
2676        .copied()
2677        .filter(|o| o.status == BetfairOrderStatus::Executable)
2678        .collect();
2679
2680    let pool = if executable.is_empty() {
2681        candidates
2682    } else {
2683        executable
2684    };
2685
2686    // Tiebreaker: most recently placed bet. Picks the replacement over the
2687    // predecessor even when both are already terminal by poll time.
2688    pool.into_iter()
2689        .max_by(|a, b| a.placed_date.cmp(&b.placed_date))
2690}
2691
2692async fn list_current_orders_with_retry(
2693    http_client: &Arc<BetfairHttpClient>,
2694    params: &ListCurrentOrdersParams,
2695) -> anyhow::Result<CurrentOrderSummaryReport> {
2696    const RATE_LIMIT_RETRY_DELAY_SECS: u64 = 5;
2697
2698    match http_client
2699        .send_betting(METHOD_LIST_CURRENT_ORDERS, params)
2700        .await
2701    {
2702        Ok(r) => Ok(r),
2703        Err(e) if e.is_session_error() || e.is_rate_limit_error() => {
2704            if e.is_rate_limit_error() {
2705                log::warn!("Rate limited, retrying in {RATE_LIMIT_RETRY_DELAY_SECS}s");
2706                tokio::time::sleep(tokio::time::Duration::from_secs(
2707                    RATE_LIMIT_RETRY_DELAY_SECS,
2708                ))
2709                .await;
2710            } else {
2711                log::warn!("Session error, refreshing session");
2712
2713                if http_client.keep_alive().await.is_err() {
2714                    let _ = http_client.reconnect().await;
2715                }
2716            }
2717            http_client
2718                .send_betting(METHOD_LIST_CURRENT_ORDERS, params)
2719                .await
2720                .map_err(|e| anyhow::anyhow!("{e}"))
2721        }
2722        Err(e) => Err(anyhow::anyhow!("{e}")),
2723    }
2724}
2725
2726// Claims and emits the HTTP place acceptance while holding the `OcmState` lock. The OCM
2727// handler emits under the same lock, so a racing OCM fill cannot enqueue ahead of this
2728// acceptance. No-op if the OCM stream already reported the order or the accept was claimed.
2729fn emit_http_accept_if_claimed(
2730    ocm_state: &Arc<Mutex<OcmState>>,
2731    client_order_id: &ClientOrderId,
2732    emit: impl FnOnce(),
2733) {
2734    let Ok(mut state) = ocm_state.lock() else {
2735        log::error!("OcmState mutex poisoned");
2736        emit();
2737        return;
2738    };
2739
2740    if state
2741        .stream_reported_client_orders
2742        .contains(client_order_id)
2743    {
2744        log::debug!(
2745            "Suppressing late HTTP acceptance for {client_order_id}: OCM already reported order state"
2746        );
2747        return;
2748    }
2749
2750    if state.mark_accepted(*client_order_id) {
2751        emit();
2752    }
2753}
2754
2755// Returns `false` if the OCM stream already reported on this order, so the
2756// HTTP rejection event should be suppressed to avoid an `InvalidStateTrigger`
2757// against the local order state machine.
2758fn should_emit_http_reject(
2759    ocm_state: &Arc<Mutex<OcmState>>,
2760    client_order_id: &ClientOrderId,
2761    reason: &str,
2762) -> bool {
2763    let Ok(state) = ocm_state.lock() else {
2764        log::error!("OcmState mutex poisoned");
2765        return true;
2766    };
2767
2768    if state
2769        .stream_reported_client_orders
2770        .contains(client_order_id)
2771    {
2772        log::debug!(
2773            "Suppressing late HTTP rejection for {client_order_id}: OCM already reported order state ({reason})"
2774        );
2775        return false;
2776    }
2777
2778    true
2779}
2780
2781fn format_betfair_reason(
2782    error_message: Option<&str>,
2783    error_code: Option<impl fmt::Debug>,
2784    fallback: Option<String>,
2785    unknown: &str,
2786) -> String {
2787    if let Some(message) = error_message
2788        .map(str::trim)
2789        .filter(|message| !message.is_empty())
2790    {
2791        return match error_code {
2792            Some(code) => format!("{message} ({code:?})"),
2793            None => message.to_string(),
2794        };
2795    }
2796
2797    error_code
2798        .map(|code| format!("{code:?}"))
2799        .or(fallback.filter(|s| !s.trim().is_empty()))
2800        .unwrap_or_else(|| unknown.to_string())
2801}
2802
2803fn format_place_instruction_reason(
2804    instruction_report: &PlaceInstructionReport,
2805    report: &PlaceExecutionReport,
2806) -> String {
2807    format_betfair_reason(
2808        instruction_report.error_message.as_deref(),
2809        instruction_report.error_code,
2810        report_fallback(report.error_message.as_deref(), report.error_code),
2811        "unknown error",
2812    )
2813}
2814
2815fn format_cancel_instruction_reason(
2816    error_message: Option<&str>,
2817    error_code: Option<InstructionReportErrorCode>,
2818    report_error_message: Option<&str>,
2819    report_error_code: Option<ExecutionReportErrorCode>,
2820) -> String {
2821    format_betfair_reason(
2822        error_message,
2823        error_code,
2824        report_fallback(report_error_message, report_error_code),
2825        "unknown instruction error",
2826    )
2827}
2828
2829fn format_replace_instruction_reason(
2830    instruction_report: &ReplaceInstructionReport,
2831    report: &ReplaceExecutionReport,
2832) -> String {
2833    let nested_reason = instruction_report
2834        .place_instruction_report
2835        .as_ref()
2836        .and_then(|ir| instruction_fallback(ir.error_message.as_deref(), ir.error_code))
2837        .or_else(|| {
2838            instruction_report
2839                .cancel_instruction_report
2840                .as_ref()
2841                .and_then(|ir| instruction_fallback(ir.error_message.as_deref(), ir.error_code))
2842        });
2843
2844    format_betfair_reason(
2845        instruction_report.error_message.as_deref(),
2846        instruction_report.error_code,
2847        nested_reason
2848            .or_else(|| report_fallback(report.error_message.as_deref(), report.error_code)),
2849        "unknown instruction error",
2850    )
2851}
2852
2853fn report_fallback(
2854    error_message: Option<&str>,
2855    error_code: Option<ExecutionReportErrorCode>,
2856) -> Option<String> {
2857    error_message
2858        .map(str::trim)
2859        .filter(|s| !s.is_empty())
2860        .map(str::to_string)
2861        .or_else(|| error_code.map(|code| format!("{code:?}")))
2862}
2863
2864fn instruction_fallback(
2865    error_message: Option<&str>,
2866    error_code: Option<InstructionReportErrorCode>,
2867) -> Option<String> {
2868    error_message
2869        .map(str::trim)
2870        .filter(|s| !s.is_empty())
2871        .map(str::to_string)
2872        .or_else(|| error_code.map(|code| format!("{code:?}")))
2873}
2874
2875#[cfg(test)]
2876mod tests {
2877    use nautilus_common::messages::{ExecutionEvent, ExecutionReport};
2878    use nautilus_model::{
2879        identifiers::{StrategyId, TraderId},
2880        types::Quantity,
2881    };
2882    use rstest::rstest;
2883    use rust_decimal::Decimal;
2884
2885    use super::*;
2886
2887    #[rstest]
2888    #[case(
2889        Some("Price out of range"),
2890        Some(InstructionReportErrorCode::InvalidOdds),
2891        None,
2892        "unknown",
2893        "Price out of range (InvalidOdds)"
2894    )]
2895    #[case(
2896        Some("Price out of range"),
2897        None,
2898        None,
2899        "unknown",
2900        "Price out of range"
2901    )]
2902    #[case(
2903        None,
2904        Some(InstructionReportErrorCode::ErrorInOrder),
2905        None,
2906        "unknown",
2907        "ErrorInOrder"
2908    )]
2909    #[case(None, None, Some("report-level msg".to_string()), "unknown", "report-level msg")]
2910    #[case(None, None, None, "unknown error", "unknown error")]
2911    #[case(
2912        Some("  "),
2913        Some(InstructionReportErrorCode::ErrorInOrder),
2914        None,
2915        "unknown",
2916        "ErrorInOrder"
2917    )]
2918    #[case(Some(""), None, Some(String::new()), "fallback", "fallback")]
2919    #[case(Some("  \n "), None, Some("  ".to_string()), "unknown", "unknown")]
2920    fn test_format_betfair_reason(
2921        #[case] error_message: Option<&str>,
2922        #[case] error_code: Option<InstructionReportErrorCode>,
2923        #[case] fallback: Option<String>,
2924        #[case] unknown: &str,
2925        #[case] expected: &str,
2926    ) {
2927        assert_eq!(
2928            format_betfair_reason(error_message, error_code, fallback, unknown),
2929            expected,
2930        );
2931    }
2932
2933    #[rstest]
2934    fn test_ocm_state_register_and_resolve() {
2935        let mut state = OcmState::default();
2936        let client_oid = ClientOrderId::from("O-20240101-001");
2937
2938        state.register_customer_order_ref(client_oid);
2939
2940        let rfo = make_customer_order_ref(client_oid.as_str());
2941        let resolved = state.resolve_client_order_id(Some(&rfo));
2942        assert_eq!(resolved, Some(client_oid));
2943    }
2944
2945    #[rstest]
2946    fn test_ocm_state_resolve_none_for_unknown_rfo() {
2947        let state = OcmState::default();
2948        assert!(state.resolve_client_order_id(Some("unknown")).is_none());
2949        assert!(state.resolve_client_order_id(None).is_none());
2950    }
2951
2952    #[rstest]
2953    fn test_ocm_state_register_with_legacy() {
2954        let mut state = OcmState::default();
2955        let id = "O-20240101-550e8400-e29b-41d4-a716-446655440000";
2956        let client_oid = ClientOrderId::from(id);
2957
2958        state.register_customer_order_ref_with_legacy(client_oid);
2959
2960        let rfo_current = make_customer_order_ref(id);
2961        let rfo_legacy = make_customer_order_ref_legacy(id);
2962        assert_ne!(rfo_current, rfo_legacy);
2963
2964        assert_eq!(
2965            state.resolve_client_order_id(Some(&rfo_current)),
2966            Some(client_oid)
2967        );
2968        assert_eq!(
2969            state.resolve_client_order_id(Some(&rfo_legacy)),
2970            Some(client_oid)
2971        );
2972    }
2973
2974    #[rstest]
2975    fn test_ocm_state_remove_customer_order_refs() {
2976        let mut state = OcmState::default();
2977        let id = "O-20240101-550e8400-e29b-41d4-a716-446655440000";
2978        let client_oid = ClientOrderId::from(id);
2979
2980        state.register_customer_order_ref_with_legacy(client_oid);
2981        state.remove_customer_order_refs(&client_oid);
2982
2983        let rfo_current = make_customer_order_ref(id);
2984        let rfo_legacy = make_customer_order_ref_legacy(id);
2985        assert!(state.resolve_client_order_id(Some(&rfo_current)).is_none());
2986        assert!(state.resolve_client_order_id(Some(&rfo_legacy)).is_none());
2987    }
2988
2989    #[rstest]
2990    fn test_http_accept_emits_when_unclaimed() {
2991        let state = Arc::new(Mutex::new(OcmState::default()));
2992        let client_oid = ClientOrderId::from("O-001");
2993
2994        let mut emitted = false;
2995        emit_http_accept_if_claimed(&state, &client_oid, || emitted = true);
2996        assert!(emitted, "first HTTP accept must emit");
2997
2998        // A second claim for the same order is suppressed (already accepted).
2999        let mut emitted_again = false;
3000        emit_http_accept_if_claimed(&state, &client_oid, || emitted_again = true);
3001        assert!(!emitted_again, "already-accepted order must not re-emit");
3002    }
3003
3004    #[rstest]
3005    fn test_http_accept_suppressed_after_stream_report() {
3006        let client_oid = ClientOrderId::from("O-001");
3007        let mut inner = OcmState::default();
3008        inner.stream_reported_client_orders.insert(client_oid);
3009        let state = Arc::new(Mutex::new(inner));
3010
3011        let mut emitted = false;
3012        emit_http_accept_if_claimed(&state, &client_oid, || emitted = true);
3013        assert!(!emitted, "OCM-reported order must suppress the HTTP accept");
3014    }
3015
3016    #[rstest]
3017    fn test_should_emit_http_reject_without_stream_report() {
3018        let state = Arc::new(Mutex::new(OcmState::default()));
3019        let client_oid = ClientOrderId::from("O-001");
3020
3021        assert!(should_emit_http_reject(
3022            &state,
3023            &client_oid,
3024            "BetLapsedPriceImprovementTooLarge",
3025        ));
3026    }
3027
3028    #[rstest]
3029    fn test_should_not_emit_http_reject_after_stream_report() {
3030        // OCM-first race: the stream has already moved the order through
3031        // a terminal state (e.g. lapsed). A late HTTP rejection would
3032        // hit InvalidStateTrigger and pollute the own book audit log.
3033        let client_oid = ClientOrderId::from("O-001");
3034        let mut inner = OcmState::default();
3035        inner.stream_reported_client_orders.insert(client_oid);
3036        let state = Arc::new(Mutex::new(inner));
3037
3038        assert!(!should_emit_http_reject(
3039            &state,
3040            &client_oid,
3041            "BetLapsedPriceImprovementTooLarge",
3042        ));
3043    }
3044
3045    fn cancel_unmatched_order(
3046        bet_id: &str,
3047        rfo: Option<String>,
3048    ) -> crate::stream::messages::UnmatchedOrder {
3049        crate::stream::messages::UnmatchedOrder {
3050            id: bet_id.to_string(),
3051            p: Decimal::new(30, 1),
3052            s: Decimal::new(20, 0),
3053            side: crate::common::enums::StreamingSide::Back,
3054            status: crate::common::enums::StreamingOrderStatus::ExecutionComplete,
3055            pt: Some(crate::common::enums::StreamingPersistenceType::Lapse),
3056            ot: crate::common::enums::StreamingOrderType::Limit,
3057            pd: 1617863365000,
3058            bsp: None,
3059            rfo,
3060            rfs: None,
3061            rc: None,
3062            rac: None,
3063            md: None,
3064            cd: None,
3065            ld: None,
3066            avp: None,
3067            sm: None,
3068            sr: None,
3069            sl: None,
3070            sc: Some(Decimal::new(20, 0)),
3071            sv: None,
3072            lsrc: Some(crate::common::enums::LapseStatusReasonCode::SpInPlay),
3073        }
3074    }
3075
3076    fn emitter_with_receiver(
3077        account_id: AccountId,
3078    ) -> (
3079        ExecutionEventEmitter,
3080        tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
3081    ) {
3082        let clock = get_atomic_clock_realtime();
3083        let mut emitter = ExecutionEventEmitter::new(
3084            clock,
3085            TraderId::from("TESTER-001"),
3086            account_id,
3087            AccountType::Betting,
3088            None,
3089        );
3090        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
3091        emitter.set_sender(tx);
3092        (emitter, rx)
3093    }
3094
3095    #[rstest]
3096    fn test_tracked_cancel_emits_direct_order_canceled() {
3097        // A tracked cancel must emit a direct OrderCanceled, not a deferrable report.
3098        let account_id = AccountId::from("BETFAIR-001");
3099        let client_order_id = ClientOrderId::from("O-CANCEL-001");
3100        let strategy_id = StrategyId::from("S-QUOTER");
3101
3102        let mut inner = OcmState::default();
3103        inner.register_customer_order_ref(client_order_id);
3104        inner.register_order_identity(client_order_id, strategy_id);
3105        inner.mark_accepted(client_order_id); // accepted before cancel (no synthesized accept)
3106        let ocm_state = Arc::new(Mutex::new(inner));
3107
3108        let (emitter, mut rx) = emitter_with_receiver(account_id);
3109        let rfo = make_customer_order_ref(client_order_id.as_str());
3110        let uo = cancel_unmatched_order("bet_cancel", Some(rfo));
3111
3112        let processed = BetfairExecutionClient::process_unmatched_order(
3113            &uo,
3114            InstrumentId::from("1.234567-12345-0.0.BETFAIR"),
3115            account_id,
3116            Currency::from("GBP"),
3117            &emitter,
3118            &ocm_state,
3119            UnixNanos::default(),
3120            UnixNanos::default(),
3121        );
3122
3123        assert!(processed);
3124
3125        match rx.try_recv().expect("expected an execution event") {
3126            ExecutionEvent::Order(OrderEventAny::Canceled(canceled)) => {
3127                assert_eq!(canceled.client_order_id, client_order_id);
3128                assert_eq!(canceled.strategy_id, strategy_id);
3129                assert_eq!(
3130                    canceled.venue_order_id,
3131                    Some(VenueOrderId::from("bet_cancel"))
3132                );
3133            }
3134            other => panic!("expected a direct OrderCanceled event, was {other:?}"),
3135        }
3136
3137        assert!(
3138            rx.try_recv().is_err(),
3139            "tracked cancel must not also emit a status report",
3140        );
3141    }
3142
3143    #[rstest]
3144    fn test_untracked_cancel_emits_status_report() {
3145        // An order with no registered identity is external: its cancel takes the report path.
3146        let account_id = AccountId::from("BETFAIR-001");
3147        let ocm_state = Arc::new(Mutex::new(OcmState::default()));
3148
3149        let (emitter, mut rx) = emitter_with_receiver(account_id);
3150        let uo = cancel_unmatched_order("bet_external", Some("EXTERNAL-REF".to_string()));
3151
3152        let processed = BetfairExecutionClient::process_unmatched_order(
3153            &uo,
3154            InstrumentId::from("1.234567-12345-0.0.BETFAIR"),
3155            account_id,
3156            Currency::from("GBP"),
3157            &emitter,
3158            &ocm_state,
3159            UnixNanos::default(),
3160            UnixNanos::default(),
3161        );
3162
3163        assert!(processed);
3164
3165        match rx.try_recv().expect("expected an execution event") {
3166            ExecutionEvent::Report(ExecutionReport::Order(report)) => {
3167                assert_eq!(report.order_status, OrderStatus::Canceled);
3168                assert_eq!(report.venue_order_id, VenueOrderId::from("bet_external"));
3169            }
3170            other => panic!("expected an OrderStatusReport, was {other:?}"),
3171        }
3172    }
3173
3174    #[rstest]
3175    fn test_untracked_order_with_fill_reports_with_resolved_id() {
3176        // Resolved rfo but untracked: the fill is reported with the resolved
3177        // client_order_id patched on.
3178        let account_id = AccountId::from("BETFAIR-001");
3179        let client_order_id = ClientOrderId::from("O-EXT-FILL");
3180
3181        let mut inner = OcmState::default();
3182        inner.register_customer_order_ref(client_order_id); // resolves, but no identity
3183        let ocm_state = Arc::new(Mutex::new(inner));
3184
3185        let (emitter, mut rx) = emitter_with_receiver(account_id);
3186        let rfo = make_customer_order_ref(client_order_id.as_str());
3187        let uo = fill_unmatched_order("bet_ext_fill", Some(rfo), Decimal::new(10, 0));
3188
3189        let processed = BetfairExecutionClient::process_unmatched_order(
3190            &uo,
3191            InstrumentId::from("1.234567-12345-0.0.BETFAIR"),
3192            account_id,
3193            Currency::from("GBP"),
3194            &emitter,
3195            &ocm_state,
3196            UnixNanos::default(),
3197            UnixNanos::default(),
3198        );
3199
3200        assert!(processed);
3201
3202        match rx.try_recv().expect("expected a fill report") {
3203            ExecutionEvent::Report(ExecutionReport::Fill(fill)) => {
3204                assert_eq!(fill.client_order_id, Some(client_order_id));
3205                assert_eq!(fill.last_qty.as_decimal(), Decimal::new(10, 0));
3206            }
3207            other => panic!("expected a FillReport for untracked fill, was {other:?}"),
3208        }
3209
3210        match rx.try_recv().expect("expected a status report") {
3211            ExecutionEvent::Report(ExecutionReport::Order(_)) => {}
3212            other => panic!("expected an OrderStatusReport, was {other:?}"),
3213        }
3214    }
3215
3216    #[rstest]
3217    fn test_unparsable_order_returns_false_without_emitting() {
3218        let account_id = AccountId::from("BETFAIR-001");
3219        let ocm_state = Arc::new(Mutex::new(OcmState::default()));
3220        let (emitter, mut rx) = emitter_with_receiver(account_id);
3221
3222        let mut uo = cancel_unmatched_order("bet_bad", None);
3223        uo.pt = None; // LIMIT without persistence type -> parse error
3224
3225        let processed = BetfairExecutionClient::process_unmatched_order(
3226            &uo,
3227            InstrumentId::from("1.234567-12345-0.0.BETFAIR"),
3228            account_id,
3229            Currency::from("GBP"),
3230            &emitter,
3231            &ocm_state,
3232            UnixNanos::default(),
3233            UnixNanos::default(),
3234        );
3235
3236        assert!(!processed);
3237        assert!(rx.try_recv().is_err(), "unparsable order must emit nothing");
3238    }
3239
3240    #[rstest]
3241    fn test_process_ocm_voided_emits_custom_data() {
3242        // An ExecutionComplete order with size_voided > 0 publishes a
3243        // BetfairOrderVoided custom data event.
3244        let data = crate::common::testing::load_test_json("stream/ocm_VOIDED.json");
3245        let ocm = match crate::stream::messages::stream_decode(data.as_bytes()).unwrap() {
3246            crate::stream::messages::StreamMessage::OrderChange(ocm) => ocm,
3247            other => panic!("expected an OCM stream message, was {other:?}"),
3248        };
3249
3250        let account_id = AccountId::from("BETFAIR-001");
3251        let ocm_state = Arc::new(Mutex::new(OcmState::default()));
3252        let (emitter, _rx) = emitter_with_receiver(account_id);
3253        let (data_tx, mut data_rx) = tokio::sync::mpsc::unbounded_channel();
3254
3255        BetfairExecutionClient::process_ocm(
3256            &ocm,
3257            account_id,
3258            Currency::from("GBP"),
3259            &emitter,
3260            &ocm_state,
3261            &data_tx,
3262            None,
3263            false,
3264        );
3265
3266        let voided = std::iter::from_fn(|| data_rx.try_recv().ok())
3267            .any(|event| matches!(event, DataEvent::Data(Data::Custom(_))));
3268        assert!(voided, "voided OCM must publish a custom voided data event");
3269    }
3270
3271    fn fill_unmatched_order(
3272        bet_id: &str,
3273        rfo: Option<String>,
3274        size_matched: Decimal,
3275    ) -> crate::stream::messages::UnmatchedOrder {
3276        crate::stream::messages::UnmatchedOrder {
3277            id: bet_id.to_string(),
3278            p: Decimal::new(20, 1),
3279            s: Decimal::new(20, 0),
3280            side: crate::common::enums::StreamingSide::Back,
3281            status: crate::common::enums::StreamingOrderStatus::Executable,
3282            pt: Some(crate::common::enums::StreamingPersistenceType::Lapse),
3283            ot: crate::common::enums::StreamingOrderType::Limit,
3284            pd: 1617863365000,
3285            bsp: None,
3286            rfo,
3287            rfs: None,
3288            rc: None,
3289            rac: None,
3290            md: None,
3291            cd: None,
3292            ld: None,
3293            avp: Some(Decimal::new(20, 1)),
3294            sm: Some(size_matched),
3295            sr: None,
3296            sl: None,
3297            sc: None,
3298            sv: None,
3299            lsrc: None,
3300        }
3301    }
3302
3303    #[rstest]
3304    fn test_tracked_fill_synthesizes_accept_then_emits_filled() {
3305        // First OCM is a fill with no prior accept: synthesize OrderAccepted, then OrderFilled.
3306        let account_id = AccountId::from("BETFAIR-001");
3307        let client_order_id = ClientOrderId::from("O-FILL-100");
3308        let strategy_id = StrategyId::from("S-QUOTER");
3309
3310        let mut inner = OcmState::default();
3311        inner.register_customer_order_ref(client_order_id);
3312        inner.register_order_identity(client_order_id, strategy_id);
3313        let ocm_state = Arc::new(Mutex::new(inner));
3314
3315        let (emitter, mut rx) = emitter_with_receiver(account_id);
3316        let rfo = make_customer_order_ref(client_order_id.as_str());
3317        let uo = fill_unmatched_order("bet_fill", Some(rfo), Decimal::new(10, 0));
3318
3319        let processed = BetfairExecutionClient::process_unmatched_order(
3320            &uo,
3321            InstrumentId::from("1.234567-12345-0.0.BETFAIR"),
3322            account_id,
3323            Currency::from("GBP"),
3324            &emitter,
3325            &ocm_state,
3326            UnixNanos::default(),
3327            UnixNanos::default(),
3328        );
3329
3330        assert!(processed);
3331
3332        match rx.try_recv().expect("expected an accepted event") {
3333            ExecutionEvent::Order(OrderEventAny::Accepted(accepted)) => {
3334                assert_eq!(accepted.client_order_id, client_order_id);
3335                assert_eq!(accepted.strategy_id, strategy_id);
3336            }
3337            other => panic!("expected a synthesized OrderAccepted, was {other:?}"),
3338        }
3339
3340        match rx.try_recv().expect("expected a filled event") {
3341            ExecutionEvent::Order(OrderEventAny::Filled(filled)) => {
3342                assert_eq!(filled.client_order_id, client_order_id);
3343                assert_eq!(filled.strategy_id, strategy_id);
3344                assert_eq!(filled.venue_order_id, VenueOrderId::from("bet_fill"));
3345                assert_eq!(filled.last_qty.as_decimal(), Decimal::new(10, 0));
3346            }
3347            other => panic!("expected a direct OrderFilled event, was {other:?}"),
3348        }
3349
3350        assert!(
3351            rx.try_recv().is_err(),
3352            "tracked fill must not also emit a report",
3353        );
3354    }
3355
3356    #[rstest]
3357    fn test_tracked_fill_on_accepted_order_skips_synth_accept() {
3358        // Already accepted via HTTP ack: the tracked fill emits only OrderFilled,
3359        // no duplicate accept.
3360        let account_id = AccountId::from("BETFAIR-001");
3361        let client_order_id = ClientOrderId::from("O-FILL-200");
3362        let strategy_id = StrategyId::from("S-QUOTER");
3363
3364        let mut inner = OcmState::default();
3365        inner.register_customer_order_ref(client_order_id);
3366        inner.register_order_identity(client_order_id, strategy_id);
3367        inner.mark_accepted(client_order_id); // already accepted via HTTP place ack
3368        let ocm_state = Arc::new(Mutex::new(inner));
3369
3370        let (emitter, mut rx) = emitter_with_receiver(account_id);
3371        let rfo = make_customer_order_ref(client_order_id.as_str());
3372        let uo = fill_unmatched_order("bet_fill2", Some(rfo), Decimal::new(10, 0));
3373
3374        let processed = BetfairExecutionClient::process_unmatched_order(
3375            &uo,
3376            InstrumentId::from("1.234567-12345-0.0.BETFAIR"),
3377            account_id,
3378            Currency::from("GBP"),
3379            &emitter,
3380            &ocm_state,
3381            UnixNanos::default(),
3382            UnixNanos::default(),
3383        );
3384
3385        assert!(processed);
3386
3387        match rx.try_recv().expect("expected a filled event") {
3388            ExecutionEvent::Order(OrderEventAny::Filled(filled)) => {
3389                assert_eq!(filled.client_order_id, client_order_id);
3390                assert_eq!(filled.last_qty.as_decimal(), Decimal::new(10, 0));
3391            }
3392            other => panic!("expected OrderFilled with no preceding accept, was {other:?}"),
3393        }
3394
3395        assert!(
3396            rx.try_recv().is_err(),
3397            "already-accepted order must not re-synthesize OrderAccepted",
3398        );
3399    }
3400
3401    #[rstest]
3402    fn test_ocm_state_suppress_cancel_for_replaced() {
3403        let mut state = OcmState::default();
3404        let client_oid = ClientOrderId::from("O-001");
3405
3406        state.replaced_venue_order_ids.insert("old_bet".to_string());
3407        assert!(state.should_suppress_cancel(&client_oid, "old_bet"));
3408        assert!(!state.should_suppress_cancel(&client_oid, "new_bet"));
3409    }
3410
3411    #[rstest]
3412    fn test_ocm_state_suppress_cancel_for_pending_replace() {
3413        let mut state = OcmState::default();
3414        let client_oid = ClientOrderId::from("O-001");
3415
3416        state
3417            .pending_update_keys
3418            .insert((client_oid, "old_bet".to_string()));
3419
3420        assert!(state.should_suppress_cancel(&client_oid, "old_bet"));
3421        assert!(!state.should_suppress_cancel(&client_oid, "other_bet"));
3422    }
3423
3424    #[rstest]
3425    fn test_ocm_state_cleanup_terminal_with_pending_replace() {
3426        let mut state = OcmState::default();
3427        let client_oid = ClientOrderId::from("O-001");
3428
3429        state.register_customer_order_ref(client_oid);
3430        state
3431            .pending_update_keys
3432            .insert((client_oid, "old_bet".to_string()));
3433
3434        // Should NOT remove refs because replace is pending
3435        state.cleanup_terminal_order(&client_oid);
3436        let rfo = make_customer_order_ref(client_oid.as_str());
3437        assert!(state.resolve_client_order_id(Some(&rfo)).is_some());
3438    }
3439
3440    #[rstest]
3441    fn test_ocm_state_cleanup_terminal_without_pending() {
3442        let mut state = OcmState::default();
3443        let client_oid = ClientOrderId::from("O-001");
3444
3445        state.register_customer_order_ref(client_oid);
3446
3447        // Should remove refs because no pending replace
3448        state.cleanup_terminal_order(&client_oid);
3449        let rfo = make_customer_order_ref(client_oid.as_str());
3450        assert!(state.resolve_client_order_id(Some(&rfo)).is_none());
3451    }
3452
3453    #[rstest]
3454    fn test_ocm_state_sync_from_orders() {
3455        let mut state = OcmState::default();
3456
3457        let orders = vec![
3458            OrderSyncEntry {
3459                bet_id: "bet1".to_string(),
3460                client_order_id: ClientOrderId::from("O-001"),
3461                filled_qty: Decimal::new(10, 0),
3462                avg_px: Decimal::new(25, 1),
3463                is_closed: false,
3464                trade_ids: Vec::new(),
3465            },
3466            OrderSyncEntry {
3467                bet_id: "bet2".to_string(),
3468                client_order_id: ClientOrderId::from("O-002"),
3469                filled_qty: Decimal::new(5, 0),
3470                avg_px: Decimal::new(30, 1),
3471                is_closed: true,
3472                trade_ids: Vec::new(),
3473            },
3474        ];
3475
3476        state.sync_from_orders(&orders);
3477
3478        // Open order: should have customer_order_ref registered
3479        let rfo1 = make_customer_order_ref("O-001");
3480        assert!(state.resolve_client_order_id(Some(&rfo1)).is_some());
3481
3482        // Closed order: should be in terminal_orders, no customer_order_ref
3483        assert!(state.terminal_orders.contains("bet2"));
3484        let rfo2 = make_customer_order_ref("O-002");
3485        assert!(state.resolve_client_order_id(Some(&rfo2)).is_none());
3486    }
3487
3488    #[rstest]
3489    fn test_reconnect_signal_not_sent_on_initial_connection() {
3490        let has_initial_connection = Arc::new(AtomicBool::new(false));
3491        let (reconnect_tx, mut reconnect_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
3492
3493        let has_initial = Arc::clone(&has_initial_connection);
3494        let handler = move |_data: &[u8]| {
3495            if has_initial.swap(true, Ordering::SeqCst) {
3496                let _ = reconnect_tx.send(());
3497            }
3498        };
3499
3500        // First connection message: no signal
3501        handler(br#"{"op":"connection","connectionId":"abc"}"#);
3502        assert!(reconnect_rx.try_recv().is_err());
3503        assert!(has_initial_connection.load(Ordering::SeqCst));
3504    }
3505
3506    #[rstest]
3507    fn test_reconnect_signal_sent_on_subsequent_connection() {
3508        let has_initial_connection = Arc::new(AtomicBool::new(false));
3509        let (reconnect_tx, mut reconnect_rx) = tokio::sync::mpsc::unbounded_channel::<()>();
3510
3511        let has_initial = Arc::clone(&has_initial_connection);
3512        let tx = reconnect_tx;
3513        let handler = move |_data: &[u8]| {
3514            if has_initial.swap(true, Ordering::SeqCst) {
3515                let _ = tx.send(());
3516            }
3517        };
3518
3519        // First connection: no signal
3520        handler(br#"{"op":"connection","connectionId":"abc"}"#);
3521        assert!(reconnect_rx.try_recv().is_err());
3522
3523        // Second connection: signal sent
3524        handler(br#"{"op":"connection","connectionId":"def"}"#);
3525        assert!(reconnect_rx.try_recv().is_ok());
3526
3527        // Third connection: signal sent again
3528        handler(br#"{"op":"connection","connectionId":"ghi"}"#);
3529        assert!(reconnect_rx.try_recv().is_ok());
3530    }
3531
3532    #[rstest]
3533    fn test_reconnect_sets_is_reconciling_flag() {
3534        // Mirrors the production handler: a second Connection raises both flags.
3535        let has_initial_connection = Arc::new(AtomicBool::new(false));
3536        let pending_resync = Arc::new(AtomicBool::new(false));
3537        let is_reconciling = Arc::new(AtomicBool::new(false));
3538
3539        let has_initial = Arc::clone(&has_initial_connection);
3540        let pending = Arc::clone(&pending_resync);
3541        let reconciling = Arc::clone(&is_reconciling);
3542        let handler = move |_data: &[u8]| {
3543            if has_initial.swap(true, Ordering::SeqCst) {
3544                pending.store(true, Ordering::Release);
3545                reconciling.store(true, Ordering::Release);
3546            }
3547        };
3548
3549        handler(br#"{"op":"connection","connectionId":"first"}"#);
3550        assert!(!pending_resync.load(Ordering::Acquire));
3551        assert!(!is_reconciling.load(Ordering::Acquire));
3552
3553        handler(br#"{"op":"connection","connectionId":"second"}"#);
3554        assert!(pending_resync.load(Ordering::Acquire));
3555        assert!(is_reconciling.load(Ordering::Acquire));
3556    }
3557
3558    #[rstest]
3559    fn test_ocm_state_persists_across_reconnections() {
3560        let ocm_state = Arc::new(Mutex::new(OcmState::default()));
3561
3562        // Populate state before "reconnect"
3563        {
3564            let mut state = ocm_state.lock().unwrap();
3565            let orders = vec![
3566                OrderSyncEntry {
3567                    bet_id: "bet1".to_string(),
3568                    client_order_id: ClientOrderId::from("O-001"),
3569                    filled_qty: Decimal::new(10, 0),
3570                    avg_px: Decimal::new(25, 1),
3571                    is_closed: false,
3572                    trade_ids: Vec::new(),
3573                },
3574                OrderSyncEntry {
3575                    bet_id: "bet2".to_string(),
3576                    client_order_id: ClientOrderId::from("O-002"),
3577                    filled_qty: Decimal::ZERO,
3578                    avg_px: Decimal::ZERO,
3579                    is_closed: true,
3580                    trade_ids: Vec::new(),
3581                },
3582            ];
3583            state.sync_from_orders(&orders);
3584        }
3585
3586        // Verify state survives (simulates reconnection where Arc<Mutex<OcmState>> persists)
3587        let state = ocm_state.lock().unwrap();
3588        let rfo = make_customer_order_ref("O-001");
3589        assert_eq!(
3590            state.resolve_client_order_id(Some(&rfo)),
3591            Some(ClientOrderId::from("O-001")),
3592        );
3593        assert!(state.terminal_orders.contains("bet2"));
3594        assert!(!state.terminal_orders.contains("bet1"));
3595    }
3596
3597    #[rstest]
3598    fn test_ocm_state_sync_from_orders_populates_fill_tracker() {
3599        let mut state = OcmState::default();
3600
3601        let orders = vec![OrderSyncEntry {
3602            bet_id: "bet_fill".to_string(),
3603            client_order_id: ClientOrderId::from("O-FILL-001"),
3604            filled_qty: Decimal::new(15, 0),
3605            avg_px: Decimal::new(30, 1),
3606            is_closed: false,
3607            trade_ids: Vec::new(),
3608        }];
3609
3610        state.sync_from_orders(&orders);
3611
3612        // Fill tracker should be pre-populated so that a stream update with
3613        // sm=15 does NOT produce a duplicate fill
3614        let uo = crate::stream::messages::UnmatchedOrder {
3615            id: "bet_fill".to_string(),
3616            p: Decimal::new(30, 1),
3617            s: Decimal::new(20, 0),
3618            side: crate::common::enums::StreamingSide::Back,
3619            status: crate::common::enums::StreamingOrderStatus::Executable,
3620            pt: Some(crate::common::enums::StreamingPersistenceType::Lapse),
3621            ot: crate::common::enums::StreamingOrderType::Limit,
3622            pd: 1617863365000,
3623            bsp: None,
3624            rfo: Some("O-FILL-001".to_string()),
3625            rfs: None,
3626            rc: None,
3627            rac: None,
3628            md: None,
3629            cd: None,
3630            ld: None,
3631            avp: Some(Decimal::new(30, 1)),
3632            sm: Some(Decimal::new(15, 0)),
3633            sr: None,
3634            sl: None,
3635            sc: None,
3636            sv: None,
3637            lsrc: None,
3638        };
3639
3640        let instrument_id = InstrumentId::from("1.234567-12345-0.0.BETFAIR");
3641        let result = state.fill_tracker.maybe_fill_report(
3642            &uo,
3643            uo.s,
3644            instrument_id,
3645            AccountId::from("BETFAIR-001"),
3646            Currency::from("GBP"),
3647            UnixNanos::default(),
3648            UnixNanos::default(),
3649        );
3650
3651        assert!(
3652            result.is_none(),
3653            "synced fill should prevent duplicate fill report"
3654        );
3655    }
3656
3657    #[rstest]
3658    fn test_ocm_state_sync_seeds_published_trade_ids_from_cache() {
3659        use crate::stream::parse::make_trade_id;
3660
3661        let mut state = OcmState::default();
3662
3663        // Seeded filled_qty < sm: cumulative-size gate would pass, so only
3664        // the trade-ids seeded by sync_from_orders can block the fill
3665        let uo_for_id = crate::stream::messages::UnmatchedOrder {
3666            id: "bet_seed".to_string(),
3667            p: Decimal::new(25, 1),
3668            s: Decimal::new(20, 0),
3669            side: crate::common::enums::StreamingSide::Back,
3670            status: crate::common::enums::StreamingOrderStatus::Executable,
3671            pt: Some(crate::common::enums::StreamingPersistenceType::Lapse),
3672            ot: crate::common::enums::StreamingOrderType::Limit,
3673            pd: 1617863365000,
3674            bsp: None,
3675            rfo: Some("O-SEED-001".to_string()),
3676            rfs: None,
3677            rc: None,
3678            rac: None,
3679            md: None,
3680            cd: None,
3681            ld: None,
3682            avp: Some(Decimal::new(25, 1)),
3683            sm: Some(Decimal::new(8, 0)),
3684            sr: None,
3685            sl: None,
3686            sc: None,
3687            sv: None,
3688            lsrc: None,
3689        };
3690        let trade_id = make_trade_id(&uo_for_id).to_string();
3691
3692        let orders = vec![OrderSyncEntry {
3693            bet_id: "bet_seed".to_string(),
3694            client_order_id: ClientOrderId::from("O-SEED-001"),
3695            filled_qty: Decimal::new(4, 0),
3696            avg_px: Decimal::new(25, 1),
3697            is_closed: false,
3698            trade_ids: vec![trade_id],
3699        }];
3700
3701        state.sync_from_orders(&orders);
3702
3703        let result = state.fill_tracker.maybe_fill_report(
3704            &uo_for_id,
3705            uo_for_id.s,
3706            InstrumentId::from("1.234567-12345-0.0.BETFAIR"),
3707            AccountId::from("BETFAIR-001"),
3708            Currency::from("GBP"),
3709            UnixNanos::default(),
3710            UnixNanos::default(),
3711        );
3712
3713        assert!(
3714            result.is_none(),
3715            "sync_from_orders must seed trade-ids so a known fill is not re-emitted \
3716             even when the cumulative-size gate would pass",
3717        );
3718    }
3719
3720    #[rstest]
3721    fn test_ocm_state_sync_from_orders_incremental_fill_after_sync() {
3722        let mut state = OcmState::default();
3723
3724        let orders = vec![OrderSyncEntry {
3725            bet_id: "bet_inc".to_string(),
3726            client_order_id: ClientOrderId::from("O-INC-001"),
3727            filled_qty: Decimal::new(10, 0),
3728            avg_px: Decimal::new(25, 1),
3729            is_closed: false,
3730            trade_ids: Vec::new(),
3731        }];
3732
3733        state.sync_from_orders(&orders);
3734
3735        // Stream update with sm=18 (8 more than synced 10)
3736        let uo = crate::stream::messages::UnmatchedOrder {
3737            id: "bet_inc".to_string(),
3738            p: Decimal::new(25, 1),
3739            s: Decimal::new(20, 0),
3740            side: crate::common::enums::StreamingSide::Lay,
3741            status: crate::common::enums::StreamingOrderStatus::Executable,
3742            pt: Some(crate::common::enums::StreamingPersistenceType::Persist),
3743            ot: crate::common::enums::StreamingOrderType::Limit,
3744            pd: 1617863365000,
3745            bsp: None,
3746            rfo: Some("O-INC-001".to_string()),
3747            rfs: None,
3748            rc: None,
3749            rac: None,
3750            md: None,
3751            cd: None,
3752            ld: None,
3753            avp: Some(Decimal::new(26, 1)),
3754            sm: Some(Decimal::new(18, 0)),
3755            sr: None,
3756            sl: None,
3757            sc: None,
3758            sv: None,
3759            lsrc: None,
3760        };
3761
3762        let instrument_id = InstrumentId::from("1.234567-12345-0.0.BETFAIR");
3763        let result = state.fill_tracker.maybe_fill_report(
3764            &uo,
3765            uo.s,
3766            instrument_id,
3767            AccountId::from("BETFAIR-001"),
3768            Currency::from("GBP"),
3769            UnixNanos::default(),
3770            UnixNanos::default(),
3771        );
3772
3773        let fill = result.expect("should produce incremental fill of 8");
3774        assert_eq!(fill.last_qty, Quantity::from("8.00"));
3775    }
3776
3777    #[rstest]
3778    fn test_ocm_state_sync_from_orders_zero_filled_not_synced() {
3779        let mut state = OcmState::default();
3780
3781        let orders = vec![OrderSyncEntry {
3782            bet_id: "bet_zero".to_string(),
3783            client_order_id: ClientOrderId::from("O-ZERO-001"),
3784            filled_qty: Decimal::ZERO,
3785            avg_px: Decimal::ZERO,
3786            is_closed: false,
3787            trade_ids: Vec::new(),
3788        }];
3789
3790        state.sync_from_orders(&orders);
3791
3792        // RFO should still be registered even if no fills
3793        let rfo = make_customer_order_ref("O-ZERO-001");
3794        assert!(state.resolve_client_order_id(Some(&rfo)).is_some());
3795
3796        // A stream update with sm=5 should produce a fill (not blocked by sync)
3797        let uo = crate::stream::messages::UnmatchedOrder {
3798            id: "bet_zero".to_string(),
3799            p: Decimal::new(30, 1),
3800            s: Decimal::new(10, 0),
3801            side: crate::common::enums::StreamingSide::Back,
3802            status: crate::common::enums::StreamingOrderStatus::Executable,
3803            pt: Some(crate::common::enums::StreamingPersistenceType::Lapse),
3804            ot: crate::common::enums::StreamingOrderType::Limit,
3805            pd: 1617863365000,
3806            bsp: None,
3807            rfo: None,
3808            rfs: None,
3809            rc: None,
3810            rac: None,
3811            md: None,
3812            cd: None,
3813            ld: None,
3814            avp: Some(Decimal::new(30, 1)),
3815            sm: Some(Decimal::new(5, 0)),
3816            sr: None,
3817            sl: None,
3818            sc: None,
3819            sv: None,
3820            lsrc: None,
3821        };
3822        let instrument_id = InstrumentId::from("1.234567-12345-0.0.BETFAIR");
3823        let result = state.fill_tracker.maybe_fill_report(
3824            &uo,
3825            uo.s,
3826            instrument_id,
3827            AccountId::from("BETFAIR-001"),
3828            Currency::from("GBP"),
3829            UnixNanos::default(),
3830            UnixNanos::default(),
3831        );
3832        assert!(
3833            result.is_some(),
3834            "zero-filled order should not block new fills"
3835        );
3836    }
3837
3838    #[rstest]
3839    fn test_ocm_state_sync_multiple_open_and_closed() {
3840        let mut state = OcmState::default();
3841
3842        let orders = vec![
3843            OrderSyncEntry {
3844                bet_id: "bet_a".to_string(),
3845                client_order_id: ClientOrderId::from("O-A"),
3846                filled_qty: Decimal::new(5, 0),
3847                avg_px: Decimal::new(20, 1),
3848                is_closed: false,
3849                trade_ids: Vec::new(),
3850            },
3851            OrderSyncEntry {
3852                bet_id: "bet_b".to_string(),
3853                client_order_id: ClientOrderId::from("O-B"),
3854                filled_qty: Decimal::ZERO,
3855                avg_px: Decimal::ZERO,
3856                is_closed: true,
3857                trade_ids: Vec::new(),
3858            },
3859            OrderSyncEntry {
3860                bet_id: "bet_c".to_string(),
3861                client_order_id: ClientOrderId::from("O-C"),
3862                filled_qty: Decimal::new(100, 0),
3863                avg_px: Decimal::new(15, 1),
3864                is_closed: true,
3865                trade_ids: Vec::new(),
3866            },
3867            OrderSyncEntry {
3868                bet_id: "bet_d".to_string(),
3869                client_order_id: ClientOrderId::from("O-D"),
3870                filled_qty: Decimal::ZERO,
3871                avg_px: Decimal::ZERO,
3872                is_closed: false,
3873                trade_ids: Vec::new(),
3874            },
3875        ];
3876
3877        state.sync_from_orders(&orders);
3878
3879        // Open orders have RFO registered
3880        assert!(
3881            state
3882                .resolve_client_order_id(Some(&make_customer_order_ref("O-A")))
3883                .is_some()
3884        );
3885        assert!(
3886            state
3887                .resolve_client_order_id(Some(&make_customer_order_ref("O-D")))
3888                .is_some()
3889        );
3890
3891        // Closed orders are terminal
3892        assert!(state.terminal_orders.contains("bet_b"));
3893        assert!(state.terminal_orders.contains("bet_c"));
3894        assert!(!state.terminal_orders.contains("bet_a"));
3895        assert!(!state.terminal_orders.contains("bet_d"));
3896
3897        // Closed orders do NOT get RFO registered
3898        assert!(
3899            state
3900                .resolve_client_order_id(Some(&make_customer_order_ref("O-B")))
3901                .is_none()
3902        );
3903    }
3904
3905    fn make_summary(
3906        bet_id: &str,
3907        market_id: &str,
3908        selection_id: u64,
3909        handicap: Decimal,
3910        status: BetfairOrderStatus,
3911        placed_date: &str,
3912    ) -> CurrentOrderSummary {
3913        CurrentOrderSummary {
3914            bet_id: bet_id.to_string(),
3915            market_id: market_id.to_string(),
3916            selection_id,
3917            handicap,
3918            price_size: crate::http::models::PriceSize {
3919                price: Decimal::new(20, 1),
3920                size: Decimal::new(10, 0),
3921            },
3922            bsp_liability: Decimal::ZERO,
3923            side: BetfairSide::Back,
3924            status,
3925            persistence_type: PersistenceType::Lapse,
3926            order_type: BetfairOrderType::Limit,
3927            placed_date: placed_date.to_string(),
3928            matched_date: None,
3929            average_price_matched: None,
3930            size_matched: None,
3931            size_remaining: Some(Decimal::new(10, 0)),
3932            size_lapsed: None,
3933            size_cancelled: None,
3934            size_voided: None,
3935            regulator_auth_code: None,
3936            regulator_code: None,
3937            customer_order_ref: None,
3938            customer_strategy_ref: None,
3939        }
3940    }
3941
3942    #[rstest]
3943    fn test_select_order_for_query_single_executable() {
3944        let cid = ClientOrderId::from("O-001");
3945        let orders = vec![make_summary(
3946            "bet_1",
3947            "1.100",
3948            12345,
3949            Decimal::ZERO,
3950            BetfairOrderStatus::Executable,
3951            "2026-04-18T10:00:00Z",
3952        )];
3953        let expected = make_instrument_id("1.100", 12345, Decimal::ZERO);
3954
3955        let selected = select_order_for_query(&orders, expected, cid, None);
3956        assert_eq!(selected.map(|o| o.bet_id.as_str()), Some("bet_1"));
3957    }
3958
3959    #[rstest]
3960    fn test_select_order_for_query_single_terminal() {
3961        let cid = ClientOrderId::from("O-001");
3962        let orders = vec![make_summary(
3963            "bet_1",
3964            "1.100",
3965            12345,
3966            Decimal::ZERO,
3967            BetfairOrderStatus::ExecutionComplete,
3968            "2026-04-18T10:00:00Z",
3969        )];
3970        let expected = make_instrument_id("1.100", 12345, Decimal::ZERO);
3971
3972        let selected = select_order_for_query(&orders, expected, cid, None);
3973        assert_eq!(selected.map(|o| o.bet_id.as_str()), Some("bet_1"));
3974    }
3975
3976    #[rstest]
3977    fn test_select_order_for_query_replace_prefers_executable() {
3978        let cid = ClientOrderId::from("O-001");
3979        let orders = vec![
3980            make_summary(
3981                "bet_old",
3982                "1.100",
3983                12345,
3984                Decimal::ZERO,
3985                BetfairOrderStatus::ExecutionComplete,
3986                "2026-04-18T10:00:00Z",
3987            ),
3988            make_summary(
3989                "bet_new",
3990                "1.100",
3991                12345,
3992                Decimal::ZERO,
3993                BetfairOrderStatus::Executable,
3994                "2026-04-18T10:05:00Z",
3995            ),
3996        ];
3997        let expected = make_instrument_id("1.100", 12345, Decimal::ZERO);
3998
3999        let selected = select_order_for_query(&orders, expected, cid, None);
4000        assert_eq!(selected.map(|o| o.bet_id.as_str()), Some("bet_new"));
4001    }
4002
4003    #[rstest]
4004    fn test_select_order_for_query_multiple_executable_prefers_most_recent() {
4005        let cid = ClientOrderId::from("O-001");
4006        let orders = vec![
4007            make_summary(
4008                "bet_old",
4009                "1.100",
4010                12345,
4011                Decimal::ZERO,
4012                BetfairOrderStatus::Executable,
4013                "2026-04-18T10:00:00Z",
4014            ),
4015            make_summary(
4016                "bet_new",
4017                "1.100",
4018                12345,
4019                Decimal::ZERO,
4020                BetfairOrderStatus::Executable,
4021                "2026-04-18T10:05:00Z",
4022            ),
4023        ];
4024        let expected = make_instrument_id("1.100", 12345, Decimal::ZERO);
4025
4026        let selected = select_order_for_query(&orders, expected, cid, None);
4027        assert_eq!(selected.map(|o| o.bet_id.as_str()), Some("bet_new"));
4028    }
4029
4030    #[rstest]
4031    fn test_select_order_for_query_multiple_terminal_prefers_most_recent() {
4032        let cid = ClientOrderId::from("O-001");
4033        let orders = vec![
4034            make_summary(
4035                "bet_old",
4036                "1.100",
4037                12345,
4038                Decimal::ZERO,
4039                BetfairOrderStatus::ExecutionComplete,
4040                "2026-04-18T10:00:00Z",
4041            ),
4042            make_summary(
4043                "bet_new",
4044                "1.100",
4045                12345,
4046                Decimal::ZERO,
4047                BetfairOrderStatus::ExecutionComplete,
4048                "2026-04-18T10:05:00Z",
4049            ),
4050        ];
4051        let expected = make_instrument_id("1.100", 12345, Decimal::ZERO);
4052
4053        let selected = select_order_for_query(&orders, expected, cid, None);
4054        assert_eq!(selected.map(|o| o.bet_id.as_str()), Some("bet_new"));
4055    }
4056
4057    #[rstest]
4058    fn test_select_order_for_query_foreign_only_without_vid_returns_none() {
4059        let cid = ClientOrderId::from("O-001");
4060        let orders = vec![make_summary(
4061            "bet_foreign",
4062            "1.999",
4063            99999,
4064            Decimal::ZERO,
4065            BetfairOrderStatus::Executable,
4066            "2026-04-18T10:00:00Z",
4067        )];
4068        let expected = make_instrument_id("1.100", 12345, Decimal::ZERO);
4069
4070        let selected = select_order_for_query(&orders, expected, cid, None);
4071        assert!(selected.is_none());
4072    }
4073
4074    #[rstest]
4075    fn test_select_order_for_query_foreign_only_with_vid_match_returns_match() {
4076        let cid = ClientOrderId::from("O-001");
4077        let orders = vec![make_summary(
4078            "bet_foreign",
4079            "1.999",
4080            99999,
4081            Decimal::ZERO,
4082            BetfairOrderStatus::Executable,
4083            "2026-04-18T10:00:00Z",
4084        )];
4085        let expected = make_instrument_id("1.100", 12345, Decimal::ZERO);
4086        let vid = VenueOrderId::from("bet_foreign");
4087
4088        let selected = select_order_for_query(&orders, expected, cid, Some(vid));
4089        assert_eq!(selected.map(|o| o.bet_id.as_str()), Some("bet_foreign"));
4090    }
4091
4092    #[rstest]
4093    fn test_select_order_for_query_foreign_only_vid_mismatch_returns_none() {
4094        let cid = ClientOrderId::from("O-001");
4095        let orders = vec![
4096            make_summary(
4097                "bet_foreign_1",
4098                "1.999",
4099                99999,
4100                Decimal::ZERO,
4101                BetfairOrderStatus::Executable,
4102                "2026-04-18T10:00:00Z",
4103            ),
4104            make_summary(
4105                "bet_foreign_2",
4106                "1.888",
4107                88888,
4108                Decimal::ZERO,
4109                BetfairOrderStatus::Executable,
4110                "2026-04-18T10:05:00Z",
4111            ),
4112        ];
4113        let expected = make_instrument_id("1.100", 12345, Decimal::ZERO);
4114        let vid = VenueOrderId::from("bet_unknown");
4115
4116        let selected = select_order_for_query(&orders, expected, cid, Some(vid));
4117        assert!(selected.is_none());
4118    }
4119
4120    #[rstest]
4121    fn test_select_order_for_query_mixed_returns_matching_instrument() {
4122        let cid = ClientOrderId::from("O-001");
4123        let orders = vec![
4124            make_summary(
4125                "bet_foreign",
4126                "1.999",
4127                99999,
4128                Decimal::ZERO,
4129                BetfairOrderStatus::Executable,
4130                "2026-04-18T10:05:00Z",
4131            ),
4132            make_summary(
4133                "bet_match",
4134                "1.100",
4135                12345,
4136                Decimal::ZERO,
4137                BetfairOrderStatus::ExecutionComplete,
4138                "2026-04-18T10:00:00Z",
4139            ),
4140        ];
4141        let expected = make_instrument_id("1.100", 12345, Decimal::ZERO);
4142
4143        let selected = select_order_for_query(&orders, expected, cid, None);
4144        assert_eq!(selected.map(|o| o.bet_id.as_str()), Some("bet_match"));
4145    }
4146
4147    #[rstest]
4148    fn test_extend_unique_filters_duplicates() {
4149        let mut candidates: Vec<CurrentOrderSummary> = Vec::new();
4150        let mut seen: AHashSet<String> = AHashSet::new();
4151
4152        let orders = vec![
4153            make_summary(
4154                "bet_1",
4155                "1.100",
4156                12345,
4157                Decimal::ZERO,
4158                BetfairOrderStatus::Executable,
4159                "2026-04-18T10:00:00Z",
4160            ),
4161            make_summary(
4162                "bet_1",
4163                "1.100",
4164                12345,
4165                Decimal::ZERO,
4166                BetfairOrderStatus::Executable,
4167                "2026-04-18T10:01:00Z",
4168            ),
4169            make_summary(
4170                "bet_2",
4171                "1.100",
4172                12345,
4173                Decimal::ZERO,
4174                BetfairOrderStatus::Executable,
4175                "2026-04-18T10:02:00Z",
4176            ),
4177        ];
4178
4179        extend_unique(&mut candidates, &mut seen, orders);
4180
4181        assert_eq!(candidates.len(), 2);
4182        assert_eq!(candidates[0].bet_id, "bet_1");
4183        assert_eq!(candidates[0].placed_date, "2026-04-18T10:00:00Z");
4184        assert_eq!(candidates[1].bet_id, "bet_2");
4185        assert!(seen.contains("bet_1"));
4186        assert!(seen.contains("bet_2"));
4187    }
4188
4189    #[rstest]
4190    fn test_extend_unique_skips_already_seen() {
4191        let mut candidates: Vec<CurrentOrderSummary> = vec![make_summary(
4192            "bet_1",
4193            "1.100",
4194            12345,
4195            Decimal::ZERO,
4196            BetfairOrderStatus::Executable,
4197            "2026-04-18T10:00:00Z",
4198        )];
4199        let mut seen: AHashSet<String> = AHashSet::new();
4200        seen.insert("bet_1".to_string());
4201
4202        let orders = vec![make_summary(
4203            "bet_1",
4204            "1.100",
4205            12345,
4206            Decimal::ZERO,
4207            BetfairOrderStatus::Executable,
4208            "2026-04-18T10:05:00Z",
4209        )];
4210
4211        extend_unique(&mut candidates, &mut seen, orders);
4212
4213        assert_eq!(candidates.len(), 1);
4214        assert_eq!(candidates[0].placed_date, "2026-04-18T10:00:00Z");
4215    }
4216
4217    #[rstest]
4218    fn test_list_current_orders_filter_bet_id_sets_only_bet_ids() {
4219        let params = list_current_orders_filter_bet_id("bet_abc".to_string());
4220
4221        assert_eq!(
4222            params.bet_ids.as_deref(),
4223            Some(&["bet_abc".to_string()][..])
4224        );
4225        assert!(params.customer_order_refs.is_none());
4226        assert!(params.market_ids.is_none());
4227        assert!(params.order_projection.is_none());
4228        assert!(params.customer_strategy_refs.is_none());
4229        assert!(params.date_range.is_none());
4230        assert!(params.order_by.is_none());
4231        assert!(params.sort_dir.is_none());
4232        assert!(params.from_record.is_none());
4233        assert!(params.record_count.is_none());
4234    }
4235
4236    #[rstest]
4237    fn test_list_current_orders_filter_ref_sets_only_customer_order_refs() {
4238        let params = list_current_orders_filter_ref("rfo_abc".to_string());
4239
4240        assert_eq!(
4241            params.customer_order_refs.as_deref(),
4242            Some(&["rfo_abc".to_string()][..])
4243        );
4244        assert!(params.bet_ids.is_none());
4245        assert!(params.market_ids.is_none());
4246        assert!(params.order_projection.is_none());
4247        assert!(params.customer_strategy_refs.is_none());
4248        assert!(params.date_range.is_none());
4249        assert!(params.order_by.is_none());
4250        assert!(params.sort_dir.is_none());
4251        assert!(params.from_record.is_none());
4252        assert!(params.record_count.is_none());
4253    }
4254}
nautilus_betfair/execution.rs

nautilus_betfair/
execution.rs
