Sik/tradingbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Change code to run strategy in parallel

Browse Source
This commit is contained in:
Sik Yoon 2023-08-05 03:07:16 +09:00
parent b0631921d3
commit 7bdb130865
1 changed files with 72 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

139
src/coex/strategy_team.rs
View File

@ -35,7 +35,6 @@ pub enum MA {
#[derive(Debug)] #[derive(Debug)]
pub struct AllData { pub struct AllData {
pub price_vec: Vec<CoinPriceData>,
pub valid_symbol_vec: Vec<String>, pub valid_symbol_vec: Vec<String>,
pub rt_price_1m_vec: Vec<(String, Vec<RealtimePriceData>)>, pub rt_price_1m_vec: Vec<(String, Vec<RealtimePriceData>)>,
@ -123,7 +122,6 @@ pub struct AllData {
impl AllData { impl AllData {
pub fn new() -> AllData { pub fn new() -> AllData {
let a = AllData { let a = AllData {
price_vec: Vec::new(),
valid_symbol_vec: Vec::new(), valid_symbol_vec: Vec::new(),
rt_price_1m_vec: Vec::new(), rt_price_1m_vec: Vec::new(),
rt_price_30m_vec: Vec::new(), rt_price_30m_vec: Vec::new(),
@ -278,7 +276,7 @@ pub async fn execute_strategists(
// strategist_015(all_data).await; // strategist_015(all_data).await;
// strategist_016(all_data).await; // strategist_016(all_data).await;
execute_strategist_for_test_temp(all_data).await; execute_strategist_for_test_temp(all_data).await?;
// execute_strategist_for_test1(all_data).await; // execute_strategist_for_test1(all_data).await;
// execute_strategist_for_test2(all_data).await; // execute_strategist_for_test2(all_data).await;
Ok(()) Ok(())
@ -930,6 +928,10 @@ pub async fn execute_strategists(
pub async fn execute_strategist_for_test_temp( pub async fn execute_strategist_for_test_temp(
alldata: &AllData, alldata: &AllData,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// print rt_price for debugging
// let a = alldata.rt_price_30m_vec.iter().position(|a| a.0 == "BTCUSDT");
// println!("BTCUSDT: {:?}", alldata.rt_price_30m_vec[a.unwrap()].1.last().unwrap());
// 1st filtering: supertrend(ATR period 10, multiplier: 1.3, 30m close price), the area should be in SELL area. // 1st filtering: supertrend(ATR period 10, multiplier: 1.3, 30m close price), the area should be in SELL area.
let mut filtered_2nd_symbols: Vec<(String, i64)> = Vec::new(); let mut filtered_2nd_symbols: Vec<(String, i64)> = Vec::new();
let mut filtered_2nd_symbols_arc: Arc<Mutex<Vec<(String, i64)>>> = let mut filtered_2nd_symbols_arc: Arc<Mutex<Vec<(String, i64)>>> =
@ -976,7 +978,7 @@ pub async fn execute_strategist_for_test_temp(
})); }));
} }
try_join_all(task_vec).await?; try_join_all(task_vec).await?;
// 2nd filtering: lookup tables if the tradepair is already there // 2nd filtering: lookup tables if the tradepair is already there
let inspect_table_name_1 = String::from("buy_ordered_coin_list"); let inspect_table_name_1 = String::from("buy_ordered_coin_list");
let inspect_table_name_2 = String::from("sell_ordered_coin_list"); let inspect_table_name_2 = String::from("sell_ordered_coin_list");
@ -1023,32 +1025,31 @@ pub async fn execute_strategist_for_test_temp(
})); }));
} }
try_join_all(task_vec).await?; try_join_all(task_vec).await?;
// 3rd filtering: BollingerBand (length 10, stddev: 3.0, 30m close price) the current price should be under the lowerband of BB. // 3rd filtering: BollingerBand (length 10, stddev: 2.5, 30m close price) the current price should be under the lowerband of BB.
let filtered_3rd_iter = filtered_3rd_symbols_arc.lock().await.clone().into_iter(); let filtered_3rd_symbols_c = filtered_3rd_symbols_arc.lock().await.clone();
let filtered_3rd_symbols_mutex = filtered_3rd_symbols_arc.lock().await;
let sma10_30m_data: Vec<(String, Vec<SmaData>)> = value_estimation_team::indicators::sma::sma( let sma10_30m_data: Vec<(String, Vec<SmaData>)> = value_estimation_team::indicators::sma::sma(
10, 10,
&alldata.rt_price_30m_vec, &alldata.rt_price_30m_vec,
&filtered_3rd_symbols_mutex, &filtered_3rd_symbols_c,
) )
.await?; .await?;
let bb10_30m_data: Vec<(String, Vec<BollingerBandData>)> = let bb10_30m_data: Vec<(String, Vec<BollingerBandData>)> =
value_estimation_team::indicators::bollingerband::bollingerband( value_estimation_team::indicators::bollingerband::bollingerband(
10, 10,
3.0, 2.5,
&sma10_30m_data, &sma10_30m_data,
&alldata.rt_price_30m_vec, &alldata.rt_price_30m_vec,
&filtered_3rd_symbols_mutex, &filtered_3rd_symbols_c,
) )
.await?; .await?;
let mut task_vec = Vec::new(); let mut task_vec = Vec::new();
let mut filtered_4th_symbols: Vec<(String, i64)> = Vec::new(); let mut filtered_4th_symbols: Vec<(String, i64)> = Vec::new();
let mut filtered_4th_symbols_arc: Arc<Mutex<Vec<(String, i64)>>> = let mut filtered_4th_symbols_arc: Arc<Mutex<Vec<(String, i64)>>> =
Arc::new(Mutex::new(filtered_4th_symbols)); // (symbol, closetime) Arc::new(Mutex::new(filtered_4th_symbols)); // (symbol, closetime)
for element in filtered_3rd_iter { for element in filtered_3rd_symbols_c {
let mut bb10_30m_vec: Vec<BollingerBandData> = Vec::new(); let mut bb10_30m_vec: Vec<BollingerBandData> = Vec::new();
let bb10_30m_option = bb10_30m_data.iter().position(|x| *x.0 == element.0); let bb10_30m_option = bb10_30m_data.iter().position(|x| *x.0 == element.0);
let bb10_30m_option_c = bb10_30m_option.clone(); let bb10_30m_option_c = bb10_30m_option.clone();
@ -1060,7 +1061,7 @@ pub async fn execute_strategist_for_test_temp(
if bb10_30m_vec.len() >= 3 { if bb10_30m_vec.len() >= 3 {
let bb10_30m_vec_c = bb10_30m_vec.clone(); let bb10_30m_vec_c = bb10_30m_vec.clone();
let current_price = get_current_price(&element_c.0, &alldata.price_vec) let current_price = get_current_price(&element_c.0, &alldata.rt_price_30m_vec)
.await .await
.unwrap(); .unwrap();
task_vec.push(tokio::spawn(async move { task_vec.push(tokio::spawn(async move {
@ -1073,7 +1074,7 @@ pub async fn execute_strategist_for_test_temp(
close_time, close_time,
}| close_time, }| close_time,
); );
if bb_search_result.is_ok() { if bb_search_result.is_ok() {
if bb10_30m_vec_c[bb_search_result.unwrap()].lowerband > current_price { if bb10_30m_vec_c[bb_search_result.unwrap()].lowerband > current_price {
let mut filtered_4th_symbols_lock = let mut filtered_4th_symbols_lock =
filtered_4th_symbols_arc_c.lock().await; filtered_4th_symbols_arc_c.lock().await;
@ -1085,25 +1086,25 @@ pub async fn execute_strategist_for_test_temp(
} }
} }
try_join_all(task_vec).await?; try_join_all(task_vec).await?;
// 4th filtering: RSI (length: 10, 30m close price) the current index should be lower than 30. // 4th filtering: RSI (length: 10, 30m close price) the current index should be lower than 30.
let filtered_4th_iter = filtered_4th_symbols_arc.lock().await.clone().into_iter(); let filtered_4th_symbol_c = filtered_4th_symbols_arc.lock().await.clone();
let mut rsi10_30m_data: Vec<(String, Vec<RsiData>)> = Vec::new(); let mut rsi10_30m_data: Vec<(String, Vec<RsiData>)> = Vec::new();
value_estimation_team::indicators::rsi::rsi( value_estimation_team::indicators::rsi::rsi(
10, 10,
&alldata.rt_price_30m_vec, &alldata.rt_price_30m_vec,
&mut rsi10_30m_data, &mut rsi10_30m_data,
&alldata.valid_symbol_vec, &filtered_4th_symbol_c,
) )
.await?; .await?;
let mut rsi10_30m_vec: Vec<RsiData> = Vec::new(); let mut rsi10_30m_vec: Vec<RsiData> = Vec::new();
let mut filtered_5th_symbols: Vec<(String, i64)> = Vec::new(); // (symbol, closetime) let mut filtered_5th_symbols: Vec<(String, i64)> = Vec::new(); // (symbol, closetime)
for element in filtered_4th_iter { for element in filtered_4th_symbol_c {
let rsi10_30m_option = rsi10_30m_data.iter().position(|x| *x.0 == element.0); let rsi10_30m_option = rsi10_30m_data.iter().position(|x| *x.0 == element.0);
if rsi10_30m_option.is_some() { if rsi10_30m_option.is_some() {
rsi10_30m_vec = rsi10_30m_data[rsi10_30m_option.unwrap()].1.clone(); rsi10_30m_vec = rsi10_30m_data[rsi10_30m_option.unwrap()].1.clone();
if rsi10_30m_vec.len() >= 3 { if rsi10_30m_vec.len() >= 3 {
let rsi_search_result = rsi10_30m_vec.binary_search_by_key( let rsi_search_result = rsi10_30m_vec.binary_search_by_key(
&element.1, &element.1,
@ -1120,7 +1121,7 @@ pub async fn execute_strategist_for_test_temp(
} }
} }
} }
// 5th filtering: heatmap volume(MA length 10, std length 10, 30m close price), the current candle should be over than high at least. // 5th filtering: heatmap volume(MA length 10, std length 10, 30m close price), the current candle should be over than high at least.
let mut filtered_6th_symbols: Vec<(String, i64)> = Vec::new(); // (symbol, closetime) let mut filtered_6th_symbols: Vec<(String, i64)> = Vec::new(); // (symbol, closetime)
for element in filtered_5th_symbols { for element in filtered_5th_symbols {
@ -1159,6 +1160,28 @@ pub async fn execute_strategist_for_test_temp(
} }
} }
// final job: adding price information to filtered results
let mut filtered_symbols: Vec<(String, i64, f64)> = Vec::new(); // (symbol, closetime, current price)
let mut filtered_symbols_arc = Arc::new(Mutex::new(filtered_symbols));
let mut task_vec = Vec::new();
for element in filtered_6th_symbols {
let mut filtered_symbols_arc_c = Arc::clone(&filtered_symbols_arc);
let rt_price_30m_vec_c = alldata.rt_price_30m_vec.clone();
let elememt_c = element.clone();
task_vec.push(tokio::spawn(async move {
let opclo_30m_option = rt_price_30m_vec_c.iter().position(|x| *x.0 == element.0);
if opclo_30m_option.is_some() {
if rt_price_30m_vec_c[opclo_30m_option.unwrap()].1.last().is_some() {
let mut filtered_symbols_lock: tokio::sync::MutexGuard<'_, Vec<(String, i64, f64)>> = filtered_symbols_arc_c.lock().await;
filtered_symbols_lock.push((elememt_c.0, elememt_c.1, rt_price_30m_vec_c[opclo_30m_option.unwrap()].1.last().unwrap().close_price));
}
}
}));
}
try_join_all(task_vec).await?;
// 6th filtering condition: MACD // 6th filtering condition: MACD
// let mut opclo_30m_vec: Vec<RealtimePriceData> = Vec::new(); // let mut opclo_30m_vec: Vec<RealtimePriceData> = Vec::new();
// let mut ema3_1d_vec: &Vec<EmaData> = &Vec::new(); // let mut ema3_1d_vec: &Vec<EmaData> = &Vec::new();
@ -1185,8 +1208,8 @@ pub async fn execute_strategist_for_test_temp(
// } // }
// } // }
// } // }
let a = filtered_symbols_arc.lock().await.clone();
// insert_pre_suggested_coins(1, &filtered_6th_symbols, alldata).await; insert_pre_suggested_coins(1, &a, alldata).await;
Ok(()) Ok(())
} }
@ -6550,17 +6573,23 @@ pub async fn execute_strategist_for_test_temp(
// } // }
// useful functions for strategists // useful functions for strategists
pub async fn get_current_price(symbol: &String, price_vec: &Vec<CoinPriceData>) -> Option<f64> { pub async fn get_current_price(symbol: &String, rt_price_vec: &Vec<(String, Vec<RealtimePriceData>)>) -> Option<f64> {
let index_result = price_vec.iter().position(|x| *x.symbol == *symbol); let index_result = rt_price_vec.iter().position(|x| *x.0 == *symbol);
match index_result { match index_result {
Some(T) => Some(price_vec[T].current_price), Some(T) => {
if rt_price_vec[T].1.last().is_some() {
Some(rt_price_vec[T].1.last().unwrap().close_price)
} else {
None
}
},
None => None, None => None,
} }
} }
async fn insert_pre_suggested_coins( async fn insert_pre_suggested_coins(
registerer: i32, registerer: i32,
filtered_symbols: &Vec<(String, i64)>, filtered_symbols: &Vec<(String, i64, f64)>,
alldata: &AllData, alldata: &AllData,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// Check the existance of record that is registered by this strategist // Check the existance of record that is registered by this strategist
@ -6665,14 +6694,11 @@ async fn insert_pre_suggested_coins(
} }
if is_dupe == false { if is_dupe == false {
let current_price = get_current_price(&filtered_element.0, &alldata.price_vec)
.await
.unwrap();
let insert_values = vec![ let insert_values = vec![
filtered_element.0.clone(), // symbol filtered_element.0.clone(), // symbol
filtered_element.1.to_string(), // close_time filtered_element.1.to_string(), // close_time
current_price.to_string(), // suggested_price filtered_element.2.to_string(), // suggested_price
current_price.to_string(), // current_price filtered_element.2.to_string(), // current_price
server_epoch().await.to_string(), // registered_server_epoch server_epoch().await.to_string(), // registered_server_epoch
0.0.to_string(), // profit_percent 0.0.to_string(), // profit_percent
0.0.to_string(), // minimum_profit_percent 0.0.to_string(), // minimum_profit_percent
@ -6745,14 +6771,11 @@ async fn insert_pre_suggested_coins(
} }
if is_dupe == false { if is_dupe == false {
let current_price = get_current_price(&filtered_element.0, &alldata.price_vec)
.await
.unwrap();
let insert_values = vec![ let insert_values = vec![
filtered_element.0.clone(), // symbol filtered_element.0.clone(), // symbol
filtered_element.1.to_string(), // close_time filtered_element.1.to_string(), // close_time
current_price.to_string(), // suggested_price filtered_element.2.to_string(), // suggested_price
current_price.to_string(), // current_price filtered_element.2.to_string(), // current_price
server_epoch().await.to_string(), // registered_server_epoch server_epoch().await.to_string(), // registered_server_epoch
0.0.to_string(), // profit_percent 0.0.to_string(), // profit_percent
0.0.to_string(), // minimum_profit_percent 0.0.to_string(), // minimum_profit_percent
@ -6815,14 +6838,11 @@ async fn insert_pre_suggested_coins(
} }
if is_dupe == false { if is_dupe == false {
let current_price = get_current_price(&filtered_element.0, &alldata.price_vec)
.await
.unwrap();
let insert_values = vec![ let insert_values = vec![
filtered_element.0.clone(), // symbol filtered_element.0.clone(), // symbol
filtered_element.1.to_string(), // close_time filtered_element.1.to_string(), // close_time
current_price.to_string(), // suggested_price filtered_element.2.to_string(), // suggested_price
current_price.to_string(), // current_price filtered_element.2.to_string(), // current_price
server_epoch().await.to_string(), // registered_server_epoch server_epoch().await.to_string(), // registered_server_epoch
0.0.to_string(), // profit_percent 0.0.to_string(), // profit_percent
0.0.to_string(), // minimum_profit_percent 0.0.to_string(), // minimum_profit_percent
@ -6869,14 +6889,11 @@ async fn insert_pre_suggested_coins(
} }
if is_dupe == false { if is_dupe == false {
let current_price = get_current_price(&filtered_element.0, &alldata.price_vec)
.await
.unwrap();
let insert_values = vec![ let insert_values = vec![
filtered_element.0.clone(), // symbol filtered_element.0.clone(), // symbol
filtered_element.1.to_string(), // close_time filtered_element.1.to_string(), // close_time
current_price.to_string(), // suggested_price filtered_element.2.to_string(), // suggested_price
current_price.to_string(), // current_price filtered_element.2.to_string(), // current_price
server_epoch().await.to_string(), // registered_server_epoch server_epoch().await.to_string(), // registered_server_epoch
0.0.to_string(), // profit_percent 0.0.to_string(), // profit_percent
0.0.to_string(), // minimum_profit_percent 0.0.to_string(), // minimum_profit_percent
@ -6949,14 +6966,11 @@ async fn insert_pre_suggested_coins(
} }
if is_dupe == false { if is_dupe == false {
let current_price = get_current_price(&filtered_element.0, &alldata.price_vec)
.await
.unwrap();
let insert_values = vec![ let insert_values = vec![
filtered_element.0.clone(), // symbol filtered_element.0.clone(), // symbol
filtered_element.1.to_string(), // close_time filtered_element.1.to_string(), // close_time
current_price.to_string(), // suggested_price filtered_element.2.to_string(), // suggested_price
current_price.to_string(), // current_price filtered_element.2.to_string(), // current_price
server_epoch().await.to_string(), // registered_server_epoch server_epoch().await.to_string(), // registered_server_epoch
0.0.to_string(), // profit_percent 0.0.to_string(), // profit_percent
0.0.to_string(), // minimum_profit_percent 0.0.to_string(), // minimum_profit_percent
@ -7013,14 +7027,11 @@ async fn insert_pre_suggested_coins(
} }
if is_dupe == false { if is_dupe == false {
let current_price = get_current_price(&filtered_element.0, &alldata.price_vec)
.await
.unwrap();
let insert_values = vec![ let insert_values = vec![
filtered_element.0.clone(), // symbol filtered_element.0.clone(), // symbol
filtered_element.1.to_string(), // close_time filtered_element.1.to_string(), // close_time
current_price.to_string(), // suggested_price filtered_element.2.to_string(), // suggested_price
current_price.to_string(), // current_price filtered_element.2.to_string(), // current_price
server_epoch().await.to_string(), // registered_server_epoch server_epoch().await.to_string(), // registered_server_epoch
0.0.to_string(), // profit_percent 0.0.to_string(), // profit_percent
0.0.to_string(), // minimum_profit_percent 0.0.to_string(), // minimum_profit_percent
@ -7067,14 +7078,11 @@ async fn insert_pre_suggested_coins(
} }
if is_dupe == false { if is_dupe == false {
let current_price = get_current_price(&filtered_element.0, &alldata.price_vec)
.await
.unwrap();
let insert_values = vec![ let insert_values = vec![
filtered_element.0.clone(), // symbol filtered_element.0.clone(), // symbol
filtered_element.1.to_string(), // close_time filtered_element.1.to_string(), // close_time
current_price.to_string(), // suggested_price filtered_element.2.to_string(), // suggested_price
current_price.to_string(), // current_price filtered_element.2.to_string(), // current_price
server_epoch().await.to_string(), // registered_server_epoch server_epoch().await.to_string(), // registered_server_epoch
0.0.to_string(), // profit_percent 0.0.to_string(), // profit_percent
0.0.to_string(), // minimum_profit_percent 0.0.to_string(), // minimum_profit_percent
@ -7086,14 +7094,11 @@ async fn insert_pre_suggested_coins(
} }
} else { } else {
for filtered_element in filtered_symbols { for filtered_element in filtered_symbols {
let current_price = get_current_price(&filtered_element.0, &alldata.price_vec)
.await
.unwrap();
let insert_values = vec![ let insert_values = vec![
filtered_element.0.clone(), // symbol filtered_element.0.clone(), // symbol
filtered_element.1.to_string(), // close_time filtered_element.1.to_string(), // close_time
current_price.to_string(), // suggested_price filtered_element.2.to_string(), // suggested_price
current_price.to_string(), // current_price filtered_element.2.to_string(), // current_price
server_epoch().await.to_string(), // registered_server_epoch server_epoch().await.to_string(), // registered_server_epoch
0.0.to_string(), // profit_percent 0.0.to_string(), // profit_percent
0.0.to_string(), // minimum_profit_percent 0.0.to_string(), // minimum_profit_percent