rustrade_execution/client/ibkr/

execution.rs

use crate::{
    order::id::{ClientOrderId, StrategyId},
    trade::{AssetFees, Trade, TradeId},
};
use chrono::{DateTime, NaiveDateTime, TimeZone, Utc};
use chrono_tz::Tz;
use fnv::FnvHashMap;
use ibapi::orders::{CommissionReport, ExecutionData, ExecutionSide};
use parking_lot::Mutex;
use rust_decimal::Decimal;
use rustrade_instrument::{
    Side, asset::name::AssetNameExchange, instrument::name::InstrumentNameExchange,
};
use smol_str::SmolStr;
use std::{cell::RefCell, sync::Arc};
use tracing::warn;

// Thread-local cache for parsed IANA timezones. IB uses per-exchange timezones,
// but most portfolios only see a handful (US/Eastern, Europe/London, etc.).
//
// Note: This cache is per-thread. In the dedicated `ibkr-order-stream` thread
// spawned by `account_stream`, the cache is fully effective. In `spawn_blocking`
// contexts (e.g., `fetch_trades`), Tokio's thread pool means each pool thread
// maintains its own cache — still beneficial but less effective than a single
// dedicated thread.
thread_local! {
    static TZ_CACHE: RefCell<FnvHashMap<SmolStr, Tz>> = RefCell::new(FnvHashMap::default());
}

/// Buffers IB executions until their commission reports arrive.
///
/// IB sends `ExecutionData` and `CommissionReport` as separate events.
/// This buffer holds executions until we can match them with commissions
/// to produce complete `Trade` events.
#[derive(Debug, Clone)]
pub struct ExecutionBuffer {
    inner: Arc<Mutex<ExecutionBufferInner>>,
}

#[derive(Debug, Default)]
struct ExecutionBufferInner {
    pending: FnvHashMap<String, PendingExecution>,
}

#[derive(Debug, Clone)]
struct PendingExecution {
    execution: ExecutionData,
    instrument: InstrumentNameExchange,
    client_order_id: ClientOrderId,
}

impl ExecutionBuffer {
    pub fn new() -> Self {
        Self {
            inner: Arc::new(Mutex::new(ExecutionBufferInner::default())),
        }
    }

    /// Buffer an execution, waiting for its commission report.
    pub fn add_execution(
        &self,
        execution: ExecutionData,
        instrument: InstrumentNameExchange,
        client_order_id: ClientOrderId,
    ) {
        let exec_id = execution.execution.execution_id.clone();
        let mut inner = self.inner.lock();
        inner.pending.insert(
            exec_id,
            PendingExecution {
                execution,
                instrument,
                client_order_id,
            },
        );

        // Warn if buffer is growing unexpectedly large (possible commission report leak)
        let pending_count = inner.pending.len();
        if pending_count > 1000 && pending_count.is_multiple_of(100) {
            warn!(
                pending_count,
                "ExecutionBuffer has >1000 pending entries; commission reports may be delayed or lost"
            );
        }
    }

    /// Try to complete a trade with a commission report.
    /// Returns the completed Trade if the matching execution was buffered.
    pub fn complete_with_commission(
        &self,
        report: &CommissionReport,
    ) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
        let pending = {
            let mut inner = self.inner.lock();
            inner.pending.remove(&report.execution_id)?
        };

        Some(build_trade(pending, report))
    }

    /// Move every pending execution into `target`, returning how many moved.
    ///
    /// Fill recovery buffers what it reads in a buffer of its own, then hands over whatever is
    /// still waiting for a commission report, so a report that reaches the account stream later
    /// can complete it there.
    ///
    /// Both buffers stay locked for the move, so a commission report the stream reads meanwhile
    /// finds the execution in one of them. `target` is locked first. The stream locks only its own
    /// buffer, so the order cannot deadlock against it.
    pub(super) fn drain_into(&self, target: &ExecutionBuffer) -> usize {
        let mut target = target.inner.lock();
        let drained = std::mem::take(&mut self.inner.lock().pending);
        let count = drained.len();
        target.pending.extend(drained);
        count
    }

    /// Get number of pending executions (for diagnostics).
    pub fn pending_count(&self) -> usize {
        self.inner.lock().pending.len()
    }

    /// Clear stale executions older than the given duration.
    ///
    /// Returns number of cleared entries.
    ///
    /// # Caller Responsibility
    ///
    /// This method is not called automatically. Callers should invoke it
    /// periodically to prevent unbounded growth if commission reports are
    /// delayed or lost.
    pub fn clear_stale(&self, max_age: std::time::Duration) -> usize {
        let now = Utc::now();
        let mut inner = self.inner.lock();
        let before = inner.pending.len();

        let max_age_secs = i64::try_from(max_age.as_secs()).unwrap_or(i64::MAX);
        inner.pending.retain(|_, pending| {
            if let Some(exec_time) = parse_ib_timestamp(&pending.execution.execution.time) {
                let age = now.signed_duration_since(exec_time);
                age.num_seconds() < max_age_secs
            } else {
                // Evict entries with unparseable timestamps to prevent memory leak
                false
            }
        });

        before - inner.pending.len()
    }
}

impl Default for ExecutionBuffer {
    fn default() -> Self {
        Self::new()
    }
}

/// Build a rustrade Trade from IB execution + commission data.
fn build_trade(
    pending: PendingExecution,
    commission: &CommissionReport,
) -> Trade<AssetNameExchange, InstrumentNameExchange> {
    let exec = &pending.execution.execution;

    // `ExecutionSide` is a closed two-variant enum in ibapi 3.x (the decoder
    // rejects unknown wire values upstream), so the mapping is total.
    let side = match exec.side {
        ExecutionSide::Bought => Side::Buy,
        ExecutionSide::Sold => Side::Sell,
    };

    let price = parse_decimal_or_warn(exec.price, "exec.price");
    let quantity = parse_decimal_or_warn(exec.shares, "exec.shares");
    let commission_amount = parse_decimal_or_warn(commission.commission, "commission");

    let time_exchange = parse_ib_timestamp(&exec.time).unwrap_or_else(Utc::now);

    Trade {
        id: TradeId::new(&exec.execution_id),
        order_id: crate::order::id::OrderId::new(&pending.client_order_id.0),
        instrument: pending.instrument,
        strategy: StrategyId::unknown(),
        time_exchange,
        side,
        price,
        quantity,
        // IB reports the order's running total on every execution, so the order advances from the
        // fill itself rather than waiting on the next `OrderStatus`.
        order_filled_quantity: Some(parse_decimal_or_warn(
            exec.cumulative_quantity,
            "exec.cumulative_quantity",
        )),
        fees: AssetFees {
            asset: AssetNameExchange::from(commission.currency.as_str()),
            fees: commission_amount,
            fees_quote: None, // Indexer computes based on fee asset vs instrument quote
        },
    }
}

/// Convert f64 to Decimal, logging a warning if conversion fails (NaN/Inf).
///
/// Returns `Decimal::ZERO` for invalid values. This is acceptable because:
/// - IB's API should never return NaN/Inf for prices, quantities, or commissions
/// - If it does, something is fundamentally broken and the warning log surfaces it
/// - Callers processing trades in bulk shouldn't abort on one corrupted record
///
/// For stricter handling, callers can check for zero in critical fields.
pub fn parse_decimal_or_warn(value: f64, field_name: &str) -> Decimal {
    Decimal::try_from(value).unwrap_or_else(|e| {
        warn!(field = %field_name, value = %value, error = %e, "Invalid f64 for Decimal, using zero");
        Decimal::ZERO
    })
}

/// Parse IB timestamp format (YYYYMMDD HH:MM:SS timezone).
///
/// IB sends timestamps like "20250418 10:30:00 US/Eastern". This function
/// parses the timezone and converts to UTC.
///
/// # Fallback Behavior
///
/// - Unknown timezone string: treats as UTC (logs warning)
/// - DST-ambiguous time (during "fall back" transition): returns `None`, caller
///   typically falls back to `Utc::now()`. This affects ~1 second per timezone
///   per year and is unlikely to occur in practice.
pub fn parse_ib_timestamp(s: &str) -> Option<DateTime<Utc>> {
    // Use iterator to avoid Vec allocation
    let mut parts = s.split_whitespace();
    let date_part = parts.next()?;
    let time_part = parts.next()?;
    let tz_part = parts.next();

    // Find the space between date and time by byte offset to avoid format! allocation
    let datetime_end = date_part.len() + 1 + time_part.len();
    let datetime_str = &s[..datetime_end.min(s.len())];

    let naive = NaiveDateTime::parse_from_str(datetime_str, "%Y%m%d %H:%M:%S").ok()?;

    // Try to parse timezone, fall back to UTC
    if let Some(tz_str) = tz_part {
        // Use cached timezone to avoid re-parsing IANA database on every call
        let tz_opt = TZ_CACHE.with(|cache| {
            let mut cache = cache.borrow_mut();
            if let Some(tz) = cache.get(tz_str) {
                return Some(*tz);
            }
            if let Ok(tz) = tz_str.parse::<Tz>() {
                cache.insert(SmolStr::new(tz_str), tz);
                return Some(tz);
            }
            None
        });

        if let Some(tz) = tz_opt {
            return tz
                .from_local_datetime(&naive)
                .single()
                .map(|dt| dt.with_timezone(&Utc));
        }
        warn!(timezone = %tz_str, "Unknown timezone in IB timestamp, treating as UTC");
    }

    Some(naive.and_utc())
}

#[cfg(test)]
#[allow(clippy::unwrap_used)] // Test code: panics are the correct failure mode
mod tests {
    use super::*;
    use chrono::{Datelike, Timelike};

    #[test]
    fn test_parse_ib_timestamp_with_timezone() {
        // US/Eastern is UTC-4 during daylight saving time (April)
        let ts = parse_ib_timestamp("20250418 10:30:00 US/Eastern");
        assert!(ts.is_some());
        let dt = ts.unwrap();
        assert_eq!(dt.year(), 2025);
        assert_eq!(dt.month(), 4);
        assert_eq!(dt.day(), 18);
        // 10:30 Eastern = 14:30 UTC (EDT is UTC-4)
        assert_eq!(dt.hour(), 14);
        assert_eq!(dt.minute(), 30);
    }

    #[test]
    fn test_parse_ib_timestamp_no_timezone() {
        // Without timezone, treat as UTC
        let ts = parse_ib_timestamp("20250418 10:30:00");
        assert!(ts.is_some());
        let dt = ts.unwrap();
        assert_eq!(dt.hour(), 10);
    }

    #[test]
    fn drain_into_moves_everything_and_keeps_the_target() {
        let execution = |exec_id: &str| ExecutionData {
            execution: ibapi::orders::Execution {
                execution_id: exec_id.to_string(),
                ..Default::default()
            },
            ..Default::default()
        };
        let add = |buffer: &ExecutionBuffer, exec_id: &str| {
            buffer.add_execution(
                execution(exec_id),
                InstrumentNameExchange::new("AAPL"),
                ClientOrderId::new("cid"),
            );
        };
        let source = ExecutionBuffer::new();
        let target = ExecutionBuffer::new();
        add(&source, "a");
        add(&source, "b");
        add(&target, "c");

        assert_eq!(source.drain_into(&target), 2);
        assert_eq!(source.pending_count(), 0);
        assert_eq!(target.pending_count(), 3);
    }

    #[test]
    fn test_parse_ib_timestamp_invalid() {
        assert!(parse_ib_timestamp("invalid").is_none());
        assert!(parse_ib_timestamp("").is_none());
    }
}