1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
//! Pricing endpoints: current prices and the pricing stream.
use serde::{Deserialize, Serialize};
use crate::client::Client;
use crate::error::Error;
use crate::models::{AccountId, ClientPrice, DateTime, HomeConversions, InstrumentName};
use crate::streaming::{
PricingKind, PricingStream, StreamConfig, StreamKind, stream_config_setters,
};
impl Client {
/// Get pricing information for a list of instruments within an
/// account.
///
/// `GET /v3/accounts/{accountID}/pricing`
///
/// # Examples
///
/// ```no_run
/// # async fn run() -> Result<(), oanda_rs::Error> {
/// # let client = oanda_rs::Client::new(oanda_rs::Environment::Practice, "token");
/// let pricing = client
/// .prices("101-004-1234567-001", ["EUR_USD", "USD_JPY"])
/// .send()
/// .await?;
/// for price in pricing.prices {
/// println!("{:?}: {:?}", price.instrument, price.closeout_bid);
/// }
/// # Ok(())
/// # }
/// ```
pub fn prices<I>(&self, account_id: impl Into<AccountId>, instruments: I) -> PricesRequest
where
I: IntoIterator,
I::Item: Into<InstrumentName>,
{
PricesRequest {
client: self.clone(),
account_id: account_id.into(),
instruments: instruments.into_iter().map(Into::into).collect(),
params: Vec::new(),
}
}
/// Open a stream of price updates (plus a heartbeat every 5 seconds)
/// for the given instruments.
///
/// The returned [`PricingStream`] manages its own connection: it
/// detects stale connections via the heartbeat, reconnects with capped
/// exponential backoff, and requests a fresh price snapshot on every
/// reconnect. See [`crate::streaming`] for details and tuning.
///
/// Connects to the **streaming host**
/// (`stream-fxpractice`/`stream-fxtrade`). OANDA allows at most 20
/// active streams per IP; prices are throttled to at most 4 updates
/// per second per instrument.
///
/// `GET /v3/accounts/{accountID}/pricing/stream`
///
/// # Examples
///
/// ```no_run
/// # async fn run() -> Result<(), oanda_rs::Error> {
/// # let client = oanda_rs::Client::new(oanda_rs::Environment::Practice, "token");
/// use futures_util::StreamExt;
/// use oanda_rs::models::PriceStreamItem;
///
/// let mut stream = client
/// .pricing_stream("101-004-1234567-001", ["EUR_USD"])
/// .send()
/// .await?;
/// while let Some(item) = stream.next().await {
/// match item? {
/// PriceStreamItem::Price(price) => println!("{:?}", price.closeout_bid),
/// PriceStreamItem::Heartbeat(_) => {}
/// _ => {}
/// }
/// }
/// # Ok(())
/// # }
/// ```
pub fn pricing_stream<I>(
&self,
account_id: impl Into<AccountId>,
instruments: I,
) -> PricingStreamRequest
where
I: IntoIterator,
I::Item: Into<InstrumentName>,
{
PricingStreamRequest {
client: self.clone(),
account_id: account_id.into(),
instruments: instruments.into_iter().map(Into::into).collect(),
snapshot: None,
config: StreamConfig::default(),
}
}
}
/// Builder for [`Client::prices`].
#[derive(Debug)]
pub struct PricesRequest {
client: Client,
account_id: AccountId,
instruments: Vec<InstrumentName>,
params: Vec<(&'static str, String)>,
}
impl PricesRequest {
/// Only return prices and home conversions updated after this time
/// (the response's `time` field can be fed back here when polling).
pub fn since(mut self, since: impl Into<DateTime>) -> Self {
self.params.push(("since", since.into().0));
self
}
/// Also include home currency conversion factors in the response.
pub fn include_home_conversions(mut self, include: bool) -> Self {
self.params
.push(("includeHomeConversions", include.to_string()));
self
}
/// Performs the request.
pub async fn send(self) -> Result<PricesResponse, Error> {
let request = self
.client
.get(&["accounts", self.account_id.as_str(), "pricing"])
.query(&[(
"instruments",
super::accounts::join_names(&self.instruments),
)])
.query(&self.params);
self.client.execute(request).await
}
}
/// Response of [`Client::prices`].
#[derive(Debug, Clone, Serialize, Deserialize)]
#[non_exhaustive]
pub struct PricesResponse {
/// The list of price objects requested.
#[serde(rename = "prices", default, skip_serializing_if = "Vec::is_empty")]
pub prices: Vec<ClientPrice>,
/// The home currency conversion factors requested (only present when
/// `includeHomeConversions` was set).
#[serde(
rename = "homeConversions",
default,
skip_serializing_if = "Vec::is_empty"
)]
pub home_conversions: Vec<HomeConversions>,
/// The most recent time any of the prices/conversion factors were
/// updated; usable as the `since` parameter of the next poll.
#[serde(rename = "time", skip_serializing_if = "Option::is_none")]
pub time: Option<DateTime>,
}
/// Builder for [`Client::pricing_stream`].
#[derive(Debug)]
pub struct PricingStreamRequest {
client: Client,
account_id: AccountId,
instruments: Vec<InstrumentName>,
snapshot: Option<bool>,
config: StreamConfig,
}
impl PricingStreamRequest {
/// Whether to send a pricing snapshot of all requested instruments
/// when the stream (first) connects (OANDA defaults to `true`).
/// Reconnects always request a snapshot regardless of this setting.
pub fn snapshot(mut self, snapshot: bool) -> Self {
self.snapshot = Some(snapshot);
self
}
stream_config_setters!();
/// Connects and returns the managed stream. Fails fast when the
/// initial connection is rejected (e.g. bad token or account).
pub async fn send(self) -> Result<PricingStream, Error> {
let mut kind = PricingKind {
client: self.client,
account_id: self.account_id,
instruments: super::accounts::join_names(&self.instruments),
snapshot: self.snapshot,
};
let initial = kind.connect(false).await?;
Ok(PricingStream::new(kind, self.config, initial))
}
}