kucoin_arbitrage 0.0.11

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
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

/// Test latency between order and private channel order detection
/// Places extreme order in REST, receive extreme order in private channel
/// Please configure the buy price to either the current market price or lower for testing purpose
use kucoin_api::{
    client::{Kucoin, KucoinEnv},
    model::websocket::{WSTopic, WSType},
};
use kucoin_arbitrage::broker::symbol::kucoin::{format_subscription_list, get_symbols};
use kucoin_arbitrage::event::{order::OrderEvent, orderchange::OrderChangeEvent};
use kucoin_arbitrage::model::counter::Counter;
use kucoin_arbitrage::{
    broker::order::kucoin::task_place_order,
    model::order::{LimitOrder, OrderSide, OrderType},
};
use kucoin_arbitrage::{
    broker::orderchange::kucoin::task_pub_orderchange_event, strings::generate_uid,
};
use kucoin_arbitrage::{broker::symbol::filter::symbol_with_quotes, global};
use std::{sync::Arc, time::Duration};
use tokio::sync::broadcast::channel;
use tokio::sync::Mutex;
use tokio::time::sleep;

#[tokio::main]
async fn main() -> Result<(), failure::Error> {
    // Provides logging format
    kucoin_arbitrage::logger::log_init();
    log::info!("Log setup");

    // Declares all the system counters
    let order_counter = Arc::new(Mutex::new(Counter::new("order")));

    // config
    let config = kucoin_arbitrage::config::from_file("config.toml")?;

    let api = Kucoin::new(KucoinEnv::Live, Some(config.kucoin_credentials()))?;
    let url_private = api.clone().get_socket_endpoint(WSType::Private).await?;
    log::info!("Credentials setup");

    // Gets all symbols concurrently
    let symbol_list = get_symbols(api.clone()).await;
    log::info!("Total exchange symbols: {:?}", symbol_list.len());

    // Filters with either btc or usdt as quote
    let symbol_infos = symbol_with_quotes(&symbol_list, "BTC", "USDT");
    log::info!("Total symbols in scope: {:?}", symbol_infos.len());

    // Changes a list of SymbolInfo into a 2D list of WSTopic per session in max 100 index
    let subs = format_subscription_list(&symbol_infos);
    log::info!("Total orderbook WS sessions: {:?}", subs.len());

    // Creates broadcast channels
    // for placing order
    let (tx_order, rx_order) = channel::<OrderEvent>(16);
    // for getting private order changes
    let (tx_orderchange, _rx_orderchange) = channel::<OrderChangeEvent>(128);
    log::info!("Broadcast channels setup");

    // TODO use the tx_order to send orders
    tokio::spawn(task_place_order(
        rx_order,
        api.clone(),
        order_counter.clone(),
    ));

    // Subscribes private order change websocket
    // NOTE TradeOrdersV2's TradeReceived appears unstable.
    // Using TradeOrders's TradeOpen instead
    let mut ws_private = api.websocket();
    ws_private
        .subscribe(
            url_private.clone(),
            vec![
                WSTopic::TradeOrders,
                WSTopic::Balances,
                WSTopic::PositionChange,
            ],
        )
        .await?;
    tokio::spawn(task_pub_orderchange_event(ws_private, tx_orderchange));

    log::info!("All application tasks setup");

    global::timer::start("order_placement_network".to_string()).await;
    global::timer::start("order_placement_broadcast".to_string()).await;
    // Sends a post order
    let event = OrderEvent::PostOrder(LimitOrder {
        id: generate_uid(40),
        order_type: OrderType::Limit,
        side: OrderSide::Buy,
        symbol: "BTC-USDT".to_string(),
        amount: 0.001.to_string(),
        price: 29947.0.to_string(),
    });
    if let Err(e) = tx_order.send(event) {
        log::error!("{e}");
    }

    loop {
        // Waits 60 seconds
        sleep(Duration::from_secs(60)).await;
    }
}