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

Update random bits and bobs

Browse Source 
Signed-off-by: Nikolaos Karaolidis <nick@karaolidis.com>
This commit is contained in:
Nick Karaolidis
2024-02-15 01:09:16 +00:00
parent 4b194e168f
commit cdaa2d20a9
20 changed files with 292 additions and 283 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,6 +1,9 @@
use super::ThreadType;
use crate::{
config::{Config, ALPACA_CRYPTO_DATA_API_URL, ALPACA_STOCK_DATA_API_URL},
config::{
Config, ALPACA_CRYPTO_DATA_API_URL, ALPACA_SOURCE, ALPACA_STOCK_DATA_API_URL,
MAX_BERT_INPUTS,
},
database,
types::{
alpaca::{
@@ -30,23 +33,24 @@ pub enum Action {
Purge,
}
impl From<super::Action> for Action {
impl From<super::Action> for Option<Action> {
fn from(action: super::Action) -> Self {
match action {
super::Action::Add => Self::Backfill,
super::Action::Remove => Self::Purge,
super::Action::Add | super::Action::Enable => Some(Action::Backfill),
super::Action::Remove => Some(Action::Purge),
super::Action::Disable => None,
}
}
}
pub struct Message {
pub action: Action,
pub action: Option<Action>,
pub symbols: Vec<String>,
pub response: oneshot::Sender<()>,
}
impl Message {
pub fn new(action: Action, symbols: Vec<String>) -> (Self, oneshot::Receiver<()>) {
pub fn new(action: Option<Action>, symbols: Vec<String>) -> (Self, oneshot::Receiver<()>) {
let (sender, receiver) = oneshot::channel::<()>();
(
Self {
@@ -77,7 +81,6 @@ pub async fn run(handler: Arc<Box<dyn Handler>>, mut receiver: mpsc::Receiver<Me
loop {
let message = receiver.recv().await.unwrap();
spawn(handle_backfill_message(
handler.clone(),
backfill_jobs.clone(),
@@ -94,7 +97,7 @@ async fn handle_backfill_message(
let mut backfill_jobs = backfill_jobs.lock().await;
match message.action {
Action::Backfill => {
Some(Action::Backfill) => {
let log_string = handler.log_string();
for symbol in message.symbols {
@@ -134,7 +137,7 @@ async fn handle_backfill_message(
);
}
}
Action::Purge => {
Some(Action::Purge) => {
for symbol in &message.symbols {
if let Some(job) = backfill_jobs.remove(symbol) {
if !job.is_finished() {
@@ -150,6 +153,7 @@ async fn handle_backfill_message(
)
.unwrap();
}
None => {}
}
message.response.send(()).unwrap();
@@ -159,7 +163,6 @@ struct BarHandler {
config: Arc<Config>,
data_url: &'static str,
api_query_constructor: fn(
config: &Arc<Config>,
symbol: String,
fetch_from: OffsetDateTime,
fetch_to: OffsetDateTime,
@@ -168,7 +171,6 @@ struct BarHandler {
}
fn us_equity_query_constructor(
config: &Arc<Config>,
symbol: String,
fetch_from: OffsetDateTime,
fetch_to: OffsetDateTime,
@@ -182,7 +184,7 @@ fn us_equity_query_constructor(
limit: Some(10000),
adjustment: None,
asof: None,
feed: Some(config.alpaca_source),
feed: Some(*ALPACA_SOURCE),
currency: None,
page_token: next_page_token,
sort: Some(Sort::Asc),
@@ -190,7 +192,6 @@ fn us_equity_query_constructor(
}
fn crypto_query_constructor(
_: &Arc<Config>,
symbol: String,
fetch_from: OffsetDateTime,
fetch_to: OffsetDateTime,
@@ -226,7 +227,7 @@ impl Handler for BarHandler {
}
async fn queue_backfill(&self, symbol: &str, fetch_to: OffsetDateTime) {
if self.config.alpaca_source == Source::Iex {
if *ALPACA_SOURCE == Source::Iex {
let run_delay = duration_until(fetch_to + FIFTEEN_MINUTES + ONE_MINUTE);
info!("Queing bar backfill for {} in {:?}.", symbol, run_delay);
sleep(run_delay).await;
@@ -241,10 +242,10 @@ impl Handler for BarHandler {
loop {
let Ok(message) = api::incoming::bar::get_historical(
&self.config,
&self.config.alpaca_client,
&self.config.alpaca_rate_limiter,
self.data_url,
&(self.api_query_constructor)(
&self.config,
symbol.clone(),
fetch_from,
fetch_to,
@@ -328,7 +329,8 @@ impl Handler for NewsHandler {
loop {
let Ok(message) = api::incoming::news::get_historical(
&self.config,
&self.config.alpaca_client,
&self.config.alpaca_rate_limiter,
&api::outgoing::news::News {
symbols: vec![symbol.clone()],
start: Some(fetch_from),
@@ -367,18 +369,15 @@ impl Handler for NewsHandler {
.map(|news| format!("{}\n\n{}", news.headline, news.content))
.collect::<Vec<_>>();
let predictions = join_all(inputs.chunks(self.config.max_bert_inputs).map(|inputs| {
let sequence_classifier = self.config.sequence_classifier.clone();
async move {
let sequence_classifier = sequence_classifier.lock().await;
block_in_place(|| {
sequence_classifier
.predict(inputs.iter().map(String::as_str).collect::<Vec<_>>())
.into_iter()
.map(|label| Prediction::try_from(label).unwrap())
.collect::<Vec<_>>()
})
}
let predictions = join_all(inputs.chunks(*MAX_BERT_INPUTS).map(|inputs| async move {
let sequence_classifier = self.config.sequence_classifier.lock().await;
block_in_place(|| {
sequence_classifier
.predict(inputs.iter().map(String::as_str).collect::<Vec<_>>())
.into_iter()
.map(|label| Prediction::try_from(label).unwrap())
.collect::<Vec<_>>()
})
}))
.await
.into_iter()

26
src/threads/data/mod.rs
View File

@@ -4,7 +4,7 @@ mod websocket;
use super::clock;
use crate::{
config::{
Config, ALPACA_CRYPTO_DATA_WEBSOCKET_URL, ALPACA_NEWS_DATA_WEBSOCKET_URL,
Config, ALPACA_CRYPTO_DATA_WEBSOCKET_URL, ALPACA_NEWS_DATA_WEBSOCKET_URL, ALPACA_SOURCE,
ALPACA_STOCK_DATA_WEBSOCKET_URL,
},
create_send_await, database,
@@ -21,9 +21,12 @@ use tokio::{
use tokio_tungstenite::connect_async;
#[derive(Clone, Copy)]
#[allow(dead_code)]
pub enum Action {
Add,
Enable,
Remove,
Disable,
}
pub struct Message {
@@ -100,10 +103,7 @@ async fn init_thread(
) {
let websocket_url = match thread_type {
ThreadType::Bars(Class::UsEquity) => {
format!(
"{}/{}",
ALPACA_STOCK_DATA_WEBSOCKET_URL, &config.alpaca_source
)
format!("{}/{}", ALPACA_STOCK_DATA_WEBSOCKET_URL, *ALPACA_SOURCE)
}
ThreadType::Bars(Class::Crypto) => ALPACA_CRYPTO_DATA_WEBSOCKET_URL.into(),
ThreadType::News => ALPACA_NEWS_DATA_WEBSOCKET_URL.into(),
@@ -111,8 +111,7 @@ 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(&config, &mut websocket_sink, &mut websocket_stream)
.await;
alpaca::websocket::data::authenticate(&mut websocket_sink, &mut websocket_stream).await;
let (backfill_sender, backfill_receiver) = mpsc::channel(100);
spawn(backfill::run(
@@ -223,7 +222,8 @@ async fn handle_message(
async move {
let asset_future = async {
alpaca::api::incoming::asset::get_by_symbol(
&config,
&config.alpaca_client,
&config.alpaca_rate_limiter,
&symbol,
Some(backoff::infinite()),
)
@@ -233,7 +233,8 @@ async fn handle_message(
let position_future = async {
alpaca::api::incoming::position::get_by_symbol(
&config,
&config.alpaca_rate_limiter,
&config.alpaca_client,
&symbol,
Some(backoff::infinite()),
)
@@ -256,6 +257,7 @@ async fn handle_message(
.await
.unwrap();
}
_ => {}
}
message.response.send(()).unwrap();
@@ -292,7 +294,7 @@ async fn handle_clock_message(
create_send_await!(
bars_us_equity_backfill_sender,
backfill::Message::new,
backfill::Action::Backfill,
Some(backfill::Action::Backfill),
us_equity_symbols.clone()
);
};
@@ -301,7 +303,7 @@ async fn handle_clock_message(
create_send_await!(
bars_crypto_backfill_sender,
backfill::Message::new,
backfill::Action::Backfill,
Some(backfill::Action::Backfill),
crypto_symbols.clone()
);
};
@@ -310,7 +312,7 @@ async fn handle_clock_message(
create_send_await!(
news_backfill_sender,
backfill::Message::new,
backfill::Action::Backfill,
Some(backfill::Action::Backfill),
symbols
);
};

15
src/threads/data/websocket.rs
View File

@@ -26,23 +26,23 @@ pub enum Action {
Unsubscribe,
}
impl From<super::Action> for Action {
impl From<super::Action> for Option<Action> {
fn from(action: super::Action) -> Self {
match action {
super::Action::Add => Self::Subscribe,
super::Action::Remove => Self::Unsubscribe,
super::Action::Add | super::Action::Enable => Some(Action::Subscribe),
super::Action::Remove | super::Action::Disable => Some(Action::Unsubscribe),
}
}
}
pub struct Message {
pub action: Action,
pub action: Option<Action>,
pub symbols: Vec<String>,
pub response: oneshot::Sender<()>,
}
impl Message {
pub fn new(action: Action, symbols: Vec<String>) -> (Self, oneshot::Receiver<()>) {
pub fn new(action: Option<Action>, symbols: Vec<String>) -> (Self, oneshot::Receiver<()>) {
let (sender, receiver) = oneshot::channel();
(
Self {
@@ -115,7 +115,7 @@ async fn handle_message(
message: Message,
) {
match message.action {
Action::Subscribe => {
Some(Action::Subscribe) => {
let (pending_subscriptions, receivers): (Vec<_>, Vec<_>) = message
.symbols
.iter()
@@ -144,7 +144,7 @@ async fn handle_message(
join_all(receivers).await;
}
Action::Unsubscribe => {
Some(Action::Unsubscribe) => {
let (pending_unsubscriptions, receivers): (Vec<_>, Vec<_>) = message
.symbols
.iter()
@@ -173,6 +173,7 @@ async fn handle_message(
join_all(receivers).await;
}
None => {}
}
message.response.send(()).unwrap();