ibapi/orders/async.rs
1//! Asynchronous implementation of order management functionality
2
3use time::OffsetDateTime;
4
5use crate::common::request_helpers::{self, expect_proto};
6use crate::messages::OutgoingMessages;
7use crate::protocol::{check_version, Features};
8use crate::subscriptions::Subscription;
9use crate::{Client, Error};
10
11use super::common::{decoders, encoders, verify};
12use super::*;
13
14impl Client {
15 /// Start building an order for the given contract
16 ///
17 /// This is the primary API for creating orders, providing a fluent interface
18 /// that guides you through the order creation process.
19 ///
20 /// # Examples
21 /// ```no_run
22 /// use ibapi::Client;
23 /// use ibapi::contracts::Contract;
24 ///
25 /// #[tokio::main]
26 /// async fn main() {
27 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
28 /// let contract = Contract::stock("AAPL").build();
29 ///
30 /// let order_id = client.order(&contract)
31 /// .buy(100)
32 /// .limit(50.0)
33 /// .submit().await.expect("order submission failed");
34 /// }
35 /// ```
36 pub fn order<'a>(&'a self, contract: &'a Contract) -> OrderBuilder<'a, Self> {
37 OrderBuilder::new(self, contract)
38 }
39
40 /// Subscribes to order update events. Only one subscription can be active at a time.
41 ///
42 /// Order-bound TWS errors and warnings (e.g. rejections, code 399 order
43 /// messages) arrive as [`SubscriptionItem::Notice`](crate::subscriptions::SubscriptionItem),
44 /// not as [`OrderUpdate`] variants. They surface via `next()` as below;
45 /// `filter_data()` drops them (logging at `warn!` level), so match on
46 /// notices explicitly when monitoring fire-and-forget orders for rejection.
47 ///
48 /// To pair a [`CommissionReport`] with the
49 /// [`ExecutionData`] it belongs to, join on
50 /// `execution_id` — the commission follows its execution and shares that key. See
51 /// the [`CommissionReport`] docs for the idiom.
52 ///
53 /// # Reconnection
54 ///
55 /// The stream survives the client's automatic reconnects: the same
56 /// subscription keeps delivering once the connection returns. Updates TWS
57 /// emitted during the outage are **not** replayed, and no marker appears
58 /// in the stream itself, so a quiet stream is indistinguishable from
59 /// missed activity. To detect a gap, watch [`Self::notice_stream`] for
60 /// the connectivity notices (codes 1100 connectivity lost, 1101 restored
61 /// with data lost, 1102 restored with data maintained, 1300 socket reset)
62 /// and reconcile open-order state via [`Self::open_orders`] after
63 /// restoration.
64 ///
65 /// # Examples
66 ///
67 /// ```no_run
68 /// use ibapi::prelude::*;
69 ///
70 /// #[tokio::main]
71 /// async fn main() {
72 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
73 /// let mut stream = client.order_update_stream().await.expect("failed to create stream");
74 /// while let Some(item) = stream.next().await {
75 /// match item {
76 /// Ok(SubscriptionItem::Data(OrderUpdate::OrderStatus(s))) => println!("status: {s:?}"),
77 /// Ok(SubscriptionItem::Data(update)) => println!("update: {update:?}"),
78 /// Ok(SubscriptionItem::Notice(notice)) if notice.is_error() => {
79 /// eprintln!("order {:?} rejected: {}", notice.request_id, notice.message);
80 /// }
81 /// Ok(SubscriptionItem::Notice(notice)) => println!("notice: {}", notice.message),
82 /// Err(e) => { eprintln!("err: {e:?}"); break; }
83 /// }
84 /// }
85 /// }
86 /// ```
87 pub async fn order_update_stream(&self) -> Result<Subscription<OrderUpdate>, Error> {
88 let internal_subscription = self.create_order_update_subscription().await?;
89 Ok(Subscription::new_from_internal_simple::<OrderUpdate>(
90 internal_subscription,
91 self.message_bus.clone(),
92 self.decoder_context(),
93 ))
94 }
95
96 /// Submits an Order (fire-and-forget).
97 ///
98 /// # Examples
99 ///
100 /// ```no_run
101 /// use ibapi::prelude::*;
102 ///
103 /// #[tokio::main]
104 /// async fn main() {
105 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
106 /// let contract = Contract::stock("AAPL").build();
107 /// let order = client
108 /// .order(&contract)
109 /// .buy(100)
110 /// .market()
111 /// .build()
112 /// .expect("order build");
113 /// let order_id = client.next_valid_order_id().await.expect("next id");
114 /// client.submit_order(order_id, &contract, &order).await.expect("submit failed");
115 /// }
116 /// ```
117 pub async fn submit_order(&self, order_id: i32, contract: &Contract, order: &Order) -> Result<(), Error> {
118 verify::verify_order(self, order, order_id)?;
119 verify::verify_order_contract(self, contract, order_id)?;
120
121 let request = encoders::encode_place_order(order_id, contract, order)?;
122 self.send_message(request).await?;
123
124 Ok(())
125 }
126
127 /// Submits an Order with a subscription for updates.
128 ///
129 /// # Examples
130 ///
131 /// ```no_run
132 /// use ibapi::prelude::*;
133 ///
134 /// #[tokio::main]
135 /// async fn main() {
136 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
137 /// let contract = Contract::stock("AAPL").build();
138 /// let order = client
139 /// .order(&contract)
140 /// .buy(100)
141 /// .market()
142 /// .build()
143 /// .expect("order build");
144 /// let order_id = client.next_valid_order_id().await.expect("next id");
145 /// let subscription = client.place_order(order_id, &contract, &order).await.expect("place");
146 /// let mut updates = subscription.filter_data();
147 /// while let Some(update) = updates.next().await {
148 /// println!("{update:?}");
149 /// }
150 /// }
151 /// ```
152 pub async fn place_order(&self, order_id: i32, contract: &Contract, order: &Order) -> Result<Subscription<PlaceOrder>, Error> {
153 verify::verify_order(self, order, order_id)?;
154 verify::verify_order_contract(self, contract, order_id)?;
155
156 let request = encoders::encode_place_order(order_id, contract, order)?;
157 let internal_subscription = self.send_order(order_id, request).await?;
158
159 Ok(Subscription::new_from_internal_simple::<PlaceOrder>(
160 internal_subscription,
161 self.message_bus.clone(),
162 self.decoder_context(),
163 ))
164 }
165
166 /// Cancels an open [Order].
167 ///
168 /// The confirmation (TWS code 202) arrives as a non-terminal
169 /// [`SubscriptionItem::Notice`](crate::subscriptions::SubscriptionItem);
170 /// the subscription stays open until dropped, so break once cancellation
171 /// is observed.
172 ///
173 /// # Examples
174 ///
175 /// ```no_run
176 /// use ibapi::prelude::*;
177 ///
178 /// #[tokio::main]
179 /// async fn main() {
180 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
181 /// // `""` selects immediate cancel (no manual order time).
182 /// let mut subscription = client.cancel_order(42, "").await.expect("cancel failed");
183 /// while let Some(item) = subscription.next().await {
184 /// match item {
185 /// Ok(SubscriptionItem::Data(event)) => println!("status: {event:?}"),
186 /// Ok(SubscriptionItem::Notice(n)) if n.is_cancellation() => {
187 /// println!("cancelled: {n}");
188 /// break;
189 /// }
190 /// Ok(SubscriptionItem::Notice(n)) => println!("notice: {n}"),
191 /// Err(e) => { eprintln!("cancel err: {e:?}"); break; }
192 /// }
193 /// }
194 /// }
195 /// ```
196 pub async fn cancel_order(&self, order_id: i32, manual_order_cancel_time: &str) -> Result<Subscription<CancelOrder>, Error> {
197 if !manual_order_cancel_time.is_empty() {
198 check_version(self.server_version(), Features::MANUAL_ORDER_TIME)?;
199 }
200
201 let request = encoders::encode_cancel_order(order_id, manual_order_cancel_time)?;
202 let internal_subscription = self.send_order(order_id, request).await?;
203
204 Ok(Subscription::new_from_internal_simple::<CancelOrder>(
205 internal_subscription,
206 self.message_bus.clone(),
207 self.decoder_context(),
208 ))
209 }
210
211 /// Cancels all open [Order]s.
212 ///
213 /// # Examples
214 ///
215 /// ```no_run
216 /// use ibapi::prelude::*;
217 ///
218 /// #[tokio::main]
219 /// async fn main() {
220 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
221 /// client.global_cancel().await.expect("global_cancel failed");
222 /// }
223 /// ```
224 pub async fn global_cancel(&self) -> Result<(), Error> {
225 check_version(self.server_version(), Features::REQ_GLOBAL_CANCEL)?;
226
227 let message = encoders::encode_global_cancel()?;
228 self.send_message(message).await?;
229
230 Ok(())
231 }
232
233 /// Gets next valid order id
234 ///
235 /// The returned value also raises the client's order-ID generator to at
236 /// least that value — monotonically, never lowering it below locally
237 /// allocated order IDs, including IDs whose order has not yet reached the
238 /// server.
239 ///
240 /// # Examples
241 ///
242 /// ```no_run
243 /// use ibapi::prelude::*;
244 ///
245 /// #[tokio::main]
246 /// async fn main() {
247 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
248 /// let next = client.next_valid_order_id().await.expect("next id failed");
249 /// println!("next_valid_order_id: {next}");
250 /// }
251 /// ```
252 pub async fn next_valid_order_id(&self) -> Result<i32, Error> {
253 let next_order_id = request_helpers::one_shot_shared(
254 self,
255 OutgoingMessages::RequestIds,
256 encoders::encode_next_valid_order_id,
257 expect_proto(decoders::decode_next_valid_id_proto),
258 )
259 .await?;
260
261 self.raise_next_order_id(next_order_id);
262 Ok(next_order_id)
263 }
264
265 /// Requests completed [Order]s.
266 ///
267 /// # Examples
268 ///
269 /// ```no_run
270 /// use ibapi::prelude::*;
271 ///
272 /// #[tokio::main]
273 /// async fn main() {
274 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
275 /// let subscription = client.completed_orders(true).await.expect("completed_orders failed");
276 /// let mut orders = subscription.filter_data();
277 /// while let Some(order) = orders.next().await {
278 /// println!("{order:?}");
279 /// }
280 /// }
281 /// ```
282 pub async fn completed_orders(&self, api_only: bool) -> Result<Subscription<Orders>, Error> {
283 check_version(self.server_version(), Features::COMPLETED_ORDERS)?;
284
285 let request = encoders::encode_completed_orders(api_only)?;
286
287 let internal_subscription = self.send_shared_request(OutgoingMessages::RequestCompletedOrders, request).await?;
288 Ok(Subscription::new_from_internal_simple::<Orders>(
289 internal_subscription,
290 self.message_bus.clone(),
291 self.decoder_context(),
292 ))
293 }
294
295 /// Requests all open orders placed by this specific API client.
296 ///
297 /// # Examples
298 ///
299 /// ```no_run
300 /// use ibapi::prelude::*;
301 ///
302 /// #[tokio::main]
303 /// async fn main() {
304 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
305 /// let subscription = client.open_orders().await.expect("open_orders failed");
306 /// let mut orders = subscription.filter_data();
307 /// while let Some(order) = orders.next().await {
308 /// println!("{order:?}");
309 /// }
310 /// }
311 /// ```
312 pub async fn open_orders(&self) -> Result<Subscription<Orders>, Error> {
313 let request = encoders::encode_open_orders()?;
314
315 let internal_subscription = self.send_shared_request(OutgoingMessages::RequestOpenOrders, request).await?;
316 Ok(Subscription::new_from_internal_simple::<Orders>(
317 internal_subscription,
318 self.message_bus.clone(),
319 self.decoder_context(),
320 ))
321 }
322
323 /// Requests all *current* open orders in associated accounts.
324 ///
325 /// # Examples
326 ///
327 /// ```no_run
328 /// use ibapi::prelude::*;
329 ///
330 /// #[tokio::main]
331 /// async fn main() {
332 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
333 /// let subscription = client.all_open_orders().await.expect("all_open_orders failed");
334 /// let mut orders = subscription.filter_data();
335 /// while let Some(order) = orders.next().await {
336 /// println!("{order:?}");
337 /// }
338 /// }
339 /// ```
340 pub async fn all_open_orders(&self) -> Result<Subscription<Orders>, Error> {
341 let request = encoders::encode_all_open_orders()?;
342
343 let internal_subscription = self.send_shared_request(OutgoingMessages::RequestAllOpenOrders, request).await?;
344 Ok(Subscription::new_from_internal_simple::<Orders>(
345 internal_subscription,
346 self.message_bus.clone(),
347 self.decoder_context(),
348 ))
349 }
350
351 /// Requests status updates about future orders placed from TWS.
352 ///
353 /// # Examples
354 ///
355 /// ```no_run
356 /// use ibapi::prelude::*;
357 ///
358 /// #[tokio::main]
359 /// async fn main() {
360 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
361 /// let subscription = client.auto_open_orders(true).await.expect("auto_open_orders failed");
362 /// let mut orders = subscription.filter_data();
363 /// while let Some(order) = orders.next().await {
364 /// println!("{order:?}");
365 /// }
366 /// }
367 /// ```
368 pub async fn auto_open_orders(&self, auto_bind: bool) -> Result<Subscription<Orders>, Error> {
369 let request = encoders::encode_auto_open_orders(auto_bind)?;
370
371 let internal_subscription = self.send_shared_request(OutgoingMessages::RequestAutoOpenOrders, request).await?;
372 Ok(Subscription::new_from_internal_simple::<Orders>(
373 internal_subscription,
374 self.message_bus.clone(),
375 self.decoder_context(),
376 ))
377 }
378
379 /// Requests current day's executions matching the filter.
380 ///
381 /// Both [`ExecutionData`] and
382 /// [`CommissionReport`] are delivered on this
383 /// stream. Join a commission to its execution by `execution_id`
384 /// (see the [`CommissionReport`] docs) — the commission follows its execution.
385 ///
386 /// # Examples
387 ///
388 /// ```no_run
389 /// use ibapi::Client;
390 /// use ibapi::orders::{ExecutionFilter, ExecutionFilterSide};
391 /// use ibapi::subscriptions::SubscriptionItem;
392 /// use futures::StreamExt;
393 ///
394 /// #[tokio::main]
395 /// async fn main() {
396 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
397 /// let filter = ExecutionFilter {
398 /// side: Some(ExecutionFilterSide::Buy),
399 /// ..ExecutionFilter::default()
400 /// };
401 /// let mut subscription = client.executions(filter).await.expect("request failed");
402 ///
403 /// while let Some(item) = subscription.next().await {
404 /// match item {
405 /// Ok(SubscriptionItem::Data(ex)) => println!("{ex:?}"),
406 /// Ok(SubscriptionItem::Notice(n)) => eprintln!("notice: {n}"),
407 /// Err(e) => eprintln!("Error: {e}"),
408 /// }
409 /// }
410 /// }
411 /// ```
412 pub async fn executions(&self, filter: ExecutionFilter) -> Result<Subscription<Executions>, Error> {
413 let request_id = self.next_request_id();
414 let request = encoders::encode_executions(request_id, &filter)?;
415 let internal_subscription = self.send_request(request_id, request).await?;
416 Ok(Subscription::new_from_internal_simple::<Executions>(
417 internal_subscription,
418 self.message_bus.clone(),
419 self.decoder_context(),
420 ))
421 }
422
423 /// Exercise an option contract.
424 ///
425 /// # Examples
426 ///
427 /// ```no_run
428 /// use ibapi::prelude::*;
429 /// use ibapi::orders::ExerciseAction;
430 ///
431 /// #[tokio::main]
432 /// async fn main() {
433 /// let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
434 /// let contract = Contract::option("AAPL", "20251219", 150.0, OptionRight::Call);
435 /// let subscription = client
436 /// .exercise_options(&contract, ExerciseAction::Exercise, 1, "DU000001", false, None)
437 /// .await
438 /// .expect("exercise_options failed");
439 /// // Consume the subscription so execution updates and commission reports surface.
440 /// let mut events = subscription.filter_data();
441 /// while let Some(event) = events.next().await {
442 /// match event {
443 /// Ok(item) => println!("exercise event: {item:?}"),
444 /// Err(e) => { eprintln!("exercise err: {e:?}"); break; }
445 /// }
446 /// }
447 /// }
448 /// ```
449 pub async fn exercise_options(
450 &self,
451 contract: &Contract,
452 exercise_action: ExerciseAction,
453 exercise_quantity: i32,
454 account: &str,
455 ovrd: bool,
456 manual_order_time: Option<OffsetDateTime>,
457 ) -> Result<Subscription<ExerciseOptions>, Error> {
458 let order_id = self.next_order_id();
459 let request = encoders::encode_exercise_options(order_id, contract, exercise_action, exercise_quantity, account, ovrd, manual_order_time)?;
460 let internal_subscription = self.send_order(order_id, request).await?;
461 Ok(Subscription::new_from_internal_simple::<ExerciseOptions>(
462 internal_subscription,
463 self.message_bus.clone(),
464 self.decoder_context(),
465 ))
466 }
467}
468
469#[cfg(test)]
470#[path = "async_tests.rs"]
471mod tests;