1use crate::{
2 order::id::{ClientOrderId, OrderId, StrategyId},
3 trade::{AssetFees, Trade, TradeId, TradeOrigin},
4};
5use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
6use chrono_tz::Tz;
7use fnv::FnvHashMap;
8use ibapi::orders::{CommissionReport, Execution, ExecutionData, ExecutionFilter, ExecutionSide};
9use parking_lot::Mutex;
10use rust_decimal::Decimal;
11use rustrade_instrument::{
12 Side, asset::name::AssetNameExchange, instrument::name::InstrumentNameExchange,
13};
14use smol_str::{SmolStr, format_smolstr};
15use std::{cell::RefCell, collections::hash_map::Entry, sync::Arc, time::Duration};
16use tracing::{debug, warn};
17
18thread_local! {
27 static TZ_CACHE: RefCell<FnvHashMap<SmolStr, Tz>> = RefCell::new(FnvHashMap::default());
28}
29
30#[derive(Debug, Clone)]
36pub struct ExecutionBuffer {
37 inner: Arc<Mutex<ExecutionBufferInner>>,
38}
39
40#[derive(Debug, Default)]
41struct ExecutionBufferInner {
42 pending: FnvHashMap<String, PendingExecution>,
43}
44
45#[derive(Debug, Clone)]
46struct PendingExecution {
47 execution: ExecutionData,
48 resolved: ResolvedExecution,
49}
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
53pub(super) struct ExecutionScope {
54 pub(super) api_client_id: i32,
56 pub(super) other_clients: bool,
59}
60
61impl ExecutionScope {
62 pub(super) fn classify(self, execution: &Execution) -> Option<(OrderId, TradeOrigin)> {
70 let own = execution.client_id == self.api_client_id;
71 let liquidation = execution.liquidation != 0;
72 if !own && !liquidation && !self.other_clients {
73 return None;
74 }
75 let origin = if liquidation {
76 TradeOrigin::Liquidation
77 } else if own {
78 TradeOrigin::Order
79 } else {
80 TradeOrigin::External
81 };
82 let order_id = if own {
83 ib_order_id(execution.order_id)
84 } else {
85 foreign_order_id(execution)
86 };
87 Some((order_id, origin))
88 }
89}
90
91#[derive(Debug, Clone, PartialEq, Eq)]
93pub(super) struct ResolvedExecution {
94 pub(super) instrument: InstrumentNameExchange,
95 pub(super) order_id: OrderId,
96 pub(super) origin: TradeOrigin,
97}
98
99impl ExecutionBuffer {
100 pub fn new() -> Self {
101 Self {
102 inner: Arc::new(Mutex::new(ExecutionBufferInner::default())),
103 }
104 }
105
106 pub(super) fn add_execution(&self, execution: ExecutionData, resolved: ResolvedExecution) {
108 let exec_id = execution.execution.execution_id.clone();
109 let mut inner = self.inner.lock();
110 inner.pending.insert(
111 exec_id,
112 PendingExecution {
113 execution,
114 resolved,
115 },
116 );
117
118 let pending_count = inner.pending.len();
120 if pending_count > 1000 && pending_count.is_multiple_of(100) {
121 warn!(
122 pending_count,
123 "ExecutionBuffer has >1000 pending entries; commission reports may be delayed or lost"
124 );
125 }
126 }
127
128 pub fn complete_with_commission(
131 &self,
132 report: &CommissionReport,
133 ) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
134 let pending = {
135 let mut inner = self.inner.lock();
136 inner.pending.remove(&report.execution_id)?
137 };
138
139 Some(build_trade(pending, report))
140 }
141
142 pub(super) fn take_without_commission(
148 &self,
149 ) -> Vec<Trade<AssetNameExchange, InstrumentNameExchange>> {
150 std::mem::take(&mut self.inner.lock().pending)
151 .into_values()
152 .map(|pending| {
153 let fees = AssetFees::new(
154 AssetNameExchange::from(UNKNOWN_FEE_ASSET),
155 Decimal::ZERO,
156 None,
157 );
158 trade_with_fees(pending, fees)
159 })
160 .collect()
161 }
162
163 pub(super) fn drain_into(&self, target: &ExecutionBuffer) -> usize {
173 let mut target = target.inner.lock();
174 let drained = std::mem::take(&mut self.inner.lock().pending);
175 let count = drained.len();
176 target.pending.extend(drained);
177 count
178 }
179
180 pub fn pending_count(&self) -> usize {
182 self.inner.lock().pending.len()
183 }
184
185 pub fn clear_stale(&self, max_age: std::time::Duration) -> usize {
195 let now = Utc::now();
196 let mut inner = self.inner.lock();
197 let before = inner.pending.len();
198
199 let max_age_secs = i64::try_from(max_age.as_secs()).unwrap_or(i64::MAX);
200 inner.pending.retain(|_, pending| {
201 if let Some(exec_time) = parse_ib_timestamp(&pending.execution.execution.time) {
202 let age = now.signed_duration_since(exec_time);
203 age.num_seconds() < max_age_secs
204 } else {
205 false
207 }
208 });
209
210 before - inner.pending.len()
211 }
212}
213
214impl Default for ExecutionBuffer {
215 fn default() -> Self {
216 Self::new()
217 }
218}
219
220#[derive(Debug, Clone, Copy, PartialEq, Eq)]
227pub(super) struct ExecutionRevision<'a> {
228 pub(super) execution: &'a str,
230 digits: &'a str,
233 pub(super) revision: u32,
234}
235
236impl<'a> ExecutionRevision<'a> {
237 pub(super) fn parse(execution_id: &'a str) -> Option<Self> {
240 let (execution, digits) = execution_id.rsplit_once('.')?;
241 if digits.is_empty() || !digits.bytes().all(|b| b.is_ascii_digit()) {
242 return None;
243 }
244 Some(Self {
245 execution,
246 digits,
247 revision: digits.parse().ok()?,
248 })
249 }
250
251 pub(super) fn is_correction(&self) -> bool {
253 self.revision > 1
254 }
255
256 pub(super) fn previous_id(&self) -> Option<TradeId> {
258 self.is_correction().then(|| {
259 TradeId(format_smolstr!(
260 "{}.{:0width$}",
261 self.execution,
262 self.revision - 1,
263 width = self.digits.len()
264 ))
265 })
266 }
267}
268
269pub(super) fn revision_of(id: &TradeId) -> u32 {
272 ExecutionRevision::parse(&id.0).map_or(0, |revision| revision.revision)
273}
274
275pub(super) fn keep_latest_revisions(
279 trades: Vec<Trade<AssetNameExchange, InstrumentNameExchange>>,
280) -> Vec<Trade<AssetNameExchange, InstrumentNameExchange>> {
281 let mut latest = FnvHashMap::<SmolStr, (usize, u32)>::with_capacity_and_hasher(
283 trades.len(),
284 Default::default(),
285 );
286 let mut kept = Vec::with_capacity(trades.len());
287 for trade in trades {
288 let Some(revision) = ExecutionRevision::parse(&trade.id.0) else {
289 kept.push(trade);
290 continue;
291 };
292 let (execution, revision) = (SmolStr::new(revision.execution), revision.revision);
293 match latest.entry(execution) {
294 Entry::Vacant(entry) => {
295 entry.insert((kept.len(), revision));
296 kept.push(trade);
297 }
298 Entry::Occupied(mut entry) => {
299 let (position, kept_revision) = *entry.get();
300 if revision == kept_revision {
301 debug!(exec_id = %trade.id, "IBKR execution read twice; keeping one");
302 continue;
303 }
304 let (older, newer) = if revision > kept_revision {
305 entry.insert((position, revision));
306 (
307 std::mem::replace(&mut kept[position], trade),
308 &kept[position],
309 )
310 } else {
311 (trade, &kept[position])
312 };
313 debug!(
314 corrected = %older.id,
315 correction = %newer.id,
316 "IBKR execution corrected; keeping the correction"
317 );
318 }
319 }
320 }
321 kept
322}
323
324pub const UNKNOWN_FEE_ASSET: &str = "UNKNOWN";
329
330pub(super) fn ib_order_id(ib_order_id: i32) -> OrderId {
333 OrderId::new(format_smolstr!("{ib_order_id}"))
334}
335
336pub(super) fn foreign_order_id(execution: &Execution) -> OrderId {
343 if execution.perm_id != 0 {
344 OrderId::new(format_smolstr!("perm:{}", execution.perm_id))
345 } else {
346 OrderId::new(format_smolstr!(
347 "client:{}:{}",
348 execution.client_id,
349 execution.order_id
350 ))
351 }
352}
353
354fn build_trade(
356 pending: PendingExecution,
357 commission: &CommissionReport,
358) -> Trade<AssetNameExchange, InstrumentNameExchange> {
359 let commission_amount = parse_decimal_or_warn(commission.commission, "commission");
360 let fees = AssetFees {
361 asset: AssetNameExchange::from(commission.currency.as_str()),
362 fees: commission_amount,
363 fees_quote: None, };
365 trade_with_fees(pending, fees)
366}
367
368fn trade_with_fees(
370 pending: PendingExecution,
371 fees: AssetFees<AssetNameExchange>,
372) -> Trade<AssetNameExchange, InstrumentNameExchange> {
373 let exec = &pending.execution.execution;
374
375 let side = match exec.side {
378 ExecutionSide::Bought => Side::Buy,
379 ExecutionSide::Sold => Side::Sell,
380 };
381
382 let price = parse_decimal_or_warn(exec.price, "exec.price");
383 let quantity = parse_decimal_or_warn(exec.shares, "exec.shares");
384
385 let time_exchange = parse_ib_timestamp(&exec.time).unwrap_or_else(Utc::now);
386
387 let ResolvedExecution {
388 instrument,
389 order_id,
390 origin,
391 } = pending.resolved;
392 Trade {
393 id: TradeId::new(&exec.execution_id),
394 order_id,
397 instrument,
398 strategy: StrategyId::unknown(),
399 time_exchange,
400 side,
401 price,
402 quantity,
403 order_filled_quantity: Some(parse_decimal_or_warn(
406 exec.cumulative_quantity,
407 "exec.cumulative_quantity",
408 )),
409 fees,
410 origin,
411 position_side: None,
412 client_order_id: (!exec.order_reference.is_empty())
415 .then(|| ClientOrderId::new(exec.order_reference.as_str())),
416 }
417}
418
419pub fn parse_decimal_or_warn(value: f64, field_name: impl std::fmt::Display) -> Decimal {
429 try_decimal_or_warn(value, field_name).unwrap_or(Decimal::ZERO)
430}
431
432pub fn try_decimal_or_warn(value: f64, field_name: impl std::fmt::Display) -> Option<Decimal> {
438 Decimal::try_from(value)
439 .map_err(
440 |e| warn!(field = %field_name, value = %value, error = %e, "Invalid f64 for Decimal"),
441 )
442 .ok()
443}
444
445pub fn parse_ib_timestamp(s: &str) -> Option<DateTime<Utc>> {
457 let mut parts = s.split_whitespace();
459 let date_part = parts.next()?;
460 let time_part = parts.next()?;
461 let tz_part = parts.next();
462
463 let datetime_end = date_part.len() + 1 + time_part.len();
465 let datetime_str = &s[..datetime_end.min(s.len())];
466
467 let naive = NaiveDateTime::parse_from_str(datetime_str, "%Y%m%d %H:%M:%S").ok()?;
468
469 if let Some(tz_str) = tz_part {
471 let tz_opt = TZ_CACHE.with(|cache| {
473 let mut cache = cache.borrow_mut();
474 if let Some(tz) = cache.get(tz_str) {
475 return Some(*tz);
476 }
477 if let Ok(tz) = tz_str.parse::<Tz>() {
478 cache.insert(SmolStr::new(tz_str), tz);
479 return Some(tz);
480 }
481 None
482 });
483
484 if let Some(tz) = tz_opt {
485 return tz
486 .from_local_datetime(&naive)
487 .single()
488 .map(|dt| dt.with_timezone(&Utc));
489 }
490 warn!(timezone = %tz_str, "Unknown timezone in IB timestamp, treating as UTC");
491 }
492
493 Some(naive.and_utc())
494}
495
496pub(super) const EXECUTION_DAYS: i32 = 7;
499
500pub(super) const EXECUTIONS_COVER: Duration = Duration::from_secs(5 * 24 * 60 * 60);
506
507pub(super) fn executions_filter() -> ExecutionFilter {
509 ExecutionFilter {
510 last_n_days: EXECUTION_DAYS,
511 ..ExecutionFilter::default()
512 }
513}
514
515pub(super) fn executions_held_since(
519 start: DateTime<Utc>,
520 now: DateTime<Utc>,
521) -> Option<DateTime<Utc>> {
522 let covered_from = now - EXECUTIONS_COVER;
523 (start < covered_from).then_some(covered_from)
524}
525
526#[cfg(test)]
527#[allow(clippy::unwrap_used)] mod tests {
529 use super::*;
530 use chrono::{Datelike, Timelike};
531
532 #[test]
533 fn test_parse_ib_timestamp_with_timezone() {
534 let ts = parse_ib_timestamp("20250418 10:30:00 US/Eastern");
536 assert!(ts.is_some());
537 let dt = ts.unwrap();
538 assert_eq!(dt.year(), 2025);
539 assert_eq!(dt.month(), 4);
540 assert_eq!(dt.day(), 18);
541 assert_eq!(dt.hour(), 14);
543 assert_eq!(dt.minute(), 30);
544 }
545
546 #[test]
547 fn test_parse_ib_timestamp_no_timezone() {
548 let ts = parse_ib_timestamp("20250418 10:30:00");
550 assert!(ts.is_some());
551 let dt = ts.unwrap();
552 assert_eq!(dt.hour(), 10);
553 }
554
555 #[test]
558 fn a_fill_carries_the_order_reference_as_its_client_order_id() {
559 let buffer = ExecutionBuffer::new();
560 for (exec_id, reference) in [("a", "outside-1"), ("b", "")] {
561 buffer.add_execution(
562 ExecutionData {
563 execution: ibapi::orders::Execution {
564 execution_id: exec_id.to_string(),
565 order_reference: reference.to_string(),
566 ..Default::default()
567 },
568 ..Default::default()
569 },
570 ResolvedExecution {
571 instrument: InstrumentNameExchange::new("AAPL"),
572 order_id: ib_order_id(1),
573 origin: TradeOrigin::Order,
574 },
575 );
576 }
577
578 let mut trades = buffer.take_without_commission();
579 trades.sort_by(|a, b| a.id.cmp(&b.id));
580 assert_eq!(
581 trades[0].client_order_id,
582 Some(ClientOrderId::new("outside-1"))
583 );
584 assert_eq!(trades[1].client_order_id, None);
585 }
586
587 #[test]
588 fn drain_into_moves_everything_and_keeps_the_target() {
589 let execution = |exec_id: &str| ExecutionData {
590 execution: ibapi::orders::Execution {
591 execution_id: exec_id.to_string(),
592 ..Default::default()
593 },
594 ..Default::default()
595 };
596 let add = |buffer: &ExecutionBuffer, exec_id: &str| {
597 buffer.add_execution(
598 execution(exec_id),
599 ResolvedExecution {
600 instrument: InstrumentNameExchange::new("AAPL"),
601 order_id: ib_order_id(1),
602 origin: TradeOrigin::Order,
603 },
604 );
605 };
606 let source = ExecutionBuffer::new();
607 let target = ExecutionBuffer::new();
608 add(&source, "a");
609 add(&source, "b");
610 add(&target, "c");
611
612 assert_eq!(source.drain_into(&target), 2);
613 assert_eq!(source.pending_count(), 0);
614 assert_eq!(target.pending_count(), 3);
615 }
616
617 #[test]
618 fn execution_revision_reads_the_digits_after_the_final_period() {
619 let original = ExecutionRevision::parse("0000e0d5.5f8b1c2a.01.01").unwrap();
620 assert_eq!(original.execution, "0000e0d5.5f8b1c2a.01");
621 assert_eq!(original.revision, 1);
622 assert!(!original.is_correction());
623 assert_eq!(original.previous_id(), None);
624
625 let correction = ExecutionRevision::parse("0000e0d5.5f8b1c2a.01.02").unwrap();
626 assert_eq!(correction.execution, "0000e0d5.5f8b1c2a.01");
627 assert!(correction.is_correction());
628 assert_eq!(
629 correction.previous_id(),
630 Some(TradeId::new("0000e0d5.5f8b1c2a.01.01"))
631 );
632
633 assert_eq!(
635 ExecutionRevision::parse("x.10").unwrap().previous_id(),
636 Some(TradeId::new("x.09"))
637 );
638 assert_eq!(
639 ExecutionRevision::parse("x.3").unwrap().previous_id(),
640 Some(TradeId::new("x.2"))
641 );
642 }
643
644 #[test]
645 fn execution_id_without_a_revision_is_never_a_correction() {
646 for id in ["e1", "abc.", "abc.0x", "abc.1a", "", ".", "abc.99999999999"] {
647 assert_eq!(ExecutionRevision::parse(id), None, "{id:?}");
648 }
649 assert_eq!(revision_of(&TradeId::new("e1")), 0);
650 assert_eq!(revision_of(&TradeId::new("a.b.02")), 2);
651 }
652
653 fn trade(id: &str, price: i64) -> Trade<AssetNameExchange, InstrumentNameExchange> {
654 Trade {
655 id: TradeId::new(id),
656 order_id: crate::order::id::OrderId::new("cid"),
657 instrument: InstrumentNameExchange::new("AAPL"),
658 strategy: StrategyId::unknown(),
659 time_exchange: DateTime::<Utc>::MIN_UTC,
660 side: Side::Buy,
661 price: Decimal::from(price),
662 quantity: Decimal::ONE,
663 order_filled_quantity: Some(Decimal::ONE),
664 fees: AssetFees::new(AssetNameExchange::from("USD"), Decimal::ZERO, None),
665 origin: TradeOrigin::Order,
666 position_side: None,
667 client_order_id: None,
668 }
669 }
670
671 #[test]
672 fn keep_latest_revisions_reads_a_corrected_execution_once() {
673 let kept = keep_latest_revisions(vec![
674 trade("a.01.01", 100),
675 trade("b.01.01", 200),
676 trade("a.01.02", 101),
677 trade("a.01.02", 101),
678 trade("plain", 300),
679 trade("b.01.03", 202),
681 trade("b.01.02", 201),
682 ]);
683 let read: Vec<_> = kept
684 .iter()
685 .map(|trade| (trade.id.0.as_str(), trade.price))
686 .collect();
687 assert_eq!(
688 read,
689 [
690 ("a.01.02", Decimal::from(101)),
691 ("b.01.03", Decimal::from(202)),
692 ("plain", Decimal::from(300)),
693 ]
694 );
695 }
696
697 #[test]
698 fn test_parse_ib_timestamp_invalid() {
699 assert!(parse_ib_timestamp("invalid").is_none());
700 assert!(parse_ib_timestamp("").is_none());
701 }
702
703 #[test]
704 fn executions_held_since_is_the_cover_start_only_when_the_span_starts_before_it() {
705 let now = Utc.with_ymd_and_hms(2026, 10, 23, 9, 0, 0).unwrap();
706 let covered_from = Utc.with_ymd_and_hms(2026, 10, 18, 9, 0, 0).unwrap();
707 let before = covered_from - chrono::TimeDelta::seconds(1);
708 assert_eq!(executions_held_since(before, now), Some(covered_from));
709 assert_eq!(executions_held_since(covered_from, now), None);
710 assert_eq!(executions_held_since(now, now), None);
711 }
712
713 #[test]
714 fn executions_filter_reads_every_client_over_the_execution_days() {
715 let filter = executions_filter();
716 assert_eq!(filter.last_n_days, 7);
717 assert_eq!(filter.client_id, None);
718 assert!(filter.specific_dates.is_empty());
719 }
720}