rustrade_execution/client/ibkr/
execution.rs

1use crate::{
2    order::id::{ClientOrderId, StrategyId},
3    trade::{AssetFees, Trade, TradeId},
4};
5use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
6use chrono_tz::Tz;
7use fnv::FnvHashMap;
8use ibapi::orders::{CommissionReport, ExecutionData, 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;
15use std::{cell::RefCell, sync::Arc};
16use tracing::warn;
17
18// Thread-local cache for parsed IANA timezones. IB uses per-exchange timezones,
19// but most portfolios only see a handful (US/Eastern, Europe/London, etc.).
20//
21// Note: This cache is per-thread. In the dedicated `ibkr-order-stream` thread
22// spawned by `account_stream`, the cache is fully effective. In `spawn_blocking`
23// contexts (e.g., `fetch_trades`), Tokio's thread pool means each pool thread
24// maintains its own cache — still beneficial but less effective than a single
25// dedicated thread.
26thread_local! {
27    static TZ_CACHE: RefCell<FnvHashMap<SmolStr, Tz>> = RefCell::new(FnvHashMap::default());
28}
29
30/// Buffers IB executions until their commission reports arrive.
31///
32/// IB sends `ExecutionData` and `CommissionReport` as separate events.
33/// This buffer holds executions until we can match them with commissions
34/// to produce complete `Trade` events.
35#[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    instrument: InstrumentNameExchange,
49    client_order_id: ClientOrderId,
50}
51
52impl ExecutionBuffer {
53    pub fn new() -> Self {
54        Self {
55            inner: Arc::new(Mutex::new(ExecutionBufferInner::default())),
56        }
57    }
58
59    /// Buffer an execution, waiting for its commission report.
60    pub fn add_execution(
61        &self,
62        execution: ExecutionData,
63        instrument: InstrumentNameExchange,
64        client_order_id: ClientOrderId,
65    ) {
66        let exec_id = execution.execution.execution_id.clone();
67        let mut inner = self.inner.lock();
68        inner.pending.insert(
69            exec_id,
70            PendingExecution {
71                execution,
72                instrument,
73                client_order_id,
74            },
75        );
76
77        // Warn if buffer is growing unexpectedly large (possible commission report leak)
78        let pending_count = inner.pending.len();
79        if pending_count > 1000 && pending_count.is_multiple_of(100) {
80            warn!(
81                pending_count,
82                "ExecutionBuffer has >1000 pending entries; commission reports may be delayed or lost"
83            );
84        }
85    }
86
87    /// Try to complete a trade with a commission report.
88    /// Returns the completed Trade if the matching execution was buffered.
89    pub fn complete_with_commission(
90        &self,
91        report: &CommissionReport,
92    ) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
93        let pending = {
94            let mut inner = self.inner.lock();
95            inner.pending.remove(&report.execution_id)?
96        };
97
98        Some(build_trade(pending, report))
99    }
100
101    /// Get number of pending executions (for diagnostics).
102    pub fn pending_count(&self) -> usize {
103        self.inner.lock().pending.len()
104    }
105
106    /// Clear stale executions older than the given duration.
107    ///
108    /// Returns number of cleared entries.
109    ///
110    /// # Caller Responsibility
111    ///
112    /// This method is not called automatically. Callers should invoke it
113    /// periodically to prevent unbounded growth if commission reports are
114    /// delayed or lost.
115    pub fn clear_stale(&self, max_age: std::time::Duration) -> usize {
116        let now = Utc::now();
117        let mut inner = self.inner.lock();
118        let before = inner.pending.len();
119
120        let max_age_secs = i64::try_from(max_age.as_secs()).unwrap_or(i64::MAX);
121        inner.pending.retain(|_, pending| {
122            if let Some(exec_time) = parse_ib_timestamp(&pending.execution.execution.time) {
123                let age = now.signed_duration_since(exec_time);
124                age.num_seconds() < max_age_secs
125            } else {
126                // Evict entries with unparseable timestamps to prevent memory leak
127                false
128            }
129        });
130
131        before - inner.pending.len()
132    }
133}
134
135impl Default for ExecutionBuffer {
136    fn default() -> Self {
137        Self::new()
138    }
139}
140
141/// Build a rustrade Trade from IB execution + commission data.
142fn build_trade(
143    pending: PendingExecution,
144    commission: &CommissionReport,
145) -> Trade<AssetNameExchange, InstrumentNameExchange> {
146    let exec = &pending.execution.execution;
147
148    // `ExecutionSide` is a closed two-variant enum in ibapi 3.x (the decoder
149    // rejects unknown wire values upstream), so the mapping is total.
150    let side = match exec.side {
151        ExecutionSide::Bought => Side::Buy,
152        ExecutionSide::Sold => Side::Sell,
153    };
154
155    let price = parse_decimal_or_warn(exec.price, "exec.price");
156    let quantity = parse_decimal_or_warn(exec.shares, "exec.shares");
157    let commission_amount = parse_decimal_or_warn(commission.commission, "commission");
158
159    let time_exchange = parse_ib_timestamp(&exec.time).unwrap_or_else(Utc::now);
160
161    Trade {
162        id: TradeId::new(&exec.execution_id),
163        order_id: crate::order::id::OrderId::new(&pending.client_order_id.0),
164        instrument: pending.instrument,
165        strategy: StrategyId::unknown(),
166        time_exchange,
167        side,
168        price,
169        quantity,
170        fees: AssetFees {
171            asset: AssetNameExchange::from(commission.currency.as_str()),
172            fees: commission_amount,
173            fees_quote: None, // Indexer computes based on fee asset vs instrument quote
174        },
175    }
176}
177
178/// Convert f64 to Decimal, logging a warning if conversion fails (NaN/Inf).
179///
180/// Returns `Decimal::ZERO` for invalid values. This is acceptable because:
181/// - IB's API should never return NaN/Inf for prices, quantities, or commissions
182/// - If it does, something is fundamentally broken and the warning log surfaces it
183/// - Callers processing trades in bulk shouldn't abort on one corrupted record
184///
185/// For stricter handling, callers can check for zero in critical fields.
186pub fn parse_decimal_or_warn(value: f64, field_name: &str) -> Decimal {
187    Decimal::try_from(value).unwrap_or_else(|e| {
188        warn!(field = %field_name, value = %value, error = %e, "Invalid f64 for Decimal, using zero");
189        Decimal::ZERO
190    })
191}
192
193/// Parse IB timestamp format (YYYYMMDD HH:MM:SS timezone).
194///
195/// IB sends timestamps like "20250418 10:30:00 US/Eastern". This function
196/// parses the timezone and converts to UTC.
197///
198/// # Fallback Behavior
199///
200/// - Unknown timezone string: treats as UTC (logs warning)
201/// - DST-ambiguous time (during "fall back" transition): returns `None`, caller
202///   typically falls back to `Utc::now()`. This affects ~1 second per timezone
203///   per year and is unlikely to occur in practice.
204pub fn parse_ib_timestamp(s: &str) -> Option<DateTime<Utc>> {
205    // Use iterator to avoid Vec allocation
206    let mut parts = s.split_whitespace();
207    let date_part = parts.next()?;
208    let time_part = parts.next()?;
209    let tz_part = parts.next();
210
211    // Find the space between date and time by byte offset to avoid format! allocation
212    let datetime_end = date_part.len() + 1 + time_part.len();
213    let datetime_str = &s[..datetime_end.min(s.len())];
214
215    let naive = NaiveDateTime::parse_from_str(datetime_str, "%Y%m%d %H:%M:%S").ok()?;
216
217    // Try to parse timezone, fall back to UTC
218    if let Some(tz_str) = tz_part {
219        // Use cached timezone to avoid re-parsing IANA database on every call
220        let tz_opt = TZ_CACHE.with(|cache| {
221            let mut cache = cache.borrow_mut();
222            if let Some(tz) = cache.get(tz_str) {
223                return Some(*tz);
224            }
225            if let Ok(tz) = tz_str.parse::<Tz>() {
226                cache.insert(SmolStr::new(tz_str), tz);
227                return Some(tz);
228            }
229            None
230        });
231
232        if let Some(tz) = tz_opt {
233            return tz
234                .from_local_datetime(&naive)
235                .single()
236                .map(|dt| dt.with_timezone(&Utc));
237        }
238        warn!(timezone = %tz_str, "Unknown timezone in IB timestamp, treating as UTC");
239    }
240
241    Some(naive.and_utc())
242}
243
244#[cfg(test)]
245#[allow(clippy::unwrap_used)] // Test code: panics are the correct failure mode
246mod tests {
247    use super::*;
248    use chrono::{Datelike, Timelike};
249
250    #[test]
251    fn test_parse_ib_timestamp_with_timezone() {
252        // US/Eastern is UTC-4 during daylight saving time (April)
253        let ts = parse_ib_timestamp("20250418 10:30:00 US/Eastern");
254        assert!(ts.is_some());
255        let dt = ts.unwrap();
256        assert_eq!(dt.year(), 2025);
257        assert_eq!(dt.month(), 4);
258        assert_eq!(dt.day(), 18);
259        // 10:30 Eastern = 14:30 UTC (EDT is UTC-4)
260        assert_eq!(dt.hour(), 14);
261        assert_eq!(dt.minute(), 30);
262    }
263
264    #[test]
265    fn test_parse_ib_timestamp_no_timezone() {
266        // Without timezone, treat as UTC
267        let ts = parse_ib_timestamp("20250418 10:30:00");
268        assert!(ts.is_some());
269        let dt = ts.unwrap();
270        assert_eq!(dt.hour(), 10);
271    }
272
273    #[test]
274    fn test_parse_ib_timestamp_invalid() {
275        assert!(parse_ib_timestamp("invalid").is_none());
276        assert!(parse_ib_timestamp("").is_none());
277    }
278}
rustrade_execution/client/ibkr/execution.rs

rustrade_execution/client/ibkr/
execution.rs
