New dema function

src/value_estimation_team/indicators/dema.rs

@@ -0,0 +1,122 @@
+#![allow(unused)]
+#![allow(warnings)]
+use crate::database_control::*;
+use crate::value_estimation_team::datapoints::price_data::RealtimePriceData;
+use futures::future::try_join_all;
+use serde::Deserialize;
+use sqlx::FromRow;
+use std::{iter::zip, sync::Arc};
+use tokio::{fs::*, io::AsyncWriteExt, sync::Mutex, time::*};
+use super::FilteredDataValue;
+use std::collections::HashMap;
+use super::ema::{ema, EmaData};
+
+#[derive(Clone, Debug)]
+pub struct DemaData {
+    pub dema_value: f64,
+    pub close_time: i64,
+}
+impl DemaData {
+    fn new() -> DemaData {
+        let a = DemaData {
+            dema_value: 0.0,
+            close_time: 0,
+        };
+        a
+    }
+}
+
+// Binance DEMA (closeprice)
+pub async fn dema(
+    moving_number: usize,
+    input_rt_data: &HashMap<String, Vec<RealtimePriceData>>,
+    filtered_symbols: &HashMap<String, FilteredDataValue>,
+) -> Result<HashMap<String, Vec<DemaData>>, Box<dyn std::error::Error + Send + Sync>> {
+    if filtered_symbols.is_empty() {
+        Err("Err")?;
+    }
+    let e1 = ema(moving_number, input_rt_data, filtered_symbols).await?;
+
+    // calculate EMA(EMA(n))
+    let alpha: f64 = 2.0 / (moving_number as f64 + 1.0);
+    let mut ema_t: f64 = 0.0;
+    let mut ema_prev: f64 = 0.0;
+
+    let mut ema_data_wrapper: HashMap<String, Vec<EmaData>> = HashMap::new();
+    let mut ema_data_wrapper_arc = Arc::new(Mutex::new(ema_data_wrapper));
+    let mut task_vec = Vec::new();
+    for (symbol, e1_ema_vec) in &e1 {
+        let ema_data_wrapper_arc_c = Arc::clone(&ema_data_wrapper_arc);
+        let mut ema_data = EmaData::new();
+        let mut ema_data_vec: Vec<EmaData> = Vec::new();
+        let symbol_c = symbol.clone();
+        let e1_ema_vec_c = e1_ema_vec.clone();
+        task_vec.push(tokio::spawn(async move {
+            if e1_ema_vec_c.len() > moving_number {
+                let partial_vec1 = e1_ema_vec_c.get(..moving_number).unwrap();
+                let partial_vec2 = e1_ema_vec_c.get(moving_number..).unwrap();
+
+                let mut sma_for_initial_value = 0.0;
+                for element in partial_vec1 {
+                    sma_for_initial_value += element.ema_value;
+                }
+                sma_for_initial_value /= moving_number as f64;
+
+                ema_data.ema_value = sma_for_initial_value;
+                ema_data.close_time = partial_vec1.last().unwrap().close_time;
+                ema_data_vec.push(ema_data.clone());
+
+                ema_prev = sma_for_initial_value;
+
+                for element in partial_vec2 {
+                    ema_t = (1.0 - alpha) * ema_prev + alpha * element.ema_value;
+
+                    ema_data.ema_value = ema_t;
+                    ema_data.close_time = element.close_time;
+                    ema_data_vec.push(ema_data.clone());
+
+                    ema_prev = ema_t;
+                }
+                let mut ema_data_wrapper_lock = ema_data_wrapper_arc_c.lock().await;
+                ema_data_wrapper_lock.insert(symbol_c, ema_data_vec.clone());
+            }
+        }));
+        
+    }
+    try_join_all(task_vec).await?;
+    let e2 = ema_data_wrapper_arc.lock().await.to_owned();
+
+    let mut dema_data_wrapper: HashMap<String, Vec<DemaData>> = HashMap::new();
+    let mut dema_data_wrapper_arc = Arc::new(Mutex::new(dema_data_wrapper));
+    let mut task_vec = Vec::new();
+    for (symbol, e2_vec) in e2 {
+        if let Some(e1_vec) = e1.get(&symbol) {
+            let dema_data_wrapper_arc_c: Arc<Mutex<HashMap<String, Vec<DemaData>>>> = Arc::clone(&dema_data_wrapper_arc);
+            let symbol_c = symbol.clone();
+            let e1_vec_c = e1_vec.clone();
+            task_vec.push(tokio::spawn(async move {
+                if e2_vec.last().unwrap().close_time == e1_vec_c.last().unwrap().close_time && 
+                e2_vec.len() < e1_vec_c.len() {
+                    let mut dema_data_vec: Vec<DemaData> = Vec::new();
+                    let e1_vec_part = e1_vec_c.get(e2_vec.len()-2..).unwrap();
+                    let zipped = e1_vec_part.iter().zip(e2_vec.iter());
+                    for element in zipped {
+                        if element.0.close_time != element.1.close_time {
+                            panic!();
+                        }
+                        let mut dema_data = DemaData::new();
+                        dema_data.close_time = element.0.close_time;
+                        dema_data.dema_value = (2.0 * element.0.ema_value) - element.1.ema_value;
+                        dema_data_vec.push(dema_data);
+                    }
+                    let mut dema_data_wrapper_lock = dema_data_wrapper_arc_c.lock().await;
+                    dema_data_wrapper_lock.insert(symbol_c, dema_data_vec.clone());
+                }
+            }));
+        }
+    }
+    try_join_all(task_vec).await?;
+    let a = dema_data_wrapper_arc.lock().await.to_owned();
+
+    Ok(a)
+}

src/value_estimation_team/indicators/ema.rs

@@ -16,7 +16,7 @@ pub struct EmaData {
     pub close_time: i64,
 }
 impl EmaData {
-    fn new() -> EmaData {
+    pub fn new() -> EmaData {
         let a = EmaData {
             ema_value: 0.0,
             close_time: 0,

src/value_estimation_team/indicators/mod.rs

@@ -8,6 +8,7 @@ pub mod sma;
 pub mod stoch_rsi;
 pub mod supertrend;
 pub mod tema;
+pub mod dema;
 
 use crate::strategy_team::FilteredDataValue;
 use crate::value_estimation_team::datapoints::price_data::RealtimePriceData;