kucoin_arbitrage 0.0.12

Event-Driven Kucoin Arbitrage Framework in Async Rust
Documentation 
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

/// Test WebSocket Message Rate
/// Subscribe to messages, run for 10 seconds, get the rate respectively
use kucoin_api::failure;
use kucoin_api::futures::TryStreamExt;
use kucoin_api::{
    client::{Kucoin, KucoinEnv},
    model::websocket::{KucoinWebsocketMsg, WSTopic, WSType},
    websocket::KucoinWebsocket,
};
use kucoin_arbitrage::global::counter_helper;
use kucoin_arbitrage::model::counter::Counter;
use std::sync::Arc;
use tokio::sync::Mutex;

/// main function
#[tokio::main]
async fn main() -> Result<(), failure::Error> {
    // provide logging format
    kucoin_arbitrage::logger::log_init();
    let counter = Arc::new(Mutex::new(Counter::new("api_input")));
    log::info!("Testing Kucoin WS Message Rate");

    // config
    let config = kucoin_arbitrage::config::from_file("config.toml")?;
    let monitor_interval: u32 = config.behaviour.monitor_interval_sec;

    // Initialize the Kucoin API struct
    let api = Kucoin::new(KucoinEnv::Live, Some(config.kucoin_credentials()))?;
    let url = api.get_socket_endpoint(WSType::Public).await?;
    let mut ws = api.websocket();
    let symbols = [
        "ETH-BTC".to_string(),
        "BTC-USDT".to_string(),
        "ETH-USDT".to_string(),
    ];
    let subs = vec![WSTopic::OrderBook(symbols.to_vec())];
    ws.subscribe(url, subs).await?;

    log::info!("Async polling");
    tokio::spawn(sync_tickers(ws, counter.clone()));
    let _res = tokio::join!(kucoin_arbitrage::global::task::task_log_mps(
        vec![counter.clone()],
        monitor_interval as u64
    ));
    panic!("Program should not arrive here")
}

async fn sync_tickers(
    mut ws: KucoinWebsocket,
    counter: Arc<Mutex<Counter>>,
) -> Result<(), failure::Error> {
    while let Some(msg) = ws.try_next().await? {
        match msg {
            KucoinWebsocketMsg::OrderBookMsg(_msg) => {
                // TODO make counter more generic
                counter_helper::reset(counter.clone()).await;
            }
            KucoinWebsocketMsg::PongMsg(_) => continue,
            KucoinWebsocketMsg::WelcomeMsg(_) => continue,
            _ => {
                panic!("unexpected msgs received: {msg:?}")
            }
        }
    }
    Ok(())
}