use anyhow::Error;
use rust_decimal::Decimal;
use std::{
future::Future,
pin::Pin,
sync::{atomic::Ordering, Arc},
};
use tokio::sync::{
watch::{channel, Receiver, Sender},
Notify,
};
use crate::{
clock::{Clock, ClockBase as _},
core::Core,
data::{Crypto, OrderDraft, Quantity},
market::{mexc_enums, Market},
};
use super::{AtomicState, ErrorInStrat, State, StratStat, Strategy, StrategyFactory};
pub struct StrictBuyFactory<M: Market> {
quote: Crypto,
amount: Decimal,
asset: Crypto,
core: Option<Core<M>>,
}
impl<M: Market> StrictBuyFactory<M> {
pub fn new(quote: Crypto, amount: Decimal, asset: Crypto) -> Self {
Self {
quote,
amount,
asset,
core: None,
}
}
}
impl<M: Market> StrategyFactory for StrictBuyFactory<M> {
type _Strategy = StrictBuy<M>;
fn build(mut self) -> StrictBuy<M> {
StrictBuy::new(
self.core.take().expect("Core not set"),
self.quote,
self.amount,
self.asset,
)
}
fn with_core(self, core: Core<M>) -> Self {
Self {
core: Some(core),
..self
}
}
}
pub struct StrictBuy<M: Market> {
core: Core<M>,
future_state: Arc<AtomicState>,
state: State,
last_state: State,
notify: Arc<Notify>,
quote: Crypto,
amount: Decimal,
asset: Crypto,
tx: Sender<Result<StratStat, ErrorInStrat>>,
rx: Receiver<Result<StratStat, ErrorInStrat>>,
}
const DEFAULT_STATE: State = State::Paused;
impl<M: Market> StrictBuy<M> {
fn new(core: Core<M>, quote: Crypto, amount: Decimal, asset: Crypto) -> Self {
let future_state = Arc::new(AtomicState::new(DEFAULT_STATE));
let notify = Arc::new(Notify::new());
let (tx, rx) = channel(Err(ErrorInStrat::new(
"Not initialized".to_string(),
core.clock().now(),
)));
Self {
core,
future_state,
state: DEFAULT_STATE,
last_state: DEFAULT_STATE,
notify,
asset,
amount,
quote,
tx,
rx,
}
}
async fn get_stats(&self) -> Result<StratStat, Error> {
let asset = self.core.get_balance(self.asset).await?;
let quote = self.core.get_balance(self.quote).await?;
Ok(StratStat {
asset,
quote,
time: self.core.clock().now(),
})
}
async fn send_stats(&self) {
let stat = self
.get_stats()
.await
.map_err(|err| ErrorInStrat::new(err.to_string(), self.core.clock().now()));
if let Err(err) = self.tx.send(stat) {
tracing::error!("Failed to send stat: {}", err);
}
}
async fn await_state_change(&mut self) {
self.last_state = self.state;
loop {
self.core.clock().synchronize();
loop {
self.state = self.future_state.load(Ordering::Relaxed);
if self.state != self.last_state {
break;
}
self.notify.notified().await;
}
if self.state == State::Stats {
self.future_state.store(self.last_state, Ordering::Relaxed);
self.send_stats().await;
} else {
break;
}
}
}
}
impl<M: Market> Strategy for StrictBuy<M> {
type _Market = M;
fn state(&self) -> &Arc<AtomicState> {
&self.future_state
}
fn notify(&self) -> &Arc<Notify> {
&self.notify
}
fn stat_channel(&self) -> Receiver<Result<StratStat, ErrorInStrat>> {
self.rx.clone()
}
fn run(&mut self) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
Box::pin(async move {
while self.state != State::Starting {
self.await_state_change().await;
}
tracing::info!("Strict_buy running");
self.core
.send_order(OrderDraft {
asset: self.asset,
quote: self.quote,
side: mexc_enums::OrderSide::Buy,
order_type: mexc_enums::OrderType::Market,
amount: Quantity::Quote(self.amount),
price: None,
})
.await
.unwrap();
tracing::info!("Strict buy waiting");
while self.state == State::Starting {
self.await_state_change().await;
}
tracing::info!("Strict buy closing");
let asset_balance = self.core.get_balance(self.asset).await.unwrap();
self.core
.send_order(OrderDraft {
asset: self.asset,
quote: self.quote,
side: mexc_enums::OrderSide::Sell,
order_type: mexc_enums::OrderType::Market,
amount: Quantity::Asset(asset_balance.free),
price: None,
})
.await
.unwrap();
self.send_stats().await;
})
}
}