Repository: yasstake/rbot (language: python)

Issue:
Archiveディレクトリを複数プロセスで利用したときに変更の通知ができない
自分で認識しているよりも新しいクエリが来た時はアーカイブディレクトリに新しいファイルがないか確認するチェックを入れる必要がある。


Discussion / hints:

Fix patch:
diff --git a/modules/rbot_lib/src/db/archive.rs b/modules/rbot_lib/src/db/archive.rs
index 2f17013..7b57dff 100644
--- a/modules/rbot_lib/src/db/archive.rs
+++ b/modules/rbot_lib/src/db/archive.rs
@@ -6,6 +6,7 @@ use crate::{
     net::{check_exist, RestApi},
 };
 use anyhow::{anyhow, Context};
+use arrow::temporal_conversions::MICROSECONDS;
 use futures::StreamExt;
 use parquet::{file::reader::SerializedFileReader, record::RowAccessor};
 use reqwest::Client;
@@ -30,6 +31,8 @@ use tempfile::tempdir;
 
 const EXTENSION: &str = "parquet";
 
+const ARCHIVE_CHECK_INTERVAL: MicroSec = 10 * 60 * MICROSECONDS;
+
 ///
 ///Archive CSV format
 ///
@@ -54,7 +57,7 @@ pub struct TradeArchive {
     start_time: MicroSec,
     end_time: MicroSec,
 
-    download_per_cent: i64,
+    end_time_update_t: MicroSec,
 }
 
 impl Clone for TradeArchive {
@@ -66,7 +69,7 @@ impl Clone for TradeArchive {
             latest_archive_date: self.latest_archive_date.clone(),
             start_time: self.start_time.clone(),
             end_time: self.end_time.clone(),
-            download_per_cent: 0,
+            end_time_update_t: 0,
         };
 
         let r = archive.analyze();
@@ -87,7 +90,7 @@ impl TradeArchive {
             latest_archive_date: 0,
             start_time: 0,
             end_time: 0,
-            download_per_cent: 0,
+            end_time_update_t: 0,            
         };
 
         let r = my.analyze();
@@ -113,7 +116,8 @@ impl TradeArchive {
         self.start_time
     }
 
-    pub fn end_time(&self) -> MicroSec {
+    pub fn end_time(&mut self) -> MicroSec {
+        self.update_end_time();
         self.end_time
     }
 
@@ -261,7 +265,7 @@ impl TradeArchive {
 
     /// load from archived paquet file retrive specifed time frame.
     pub fn select_cachedf(
-        &self,
+        &mut self,
         start_time: MicroSec,
         end_time: MicroSec,
     ) -> anyhow::Result<DataFrame> {
@@ -283,7 +287,7 @@ impl TradeArchive {
     }
 
     /// load from parquet file and retrive as cachedf.
-    pub fn load_cache_df(&self, date: MicroSec) -> anyhow::Result<DataFrame> {
+    pub fn load_cache_df(&mut self, date: MicroSec) -> anyhow::Result<DataFrame> {
         let date = FLOOR_DAY(date);
 
         if date < self.start_time() {
@@ -326,7 +330,7 @@ impl TradeArchive {
 
     /// execute f for each rec in archive within specifed time frame.
     pub fn foreach<F>(
-        &self,
+        &mut self,
         start_time: MicroSec,
         end_time: MicroSec,
         f: &mut F,
@@ -384,7 +388,7 @@ impl TradeArchive {
 
     ///
     pub fn select_dates(
-        &self,
+        &mut self,
         start_time: MicroSec,
         end_time: MicroSec,
     ) -> anyhow::Result<Vec<MicroSec>> {
@@ -434,6 +438,24 @@ impl TradeArchive {
         Ok(timestamp)
     }
 
+    pub fn update_end_time(&mut self) -> anyhow::Result<()>{
+        if   self.end_time_update_t < NOW() + ARCHIVE_CHECK_INTERVAL {
+            return Ok(());
+        }
+
+        let dates = self.list_dates()?;
+
+        let len = dates.len();
+        if len == 0 {
+            return Err(anyhow!("no data in archive"));
+        }
+
+        self.end_time = dates[len -1 ] + DAYS(1);
+
+        self.end_time_update_t = NOW();
+        Ok(())
+    }
+
     /// analyze archive directory, find start_time, end_time
     /// if there is fragmented date, delete it.
     /// Note: endday means next day's begining(archive date 24:00 = next day 00:00)
@@ -473,6 +495,8 @@ impl TradeArchive {
 
                 if detect_gap {
                     log::warn!("Delete too old log {:?}", date_string(d));
+
+                    // TODO: for some exchange, this must be moved(should be backed up the data)
                     self.delete(d)?;
                 }
             }
@@ -483,7 +507,7 @@ impl TradeArchive {
         Ok(())
     }
 
-    /// get list of archive date in order(older first)
+    /// get list of archive date in order
     pub fn list_dates(&self) -> anyhow::Result<Vec<MicroSec>> {
         let mut dates: Vec<MicroSec> = vec![];
 
@@ -567,7 +591,6 @@ impl TradeArchive {
         T: RestApi,
         F: FnMut(i64, i64),
     {
-        self.download_per_cent = 0;
         let config = &self.config.clone();
 
         let has_csv_file = self.has_local_archive(date);
diff --git a/modules/rbot_lib/src/db/tradedf.rs b/modules/rbot_lib/src/db/tradedf.rs
index 072bf30..5beb84c 100644
--- a/modules/rbot_lib/src/db/tradedf.rs
+++ b/modules/rbot_lib/src/db/tradedf.rs
@@ -94,7 +94,7 @@ impl TradeDataFrame {
         self.archive.start_time()
     }
 
-    pub fn get_archive_end_time(&self) -> MicroSec {
+    pub fn get_archive_end_time(&mut self) -> MicroSec {
         self.archive.end_time()
     }
 
@@ -122,7 +122,7 @@ impl TradeDataFrame {
         db_start
     }
 
-    pub fn end_time(&self) -> MicroSec {
+    pub fn end_time(&mut self) -> MicroSec {
         let archive_end = self.archive.end_time();
 
         let db_end = self.get_db_end_time(Self::archive_emd_default());
@@ -479,7 +479,7 @@ impl TradeDataFrame {
         return Ok(df);
     }
 
-    pub fn info(&self) -> String {
+    pub fn info(&mut self) -> String {
         let min = self.start_time();
         let max = self.end_time();
 
@@ -490,7 +490,7 @@ impl TradeDataFrame {
         );
     }
 
-    pub fn _repr_html_(&self) -> String {
+    pub fn _repr_html_(&mut self) -> String {
         let min = self.start_time();
         let max = self.end_time();
 
diff --git a/modules/rbot_market/src/market.rs b/modules/rbot_market/src/market.rs
index e1142f6..7eb0cae 100644
--- a/modules/rbot_market/src/market.rs
+++ b/modules/rbot_market/src/market.rs
@@ -12,7 +12,6 @@ use rbot_lib::common::AccountCoins;
 use rbot_lib::common::LogStatus;
 use rbot_lib::common::MarketMessage;
 
-use rbot_lib::db::TradeArchive;
 use rbot_lib::db::TradeDataFrame;
 use rbot_lib::db::TradeDb;
 use rbot_lib::net::TradePage;
@@ -385,7 +384,7 @@ where
 
     fn get_archive_info(&self) -> anyhow::Result<(MicroSec, MicroSec)> {
         let db = self.get_db();
-        let lock = db.lock().unwrap();
+        let mut lock = db.lock().unwrap();
 
         let start_time = lock.get_archive_start_time();
         let end_time = lock.get_archive_end_time();
@@ -448,13 +447,13 @@ where
 
     fn info(&mut self) -> String {
         let db = self.get_db();
-        let lock = db.lock().unwrap();
+        let mut lock = db.lock().unwrap();
         lock.info()
     }
 
     fn _repr_html_(&self) -> String {
         let db = self.get_db();
-        let lock = db.lock().unwrap();
+        let mut lock = db.lock().unwrap();
         lock._repr_html_()
     }
 
@@ -601,7 +600,7 @@ where
     ) -> anyhow::Result<(MicroSec, MicroSec, MarketStream)> {
         let (sender, market_stream) = MarketStream::open();
 
-        let archive = {
+        let mut archive = {
             let db = self.get_db();
             let trade_dataframe = db.lock().unwrap();
             trade_dataframe.get_archive()


Test patch:
diff --git a/exchanges/bybit/src/archive_test.rs b/exchanges/bybit/src/archive_test.rs
index 5228782..0cf24e6 100644
--- a/exchanges/bybit/src/archive_test.rs
+++ b/exchanges/bybit/src/archive_test.rs
@@ -52,7 +52,7 @@ mod archive_test {
 
     #[test]
     fn test_foreach_count() -> anyhow::Result<()> {
-        let archive = create_archive();
+        let mut archive = create_archive();
         init_debug_log();
 
         let mut rec: i64 = 0;
@@ -101,7 +101,7 @@ mod archive_test {
 
     #[test]
     fn test_load_cache_df() -> anyhow::Result<()> {
-        let archive = create_archive();
+        let mut archive = create_archive();
         init_debug_log();
 
 
@@ -114,7 +114,7 @@ mod archive_test {
 
     #[test]
     fn test_select_cache_df() -> anyhow::Result<()> {
-        let archive = create_archive();
+        let mut archive = create_archive();
         init_debug_log();
 
         let df = archive.select_cachedf(0, 0)?;
@@ -244,7 +244,7 @@ mod archive_test {
     #[test]
     fn test_select_dates() {
         init_debug_log();
-        let archive = create_archive();
+        let mut archive = create_archive();
 
         let dates = archive.select_dates(0, 0);
         log::debug!("{:?}", dates);
@@ -259,7 +259,7 @@ mod archive_test {
     #[test]
     fn test_select_df_perf() -> anyhow::Result<()> {
         init_debug_log();
-        let archive = create_archive();
+        let mut archive = create_archive();
 
         let now = NOW();