karaolidis/qrust
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix bad request response handling

Browse Source 
Signed-off-by: Nikolaos Karaolidis <nick@karaolidis.com>
This commit is contained in:
Nick Karaolidis
2024-02-17 20:35:50 +00:00
parent ae5044142d
commit 152a0b4682
10 changed files with 46 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/init.rs
View File

@@ -1,7 +1,7 @@
use crate::{ use crate::{
config::{Config, ALPACA_MODE}, config::{Config, ALPACA_MODE},
database, database,
types::alpaca::{self, api, shared::Sort}, types::alpaca::{self, shared::Sort},
}; };
use log::{info, warn}; use log::{info, warn};
use std::{collections::HashMap, sync::Arc}; use std::{collections::HashMap, sync::Arc};
@@ -45,11 +45,11 @@ pub async fn rehydrate_orders(config: &Arc<Config>) {
let mut orders = vec![]; let mut orders = vec![];
let mut after = OffsetDateTime::UNIX_EPOCH; let mut after = OffsetDateTime::UNIX_EPOCH;
while let Some(message) = api::incoming::order::get( while let Some(message) = alpaca::api::incoming::order::get(
&config.alpaca_client, &config.alpaca_client,
&config.alpaca_rate_limiter, &config.alpaca_rate_limiter,
&api::outgoing::order::Order { &alpaca::api::outgoing::order::Order {
status: Some(api::outgoing::order::Status::All), status: Some(alpaca::api::outgoing::order::Status::All),
limit: Some(500), limit: Some(500),
after: Some(after), after: Some(after),
until: None, until: None,
@@ -70,7 +70,7 @@ pub async fn rehydrate_orders(config: &Arc<Config>) {
let orders = orders let orders = orders
.into_iter() .into_iter()
.flat_map(&api::incoming::order::Order::normalize) .flat_map(&alpaca::api::incoming::order::Order::normalize)
.collect::<Vec<_>>(); .collect::<Vec<_>>();
database::orders::upsert_batch(&config.clickhouse_client, &orders) database::orders::upsert_batch(&config.clickhouse_client, &orders)

20
src/threads/data/backfill.rs
View File

@@ -7,7 +7,7 @@ use crate::{
database, database,
types::{ types::{
alpaca::{ alpaca::{
api, self,
shared::{Sort, Source}, shared::{Sort, Source},
}, },
news::Prediction, news::Prediction,
@@ -167,7 +167,7 @@ struct BarHandler {
fetch_from: OffsetDateTime, fetch_from: OffsetDateTime,
fetch_to: OffsetDateTime, fetch_to: OffsetDateTime,
next_page_token: Option<String>, next_page_token: Option<String>,
) -> api::outgoing::bar::Bar, ) -> alpaca::api::outgoing::bar::Bar,
} }
fn us_equity_query_constructor( fn us_equity_query_constructor(
@@ -175,14 +175,14 @@ fn us_equity_query_constructor(
fetch_from: OffsetDateTime, fetch_from: OffsetDateTime,
fetch_to: OffsetDateTime, fetch_to: OffsetDateTime,
next_page_token: Option<String>, next_page_token: Option<String>,
) -> api::outgoing::bar::Bar { ) -> alpaca::api::outgoing::bar::Bar {
api::outgoing::bar::Bar::UsEquity { alpaca::api::outgoing::bar::Bar::UsEquity {
symbols: vec![symbol], symbols: vec![symbol],
timeframe: ONE_MINUTE, timeframe: ONE_MINUTE,
start: Some(fetch_from), start: Some(fetch_from),
end: Some(fetch_to), end: Some(fetch_to),
limit: Some(10000), limit: Some(10000),
adjustment: None, adjustment: Some(alpaca::api::outgoing::bar::Adjustment::All),
asof: None, asof: None,
feed: Some(*ALPACA_SOURCE), feed: Some(*ALPACA_SOURCE),
currency: None, currency: None,
@@ -196,8 +196,8 @@ fn crypto_query_constructor(
fetch_from: OffsetDateTime, fetch_from: OffsetDateTime,
fetch_to: OffsetDateTime, fetch_to: OffsetDateTime,
next_page_token: Option<String>, next_page_token: Option<String>,
) -> api::outgoing::bar::Bar { ) -> alpaca::api::outgoing::bar::Bar {
api::outgoing::bar::Bar::Crypto { alpaca::api::outgoing::bar::Bar::Crypto {
symbols: vec![symbol], symbols: vec![symbol],
timeframe: ONE_MINUTE, timeframe: ONE_MINUTE,
start: Some(fetch_from), start: Some(fetch_from),
@@ -241,7 +241,7 @@ impl Handler for BarHandler {
let mut next_page_token = None; let mut next_page_token = None;
loop { loop {
let Ok(message) = api::incoming::bar::get_historical( let Ok(message) = alpaca::api::incoming::bar::get_historical(
&self.config.alpaca_client, &self.config.alpaca_client,
&self.config.alpaca_rate_limiter, &self.config.alpaca_rate_limiter,
self.data_url, self.data_url,
@@ -328,10 +328,10 @@ impl Handler for NewsHandler {
let mut next_page_token = None; let mut next_page_token = None;
loop { loop {
let Ok(message) = api::incoming::news::get_historical( let Ok(message) = alpaca::api::incoming::news::get_historical(
&self.config.alpaca_client, &self.config.alpaca_client,
&self.config.alpaca_rate_limiter, &self.config.alpaca_rate_limiter,
&api::outgoing::news::News { &alpaca::api::outgoing::news::News {
symbols: vec![symbol.clone()], symbols: vec![symbol.clone()],
start: Some(fetch_from), start: Some(fetch_from),
end: Some(fetch_to), end: Some(fetch_to),

4
src/types/alpaca/api/incoming/account.rs
View File

@@ -95,7 +95,9 @@ pub async fn get(
.await? .await?
.error_for_status() .error_for_status()
.map_err(|e| match e.status() { .map_err(|e| match e.status() {
Some(reqwest::StatusCode::FORBIDDEN) => backoff::Error::Permanent(e), Some(reqwest::StatusCode::BAD_REQUEST | reqwest::StatusCode::FORBIDDEN) => {
backoff::Error::Permanent(e)
}
_ => e.into(), _ => e.into(),
})? })?
.json::<Account>() .json::<Account>()

8
src/types/alpaca/api/incoming/asset.rs
View File

@@ -63,9 +63,11 @@ pub async fn get_by_symbol(
.await? .await?
.error_for_status() .error_for_status()
.map_err(|e| match e.status() { .map_err(|e| match e.status() {
Some(reqwest::StatusCode::FORBIDDEN | reqwest::StatusCode::NOT_FOUND) => { Some(
backoff::Error::Permanent(e) reqwest::StatusCode::BAD_REQUEST
} | reqwest::StatusCode::FORBIDDEN
| reqwest::StatusCode::NOT_FOUND,
) => backoff::Error::Permanent(e),
_ => e.into(), _ => e.into(),
})? })?
.json::<Asset>() .json::<Asset>()

4
src/types/alpaca/api/incoming/bar.rs
View File

@@ -68,7 +68,9 @@ pub async fn get_historical(
.await? .await?
.error_for_status() .error_for_status()
.map_err(|e| match e.status() { .map_err(|e| match e.status() {
Some(reqwest::StatusCode::FORBIDDEN) => backoff::Error::Permanent(e), Some(reqwest::StatusCode::BAD_REQUEST | reqwest::StatusCode::FORBIDDEN) => {
backoff::Error::Permanent(e)
}
_ => e.into(), _ => e.into(),
})? })?
.json::<Message>() .json::<Message>()

4
src/types/alpaca/api/incoming/clock.rs
View File

@@ -33,7 +33,9 @@ pub async fn get(
.await? .await?
.error_for_status() .error_for_status()
.map_err(|e| match e.status() { .map_err(|e| match e.status() {
Some(reqwest::StatusCode::FORBIDDEN) => backoff::Error::Permanent(e), Some(reqwest::StatusCode::BAD_REQUEST | reqwest::StatusCode::FORBIDDEN) => {
backoff::Error::Permanent(e)
}
_ => e.into(), _ => e.into(),
})? })?
.json::<Clock>() .json::<Clock>()

8
src/types/alpaca/api/incoming/news.rs
View File

@@ -88,7 +88,13 @@ pub async fn get_historical(
.query(query) .query(query)
.send() .send()
.await? .await?
.error_for_status()? .error_for_status()
.map_err(|e| match e.status() {
Some(reqwest::StatusCode::BAD_REQUEST | reqwest::StatusCode::FORBIDDEN) => {
backoff::Error::Permanent(e)
}
_ => e.into(),
})?
.json::<Message>() .json::<Message>()
.await .await
.map_err(backoff::Error::Permanent) .map_err(backoff::Error::Permanent)

4
src/types/alpaca/api/incoming/order.rs
View File

@@ -27,7 +27,9 @@ pub async fn get(
.await? .await?
.error_for_status() .error_for_status()
.map_err(|e| match e.status() { .map_err(|e| match e.status() {
Some(reqwest::StatusCode::FORBIDDEN) => backoff::Error::Permanent(e), Some(reqwest::StatusCode::BAD_REQUEST | reqwest::StatusCode::FORBIDDEN) => {
backoff::Error::Permanent(e)
}
_ => e.into(), _ => e.into(),
})? })?
.json::<Vec<Order>>() .json::<Vec<Order>>()

8
src/types/alpaca/api/incoming/position.rs
View File

@@ -81,7 +81,9 @@ pub async fn get(
.await? .await?
.error_for_status() .error_for_status()
.map_err(|e| match e.status() { .map_err(|e| match e.status() {
Some(reqwest::StatusCode::FORBIDDEN) => backoff::Error::Permanent(e), Some(reqwest::StatusCode::BAD_REQUEST | reqwest::StatusCode::FORBIDDEN) => {
backoff::Error::Permanent(e)
}
_ => e.into(), _ => e.into(),
})? })?
.json::<Vec<Position>>() .json::<Vec<Position>>()
@@ -121,7 +123,9 @@ pub async fn get_by_symbol(
response response
.error_for_status() .error_for_status()
.map_err(|e| match e.status() { .map_err(|e| match e.status() {
Some(reqwest::StatusCode::FORBIDDEN) => backoff::Error::Permanent(e), Some(reqwest::StatusCode::BAD_REQUEST | reqwest::StatusCode::FORBIDDEN) => {
backoff::Error::Permanent(e)
}
_ => e.into(), _ => e.into(),
})? })?
.json::<Position>() .json::<Position>()

1
src/types/alpaca/api/outgoing/bar.rs
View File

@@ -7,6 +7,7 @@ use std::time::Duration;
use time::OffsetDateTime; use time::OffsetDateTime;
#[derive(Serialize)] #[derive(Serialize)]
#[serde(rename_all = "snake_case")]
#[allow(dead_code)] #[allow(dead_code)]
pub enum Adjustment { pub enum Adjustment {
Raw, Raw,