request_market_data/
request_market_data.rs

1use std::time::Duration;
2
3use truefix_twsapi_client::client::{ClientConfig, TwsApiClient};
4use truefix_twsapi_client::events::Event;
5use truefix_twsapi_client::requests::MarketDataRequest;
6use truefix_twsapi_client::types::Contract;
7
8#[tokio::main]
9async fn main() -> truefix_twsapi_client::error::TwsApiResult<()> {
10    let host = std::env::var("TWS_HOST").unwrap_or_else(|_| "127.0.0.1".to_owned());
11    let port = std::env::var("TWS_PORT")
12        .ok()
13        .and_then(|value| value.parse::<u16>().ok())
14        .unwrap_or(7497);
15    let client_id = std::env::var("TWS_CLIENT_ID")
16        .ok()
17        .and_then(|value| value.parse::<i32>().ok())
18        .unwrap_or(1002);
19    let symbol = std::env::var("TWS_SYMBOL").unwrap_or_else(|_| "AAPL".to_owned());
20    let exchange = std::env::var("TWS_EXCHANGE").unwrap_or_else(|_| "SMART".to_owned());
21    let currency = std::env::var("TWS_CURRENCY").unwrap_or_else(|_| "USD".to_owned());
22    let market_data_type = std::env::var("TWS_MARKET_DATA_TYPE")
23        .ok()
24        .and_then(|value| value.parse::<i32>().ok())
25        .unwrap_or(1);
26
27    let req_id = 9001;
28    let mut client = TwsApiClient::connect(ClientConfig::new(host, port, client_id)).await?;
29    wait_until_api_ready(&mut client).await?;
30    client.req_market_data_type(market_data_type).await?;
31    client
32        .req_mkt_data(MarketDataRequest {
33            req_id,
34            contract: Contract {
35                symbol,
36                sec_type: "STK".to_owned(),
37                exchange,
38                currency,
39                ..Contract::default()
40            },
41            generic_tick_list: String::new(),
42            snapshot: false,
43            regulatory_snapshot: false,
44            market_data_options: Vec::new(),
45        })
46        .await?;
47
48    let result = tokio::time::timeout(Duration::from_secs(30), async {
49        let should_cancel = loop {
50            match client.read_event().await? {
51                Event::TickPrice {
52                    req_id: event_req_id,
53                    tick_type,
54                    price,
55                    attrib,
56                } if event_req_id == req_id => {
57                    println!("price tick_type={tick_type} price={price} attrib={attrib}");
58                    break true;
59                }
60                Event::TickSize {
61                    req_id: event_req_id,
62                    tick_type,
63                    size,
64                } if event_req_id == req_id => {
65                    println!("size tick_type={tick_type} size={size}");
66                }
67                Event::Error {
68                    req_id: event_req_id,
69                    code,
70                    message,
71                    ..
72                } if event_req_id < 0 && is_market_data_status_code(code) => {
73                    eprintln!("TWS status {code}: {message}");
74                }
75                Event::Error {
76                    req_id: event_req_id,
77                    code,
78                    message,
79                    ..
80                } if event_req_id == req_id && is_delayed_market_data_notice(code) => {
81                    eprintln!("TWS notice {code}: {message}");
82                }
83                Event::Error {
84                    req_id: event_req_id,
85                    code,
86                    message,
87                    ..
88                } if event_req_id == req_id => {
89                    eprintln!("TWS error {code}: {message}");
90                    break false;
91                }
92                Event::Error { code, message, .. } => {
93                    eprintln!("TWS notice {code}: {message}");
94                }
95                _ => {}
96            }
97        };
98        truefix_twsapi_client::error::TwsApiResult::Ok(should_cancel)
99    })
100    .await;
101
102    let should_cancel = match &result {
103        Ok(Ok(should_cancel)) => *should_cancel,
104        Ok(Err(_)) | Err(_) => true,
105    };
106    if should_cancel && let Err(error) = client.cancel_mkt_data(req_id).await {
107        eprintln!("cancelMktData failed: {error}");
108    }
109    if should_cancel {
110        tokio::time::sleep(Duration::from_millis(250)).await;
111    }
112
113    match result {
114        Ok(result) => result.map(|_| ()),
115        Err(_) => {
116            eprintln!("timed out waiting for market data");
117            Ok(())
118        }
119    }
120}
121
122fn is_market_data_status_code(code: i32) -> bool {
123    matches!(code, 2103 | 2104 | 2106 | 2107 | 2108 | 2158)
124}
125
126fn is_delayed_market_data_notice(code: i32) -> bool {
127    code == 10167
128}
129
130async fn wait_until_api_ready(
131    client: &mut TwsApiClient,
132) -> truefix_twsapi_client::error::TwsApiResult<()> {
133    let result = tokio::time::timeout(Duration::from_secs(10), async {
134        while !client.api_ready() {
135            match client.read_event().await? {
136                Event::Error { code, message, .. } if is_market_data_status_code(code) => {
137                    eprintln!("TWS status {code}: {message}");
138                }
139                Event::Error { code, message, .. } => {
140                    eprintln!("TWS notice {code}: {message}");
141                }
142                _ => {}
143            }
144        }
145        truefix_twsapi_client::error::TwsApiResult::Ok(())
146    })
147    .await;
148
149    match result {
150        Ok(result) => result,
151        Err(_) => {
152            eprintln!("timed out waiting for initial API callbacks");
153            Ok(())
154        }
155    }
156}
request_market_data/request_market_data.rs

request_market_data/
request_market_data.rs
