From 835a9e477aa821d37d012bb229c050da732efaab Mon Sep 17 00:00:00 2001 From: Sho Kaneko Date: Mon, 24 Jul 2023 02:32:07 +0100 Subject: [PATCH 1/3] latency MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit I am currently reading order_placement_broadcast: 75.293µs order_placement_network: 317.955963ms which is not good --- src/bin/test_kucoin_order_private_latency.rs | 6 +++-- src/broker/order/kucoin.rs | 5 +++++ src/broker/orderchange/kucoin.rs | 8 ++++--- src/global/mod.rs | 2 ++ src/global/timer.rs | 23 ++++++++++++++++++++ 5 files changed, 39 insertions(+), 5 deletions(-) create mode 100644 src/global/timer.rs diff --git a/src/bin/test_kucoin_order_private_latency.rs b/src/bin/test_kucoin_order_private_latency.rs index 8f78451..81cf046 100644 --- a/src/bin/test_kucoin_order_private_latency.rs +++ b/src/bin/test_kucoin_order_private_latency.rs @@ -5,7 +5,7 @@ use kucoin_api::{ client::{Kucoin, KucoinEnv}, model::websocket::{WSTopic, WSType}, }; -use kucoin_arbitrage::broker::symbol::filter::symbol_with_quotes; +use kucoin_arbitrage::{broker::symbol::filter::symbol_with_quotes, global}; 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; @@ -80,6 +80,8 @@ async fn main() -> Result<(), kucoin_api::failure::Error> { 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), @@ -87,7 +89,7 @@ async fn main() -> Result<(), kucoin_api::failure::Error> { side: OrderSide::Buy, symbol: "BTC-USDT".to_string(), amount: 0.001.to_string(), - price: 29850.0.to_string(), + price: 29947.0.to_string(), }); if let Err(e) = tx_order.send(event) { log::error!("{e}"); diff --git a/src/broker/order/kucoin.rs b/src/broker/order/kucoin.rs index 606887d..b430d10 100644 --- a/src/broker/order/kucoin.rs +++ b/src/broker/order/kucoin.rs @@ -1,5 +1,6 @@ use std::sync::Arc; +use crate::global; use crate::model::counter::Counter; use crate::model::order::Order; use crate::{event::order::OrderEvent, global::counter_helper}; @@ -45,6 +46,10 @@ pub async fn task_place_order( // } } OrderEvent::PostOrder(order) => { + // gge the broadcast duration + let time = global::timer::stop("order_placement_broadcast".to_string()).await.unwrap(); + log::info!("order_placement_broadcast: {time:?}"); + log::info!("order placement\n{order:?}"); if let Err(e) = kucoin .post_limit_order( diff --git a/src/broker/orderchange/kucoin.rs b/src/broker/orderchange/kucoin.rs index bbc3745..b014218 100644 --- a/src/broker/orderchange/kucoin.rs +++ b/src/broker/orderchange/kucoin.rs @@ -1,7 +1,7 @@ use crate::event::orderchange::OrderChangeEvent; +use crate::global; use kucoin_api::futures::TryStreamExt; use kucoin_api::{model::websocket::KucoinWebsocketMsg, websocket::KucoinWebsocket}; -use std::time::SystemTime; use tokio::sync::broadcast::Sender; /// Task to publish order change events. @@ -26,8 +26,10 @@ pub async fn task_pub_orderchange_event( // Currently using a more stable TradeOpenMsg, although TradeReceived is always ahead of TradeOpen log::info!("TradeReceivedMsg: {:?}\n{:#?}", msg.topic, msg.data); } else if let KucoinWebsocketMsg::TradeOpenMsg(msg) = msg { - let time_received = SystemTime::now(); - log::info!("time_received: {time_received:?}"); + + let time = global::timer::stop("order_placement_network".to_string()).await.unwrap(); + log::info!("order_placement_network: {time:?}"); + log::info!("TradeOpenMsg: {:?}\n{:#?}", msg.topic, msg.data); // TODO optimize below to something more insightful diff --git a/src/global/mod.rs b/src/global/mod.rs index e73a592..484719d 100644 --- a/src/global/mod.rs +++ b/src/global/mod.rs @@ -4,3 +4,5 @@ pub mod config; pub mod counter_helper; /// Routine tasks pub mod task; +/// Timers (for debugging) +pub mod timer; diff --git a/src/global/timer.rs b/src/global/timer.rs new file mode 100644 index 0000000..e512439 --- /dev/null +++ b/src/global/timer.rs @@ -0,0 +1,23 @@ +use std::collections::HashMap; +use tokio::sync::Mutex; +use tokio::time::{Duration, Instant}; + +lazy_static::lazy_static! { + static ref TIMERS: Mutex> = Mutex::new(HashMap::new()); +} + +/// Start +pub async fn start(name: String) { + let mut timer = TIMERS.lock().await; + timer.insert(name, Instant::now()); +} + +/// Stop +pub async fn stop(name: String) -> Result { + let now = Instant::now(); + let timer = TIMERS.lock().await; + let stat = timer + .get(&name) + .ok_or(format!("global timer [{name:?}] is not found"))?; + Ok(now.duration_since(*stat)) +} From 82ae9979018034b06de2f437cc46a90ec2a4a935 Mon Sep 17 00:00:00 2001 From: Sho Kaneko Date: Sat, 7 Oct 2023 16:29:14 +0100 Subject: [PATCH 2/3] replaced config system, return pure failure::Error at run time --- .gitignore | 1 + Cargo.lock | 158 +++++++++++------- Cargo.toml | 7 +- LICENSE | 2 +- config.ini.sample | 10 -- config_sample.toml | 13 ++ rust-toolchain.toml | 2 +- src/bin/event_orderbook.rs | 15 +- src/bin/event_triangular.rs | 27 +-- src/bin/sample_listcoin.rs | 8 +- src/bin/sample_orderbook_rest.rs | 15 +- src/bin/sample_orderbook_ws.rs | 17 +- .../test_kucoin_order_orderbook_latency.rs | 8 +- src/bin/test_kucoin_order_private_latency.rs | 11 +- src/bin/test_kucoin_orderbook_rate.rs | 16 +- src/config.rs | 37 ++++ src/error.rs | 31 ++++ src/global/config.rs | 28 ---- src/global/mod.rs | 2 - src/global/task.rs | 13 +- src/lib.rs | 4 + 21 files changed, 261 insertions(+), 164 deletions(-) delete mode 100644 config.ini.sample create mode 100644 config_sample.toml create mode 100644 src/config.rs create mode 100644 src/error.rs delete mode 100644 src/global/config.rs diff --git a/.gitignore b/.gitignore index 82f756d..af101be 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,4 @@ config.ini log/* .vscode/settings.json +config.toml diff --git a/Cargo.lock b/Cargo.lock index 53dc1e8..d1f2243 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,17 +17,6 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe" -[[package]] -name = "ahash" -version = "0.7.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fcb51a0695d8f838b1ee009b3fbf66bda078cd64590202a864a8f3e8c4315c47" -dependencies = [ - "getrandom", - "once_cell", - "version_check", -] - [[package]] name = "aho-corasick" version = "1.0.2" @@ -237,12 +226,6 @@ dependencies = [ "generic-array 0.14.7", ] -[[package]] -name = "dlv-list" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0688c2a7f92e427f44895cd63841bff7b29f8d7a1648b9e7e07a4a365b2e1257" - [[package]] name = "encoding_rs" version = "0.8.32" @@ -265,6 +248,12 @@ dependencies = [ "termcolor", ] +[[package]] +name = "equivalent" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5443807d6dff69373d433ab9ef5378ad8df50ca6298caf15de6e52e24aaf54d5" + [[package]] name = "errno" version = "0.3.1" @@ -415,7 +404,7 @@ checksum = "89ca545a94061b6365f2c7355b4b32bd20df3ff95f02da9329b34ccc3bd6ee72" dependencies = [ "proc-macro2", "quote", - "syn 2.0.27", + "syn 2.0.38", ] [[package]] @@ -496,7 +485,7 @@ dependencies = [ "futures-sink", "futures-util", "http", - "indexmap", + "indexmap 1.9.3", "slab", "tokio", "tokio-util", @@ -508,9 +497,12 @@ name = "hashbrown" version = "0.12.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" -dependencies = [ - "ahash", -] + +[[package]] +name = "hashbrown" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dfda62a12f55daeae5015f81b0baea145391cb4520f86c248fc615d72640d12" [[package]] name = "hermit-abi" @@ -659,7 +651,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bd070e393353796e801d209ad339e89596eb4c8d430d18ede6a1cced8fafbd99" dependencies = [ "autocfg", - "hashbrown", + "hashbrown 0.12.3", +] + +[[package]] +name = "indexmap" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8adf3ddd720272c6ea8bf59463c04e0f93d0bbf7c5439b691bca2987e0270897" +dependencies = [ + "equivalent", + "hashbrown 0.14.1", ] [[package]] @@ -733,6 +735,7 @@ version = "0.0.10" dependencies = [ "chrono", "env_logger", + "failure", "fern", "futures", "kucoin_api", @@ -741,8 +744,10 @@ dependencies = [ "num-traits", "ordered-float", "rand", - "rust-ini", + "serde", + "serde_derive", "tokio", + "toml", ] [[package]] @@ -898,7 +903,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.27", + "syn 2.0.38", ] [[package]] @@ -928,16 +933,6 @@ dependencies = [ "num-traits", ] -[[package]] -name = "ordered-multimap" -version = "0.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ccd746e37177e1711c20dd619a1620f34f5c8b569c53590a72dedd5344d8924a" -dependencies = [ - "dlv-list", - "hashbrown", -] - [[package]] name = "parking_lot" version = "0.12.1" @@ -984,7 +979,7 @@ checksum = "ec2e072ecce94ec471b13398d5402c188e76ac03cf74dd1a975161b23a3f6d9c" dependencies = [ "proc-macro2", "quote", - "syn 2.0.27", + "syn 2.0.38", ] [[package]] @@ -1013,9 +1008,9 @@ checksum = "5b40af805b3121feab8a3c29f04d8ad262fa8e0561883e7653e024ae4479e6de" [[package]] name = "proc-macro2" -version = "1.0.66" +version = "1.0.68" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "18fb31db3f9bddb2ea821cde30a9f70117e3f119938b5ee630b7403aa6e2ead9" +checksum = "5b1106fec09662ec6dd98ccac0f81cef56984d0b49f75c92d8cbad76e20c005c" dependencies = [ "unicode-ident", ] @@ -1154,16 +1149,6 @@ dependencies = [ "winapi", ] -[[package]] -name = "rust-ini" -version = "0.18.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f6d5f2436026b4f6e79dc829837d467cc7e9a55ee40e750d716713540715a2df" -dependencies = [ - "cfg-if", - "ordered-multimap", -] - [[package]] name = "rustc-demangle" version = "0.1.23" @@ -1270,19 +1255,22 @@ dependencies = [ [[package]] name = "serde" -version = "1.0.174" +version = "1.0.188" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3b88756493a5bd5e5395d53baa70b194b05764ab85b59e43e4b8f4e1192fa9b1" +checksum = "cf9e0fcba69a370eed61bcf2b728575f726b50b55cba78064753d708ddc7549e" +dependencies = [ + "serde_derive", +] [[package]] name = "serde_derive" -version = "1.0.174" +version = "1.0.188" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6e5c3a298c7f978e53536f95a63bdc4c4a64550582f31a0359a9afda6aede62e" +checksum = "4eca7ac642d82aa35b60049a6eccb4be6be75e599bd2e9adb5f875a737654af2" dependencies = [ "proc-macro2", "quote", - "syn 2.0.27", + "syn 2.0.38", ] [[package]] @@ -1296,6 +1284,15 @@ dependencies = [ "serde", ] +[[package]] +name = "serde_spanned" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96426c9936fd7a0124915f9185ea1d20aa9445cc9821142f0a73bc9207a2e186" +dependencies = [ + "serde", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -1404,9 +1401,9 @@ dependencies = [ [[package]] name = "syn" -version = "2.0.27" +version = "2.0.38" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b60f673f44a8255b9c8c657daf66a596d435f2da81a555b06dc644d080ba45e0" +checksum = "e96b79aaa137db8f61e26363a0c9b47d8b4ec75da28b7d1d614c2303e232408b" dependencies = [ "proc-macro2", "quote", @@ -1464,7 +1461,7 @@ checksum = "090198534930841fab3a5d1bb637cde49e339654e606195f8d9c76eeb081dc96" dependencies = [ "proc-macro2", "quote", - "syn 2.0.27", + "syn 2.0.38", ] [[package]] @@ -1521,7 +1518,7 @@ checksum = "630bdcf245f78637c13ec01ffae6187cca34625e8c63150d424b59e55af2675e" dependencies = [ "proc-macro2", "quote", - "syn 2.0.27", + "syn 2.0.38", ] [[package]] @@ -1573,6 +1570,40 @@ dependencies = [ "tracing", ] +[[package]] +name = "toml" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "185d8ab0dfbb35cf1399a6344d8484209c088f75f8f68230da55d48d95d43e3d" +dependencies = [ + "serde", + "serde_spanned", + "toml_datetime", + "toml_edit", +] + +[[package]] +name = "toml_datetime" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7cda73e2f1397b1262d6dfdcef8aafae14d1de7748d66822d3bfeeb6d03e5e4b" +dependencies = [ + "serde", +] + +[[package]] +name = "toml_edit" +version = "0.20.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "396e4d48bbb2b7554c944bde63101b5ae446cff6ec4a24227428f15eb72ef338" +dependencies = [ + "indexmap 2.0.2", + "serde", + "serde_spanned", + "toml_datetime", + "winnow", +] + [[package]] name = "tower-service" version = "0.3.2" @@ -1756,7 +1787,7 @@ dependencies = [ "once_cell", "proc-macro2", "quote", - "syn 2.0.27", + "syn 2.0.38", "wasm-bindgen-shared", ] @@ -1790,7 +1821,7 @@ checksum = "54681b18a46765f095758388f2d0cf16eb8d4169b639ab575a8f5693af210c7b" dependencies = [ "proc-macro2", "quote", - "syn 2.0.27", + "syn 2.0.38", "wasm-bindgen-backend", "wasm-bindgen-shared", ] @@ -1936,6 +1967,15 @@ version = "0.48.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1a515f5799fe4961cb532f983ce2b23082366b898e52ffbce459c86f67c8378a" +[[package]] +name = "winnow" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "037711d82167854aff2018dfd193aa0fef5370f456732f0d5a0c59b0f1b4b907" +dependencies = [ + "memchr", +] + [[package]] name = "winreg" version = "0.10.1" diff --git a/Cargo.toml b/Cargo.toml index e8a9a0d..9a03f08 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,7 +6,7 @@ authors = ["Sho Kaneko "] description = "Event-Driven Kucoin Arbitrage Framework in Async Rust" repository = "https://github.com/kanekoshoyu/kucoin_arbitrage" license = "MIT" - +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [dependencies] kucoin_api = "1.4.10" @@ -14,10 +14,13 @@ log = "0.4" fern = "0.6.2" chrono = "0.4" env_logger = "0.10.0" -rust-ini = "0.18" lazy_static = "1.4" futures = "0.3" ordered-float = "3.6.0" num-traits = "0.2.15" +serde = "1.0.188" +serde_derive = "1.0.188" +failure = "0.1.8" +toml = "0.8.2" tokio = { version = "1.27.0", features = ["full"] } rand = "0.8.5" diff --git a/LICENSE b/LICENSE index a97d535..d68f0b5 100644 --- a/LICENSE +++ b/LICENSE @@ -1,6 +1,6 @@ MIT License -Copyright (c) 2022 Sho Kaneko +Copyright (c) 2023 Sho Kaneko Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal diff --git a/config.ini.sample b/config.ini.sample deleted file mode 100644 index 32162bc..0000000 --- a/config.ini.sample +++ /dev/null @@ -1,10 +0,0 @@ -# Create a copy of config.ini.sample as config.ini in the same directory - -# Get API credentials from https://www.kucoin.com/ and paste below -[KuCoin Credentials] -api_key="YOUR_API_KEY_HERE" -secret_key="YOUR_SECRET_KEY_HERE" -passphrase="YOUR_PASSPHRASE_HERE" - -[Behaviour] -monitor_interval_sec=2 \ No newline at end of file diff --git a/config_sample.toml b/config_sample.toml new file mode 100644 index 0000000..d047744 --- /dev/null +++ b/config_sample.toml @@ -0,0 +1,13 @@ +# Create a copy of config_sample.toml as config.toml in the same directory + +# Get API credentials from https://www.kucoin.com/ and paste below +[kucoin] +api_key = "YOUR_API_KEY_HERE" +secret_key = "YOUR_SECRET_KEY_HERE" +passphrase = "YOUR_PASSPHRASE_HERE" + +[behaviour] +# Performance monitor interval in seconds +monitor_interval_sec = 120 +# max amount of USD to use in a single cyclic arbitrage +usd_cyclic_arbitrage = 100 diff --git a/rust-toolchain.toml b/rust-toolchain.toml index 271800c..31578d3 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,2 +1,2 @@ [toolchain] -channel = "nightly" \ No newline at end of file +channel = "stable" \ No newline at end of file diff --git a/src/bin/event_orderbook.rs b/src/bin/event_orderbook.rs index fd5a871..a353ac5 100644 --- a/src/bin/event_orderbook.rs +++ b/src/bin/event_orderbook.rs @@ -11,15 +11,15 @@ use tokio::sync::broadcast::channel; use tokio::sync::Mutex; #[tokio::main] -async fn main() -> Result<(), kucoin_api::failure::Error> { +async fn main() -> Result<(), failure::Error> { // provide logging format kucoin_arbitrage::logger::log_init(); log::info!("Log setup"); let counter = Arc::new(Mutex::new(Counter::new("api_input"))); - // credentials - let credentials = kucoin_arbitrage::global::config::credentials(); - let api = Kucoin::new(KucoinEnv::Live, Some(credentials))?; + // config + let config = kucoin_arbitrage::config::from_file("config.toml")?; + let api = Kucoin::new(KucoinEnv::Live, Some(config.kucoin_credentials()))?; let url = api.get_socket_endpoint(WSType::Public).await?; log::info!("Credentials setup"); @@ -55,8 +55,9 @@ async fn main() -> Result<(), kucoin_api::failure::Error> { )); log::info!("task_sync_orderbook setup"); - let _ = tokio::join!(kucoin_arbitrage::global::task::background_routine(vec![ - counter.clone(), - ])); + let _ = tokio::join!(kucoin_arbitrage::global::task::background_routine( + vec![counter.clone()], + config.behaviour.monitor_interval_sec as u64 + )); panic!("Program should not arrive here") } diff --git a/src/bin/event_triangular.rs b/src/bin/event_triangular.rs index ec82c7c..8cf80ba 100644 --- a/src/bin/event_triangular.rs +++ b/src/bin/event_triangular.rs @@ -14,7 +14,6 @@ use kucoin_arbitrage::event::{ chance::ChanceEvent, order::OrderEvent, orderbook::OrderbookEvent, orderchange::OrderChangeEvent, }; -use kucoin_arbitrage::global::config::CONFIG; use kucoin_arbitrage::model::{counter::Counter, orderbook::FullOrderbook}; use kucoin_arbitrage::strategy::all_taker_btc_usd::task_pub_chance_all_taker_btc_usd; use kucoin_arbitrage::translator::traits::OrderBookTranslator; @@ -23,7 +22,7 @@ use tokio::sync::broadcast::channel; use tokio::sync::Mutex; #[tokio::main] -async fn main() -> Result<(), kucoin_api::failure::Error> { +async fn main() -> Result<(), failure::Error> { // Provides logging format kucoin_arbitrage::logger::log_init(); log::info!("Log setup"); @@ -35,8 +34,11 @@ async fn main() -> Result<(), kucoin_api::failure::Error> { let order_counter = Arc::new(Mutex::new(Counter::new("order"))); // Credentials - let credentials = kucoin_arbitrage::global::config::credentials(); - let api = Kucoin::new(KucoinEnv::Live, Some(credentials))?; + let config = kucoin_arbitrage::config::from_file("config.toml")?; + let budget = config.behaviour.usd_cyclic_arbitrage; + let monitor_interval = config.behaviour.monitor_interval_sec; + + let api = Kucoin::new(KucoinEnv::Live, Some(config.kucoin_credentials()))?; let url_public = api.clone().get_socket_endpoint(WSType::Public).await?; let url_private = api.clone().get_socket_endpoint(WSType::Private).await?; log::info!("Credentials setup"); @@ -84,7 +86,7 @@ async fn main() -> Result<(), kucoin_api::failure::Error> { tx_chance, full_orderbook.clone(), hash_symbols, - CONFIG.usd_cyclic_arbitrage as f64, + budget as f64, best_price_counter.clone(), )); tokio::spawn(task_gatekeep_chances( @@ -158,11 +160,14 @@ async fn main() -> Result<(), kucoin_api::failure::Error> { log::info!("All application tasks setup"); // Background routine - let _ = tokio::join!(kucoin_arbitrage::global::task::background_routine(vec![ - api_input_counter.clone(), - best_price_counter.clone(), - chance_counter.clone(), - order_counter.clone() - ])); + let _ = tokio::join!(kucoin_arbitrage::global::task::background_routine( + vec![ + api_input_counter.clone(), + best_price_counter.clone(), + chance_counter.clone(), + order_counter.clone() + ], + monitor_interval as u64 + )); panic!("Program should not arrive here") } diff --git a/src/bin/sample_listcoin.rs b/src/bin/sample_listcoin.rs index a4f67df..1cbc39d 100644 --- a/src/bin/sample_listcoin.rs +++ b/src/bin/sample_listcoin.rs @@ -3,14 +3,12 @@ use kucoin_api::client::{Kucoin, KucoinEnv}; use kucoin_arbitrage::broker::symbol::{filter::symbol_with_quotes, kucoin::get_symbols}; #[tokio::main] -async fn main() -> Result<(), kucoin_api::failure::Error> { +async fn main() -> Result<(), failure::Error> { // provide logging format kucoin_arbitrage::logger::log_init(); log::info!("Hello world"); - - // set credentials - let credentials = kucoin_arbitrage::global::config::credentials(); - let api = Kucoin::new(KucoinEnv::Live, Some(credentials))?; + let config = kucoin_arbitrage::config::from_file("config.toml")?; + let api = Kucoin::new(KucoinEnv::Live, Some(config.kucoin_credentials()))?; // get symbol lists let symbol_list = get_symbols(api).await; diff --git a/src/bin/sample_orderbook_rest.rs b/src/bin/sample_orderbook_rest.rs index c24c0e6..2b82f05 100644 --- a/src/bin/sample_orderbook_rest.rs +++ b/src/bin/sample_orderbook_rest.rs @@ -3,22 +3,17 @@ use kucoin_api::client::{Kucoin, KucoinEnv}; use kucoin_api::model::market::OrderBookType; #[tokio::main] -async fn main() -> Result<(), kucoin_api::failure::Error> { +async fn main() -> Result<(), failure::Error> { // provide logging format kucoin_arbitrage::logger::log_init(); log::info!("Hello world"); - // credentials - let credentials = kucoin_arbitrage::global::config::credentials(); - let api = Kucoin::new(KucoinEnv::Live, Some(credentials))?; + let config = kucoin_arbitrage::config::from_file("config.toml")?; + let api = Kucoin::new(KucoinEnv::Live, Some(config.kucoin_credentials()))?; let symbol_name = "BTC-USDT"; let res = api.get_orderbook(symbol_name, OrderBookType::L20).await?; - if let Some(orderbook) = res.data { - log::info!("{orderbook:#?}"); - } else { - log::info!("failed obtaining the proper data") - } - + let orderbook = res.data.expect("failed obtaining the proper data"); + log::info!("{orderbook:#?}"); Ok(()) } diff --git a/src/bin/sample_orderbook_ws.rs b/src/bin/sample_orderbook_ws.rs index 643ad7b..e2f3c92 100644 --- a/src/bin/sample_orderbook_ws.rs +++ b/src/bin/sample_orderbook_ws.rs @@ -13,15 +13,17 @@ use std::sync::Arc; use tokio::sync::Mutex; #[tokio::main] -async fn main() -> Result<(), kucoin_api::failure::Error> { +async fn main() -> Result<(), failure::Error> { // provide logging format kucoin_arbitrage::logger::log_init(); log::info!("Log setup"); let counter = Arc::new(Mutex::new(Counter::new("api_input"))); - // credentials - let credentials = kucoin_arbitrage::global::config::credentials(); - let api = Kucoin::new(KucoinEnv::Live, Some(credentials))?; + // config + let config = kucoin_arbitrage::config::from_file("config.toml")?; + let monitor_interval = config.behaviour.monitor_interval_sec; + + let api = Kucoin::new(KucoinEnv::Live, Some(config.kucoin_credentials()))?; let url = api.clone().get_socket_endpoint(WSType::Public).await?; log::info!("Credentials setup"); @@ -44,9 +46,10 @@ async fn main() -> Result<(), kucoin_api::failure::Error> { tokio::spawn(sync_tickers(ws, counter.clone())); log::info!("{i:?}-th session of WS subscription setup"); } - let _res = tokio::join!(kucoin_arbitrage::global::task::background_routine(vec![ - counter.clone(), - ])); + let _res = tokio::join!(kucoin_arbitrage::global::task::background_routine( + vec![counter.clone(),], + monitor_interval as u64 + )); panic!("Program should not arrive here") } diff --git a/src/bin/test_kucoin_order_orderbook_latency.rs b/src/bin/test_kucoin_order_orderbook_latency.rs index ed6f43f..ad27474 100644 --- a/src/bin/test_kucoin_order_orderbook_latency.rs +++ b/src/bin/test_kucoin_order_orderbook_latency.rs @@ -17,10 +17,12 @@ async fn main() -> Result<(), failure::Error> { // provide logging format kucoin_arbitrage::logger::log_init(); log::info!("Testing Kucoin REST-to-WS latency"); - let credentials = kucoin_arbitrage::global::config::credentials(); - log::info!("{credentials:#?}"); + + // config + let config = kucoin_arbitrage::config::from_file("config.toml")?; + // Initialize the Kucoin API struct - let api = Kucoin::new(KucoinEnv::Live, Some(credentials))?; + let api = Kucoin::new(KucoinEnv::Live, Some(config.kucoin_credentials()))?; let url = api.get_socket_endpoint(WSType::Public).await?; let mut ws = api.websocket(); diff --git a/src/bin/test_kucoin_order_private_latency.rs b/src/bin/test_kucoin_order_private_latency.rs index 81cf046..e3165f0 100644 --- a/src/bin/test_kucoin_order_private_latency.rs +++ b/src/bin/test_kucoin_order_private_latency.rs @@ -5,7 +5,6 @@ use kucoin_api::{ client::{Kucoin, KucoinEnv}, model::websocket::{WSTopic, WSType}, }; -use kucoin_arbitrage::{broker::symbol::filter::symbol_with_quotes, global}; 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; @@ -16,13 +15,14 @@ use kucoin_arbitrage::{ 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<(), kucoin_api::failure::Error> { +async fn main() -> Result<(), failure::Error> { // Provides logging format kucoin_arbitrage::logger::log_init(); log::info!("Log setup"); @@ -30,9 +30,10 @@ async fn main() -> Result<(), kucoin_api::failure::Error> { // Declares all the system counters let order_counter = Arc::new(Mutex::new(Counter::new("order"))); - // Credentials - let credentials = kucoin_arbitrage::global::config::credentials(); - let api = Kucoin::new(KucoinEnv::Live, Some(credentials))?; + // 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"); diff --git a/src/bin/test_kucoin_orderbook_rate.rs b/src/bin/test_kucoin_orderbook_rate.rs index fdb1c2e..21dc2ab 100644 --- a/src/bin/test_kucoin_orderbook_rate.rs +++ b/src/bin/test_kucoin_orderbook_rate.rs @@ -19,10 +19,13 @@ async fn main() -> Result<(), failure::Error> { kucoin_arbitrage::logger::log_init(); let counter = Arc::new(Mutex::new(Counter::new("api_input"))); log::info!("Testing Kucoin WS Message Rate"); - let credentials = kucoin_arbitrage::global::config::credentials(); - log::info!("{credentials:#?}"); + + // 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(credentials))?; + 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 = [ @@ -35,9 +38,10 @@ async fn main() -> Result<(), failure::Error> { log::info!("Async polling"); tokio::spawn(sync_tickers(ws, counter.clone())); - let _res = tokio::join!(kucoin_arbitrage::global::task::background_routine(vec![ - counter.clone() - ])); + let _res = tokio::join!(kucoin_arbitrage::global::task::background_routine( + vec![counter.clone()], + monitor_interval as u64 + )); panic!("Program should not arrive here") } diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..63508da --- /dev/null +++ b/src/config.rs @@ -0,0 +1,37 @@ +use crate::error::Error; +use kucoin_api::client::Credentials; +use serde_derive::Deserialize; + +#[derive(Deserialize, Debug)] +pub struct Config { + pub kucoin: KuCoin, + pub behaviour: Behaviour, +} + +impl Config { + pub fn kucoin_credentials(self) -> Credentials { + Credentials::new( + &self.kucoin.api_key, + &self.kucoin.secret_key, + &self.kucoin.passphrase, + ) + } +} + +#[derive(Deserialize, Debug)] +pub struct KuCoin { + pub api_key: String, + pub secret_key: String, + pub passphrase: String, +} + +#[derive(Deserialize, Debug)] +pub struct Behaviour { + pub monitor_interval_sec: u32, + pub usd_cyclic_arbitrage: u32, +} + +pub fn from_file(filename: &str) -> Result { + let toml_str = std::fs::read_to_string(filename).map_err(Error::IoError)?; + toml::from_str(&toml_str).map_err(Error::TomlError) +} diff --git a/src/error.rs b/src/error.rs new file mode 100644 index 0000000..39b4ccd --- /dev/null +++ b/src/error.rs @@ -0,0 +1,31 @@ +use std::error; +use std::fmt; +use std::io; +use toml; + +#[derive(Debug)] +pub enum Error { + IoError(io::Error), + TomlError(toml::de::Error), + // Add more error variants as needed +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + match *self { + Error::IoError(ref err) => write!(f, "IO error: {}", err), + Error::TomlError(ref err) => write!(f, "TOML error: {}", err), + // Add more error variants as needed + } + } +} + +impl error::Error for Error { + fn source(&self) -> Option<&(dyn error::Error + 'static)> { + match *self { + Error::IoError(ref err) => Some(err), + Error::TomlError(ref err) => Some(err), + // Add more error variants as needed + } + } +} diff --git a/src/global/config.rs b/src/global/config.rs deleted file mode 100644 index 89b2f0a..0000000 --- a/src/global/config.rs +++ /dev/null @@ -1,28 +0,0 @@ -use crate::model::config::Config; -use ini::{Ini, Properties}; -use kucoin_api::client::Credentials; -use std::sync::Arc; - -lazy_static::lazy_static! { - static ref INI: Ini = Ini::load_from_file("config.ini").expect("config file not found"); - pub static ref SEC_CRED: Properties = INI.section(Some("KuCoin Credentials")).unwrap().clone(); - pub static ref SEC_BEHV: Properties = INI.section(Some("Behaviour")).unwrap().clone(); - pub static ref CONFIG: Arc = Arc::new(load_ini()); -} - -// might require macro to load the filename -pub fn load_ini() -> Config { - let str_interval = SEC_BEHV.get("monitor_interval_sec").unwrap(); - let str_budget = SEC_BEHV.get("usd_cyclic_arbitrage").unwrap(); - Config { - monitor_interval_sec: str_interval.parse::().unwrap(), - usd_cyclic_arbitrage: str_budget.parse::().unwrap(), - api_key: SEC_CRED.get("api_key").unwrap(), - secret_key: SEC_CRED.get("secret_key").unwrap(), - passphrase: SEC_CRED.get("passphrase").unwrap(), - } -} - -pub fn credentials() -> Credentials { - Credentials::new(CONFIG.api_key, CONFIG.secret_key, CONFIG.passphrase) -} diff --git a/src/global/mod.rs b/src/global/mod.rs index 484719d..945bf91 100644 --- a/src/global/mod.rs +++ b/src/global/mod.rs @@ -1,5 +1,3 @@ -/// Config parser -pub mod config; /// Generic counter pub mod counter_helper; /// Routine tasks diff --git a/src/global/task.rs b/src/global/task.rs index 9d4a9fb..7907953 100644 --- a/src/global/task.rs +++ b/src/global/task.rs @@ -1,4 +1,4 @@ -use crate::global::{config, counter_helper}; +use crate::global::counter_helper; use crate::model::counter::Counter; use std::sync::Arc; use tokio::sync::Mutex; @@ -6,6 +6,7 @@ use tokio::time::{sleep, Duration}; async fn report_status( counters: Vec>>, + interval: u64, ) -> Result<(), kucoin_api::failure::Error> { log::info!("Reporting broadcast data rate"); for counter in counters.iter() { @@ -13,10 +14,7 @@ async fn report_status( let p = counter.lock().await; (p.name, p.data_count) }; - log::info!( - "{name:?}: {count:?} points ({:?}pps)", - count / config::CONFIG.monitor_interval_sec - ); + log::info!("{name:?}: {count:?} points ({:?}pps)", count / interval); // clear the data counter_helper::reset(counter.clone()).await; } @@ -25,11 +23,12 @@ async fn report_status( pub async fn background_routine( counters: Vec>>, + interval: u64, ) -> Result<(), kucoin_api::failure::Error> { - let monitor_delay = Duration::from_secs(config::CONFIG.monitor_interval_sec); + let monitor_delay = Duration::from_secs(interval); loop { sleep(monitor_delay).await; - report_status(counters.clone()) + report_status(counters.clone(), interval) .await .expect("report status error"); } diff --git a/src/lib.rs b/src/lib.rs index cb2fd7d..b5af4b1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -14,3 +14,7 @@ pub mod strategy; pub mod strings; /// Traits/impl to convert between API crate models and internal models pub mod translator; +/// Config file reader +pub mod config; +/// Custom error +pub mod error; From 97924491bda9200389cfb1942d1f6c7741d8760e Mon Sep 17 00:00:00 2001 From: Sho Kaneko Date: Sat, 7 Oct 2023 16:30:23 +0100 Subject: [PATCH 3/3] 0.0.11 --- Cargo.lock | 2 +- Cargo.toml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d1f2243..a5c6c06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -731,7 +731,7 @@ dependencies = [ [[package]] name = "kucoin_arbitrage" -version = "0.0.10" +version = "0.0.11" dependencies = [ "chrono", "env_logger", diff --git a/Cargo.toml b/Cargo.toml index 9a03f08..963e848 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "kucoin_arbitrage" -version = "0.0.10" +version = "0.0.11" edition = "2021" authors = ["Sho Kaneko "] description = "Event-Driven Kucoin Arbitrage Framework in Async Rust"