Add shared lib

Signed-off-by: Nikolaos Karaolidis <nick@karaolidis.com>

Cargo.toml

@@ -3,6 +3,14 @@ name    = "qrust"
 version = "0.1.0"
 edition = "2021"
 
+[lib]
+name = "qrust"
+path = "src/lib/mod.rs"
+
+[[bin]]
+name = "qrust"
+path = "src/main.rs"
+
 [profile.release]
 panic         = 'abort'
 strip         = true

src/config.rs

@@ -1,6 +1,6 @@
-use crate::types::alpaca::shared::{Mode, Source};
 use governor::{DefaultDirectRateLimiter, Quota, RateLimiter};
 use lazy_static::lazy_static;
+use qrust::types::alpaca::shared::{Mode, Source};
 use reqwest::{
     header::{HeaderMap, HeaderName, HeaderValue},
     Client,
@@ -15,42 +15,23 @@ use rust_bert::{
 use std::{env, num::NonZeroU32, path::PathBuf, sync::Arc};
 use tokio::sync::{Mutex, Semaphore};
 
-pub const ALPACA_STOCK_DATA_API_URL: &str = "https://data.alpaca.markets/v2/stocks/bars";
-pub const ALPACA_CRYPTO_DATA_API_URL: &str = "https://data.alpaca.markets/v1beta3/crypto/us/bars";
-pub const ALPACA_NEWS_DATA_API_URL: &str = "https://data.alpaca.markets/v1beta1/news";
-
-pub const ALPACA_STOCK_DATA_WEBSOCKET_URL: &str = "wss://stream.data.alpaca.markets/v2";
-pub const ALPACA_CRYPTO_DATA_WEBSOCKET_URL: &str =
-    "wss://stream.data.alpaca.markets/v1beta3/crypto/us";
-pub const ALPACA_NEWS_DATA_WEBSOCKET_URL: &str = "wss://stream.data.alpaca.markets/v1beta1/news";
-
 lazy_static! {
     pub static ref ALPACA_MODE: Mode = env::var("ALPACA_MODE")
         .expect("ALPACA_MODE must be set.")
         .parse()
         .expect("ALPACA_MODE must be 'live' or 'paper'");
+    pub static ref ALPACA_API_BASE: String = match *ALPACA_MODE {
+        Mode::Live => String::from("api"),
+        Mode::Paper => String::from("paper-api"),
+    };
     pub static ref ALPACA_SOURCE: Source = env::var("ALPACA_SOURCE")
         .expect("ALPACA_SOURCE must be set.")
         .parse()
         .expect("ALPACA_SOURCE must be 'iex', 'sip', or 'otc'");
-    pub static ref ALPACA_API_KEY: String = env::var("ALPACA_API_KEY").expect("ALPACA_API_KEY must be set.");
-    pub static ref ALPACA_API_SECRET: String = env::var("ALPACA_API_SECRET").expect("ALPACA_API_SECRET must be set.");
-    #[derive(Debug)]
-    pub static ref ALPACA_API_URL: String = format!(
-        "https://{}.alpaca.markets/v2",
-        match *ALPACA_MODE {
-            Mode::Live => String::from("api"),
-            Mode::Paper => String::from("paper-api"),
-        }
-    );
-    #[derive(Debug)]
-    pub static ref ALPACA_WEBSOCKET_URL: String = format!(
-        "wss://{}.alpaca.markets/stream",
-        match *ALPACA_MODE {
-            Mode::Live => String::from("api"),
-            Mode::Paper => String::from("paper-api"),
-        }
-    );
+    pub static ref ALPACA_API_KEY: String =
+        env::var("ALPACA_API_KEY").expect("ALPACA_API_KEY must be set.");
+    pub static ref ALPACA_API_SECRET: String =
+        env::var("ALPACA_API_SECRET").expect("ALPACA_API_SECRET must be set.");
     pub static ref BERT_MAX_INPUTS: usize = env::var("BERT_MAX_INPUTS")
         .expect("BERT_MAX_INPUTS must be set.")
         .parse()

src/init.rs

@@ -1,9 +1,9 @@
 use crate::{
-    config::{Config, ALPACA_MODE},
+    config::{Config, ALPACA_API_BASE},
     database,
-    types::alpaca,
 };
 use log::{info, warn};
+use qrust::types::alpaca;
 use std::{collections::HashMap, sync::Arc};
 use time::OffsetDateTime;
 use tokio::join;
@@ -13,6 +13,7 @@ pub async fn check_account(config: &Arc<Config>) {
         &config.alpaca_client,
         &config.alpaca_rate_limiter,
         None,
+        &ALPACA_API_BASE,
     )
     .await
     .unwrap();
@@ -35,7 +36,7 @@ pub async fn check_account(config: &Arc<Config>) {
 
     warn!(
         "qrust active on {} account with {} {}, avoid transferring funds without shutting down.",
-        *ALPACA_MODE, account.currency, account.cash
+        *ALPACA_API_BASE, account.currency, account.cash
     );
 }
 
@@ -54,6 +55,7 @@ pub async fn rehydrate_orders(config: &Arc<Config>) {
             ..Default::default()
         },
         None,
+        &ALPACA_API_BASE,
     )
     .await
     .ok()
@@ -87,6 +89,7 @@ pub async fn rehydrate_positions(config: &Arc<Config>) {
             &config.alpaca_client,
             &config.alpaca_rate_limiter,
             None,
+            &ALPACA_API_BASE,
         )
         .await
         .unwrap()

src/lib/mod.rs

@@ -0,0 +1,3 @@
+pub mod database;
+pub mod types;
+pub mod utils;

src/lib/types/alpaca/api/incoming/account.rs

@@ -1,4 +1,3 @@
-use crate::config::ALPACA_API_URL;
 use backoff::{future::retry_notify, ExponentialBackoff};
 use governor::DefaultDirectRateLimiter;
 use log::warn;
@@ -84,13 +83,14 @@ pub async fn get(
     client: &Client,
     rate_limiter: &DefaultDirectRateLimiter,
     backoff: Option<ExponentialBackoff>,
+    api_base: &str,
 ) -> Result<Account, Error> {
     retry_notify(
         backoff.unwrap_or_default(),
         || async {
             rate_limiter.until_ready().await;
             client
-                .get(&format!("{}/account", *ALPACA_API_URL))
+                .get(&format!("https://{}.alpaca.markets/v2/account", api_base))
                 .send()
                 .await?
                 .error_for_status()

src/lib/types/alpaca/api/incoming/asset.rs

@@ -1,12 +1,9 @@
 use super::position::Position;
-use crate::{
-    config::ALPACA_API_URL,
-    types::{
-        self,
-        alpaca::{
-            api::outgoing,
-            shared::asset::{Class, Exchange, Status},
-        },
+use crate::types::{
+    self,
+    alpaca::{
+        api::outgoing,
+        shared::asset::{Class, Exchange, Status},
     },
 };
 use backoff::{future::retry_notify, ExponentialBackoff};
@@ -57,13 +54,14 @@ pub async fn get(
     rate_limiter: &DefaultDirectRateLimiter,
     query: &outgoing::asset::Asset,
     backoff: Option<ExponentialBackoff>,
+    api_base: &str,
 ) -> Result<Vec<Asset>, Error> {
     retry_notify(
         backoff.unwrap_or_default(),
         || async {
             rate_limiter.until_ready().await;
             client
-                .get(&format!("{}/assets", *ALPACA_API_URL))
+                .get(&format!("https://{}.alpaca.markets/v2/assets", api_base))
                 .query(query)
                 .send()
                 .await?
@@ -96,13 +94,17 @@ pub async fn get_by_symbol(
     rate_limiter: &DefaultDirectRateLimiter,
     symbol: &str,
     backoff: Option<ExponentialBackoff>,
+    api_base: &str,
 ) -> Result<Asset, Error> {
     retry_notify(
         backoff.unwrap_or_default(),
         || async {
             rate_limiter.until_ready().await;
             client
-                .get(&format!("{}/assets/{}", *ALPACA_API_URL, symbol))
+                .get(&format!(
+                    "https://{}.alpaca.markets/v2/assets/{}",
+                    api_base, symbol
+                ))
                 .send()
                 .await?
                 .error_for_status()
@@ -134,10 +136,11 @@ pub async fn get_by_symbols(
     rate_limiter: &DefaultDirectRateLimiter,
     symbols: &[String],
     backoff: Option<ExponentialBackoff>,
+    api_base: &str,
 ) -> Result<Vec<Asset>, Error> {
     if symbols.len() < 2 {
         let symbol = symbols.first().unwrap();
-        let asset = get_by_symbol(client, rate_limiter, symbol, backoff).await?;
+        let asset = get_by_symbol(client, rate_limiter, symbol, backoff, api_base).await?;
         return Ok(vec![asset]);
     }
 
@@ -150,14 +153,20 @@ pub async fn get_by_symbols(
         ..Default::default()
     };
 
-    let us_equity_assets = get(client, rate_limiter, &us_equity_query, backoff_clone);
+    let us_equity_assets = get(
+        client,
+        rate_limiter,
+        &us_equity_query,
+        backoff_clone,
+        api_base,
+    );
 
     let crypto_query = outgoing::asset::Asset {
         class: Some(Class::Crypto),
         ..Default::default()
     };
 
-    let crypto_assets = get(client, rate_limiter, &crypto_query, backoff);
+    let crypto_assets = get(client, rate_limiter, &crypto_query, backoff, api_base);
 
     let (us_equity_assets, crypto_assets) = try_join!(us_equity_assets, crypto_assets)?;
 

src/lib/types/alpaca/api/incoming/bar.rs

@@ -7,6 +7,7 @@ use serde::Deserialize;
 use std::{collections::HashMap, time::Duration};
 use time::OffsetDateTime;
 
+
 #[derive(Deserialize)]
 pub struct Bar {
     #[serde(rename = "t")]

src/lib/types/alpaca/api/incoming/calendar.rs

@@ -1,5 +1,4 @@
 use crate::{
-    config::ALPACA_API_URL,
     types::{self, alpaca::api::outgoing},
     utils::{de, time::EST_OFFSET},
 };
@@ -36,13 +35,14 @@ pub async fn get(
     rate_limiter: &DefaultDirectRateLimiter,
     query: &outgoing::calendar::Calendar,
     backoff: Option<ExponentialBackoff>,
+    api_base: &str,
 ) -> Result<Vec<Calendar>, Error> {
     retry_notify(
         backoff.unwrap_or_default(),
         || async {
             rate_limiter.until_ready().await;
             client
-                .get(&format!("{}/calendar", *ALPACA_API_URL))
+                .get(&format!("https://{}.alpaca.markets/v2/calendar", api_base))
                 .query(query)
                 .send()
                 .await?

src/lib/types/alpaca/api/incoming/clock.rs

@@ -1,4 +1,3 @@
-use crate::config::ALPACA_API_URL;
 use backoff::{future::retry_notify, ExponentialBackoff};
 use governor::DefaultDirectRateLimiter;
 use log::warn;
@@ -22,13 +21,14 @@ pub async fn get(
     client: &Client,
     rate_limiter: &DefaultDirectRateLimiter,
     backoff: Option<ExponentialBackoff>,
+    api_base: &str,
 ) -> Result<Clock, Error> {
     retry_notify(
         backoff.unwrap_or_default(),
         || async {
             rate_limiter.until_ready().await;
             client
-                .get(&format!("{}/clock", *ALPACA_API_URL))
+                .get(&format!("https://{}.alpaca.markets/v2/clock", api_base))
                 .send()
                 .await?
                 .error_for_status()

src/lib/types/alpaca/api/incoming/news.rs

@@ -1,8 +1,10 @@
 use crate::{
-    config::ALPACA_NEWS_DATA_API_URL,
     types::{
         self,
-        alpaca::{api::outgoing, shared::news::normalize_html_content},
+        alpaca::{
+            api::{outgoing, ALPACA_NEWS_DATA_API_URL},
+            shared::news::normalize_html_content,
+        },
     },
     utils::de,
 };

src/lib/types/alpaca/api/incoming/order.rs

@@ -1,27 +1,24 @@
-use crate::{
-    config::ALPACA_API_URL,
-    types::alpaca::{api::outgoing, shared},
-};
+use crate::types::alpaca::{api::outgoing, shared};
 use backoff::{future::retry_notify, ExponentialBackoff};
 use governor::DefaultDirectRateLimiter;
 use log::warn;
 use reqwest::{Client, Error};
-use std::time::Duration;
-
 pub use shared::order::Order;
+use std::time::Duration;
 
 pub async fn get(
     client: &Client,
     rate_limiter: &DefaultDirectRateLimiter,
     query: &outgoing::order::Order,
     backoff: Option<ExponentialBackoff>,
+    api_base: &str,
 ) -> Result<Vec<Order>, Error> {
     retry_notify(
         backoff.unwrap_or_default(),
         || async {
             rate_limiter.until_ready().await;
             client
-                .get(&format!("{}/orders", *ALPACA_API_URL))
+                .get(&format!("https://{}.alpaca.markets/v2/orders", api_base))
                 .query(query)
                 .send()
                 .await?

src/lib/types/alpaca/api/incoming/position.rs

@@ -1,5 +1,4 @@
 use crate::{
-    config::ALPACA_API_URL,
     types::alpaca::shared::{
         self,
         asset::{Class, Exchange},
@@ -66,17 +65,20 @@ pub struct Position {
     pub asset_marginable: bool,
 }
 
+pub const ALPACA_API_URL_TEMPLATE: &str = "";
+
 pub async fn get(
     client: &Client,
     rate_limiter: &DefaultDirectRateLimiter,
     backoff: Option<ExponentialBackoff>,
+    api_base: &str,
 ) -> Result<Vec<Position>, reqwest::Error> {
     retry_notify(
         backoff.unwrap_or_default(),
         || async {
             rate_limiter.until_ready().await;
             client
-                .get(&format!("{}/positions", *ALPACA_API_URL))
+                .get(&format!("https://{}.alpaca.markets/v2/positions", api_base))
                 .send()
                 .await?
                 .error_for_status()
@@ -106,13 +108,17 @@ pub async fn get_by_symbol(
     rate_limiter: &DefaultDirectRateLimiter,
     symbol: &str,
     backoff: Option<ExponentialBackoff>,
+    api_base: &str,
 ) -> Result<Option<Position>, reqwest::Error> {
     retry_notify(
         backoff.unwrap_or_default(),
         || async {
             rate_limiter.until_ready().await;
             let response = client
-                .get(&format!("{}/positions/{}", *ALPACA_API_URL, symbol))
+                .get(&format!(
+                    "https://{}.alpaca.markets/v2/positions/{}",
+                    api_base, symbol
+                ))
                 .send()
                 .await?;
 
@@ -149,16 +155,17 @@ pub async fn get_by_symbols(
     rate_limiter: &DefaultDirectRateLimiter,
     symbols: &[String],
     backoff: Option<ExponentialBackoff>,
+    api_base: &str,
 ) -> Result<Vec<Position>, reqwest::Error> {
     if symbols.len() < 2 {
         let symbol = symbols.first().unwrap();
-        let position = get_by_symbol(client, rate_limiter, symbol, backoff).await?;
+        let position = get_by_symbol(client, rate_limiter, symbol, backoff, api_base).await?;
         return Ok(position.into_iter().collect());
     }
 
     let symbols = symbols.iter().collect::<HashSet<_>>();
 
-    let positions = get(client, rate_limiter, backoff).await?;
+    let positions = get(client, rate_limiter, backoff, api_base).await?;
 
     Ok(positions
         .into_iter()

src/lib/types/alpaca/api/mod.rs

@@ -0,0 +1,6 @@
+pub mod incoming;
+pub mod outgoing;
+
+pub const ALPACA_US_EQUITY_DATA_API_URL: &str = "https://data.alpaca.markets/v2/stocks/bars";
+pub const ALPACA_CRYPTO_DATA_API_URL: &str = "https://data.alpaca.markets/v1beta3/crypto/us/bars";
+pub const ALPACA_NEWS_DATA_API_URL: &str = "https://data.alpaca.markets/v1beta1/news";

src/lib/types/alpaca/api/outgoing/bar.rs

@@ -1,5 +1,4 @@
 use crate::{
-    config::ALPACA_SOURCE,
     types::alpaca::shared::{Sort, Source},
     utils::{ser, ONE_MINUTE},
 };
@@ -58,7 +57,7 @@ impl Default for UsEquity {
             limit: Some(MAX_LIMIT),
             adjustment: Some(Adjustment::All),
             asof: None,
-            feed: Some(*ALPACA_SOURCE),
+            feed: Some(Source::Iex),
             currency: None,
             page_token: None,
             sort: Some(Sort::Asc),

src/lib/types/alpaca/mod.rs

@@ -1,3 +1,4 @@
 pub mod api;
 pub mod shared;
 pub mod websocket;
+

src/lib/types/alpaca/websocket/data/mod.rs

@@ -1,10 +1,7 @@
 pub mod incoming;
 pub mod outgoing;
 
-use crate::{
-    config::{ALPACA_API_KEY, ALPACA_API_SECRET},
-    types::alpaca::websocket,
-};
+use crate::types::alpaca::websocket;
 use core::panic;
 use futures_util::{
     stream::{SplitSink, SplitStream},
@@ -17,6 +14,8 @@ use tokio_tungstenite::{tungstenite::Message, MaybeTlsStream, WebSocketStream};
 pub async fn authenticate(
     sink: &mut SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>,
     stream: &mut SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
+    api_key: String,
+    api_secret: String,
 ) {
     match stream.next().await.unwrap().unwrap() {
         Message::Text(data)
@@ -32,8 +31,8 @@ pub async fn authenticate(
     sink.send(Message::Text(
         to_string(&websocket::data::outgoing::Message::Auth(
             websocket::auth::Message {
-                key: (*ALPACA_API_KEY).clone(),
-                secret: (*ALPACA_API_SECRET).clone(),
+                key: api_key,
+                secret: api_secret,
             },
         ))
         .unwrap(),

src/lib/types/alpaca/websocket/mod.rs

@@ -0,0 +1,8 @@
+pub mod auth;
+pub mod data;
+pub mod trading;
+
+pub const ALPACA_US_EQUITY_DATA_WEBSOCKET_URL: &str = "wss://stream.data.alpaca.markets/v2";
+pub const ALPACA_CRYPTO_DATA_WEBSOCKET_URL: &str =
+    "wss://stream.data.alpaca.markets/v1beta3/crypto/us";
+pub const ALPACA_NEWS_DATA_WEBSOCKET_URL: &str = "wss://stream.data.alpaca.markets/v1beta1/news";

src/lib/types/alpaca/websocket/trading/mod.rs

@@ -1,10 +1,7 @@
 pub mod incoming;
 pub mod outgoing;
 
-use crate::{
-    config::{ALPACA_API_KEY, ALPACA_API_SECRET},
-    types::alpaca::websocket,
-};
+use crate::types::alpaca::websocket;
 use core::panic;
 use futures_util::{
     stream::{SplitSink, SplitStream},
@@ -17,12 +14,14 @@ use tokio_tungstenite::{tungstenite::Message, MaybeTlsStream, WebSocketStream};
 pub async fn authenticate(
     sink: &mut SplitSink<WebSocketStream<MaybeTlsStream<TcpStream>>, Message>,
     stream: &mut SplitStream<WebSocketStream<MaybeTlsStream<TcpStream>>>,
+    api_key: String,
+    api_secret: String,
 ) {
     sink.send(Message::Text(
         to_string(&websocket::trading::outgoing::Message::Auth(
             websocket::auth::Message {
-                key: (*ALPACA_API_KEY).clone(),
-                secret: (*ALPACA_API_SECRET).clone(),
+                key: api_key,
+                secret: api_secret,
             },
         ))
         .unwrap(),

src/main.rs

@@ -3,16 +3,14 @@
 #![feature(hash_extract_if)]
 
 mod config;
-mod database;
 mod init;
 mod routes;
 mod threads;
-mod types;
-mod utils;
 
 use config::Config;
 use dotenv::dotenv;
 use log4rs::config::Deserializers;
+use qrust::{create_send_await, database};
 use tokio::{join, spawn, sync::mpsc, try_join};
 
 #[tokio::main]

src/routes/assets.rs

@@ -1,10 +1,10 @@
 use crate::{
-    config::Config,
+    config::{Config, ALPACA_API_BASE},
     create_send_await, database, threads,
-    types::{alpaca, Asset},
 };
 use axum::{extract::Path, Extension, Json};
 use http::StatusCode;
+use qrust::types::{alpaca, Asset};
 use serde::{Deserialize, Serialize};
 use std::{
     collections::{HashMap, HashSet},
@@ -74,6 +74,7 @@ pub async fn add(
         &config.alpaca_rate_limiter,
         &request.symbols,
         None,
+        &ALPACA_API_BASE,
     )
     .await
     .map_err(|e| {
@@ -147,6 +148,7 @@ pub async fn add_symbol(
         &config.alpaca_rate_limiter,
         &symbol,
         None,
+        &ALPACA_API_BASE,
     )
     .await
     .map_err(|e| {

src/threads/clock.rs

@@ -1,10 +1,12 @@
 use crate::{
-    config::Config,
+    config::{Config, ALPACA_API_BASE},
     database,
+};
+use log::info;
+use qrust::{
     types::{alpaca, Calendar},
     utils::{backoff, duration_until},
 };
-use log::info;
 use std::sync::Arc;
 use time::OffsetDateTime;
 use tokio::{join, sync::mpsc, time::sleep};
@@ -42,6 +44,7 @@ pub async fn run(config: Arc<Config>, sender: mpsc::Sender<Message>) {
                 &config.alpaca_client,
                 &config.alpaca_rate_limiter,
                 Some(backoff::infinite()),
+                &ALPACA_API_BASE,
             )
             .await
             .unwrap()
@@ -53,6 +56,7 @@ pub async fn run(config: Arc<Config>, sender: mpsc::Sender<Message>) {
                 &config.alpaca_rate_limiter,
                 &alpaca::api::outgoing::calendar::Calendar::default(),
                 Some(backoff::infinite()),
+                &ALPACA_API_BASE,
             )
             .await
             .unwrap()

src/threads/data/backfill/bars.rs

@@ -2,6 +2,11 @@ use super::Job;
 use crate::{
     config::{Config, ALPACA_SOURCE},
     database,
+};
+use async_trait::async_trait;
+use itertools::{Either, Itertools};
+use log::{error, info};
+use qrust::{
     types::{
         alpaca::{
             self,
@@ -11,9 +16,6 @@ use crate::{
     },
     utils::{duration_until, FIFTEEN_MINUTES, ONE_MINUTE},
 };
-use async_trait::async_trait;
-use itertools::{Either, Itertools};
-use log::{error, info};
 use std::{collections::HashMap, sync::Arc};
 use time::OffsetDateTime;
 use tokio::time::sleep;
@@ -41,6 +43,7 @@ pub fn us_equity_query_constructor(
         end: Some(fetch_to),
         page_token: next_page_token,
         sort: Some(Sort::Asc),
+        feed: Some(*ALPACA_SOURCE),
         ..Default::default()
     })
 }

src/threads/data/backfill/mod.rs

@@ -2,14 +2,17 @@ mod bars;
 mod news;
 
 use super::ThreadType;
-use crate::{
-    config::{Config, ALPACA_CRYPTO_DATA_API_URL, ALPACA_STOCK_DATA_API_URL},
-    types::{Backfill, Class},
-    utils::{last_minute, ONE_SECOND},
-};
+use crate::config::Config;
 use async_trait::async_trait;
 use itertools::Itertools;
 use log::{info, warn};
+use qrust::{
+    types::{
+        alpaca::api::{ALPACA_CRYPTO_DATA_API_URL, ALPACA_US_EQUITY_DATA_API_URL},
+        Backfill, Class,
+    },
+    utils::{last_minute, ONE_SECOND},
+};
 use std::{collections::HashMap, sync::Arc};
 use time::OffsetDateTime;
 use tokio::{
@@ -224,7 +227,7 @@ pub fn create_handler(thread_type: ThreadType, config: Arc<Config>) -> Box<dyn H
     match thread_type {
         ThreadType::Bars(Class::UsEquity) => Box::new(bars::Handler {
             config,
-            data_url: ALPACA_STOCK_DATA_API_URL,
+            data_url: ALPACA_US_EQUITY_DATA_API_URL,
             api_query_constructor: bars::us_equity_query_constructor,
         }),
         ThreadType::Bars(Class::Crypto) => Box::new(bars::Handler {

src/threads/data/backfill/news.rs

@@ -2,6 +2,12 @@ use super::Job;
 use crate::{
     config::{Config, ALPACA_SOURCE, BERT_MAX_INPUTS},
     database,
+};
+use async_trait::async_trait;
+use futures_util::future::join_all;
+use itertools::{Either, Itertools};
+use log::{error, info};
+use qrust::{
     types::{
         alpaca::{
             self,
@@ -12,10 +18,6 @@ use crate::{
     },
     utils::{duration_until, FIFTEEN_MINUTES, ONE_MINUTE},
 };
-use async_trait::async_trait;
-use futures_util::future::join_all;
-use itertools::{Either, Itertools};
-use log::{error, info};
 use std::{collections::HashMap, sync::Arc};
 use tokio::{task::block_in_place, time::sleep};
 

src/threads/data/mod.rs

@@ -3,16 +3,22 @@ mod websocket;
 
 use super::clock;
 use crate::{
-    config::{
-        Config, ALPACA_CRYPTO_DATA_WEBSOCKET_URL, ALPACA_NEWS_DATA_WEBSOCKET_URL, ALPACA_SOURCE,
-        ALPACA_STOCK_DATA_WEBSOCKET_URL,
-    },
+    config::{Config, ALPACA_API_BASE, ALPACA_API_KEY, ALPACA_API_SECRET, ALPACA_SOURCE},
     create_send_await, database,
-    types::{alpaca, Asset, Class},
 };
 use futures_util::StreamExt;
 use itertools::{Either, Itertools};
 use log::error;
+use qrust::types::{
+    alpaca::{
+        self,
+        websocket::{
+            ALPACA_CRYPTO_DATA_WEBSOCKET_URL, ALPACA_NEWS_DATA_WEBSOCKET_URL,
+            ALPACA_US_EQUITY_DATA_WEBSOCKET_URL,
+        },
+    },
+    Asset, Class,
+};
 use std::{collections::HashMap, sync::Arc};
 use tokio::{
     join, select, spawn,
@@ -103,7 +109,7 @@ async fn init_thread(
 ) {
     let websocket_url = match thread_type {
         ThreadType::Bars(Class::UsEquity) => {
-            format!("{}/{}", ALPACA_STOCK_DATA_WEBSOCKET_URL, *ALPACA_SOURCE)
+            format!("{}/{}", ALPACA_US_EQUITY_DATA_WEBSOCKET_URL, *ALPACA_SOURCE)
         }
         ThreadType::Bars(Class::Crypto) => ALPACA_CRYPTO_DATA_WEBSOCKET_URL.into(),
         ThreadType::News => ALPACA_NEWS_DATA_WEBSOCKET_URL.into(),
@@ -111,7 +117,13 @@ async fn init_thread(
 
     let (websocket, _) = connect_async(websocket_url).await.unwrap();
     let (mut websocket_sink, mut websocket_stream) = websocket.split();
-    alpaca::websocket::data::authenticate(&mut websocket_sink, &mut websocket_stream).await;
+    alpaca::websocket::data::authenticate(
+        &mut websocket_sink,
+        &mut websocket_stream,
+        (*ALPACA_API_KEY).to_string(),
+        (*ALPACA_API_SECRET).to_string(),
+    )
+    .await;
 
     let (backfill_sender, backfill_receiver) = mpsc::channel(100);
     spawn(backfill::run(
@@ -207,6 +219,7 @@ async fn handle_message(
                     &config.alpaca_rate_limiter,
                     &symbols,
                     None,
+                    &ALPACA_API_BASE,
                 )
                 .await
                 .unwrap()
@@ -221,6 +234,7 @@ async fn handle_message(
                     &config.alpaca_rate_limiter,
                     &symbols,
                     None,
+                    &ALPACA_API_BASE,
                 )
                 .await
                 .unwrap()

src/threads/data/websocket/bars.rs

@@ -1,11 +1,8 @@
 use super::Pending;
-use crate::{
-    config::Config,
-    database,
-    types::{alpaca::websocket, Bar},
-};
+use crate::{config::Config, database};
 use async_trait::async_trait;
 use log::{debug, error, info};
+use qrust::types::{alpaca::websocket, Bar};
 use std::{collections::HashMap, sync::Arc};
 use tokio::sync::RwLock;
 

src/threads/data/websocket/mod.rs

@@ -2,10 +2,7 @@ mod bars;
 mod news;
 
 use super::ThreadType;
-use crate::{
-    config::Config,
-    types::{alpaca::websocket, Class},
-};
+use crate::config::Config;
 use async_trait::async_trait;
 use futures_util::{
     future::join_all,
@@ -13,6 +10,7 @@ use futures_util::{
     SinkExt, StreamExt,
 };
 use log::error;
+use qrust::types::{alpaca::websocket, Class};
 use serde_json::{from_str, to_string};
 use std::{collections::HashMap, sync::Arc};
 use tokio::{

src/threads/data/websocket/news.rs

@@ -1,11 +1,8 @@
 use super::Pending;
-use crate::{
-    config::Config,
-    database,
-    types::{alpaca::websocket, news::Prediction, News},
-};
+use crate::{config::Config, database};
 use async_trait::async_trait;
 use log::{debug, error, info};
+use qrust::types::{alpaca::websocket, news::Prediction, News};
 use std::{collections::HashMap, sync::Arc};
 use tokio::{sync::RwLock, task::block_in_place};
 

src/threads/trading/mod.rs

@@ -1,19 +1,26 @@
 mod websocket;
 
-use crate::{
-    config::{Config, ALPACA_WEBSOCKET_URL},
-    types::alpaca,
-};
+use crate::config::{Config, ALPACA_API_BASE, ALPACA_API_KEY, ALPACA_API_SECRET};
 use futures_util::StreamExt;
+use qrust::types::alpaca;
 use std::sync::Arc;
 use tokio::spawn;
 use tokio_tungstenite::connect_async;
 
 pub async fn run(config: Arc<Config>) {
-    let (websocket, _) = connect_async(&*ALPACA_WEBSOCKET_URL).await.unwrap();
+    let (websocket, _) =
+        connect_async(&format!("wss://{}.alpaca.markets/stream", *ALPACA_API_BASE))
+            .await
+            .unwrap();
     let (mut websocket_sink, mut websocket_stream) = websocket.split();
 
-    alpaca::websocket::trading::authenticate(&mut websocket_sink, &mut websocket_stream).await;
+    alpaca::websocket::trading::authenticate(
+        &mut websocket_sink,
+        &mut websocket_stream,
+        (*ALPACA_API_KEY).to_string(),
+        (*ALPACA_API_SECRET).to_string(),
+    )
+    .await;
     alpaca::websocket::trading::subscribe(&mut websocket_sink, &mut websocket_stream).await;
 
     spawn(websocket::run(config, websocket_stream));

src/threads/trading/websocket.rs

@@ -1,10 +1,7 @@
-use crate::{
-    config::Config,
-    database,
-    types::{alpaca::websocket, Order},
-};
+use crate::{config::Config, database};
 use futures_util::{stream::SplitStream, StreamExt};
 use log::{debug, error};
+use qrust::types::{alpaca::websocket, Order};
 use serde_json::from_str;
 use std::sync::Arc;
 use tokio::{net::TcpStream, spawn};

src/types/alpaca/api/mod.rs

@@ -1,2 +0,0 @@
-pub mod incoming;
-pub mod outgoing;

src/types/alpaca/websocket/mod.rs

@@ -1,3 +0,0 @@
-pub mod auth;
-pub mod data;
-pub mod trading;