ibapi/market_data/historical/
sync.rs

1use std::collections::VecDeque;
2use std::sync::atomic::{AtomicBool, Ordering};
3use std::sync::{Arc, Mutex};
4
5use log::error;
6use time::OffsetDateTime;
7
8use crate::client::blocking::ClientRequestBuilders;
9use crate::common::request_helpers::{self, empty_on_end_of_stream};
10use crate::common::retry::blocking::retry_on_connection_reset;
11use crate::contracts::Contract;
12use crate::protocol::{check_version, Features};
13use crate::subscriptions::common::{RoutedItem, SubscriptionItem};
14use crate::subscriptions::log_cancel_error;
15use crate::subscriptions::sync::{FilterData, Subscription, SubscriptionItemIterExt};
16use crate::transport::{InternalSubscription, MessageBus};
17use crate::{client::sync::Client, Error};
18
19use super::common::tick::{classify, TickAction};
20use super::common::{self, encoders};
21use super::{
22    BarSize, Duration, HeadTimestampRequest, HistogramEntry, HistogramRequest, HistoricalBarUpdate, HistoricalData, HistoricalDataFrame, Schedule,
23    ScheduleRequest, TickDecoder, WhatToShow,
24};
25use crate::market_data::TradingHours;
26
27impl Client {
28    /// Returns the timestamp of earliest available historical data for a contract and data type.
29    ///
30    /// # Examples
31    ///
32    /// ```no_run
33    /// use ibapi::client::blocking::Client;
34    /// use ibapi::contracts::Contract;
35    /// use ibapi::market_data::historical::{self, WhatToShow};
36    /// use ibapi::market_data::TradingHours;
37    ///
38    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
39    ///
40    /// let contract = Contract::stock("MSFT").build();
41    /// let what_to_show = WhatToShow::Trades;
42    /// let trading_hours = TradingHours::Regular;
43    ///
44    /// let result = client.head_timestamp(&contract, what_to_show, trading_hours).expect("head timestamp failed");
45    ///
46    /// print!("head_timestamp: {result:?}");
47    /// ```
48    pub fn head_timestamp(&self, contract: &Contract, what_to_show: WhatToShow, trading_hours: TradingHours) -> Result<OffsetDateTime, Error> {
49        request_helpers::blocking::one_shot_cancelling::<HeadTimestampRequest>(self, |request_id| {
50            encoders::encode_request_head_timestamp(request_id, contract, what_to_show, trading_hours.use_rth())
51        })
52    }
53
54    /// Build a request for historical bar data.
55    ///
56    /// Required: a date spec via either [`HistoricalDataBuilder::duration`](super::HistoricalDataBuilder::duration)
57    /// (with optional [`HistoricalDataBuilder::ending`](super::HistoricalDataBuilder::ending)) or
58    /// [`HistoricalDataBuilder::between`](super::HistoricalDataBuilder::between). Terminals:
59    /// [`HistoricalDataBuilder::fetch`](super::HistoricalDataBuilder::fetch) for a one-shot
60    /// [`HistoricalData`] result; [`HistoricalDataBuilder::stream`](super::HistoricalDataBuilder::stream)
61    /// for a `Subscription<HistoricalBarUpdate>` that yields bars as they arrive.
62    ///
63    /// # Arguments
64    /// * `contract` - Contract object that is subject of query
65    /// * `bar_size` - Bar size (resolution)
66    ///
67    /// # Examples
68    ///
69    /// ```no_run
70    /// use ibapi::client::blocking::Client;
71    /// use ibapi::contracts::Contract;
72    /// use ibapi::market_data::historical::{BarSize, ToDuration, WhatToShow};
73    /// use time::macros::datetime;
74    ///
75    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
76    /// let contract = Contract::stock("AAPL").build();
77    ///
78    /// // IBKR-native: amount of data ending at a specific time (or now if `.ending` is unset)
79    /// let bars = client
80    ///     .historical_data(&contract, BarSize::Hour)
81    ///     .what_to_show(WhatToShow::Trades)
82    ///     .duration(7.days())
83    ///     .fetch()
84    ///     .expect("historical data request failed");
85    ///
86    /// // Convenience: explicit date range; bars outside it are dropped
87    /// let bars = client
88    ///     .historical_data(&contract, BarSize::Hour)
89    ///     .between(datetime!(2023-04-08 0:00 UTC), datetime!(2023-04-15 0:00 UTC))
90    ///     .fetch()
91    ///     .expect("historical data request failed");
92    /// # let _ = bars;
93    /// ```
94    pub fn historical_data<'a>(&'a self, contract: &'a Contract, bar_size: BarSize) -> super::HistoricalDataBuilder<'a, Self> {
95        super::HistoricalDataBuilder::new(self, contract, bar_size)
96    }
97
98    /// Build a request for [`Schedule`] data over the given duration.
99    ///
100    /// Defaults to anchoring at the current time. Use [`HistoricalScheduleBuilder::ending`](super::HistoricalScheduleBuilder::ending)
101    /// to anchor at a specific end date.
102    ///
103    /// # Arguments
104    /// * `contract` - [Contract] to retrieve [Schedule] for.
105    /// * `duration` - [Duration] of the interval to retrieve.
106    ///
107    /// # Examples
108    ///
109    /// ```no_run
110    /// use time::macros::datetime;
111    /// use ibapi::client::blocking::Client;
112    /// use ibapi::contracts::Contract;
113    /// use ibapi::market_data::historical::ToDuration;
114    ///
115    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
116    /// let contract = Contract::stock("GM").build();
117    ///
118    /// // Ending now:
119    /// let schedule = client
120    ///     .historical_schedules(&contract, 30.days())
121    ///     .fetch()
122    ///     .expect("historical schedule request failed");
123    ///
124    /// // Anchored to a specific end date:
125    /// let schedule = client
126    ///     .historical_schedules(&contract, 30.days())
127    ///     .ending(datetime!(2023-04-15 0:00 UTC))
128    ///     .fetch()
129    ///     .expect("historical schedule request failed");
130    ///
131    /// for session in &schedule.sessions {
132    ///     println!("{session:?}");
133    /// }
134    /// ```
135    pub fn historical_schedules<'a>(&'a self, contract: &'a Contract, duration: Duration) -> super::HistoricalScheduleBuilder<'a, Self> {
136        super::HistoricalScheduleBuilder::new(self, contract, duration)
137    }
138
139    /// Build a request for historical time & sales data (tick-by-tick).
140    ///
141    /// The terminal method selects the tick type:
142    /// [`HistoricalTicksBuilder::trade`](super::HistoricalTicksBuilder::trade) /
143    /// `.mid_point()` / `.bid_ask(IgnoreSize)`. Use
144    /// [`HistoricalTicksBuilder::starting`](super::HistoricalTicksBuilder::starting) /
145    /// `.ending()` to anchor the query (at least one is required per IBKR).
146    ///
147    /// # Arguments
148    /// * `contract` - [Contract] object that is subject of query
149    /// * `number_of_ticks` - Number of distinct data points. Max currently 1000 per request.
150    ///
151    /// # Examples
152    ///
153    /// ```no_run
154    /// use ibapi::client::blocking::Client;
155    /// use ibapi::contracts::Contract;
156    /// use ibapi::market_data::IgnoreSize;
157    /// use ibapi::market_data::TradingHours;
158    /// use time::macros::datetime;
159    ///
160    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
161    /// let contract = Contract::stock("TSLA").build();
162    ///
163    /// // Trade ticks anchored at a start date:
164    /// let trades = client
165    ///     .historical_ticks(&contract, 100)
166    ///     .starting(datetime!(2023-04-15 0:00 UTC))
167    ///     .trading_hours(TradingHours::Regular)
168    ///     .trade()
169    ///     .expect("historical ticks request failed");
170    ///
171    /// // Bid/ask ticks anchored at an end date, ignoring tick sizes:
172    /// let quotes = client
173    ///     .historical_ticks(&contract, 100)
174    ///     .ending(datetime!(2023-04-15 0:00 UTC))
175    ///     .bid_ask(IgnoreSize::Yes)
176    ///     .expect("historical ticks request failed");
177    ///
178    /// for tick in trades {
179    ///     println!("{tick:?}");
180    /// }
181    /// # let _ = quotes;
182    /// ```
183    pub fn historical_ticks<'a>(&'a self, contract: &'a Contract, number_of_ticks: i32) -> super::HistoricalTicksBuilder<'a, Self> {
184        super::HistoricalTicksBuilder::new(self, contract, number_of_ticks)
185    }
186
187    /// Cancels an in-flight historical ticks request.
188    ///
189    /// # Arguments
190    /// * `request_id` - The request ID of the historical ticks subscription to cancel.
191    /// # Examples
192    ///
193    /// ```no_run
194    /// use ibapi::client::blocking::Client;
195    /// use ibapi::contracts::Contract;
196    ///
197    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
198    ///
199    /// let contract = Contract::stock("MSFT").build();
200    /// let subscription = client.historical_ticks(&contract, 100).trade().expect("request failed");
201    ///
202    /// // Holding the subscription, `subscription.cancel()` (or dropping it) does the same.
203    /// client.cancel_historical_ticks(subscription.request_id()).expect("cancel failed");
204    /// ```
205    pub fn cancel_historical_ticks(&self, request_id: i32) -> Result<(), Error> {
206        check_version(self.server_version(), Features::CANCEL_CONTRACT_DATA)?;
207
208        let message = encoders::encode_cancel_historical_ticks(request_id)?;
209        self.send_message(message)?;
210        Ok(())
211    }
212
213    /// Requests data histogram of specified contract.
214    ///
215    /// # Arguments
216    /// * `contract`  - [Contract] to retrieve [Histogram Entries](HistogramEntry) for.
217    /// * `trading_hours` - Regular trading hours only, or include extended hours.
218    /// * `period`    - The time period of each histogram bar (e.g., `BarSize::Day`, `BarSize::Week`, `BarSize::Month`).
219    ///
220    /// # Examples
221    ///
222    /// ```no_run
223    /// use time::macros::datetime;
224    ///
225    /// use ibapi::contracts::Contract;
226    /// use ibapi::client::blocking::Client;
227    /// use ibapi::market_data::historical::BarSize;
228    /// use ibapi::market_data::TradingHours;
229    ///
230    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
231    ///
232    /// let contract = Contract::stock("GM").build();
233    ///
234    /// let histogram = client
235    ///     .histogram_data(&contract, TradingHours::Regular, BarSize::Week)
236    ///     .expect("histogram request failed");
237    ///
238    /// for item in &histogram {
239    ///     println!("{item:?}");
240    /// }
241    /// ```
242    pub fn histogram_data(&self, contract: &Contract, trading_hours: TradingHours, period: BarSize) -> Result<Vec<HistogramEntry>, Error> {
243        request_helpers::blocking::one_shot_cancelling::<HistogramRequest>(self, |request_id| {
244            encoders::encode_request_histogram_data(request_id, contract, trading_hours.use_rth(), period)
245        })
246        .or_else(empty_on_end_of_stream)
247    }
248}
249
250pub(crate) fn historical_data(
251    client: &Client,
252    contract: &Contract,
253    end_date: Option<OffsetDateTime>,
254    duration: Duration,
255    bar_size: BarSize,
256    what_to_show: WhatToShow,
257    trading_hours: TradingHours,
258) -> Result<HistoricalData, Error> {
259    common::validate_historical_data(end_date, what_to_show)?;
260    // One deadline for both reads and every resend.
261    let deadline = request_helpers::blocking::one_shot_deadline(client);
262
263    retry_on_connection_reset(client, || {
264        request_helpers::blocking::check_deadline(deadline)?;
265        let builder = client.request();
266        let request = encoders::encode_request_historical_data(
267            builder.request_id(),
268            contract,
269            end_date,
270            duration,
271            bar_size,
272            what_to_show,
273            trading_hours.use_rth(),
274            false,
275        )?;
276
277        // The window frame is a second read, which is why this is not a
278        // one-shot and cannot use the helpers in `common::request_helpers`.
279        // Dropping the subscription before it, on any error below, sends
280        // TWS's cancel; after it, nothing is sent.
281        let subscription = builder.send::<HistoricalDataFrame>(request)?;
282
283        let mut data = request_helpers::blocking::next_frame(&subscription, deadline)?.into_bars()?;
284        // Only the window frame proves the download is complete, so a close
285        // before it is an error.
286        (data.start, data.end) = request_helpers::blocking::next_frame(&subscription, deadline)?.into_window()?;
287
288        Ok(data)
289    })
290}
291
292pub(crate) fn historical_data_stream(
293    client: &Client,
294    contract: &Contract,
295    duration: Duration,
296    bar_size: BarSize,
297    what_to_show: WhatToShow,
298    trading_hours: TradingHours,
299    buffer_limit: Option<usize>,
300) -> Result<Subscription<HistoricalBarUpdate>, Error> {
301    let builder = client.request();
302    let request = encoders::encode_request_historical_data(
303        builder.request_id(),
304        contract,
305        None, // IBKR requires end_date=None when keep_up_to_date=true
306        duration,
307        bar_size,
308        what_to_show,
309        trading_hours.use_rth(),
310        true, // keep_up_to_date — the whole point of .stream()
311    )?;
312
313    builder.send_capped::<HistoricalBarUpdate>(request, buffer_limit)
314}
315
316// pub(crate) internal plumbing called from `HistoricalTicksBuilder`; the
317// public API is already a builder, so flat args here are the deliberate
318// seam between the typed builder and the wire encoder. Internal, so the param
319// budget doesn't apply; the allow is for clippy's 8-arg lint.
320#[allow(clippy::too_many_arguments)]
321pub(crate) fn historical_ticks<T: TickDecoder<T>>(
322    client: &Client,
323    contract: &Contract,
324    start: Option<OffsetDateTime>,
325    end: Option<OffsetDateTime>,
326    number_of_ticks: i32,
327    what_to_show: WhatToShow,
328    trading_hours: TradingHours,
329    ignore_size: bool,
330) -> Result<TickSubscription<T>, Error> {
331    let builder = client.request();
332    let request = encoders::encode_request_historical_ticks(
333        builder.request_id(),
334        contract,
335        start,
336        end,
337        number_of_ticks,
338        what_to_show,
339        trading_hours.use_rth(),
340        ignore_size,
341    )?;
342    let request_id = builder.request_id();
343    let subscription = builder.send_raw(request)?;
344
345    Ok(TickSubscription::new(subscription, request_id, Arc::clone(&client.message_bus)))
346}
347
348pub(crate) fn historical_schedule(
349    client: &Client,
350    contract: &Contract,
351    end_date: Option<OffsetDateTime>,
352    duration: Duration,
353) -> Result<Schedule, Error> {
354    request_helpers::blocking::one_shot_cancelling::<ScheduleRequest>(client, |request_id| {
355        encoders::encode_request_historical_data(request_id, contract, end_date, duration, BarSize::Day, WhatToShow::Schedule, true, false)
356    })
357}
358
359// TickSubscription and related types
360
361/// Shared subscription handle that decodes historical tick batches as they arrive.
362///
363/// Each [`next`](Self::next), [`try_next`](Self::try_next), or
364/// [`next_timeout`](Self::next_timeout) returns
365/// `Option<Result<SubscriptionItem<T>, Error>>`, the same shape as
366/// [`Subscription`](crate::subscriptions::Subscription):
367///
368/// * `None` — the stream has ended.
369/// * `Some(Ok(SubscriptionItem::Data(tick)))` — a decoded tick.
370/// * `Some(Ok(SubscriptionItem::Notice(n)))` — a non-fatal IB notice bound to
371///   this request; the stream stays open.
372/// * `Some(Err(e))` — terminal error (decode or transport); subsequent calls
373///   return `None`.
374///
375/// When you only care about ticks, use [`iter_data`](Self::iter_data) (or
376/// [`next_data`](Self::next_data)), which filter notices and yield
377/// `Result<T, Error>`.
378#[must_use = "TickSubscription must be polled (.next() or .iter_data()) to receive ticks; dropping it cancels the request"]
379pub struct TickSubscription<T: TickDecoder<T>> {
380    done: AtomicBool,
381    stream_ended: AtomicBool,
382    messages: InternalSubscription,
383    buffer: Mutex<VecDeque<T>>,
384    request_id: i32,
385    message_bus: Arc<dyn MessageBus>,
386    cancelled: AtomicBool,
387}
388
389impl<T: TickDecoder<T>> TickSubscription<T> {
390    fn new(messages: InternalSubscription, request_id: i32, message_bus: Arc<dyn MessageBus>) -> Self {
391        Self {
392            done: false.into(),
393            stream_ended: AtomicBool::new(false),
394            messages,
395            buffer: Mutex::new(VecDeque::new()),
396            request_id,
397            message_bus,
398            cancelled: AtomicBool::new(false),
399        }
400    }
401
402    /// The request ID TWS knows this request by, for [`Client::cancel_historical_ticks`](crate::client::blocking::Client::cancel_historical_ticks)
403    /// from code that does not hold the subscription, or to match it in logs.
404    pub fn request_id(&self) -> i32 {
405        self.request_id
406    }
407
408    /// Cancel the historical-ticks request. After completion nothing is written.
409    /// Also fired automatically on `Drop` for unfinished subscriptions; explicit calls are idempotent.
410    ///
411    /// # Examples
412    ///
413    /// ```no_run
414    /// use ibapi::client::blocking::Client;
415    /// use ibapi::contracts::Contract;
416    ///
417    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
418    /// let contract = Contract::stock("MSFT").build();
419    /// let subscription = client.historical_ticks(&contract, 100).trade().expect("request failed");
420    /// subscription.cancel();
421    /// ```
422    pub fn cancel(&self) {
423        if self.cancelled.swap(true, Ordering::Relaxed) {
424            return;
425        }
426
427        // After the last batch TWS has finished the request: nothing to
428        // write, but the registration still goes.
429        if !self.done.load(Ordering::Relaxed) {
430            match encoders::encode_cancel_historical_ticks(self.request_id) {
431                Ok(message) => {
432                    if let Err(e) = self.message_bus.send_message(&message) {
433                        log_cancel_error("historical ticks subscription", &e);
434                    }
435                }
436                Err(e) => error!("error encoding cancel historical ticks: {e}"),
437            }
438        }
439        self.messages.cancel();
440    }
441
442    /// Blocking iterator yielding `Result<SubscriptionItem<T>, Error>` — both
443    /// `Data` and `Notice` arms surface to the caller. Use
444    /// [`iter_data`](Self::iter_data) when you only want ticks.
445    ///
446    /// # Examples
447    ///
448    /// ```no_run
449    /// use ibapi::client::blocking::Client;
450    /// use ibapi::contracts::Contract;
451    /// use ibapi::subscriptions::SubscriptionItem;
452    ///
453    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
454    /// let contract = Contract::stock("MSFT").build();
455    /// let subscription = client.historical_ticks(&contract, 100).trade().expect("request failed");
456    /// for item in subscription.iter() {
457    ///     match item {
458    ///         Ok(SubscriptionItem::Data(tick))   => println!("tick: {tick:?}"),
459    ///         Ok(SubscriptionItem::Notice(note)) => eprintln!("notice: {note}"),
460    ///         Err(e)                             => { eprintln!("error: {e}"); break; }
461    ///     }
462    /// }
463    /// ```
464    pub fn iter(&self) -> TickSubscriptionIter<'_, T> {
465        TickSubscriptionIter { subscription: self }
466    }
467
468    /// Non-blocking iterator. Same `SubscriptionItem<T>` shape as [`iter`](Self::iter);
469    /// see [`try_iter_data`](Self::try_iter_data) for the data-only variant.
470    ///
471    /// # Examples
472    ///
473    /// ```no_run
474    /// use ibapi::client::blocking::Client;
475    /// use ibapi::contracts::Contract;
476    ///
477    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
478    /// let contract = Contract::stock("MSFT").build();
479    /// let subscription = client.historical_ticks(&contract, 100).trade().expect("request failed");
480    /// for tick in subscription.try_iter_data() {
481    ///     println!("tick: {:?}", tick.expect("decode error"));
482    /// }
483    /// ```
484    pub fn try_iter(&self) -> TickSubscriptionTryIter<'_, T> {
485        TickSubscriptionTryIter { subscription: self }
486    }
487
488    /// Iterator that waits up to `duration` for each item. Same `SubscriptionItem<T>`
489    /// shape as [`iter`](Self::iter); see [`timeout_iter_data`](Self::timeout_iter_data)
490    /// for the data-only variant.
491    ///
492    /// # Examples
493    ///
494    /// ```no_run
495    /// use std::time::Duration;
496    /// use ibapi::client::blocking::Client;
497    /// use ibapi::contracts::Contract;
498    ///
499    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
500    /// let contract = Contract::stock("MSFT").build();
501    /// let subscription = client.historical_ticks(&contract, 100).trade().expect("request failed");
502    /// for tick in subscription.timeout_iter_data(Duration::from_secs(5)) {
503    ///     println!("tick: {:?}", tick.expect("decode error"));
504    /// }
505    /// ```
506    pub fn timeout_iter(&self, duration: std::time::Duration) -> TickSubscriptionTimeoutIter<'_, T> {
507        TickSubscriptionTimeoutIter {
508            subscription: self,
509            timeout: duration,
510        }
511    }
512
513    /// Blocking data iterator that filters notices and yields `Result<T, Error>`.
514    /// Notices are logged at `warn!` level.
515    ///
516    /// # Examples
517    ///
518    /// ```no_run
519    /// use ibapi::client::blocking::Client;
520    /// use ibapi::contracts::Contract;
521    ///
522    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
523    /// let contract = Contract::stock("MSFT").build();
524    /// let subscription = client.historical_ticks(&contract, 100).trade().expect("request failed");
525    /// for tick in subscription.iter_data() {
526    ///     println!("tick: {:?}", tick.expect("decode error"));
527    /// }
528    /// ```
529    pub fn iter_data(&self) -> FilterData<TickSubscriptionIter<'_, T>> {
530        self.iter().filter_data()
531    }
532
533    /// Non-blocking data iterator (notices filtered).
534    pub fn try_iter_data(&self) -> FilterData<TickSubscriptionTryIter<'_, T>> {
535        self.try_iter().filter_data()
536    }
537
538    /// Timeout-bounded data iterator (notices filtered).
539    pub fn timeout_iter_data(&self, duration: std::time::Duration) -> FilterData<TickSubscriptionTimeoutIter<'_, T>> {
540        self.timeout_iter(duration).filter_data()
541    }
542
543    /// Block until the next item is available.
544    ///
545    /// Returns the `SubscriptionItem<T>` envelope; use [`next_data`](Self::next_data)
546    /// when you only want ticks.
547    ///
548    /// # Examples
549    ///
550    /// ```no_run
551    /// use ibapi::client::blocking::Client;
552    /// use ibapi::contracts::Contract;
553    /// use ibapi::subscriptions::SubscriptionItem;
554    ///
555    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
556    /// let contract = Contract::stock("MSFT").build();
557    /// let subscription = client.historical_ticks(&contract, 100).trade().expect("request failed");
558    /// while let Some(item) = subscription.next() {
559    ///     match item {
560    ///         Ok(SubscriptionItem::Data(tick))   => println!("tick: {tick:?}"),
561    ///         Ok(SubscriptionItem::Notice(note)) => eprintln!("notice: {note}"),
562    ///         Err(e)                             => { eprintln!("error: {e}"); break; }
563    ///     }
564    /// }
565    /// ```
566    pub fn next(&self) -> Option<Result<SubscriptionItem<T>, Error>> {
567        self.next_helper(|| self.messages.next_routed())
568    }
569
570    /// Attempt to fetch the next item without blocking.
571    ///
572    /// Same `SubscriptionItem<T>` shape as [`next`](Self::next); returns `None`
573    /// if nothing is queued right now.
574    ///
575    /// # Examples
576    ///
577    /// ```no_run
578    /// use ibapi::client::blocking::Client;
579    /// use ibapi::contracts::Contract;
580    ///
581    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
582    /// let contract = Contract::stock("MSFT").build();
583    /// let subscription = client.historical_ticks(&contract, 100).trade().expect("request failed");
584    /// if let Some(Ok(item)) = subscription.try_next() {
585    ///     println!("item: {item:?}");
586    /// }
587    /// ```
588    pub fn try_next(&self) -> Option<Result<SubscriptionItem<T>, Error>> {
589        self.next_helper(|| self.messages.try_next_routed())
590    }
591
592    /// Wait up to `duration` for the next item to arrive.
593    ///
594    /// Same `SubscriptionItem<T>` shape as [`next`](Self::next). `None` means
595    /// either that `duration` passed or that the stream ended; terminal errors
596    /// come as `Some(Err(_))`.
597    ///
598    /// # Examples
599    ///
600    /// ```no_run
601    /// use std::time::Duration;
602    /// use ibapi::client::blocking::Client;
603    /// use ibapi::contracts::Contract;
604    ///
605    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
606    /// let contract = Contract::stock("MSFT").build();
607    /// let subscription = client.historical_ticks(&contract, 100).trade().expect("request failed");
608    /// if let Some(Ok(item)) = subscription.next_timeout(Duration::from_secs(5)) {
609    ///     println!("item: {item:?}");
610    /// }
611    /// ```
612    pub fn next_timeout(&self, duration: std::time::Duration) -> Option<Result<SubscriptionItem<T>, Error>> {
613        self.next_helper(|| self.messages.next_timeout_routed(duration))
614    }
615
616    /// Convenience: blocking `next` that filters out notices and yields just a tick.
617    /// Equivalent to `iter_data().next()`. Filtered notices are logged at `warn!`.
618    /// Use [`next`](Self::next) instead if you want to observe `Notice` items.
619    ///
620    /// # Examples
621    ///
622    /// ```no_run
623    /// use ibapi::client::blocking::Client;
624    /// use ibapi::contracts::Contract;
625    ///
626    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
627    /// let contract = Contract::stock("MSFT").build();
628    /// let subscription = client.historical_ticks(&contract, 100).trade().expect("request failed");
629    /// while let Some(tick) = subscription.next_data() {
630    ///     println!("tick: {:?}", tick.expect("decode error"));
631    /// }
632    /// ```
633    pub fn next_data(&self) -> Option<Result<T, Error>> {
634        self.iter_data().next()
635    }
636
637    fn next_helper<F>(&self, next_routed: F) -> Option<Result<SubscriptionItem<T>, Error>>
638    where
639        F: Fn() -> Option<RoutedItem>,
640    {
641        loop {
642            if let Some(tick) = self.next_buffered() {
643                return Some(Ok(SubscriptionItem::Data(tick)));
644            }
645
646            // `done` (decoder said this was the last batch) and `stream_ended`
647            // (terminal error/EndOfStream) are distinct: the former is graceful
648            // completion, the latter is termination. Either ends the stream once
649            // the buffer is drained.
650            if self.done.load(Ordering::Relaxed) || self.stream_ended.load(Ordering::Relaxed) {
651                return None;
652            }
653
654            let item = next_routed()?;
655
656            match classify::<T>(item) {
657                TickAction::Batch(ticks, done) => {
658                    self.buffer.lock().unwrap().extend(ticks);
659                    self.done.store(done, Ordering::Relaxed);
660                }
661                TickAction::Skip => {}
662                TickAction::Notice(notice) => return Some(Ok(SubscriptionItem::Notice(notice))),
663                TickAction::EndOfStream => {
664                    self.stream_ended.store(true, Ordering::Relaxed);
665                    return None;
666                }
667                TickAction::Error(e) => {
668                    self.stream_ended.store(true, Ordering::Relaxed);
669                    return Some(Err(e));
670                }
671            }
672        }
673    }
674
675    fn next_buffered(&self) -> Option<T> {
676        let mut buffer = self.buffer.lock().unwrap();
677        buffer.pop_front()
678    }
679}
680
681impl<T: TickDecoder<T>> Drop for TickSubscription<T> {
682    fn drop(&mut self) {
683        self.cancel();
684    }
685}
686
687/// A blocking iterator over `Result<SubscriptionItem<T>, Error>`.
688pub struct TickSubscriptionIter<'a, T: TickDecoder<T>> {
689    subscription: &'a TickSubscription<T>,
690}
691
692impl<T: TickDecoder<T>> Iterator for TickSubscriptionIter<'_, T> {
693    type Item = Result<SubscriptionItem<T>, Error>;
694
695    fn next(&mut self) -> Option<Self::Item> {
696        self.subscription.next()
697    }
698}
699
700impl<'a, T: TickDecoder<T>> IntoIterator for &'a TickSubscription<T> {
701    type Item = Result<SubscriptionItem<T>, Error>;
702    type IntoIter = TickSubscriptionIter<'a, T>;
703
704    fn into_iter(self) -> Self::IntoIter {
705        self.iter()
706    }
707}
708
709/// An owned blocking iterator over `Result<SubscriptionItem<T>, Error>`.
710pub struct TickSubscriptionOwnedIter<T: TickDecoder<T>> {
711    subscription: TickSubscription<T>,
712}
713
714impl<T: TickDecoder<T>> Iterator for TickSubscriptionOwnedIter<T> {
715    type Item = Result<SubscriptionItem<T>, Error>;
716
717    fn next(&mut self) -> Option<Self::Item> {
718        self.subscription.next()
719    }
720}
721
722impl<T: TickDecoder<T>> IntoIterator for TickSubscription<T> {
723    type Item = Result<SubscriptionItem<T>, Error>;
724    type IntoIter = TickSubscriptionOwnedIter<T>;
725
726    fn into_iter(self) -> Self::IntoIter {
727        TickSubscriptionOwnedIter { subscription: self }
728    }
729}
730
731/// A non-blocking iterator over `Result<SubscriptionItem<T>, Error>`.
732pub struct TickSubscriptionTryIter<'a, T: TickDecoder<T>> {
733    subscription: &'a TickSubscription<T>,
734}
735
736impl<T: TickDecoder<T>> Iterator for TickSubscriptionTryIter<'_, T> {
737    type Item = Result<SubscriptionItem<T>, Error>;
738
739    fn next(&mut self) -> Option<Self::Item> {
740        self.subscription.try_next()
741    }
742}
743
744/// An iterator that waits for the specified timeout duration for each item.
745pub struct TickSubscriptionTimeoutIter<'a, T: TickDecoder<T>> {
746    subscription: &'a TickSubscription<T>,
747    timeout: std::time::Duration,
748}
749
750impl<T: TickDecoder<T>> Iterator for TickSubscriptionTimeoutIter<'_, T> {
751    type Item = Result<SubscriptionItem<T>, Error>;
752
753    fn next(&mut self) -> Option<Self::Item> {
754        self.subscription.next_timeout(self.timeout)
755    }
756}
757
758#[cfg(test)]
759#[path = "sync_tests.rs"]
760mod tests;
ibapi/market_data/historical/sync.rs

ibapi/market_data/historical/
sync.rs
