ibapi/market_data/realtime/async.rs
1use crate::client::ClientRequestBuilders;
2use crate::common::request_helpers::{self, empty_on_end_of_stream, expect_proto};
3use crate::contracts::{Contract, TagValue};
4use crate::messages::OutgoingMessages;
5use crate::{Client, Error};
6
7use super::common::{decoders, encoders};
8use super::{
9 Bar, DepthMarketDataDescription, MarketDataBuilder, MarketDepthBuilder, MarketDepths, RealtimeBarsBuilder, TickByTickBuilder, TickTypes,
10 WhatToShow,
11};
12use crate::market_data::{SmartDepth, TradingHours};
13use crate::subscriptions::{StreamDecoder, Subscription};
14
15impl Client {
16 /// Switches market data type returned from request_market_data requests to Live, Frozen, Delayed, or FrozenDelayed.
17 ///
18 /// # Arguments
19 /// * `market_data_type` - Type of market data to retrieve.
20 ///
21 /// # Examples
22 ///
23 /// ```no_run
24 /// use ibapi::prelude::*;
25 ///
26 /// #[tokio::main]
27 /// async fn main() {
28 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
29 ///
30 /// let market_data_type = MarketDataType::Realtime;
31 /// client.switch_market_data_type(market_data_type).await.expect("request failed");
32 /// println!("market data switched: {market_data_type:?}");
33 /// }
34 /// ```
35 pub async fn switch_market_data_type(&self, market_data_type: crate::market_data::MarketDataType) -> Result<(), Error> {
36 let message = crate::market_data::encoders::encode_request_market_data_type(market_data_type)?;
37 self.send_message(message).await
38 }
39
40 /// Returns a builder for a streaming level-1 market data subscription.
41 ///
42 /// Streams by default; `.snapshot()` switches to a one-shot request. See
43 /// [`MarketDataBuilder`] for the chained methods.
44 ///
45 /// # Examples
46 ///
47 /// ```no_run
48 /// use ibapi::prelude::*;
49 ///
50 /// #[tokio::main]
51 /// async fn main() {
52 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
53 /// let contract = Contract::stock("AAPL").build();
54 ///
55 /// // Subscribe to real-time streaming data with specific tick types
56 /// let mut subscription = client
57 /// .market_data(&contract)
58 /// .generic_ticks(&["233", "236"]) // RTVolume and Shortable
59 /// .subscribe()
60 /// .await
61 /// .expect("subscription failed");
62 ///
63 /// while let Some(item) = subscription.next().await {
64 /// match item {
65 /// Ok(SubscriptionItem::Data(TickTypes::Price(price))) => println!("Price: {price:?}"),
66 /// Ok(SubscriptionItem::Data(TickTypes::Size(size))) => println!("Size: {size:?}"),
67 /// Ok(SubscriptionItem::Data(_)) => {}
68 /// Ok(SubscriptionItem::Notice(n)) => eprintln!("notice: {n}"),
69 /// Err(e) => { eprintln!("error: {e}"); break; }
70 /// }
71 /// }
72 /// }
73 /// ```
74 ///
75 /// A one-shot snapshot, driven manually. To just collect the ticks and stop, prefer
76 /// [`MarketDataBuilder::snapshot_once`], which wraps this loop with a timeout.
77 ///
78 /// ```no_run
79 /// use ibapi::prelude::*;
80 ///
81 /// #[tokio::main]
82 /// async fn main() {
83 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
84 /// let contract = Contract::stock("AAPL").build();
85 ///
86 /// let mut subscription = client
87 /// .market_data(&contract)
88 /// .snapshot()
89 /// .subscribe()
90 /// .await
91 /// .expect("subscription failed");
92 ///
93 /// while let Some(item) = subscription.next().await {
94 /// match item {
95 /// Ok(SubscriptionItem::Data(TickTypes::SnapshotEnd)) => { println!("Snapshot complete"); break; }
96 /// Ok(SubscriptionItem::Data(tick)) => println!("tick: {tick:?}"),
97 /// Ok(SubscriptionItem::Notice(n)) => eprintln!("notice: {n}"),
98 /// Err(e) => { eprintln!("error: {e}"); break; }
99 /// }
100 /// }
101 /// }
102 /// ```
103 pub fn market_data<'a>(&'a self, contract: &'a Contract) -> MarketDataBuilder<'a, Self> {
104 MarketDataBuilder::new(self, contract)
105 }
106
107 /// Returns a builder for a real-time 5-second bar subscription.
108 ///
109 /// Defaults to `WhatToShow::Trades` and `TradingHours::Regular`. See
110 /// [`RealtimeBarsBuilder`] for the chained methods.
111 ///
112 /// # Examples
113 ///
114 /// ```no_run
115 /// use ibapi::prelude::*;
116 ///
117 /// #[tokio::main]
118 /// async fn main() {
119 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
120 /// let contract = Contract::stock("TSLA").build();
121 ///
122 /// let mut subscription = client
123 /// .realtime_bars(&contract)
124 /// .what_to_show(RealtimeWhatToShow::Trades)
125 /// .trading_hours(TradingHours::Extended)
126 /// .subscribe()
127 /// .await
128 /// .expect("realtime bars request failed");
129 ///
130 /// while let Some(item) = subscription.next().await {
131 /// match item {
132 /// Ok(SubscriptionItem::Data(bar)) => println!("{bar:?}"),
133 /// Ok(SubscriptionItem::Notice(n)) => eprintln!("notice: {n}"),
134 /// Err(e) => { eprintln!("error: {e}"); break; }
135 /// }
136 /// }
137 /// }
138 /// ```
139 pub fn realtime_bars<'a>(&'a self, contract: &'a Contract) -> RealtimeBarsBuilder<'a, Self> {
140 RealtimeBarsBuilder::new(self, contract)
141 }
142
143 /// Returns a builder for a tick-by-tick real-time subscription.
144 ///
145 /// Pick the tick stream with the terminal — `.last()` / `.all_last()` /
146 /// `.bid_ask(IgnoreSize)` / `.mid_point()`. See [`TickByTickBuilder`].
147 ///
148 /// # Examples
149 ///
150 /// ```no_run
151 /// use ibapi::prelude::*;
152 ///
153 /// #[tokio::main]
154 /// async fn main() {
155 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
156 /// let contract = Contract::stock("AAPL").build();
157 ///
158 /// let mut quotes = client
159 /// .tick_by_tick(&contract, 10)
160 /// .bid_ask(IgnoreSize::No)
161 /// .await
162 /// .expect("tick-by-tick bid/ask request failed");
163 ///
164 /// while let Some(item) = quotes.next().await {
165 /// match item {
166 /// Ok(SubscriptionItem::Data(q)) => println!("{q:?}"),
167 /// Ok(SubscriptionItem::Notice(n)) => eprintln!("notice: {n}"),
168 /// Err(e) => { eprintln!("error: {e}"); break; }
169 /// }
170 /// }
171 /// }
172 /// ```
173 pub fn tick_by_tick<'a>(&'a self, contract: &'a Contract, number_of_ticks: i32) -> TickByTickBuilder<'a, Self> {
174 TickByTickBuilder::new(self, contract, number_of_ticks)
175 }
176
177 /// Returns a builder for a level-2 market-depth (order book) subscription.
178 ///
179 /// Defaults to `SmartDepth::No`. See [`MarketDepthBuilder`] for the chained
180 /// methods, and [`MarketDepths`] for handling a book reset (317) and halt
181 /// (316).
182 ///
183 /// # Examples
184 ///
185 /// ```no_run
186 /// use ibapi::prelude::*;
187 ///
188 /// #[tokio::main]
189 /// async fn main() {
190 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
191 /// let contract = Contract::stock("AAPL").build();
192 ///
193 /// let mut subscription = client
194 /// .market_depth(&contract, 5)
195 /// .smart_depth(SmartDepth::Yes)
196 /// .subscribe()
197 /// .await
198 /// .expect("market depth request failed");
199 ///
200 /// while let Some(item) = subscription.next().await {
201 /// match item {
202 /// Ok(SubscriptionItem::Data(row)) => println!("{row:?}"),
203 /// Ok(SubscriptionItem::Notice(n)) => eprintln!("notice: {n}"),
204 /// Err(e) => { eprintln!("error: {e}"); break; }
205 /// }
206 /// }
207 /// }
208 /// ```
209 pub fn market_depth<'a>(&'a self, contract: &'a Contract, number_of_rows: i32) -> MarketDepthBuilder<'a, Self> {
210 MarketDepthBuilder::new(self, contract, number_of_rows)
211 }
212
213 /// Requests venues for which market data is returned to market_depth (those with market makers)
214 ///
215 /// # Examples
216 ///
217 /// ```no_run
218 /// use ibapi::prelude::*;
219 ///
220 /// #[tokio::main]
221 /// async fn main() {
222 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
223 /// let exchanges = client.market_depth_exchanges().await.expect("error requesting market depth exchanges");
224 /// for exchange in &exchanges {
225 /// println!("{exchange:?}");
226 /// }
227 /// }
228 /// ```
229 pub async fn market_depth_exchanges(&self) -> Result<Vec<DepthMarketDataDescription>, Error> {
230 request_helpers::one_shot_shared(
231 self,
232 OutgoingMessages::RequestMktDepthExchanges,
233 encoders::encode_request_market_depth_exchanges,
234 expect_proto(decoders::decode_market_depth_exchanges_proto),
235 )
236 .await
237 .or_else(empty_on_end_of_stream)
238 }
239}
240
241pub(crate) async fn market_data(
242 client: &Client,
243 contract: &Contract,
244 generic_ticks: &[&str],
245 snapshot: bool,
246 regulatory_snapshot: bool,
247) -> Result<Subscription<TickTypes>, Error> {
248 let builder = client.request();
249 let request = encoders::encode_request_market_data(builder.request_id(), contract, generic_ticks, snapshot, regulatory_snapshot)?;
250
251 builder.send::<TickTypes>(request).await
252}
253
254pub(crate) async fn realtime_bars(
255 client: &Client,
256 contract: &Contract,
257 what_to_show: &WhatToShow,
258 trading_hours: TradingHours,
259 options: &[TagValue],
260) -> Result<Subscription<Bar>, Error> {
261 let builder = client.request();
262 let request = encoders::encode_request_realtime_bars(builder.request_id(), contract, what_to_show, trading_hours.use_rth(), options)?;
263
264 builder.send::<Bar>(request).await
265}
266
267pub(crate) async fn market_depth(
268 client: &Client,
269 contract: &Contract,
270 number_of_rows: i32,
271 smart_depth: SmartDepth,
272) -> Result<Subscription<MarketDepths>, Error> {
273 let is_smart_depth = matches!(smart_depth, SmartDepth::Yes);
274
275 let builder = client.request();
276 let request = encoders::encode_request_market_depth(builder.request_id(), contract, number_of_rows, is_smart_depth)?;
277 builder
278 .send_with_context::<MarketDepths>(request, client.decoder_context().with_smart_depth(is_smart_depth))
279 .await
280}
281
282pub(crate) async fn tick_by_tick<T: StreamDecoder<T> + Send + 'static>(
283 client: &Client,
284 contract: &Contract,
285 tick_type: &str,
286 number_of_ticks: i32,
287 ignore_size: bool,
288) -> Result<Subscription<T>, Error> {
289 let builder = client.request();
290 let request = encoders::encode_tick_by_tick(builder.request_id(), contract, tick_type, number_of_ticks, ignore_size)?;
291 builder.send::<T>(request).await
292}
293
294#[cfg(test)]
295#[path = "async_tests.rs"]
296mod tests;