Repository: yasstake/rbot (language: python)

Issue:
DataBase dirtyフラグの設定
データーベスをTESTNETとREALNETに繋げた場合、
データが混ざって汚れる。

フラグを立てて、Warningを出す必要がある。

単純にDIRTYフラグだけでOKかも？


Discussion / hints:
トランザクションIDにTESTというPREFIXをつけてもいいかも。

Fix patch:
diff --git a/src/common/order.rs b/src/common/order.rs
index 232c96bf..96b51b4a 100644
--- a/src/common/order.rs
+++ b/src/common/order.rs
@@ -197,6 +197,58 @@ impl From<&String> for OrderType {
     }
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Display, Serialize, Deserialize)]
+#[pyclass]
+pub enum LogStatus {
+    UnFix,           // データはWebSocketなどから取得されたが、まだ確定していない
+    FixBlockStart,   // データが確定(アーカイブ）し、ブロックの開始を表す
+    FixArchiveBlock, // データが確定(アーカイブ）し、ブロックの中間を表す（アーカイブファイル）
+    FixBlockEnd,     // データが確定(アーカイブ）し、ブロックの終了を表す
+    FixRestApiStart,
+    FixRestApiBlock, // データが確定(アーカイブ）し、ブロックの中間を表す（REST API）
+    FixRestApiEnd,
+    Unknown, // 未知のステータス / 未確定のステータス
+}
+
+impl Default for LogStatus {
+    fn default() -> Self {
+        LogStatus::UnFix
+    }
+}
+
+impl From<&str> for LogStatus {
+    fn from(status: &str) -> Self {
+        match status.to_uppercase().as_str() {
+            "U" => LogStatus::UnFix,
+            "S" => LogStatus::FixBlockStart,
+            "A" => LogStatus::FixArchiveBlock,
+            "E" => LogStatus::FixBlockEnd,
+            "s" => LogStatus::FixRestApiStart,
+            "a" => LogStatus::FixRestApiBlock,
+            "e" => LogStatus::FixRestApiEnd,
+            _ => {
+                log::error!("Unknown log status: {:?}", status);
+                LogStatus::Unknown
+            }
+        }
+    }
+}
+
+impl LogStatus {
+    pub fn to_string(&self) -> String {
+        match self {
+            LogStatus::UnFix => "U".to_string(),
+            LogStatus::FixBlockStart => "S".to_string(),
+            LogStatus::FixArchiveBlock => "A".to_string(),
+            LogStatus::FixBlockEnd => "E".to_string(),
+            LogStatus::FixRestApiStart => "s".to_string(),
+            LogStatus::FixRestApiBlock => "a".to_string(),
+            LogStatus::FixRestApiEnd => "e".to_string(),
+            LogStatus::Unknown => "X".to_string(),
+        }
+    }
+}
+
 // Represent one Trade execution.
 #[pyclass]
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -210,6 +262,8 @@ pub struct Trade {
     pub price: Decimal,
     /// The size of the trade.
     pub size: Decimal,
+    /// trade status
+    pub status: LogStatus,
     /// The unique identifier for the trade.
     pub id: String,
 }
@@ -222,6 +276,7 @@ impl Trade {
         order_side: OrderSide,
         price: Decimal,
         size: Decimal,
+        status: LogStatus,
         id: String,
     ) -> Self {
         return Trade {
@@ -229,14 +284,15 @@ impl Trade {
             order_side,
             price,
             size,
+            status,
             id,
         };
     }
 
     pub fn to_csv(&self) -> String {
         format!(
-            "{:?}, {:?}, {:?}, {:?}, {:?}\n",
-            self.time, self.order_side, self.price, self.size, self.id
+            "{:?}, {:?}, {:?}, {:?}, {:?}, {:?}\n",
+            self.time, self.order_side, self.price, self.size, self.status, self.id
         )
     }
 
@@ -818,14 +874,13 @@ impl Order {
 
             // update foreign
             self.free_foreign_change = self.foreign_change;
-            self.lock_foreign_change = dec![0.0];            
+            self.lock_foreign_change = dec![0.0];
 
             if self.is_maker {
-                // Maker -> homeでlockしたものがforeignに変化する。freeは変化しない。                
-                self.lock_home_change = - filled_quote_vol;
-                self.free_home_change = dec![0.0];                
-            }
-            else {
+                // Maker -> homeでlockしたものがforeignに変化する。freeは変化しない。
+                self.lock_home_change = -filled_quote_vol;
+                self.free_home_change = dec![0.0];
+            } else {
                 // Taker -> lockは０。freeだったものからforeignに変化する。
                 self.lock_home_change = dec![0.0];
                 self.free_home_change = self.home_change;
@@ -833,7 +888,7 @@ impl Order {
         } else {
             // Sell
             // move foreign to home
-            self.foreign_change = (-filled_size) - self.commission_foreign;            
+            self.foreign_change = (-filled_size) - self.commission_foreign;
             self.home_change = filled_quote_vol - self.commission_home;
 
             // update home
@@ -843,22 +898,25 @@ impl Order {
             if self.is_maker {
                 self.lock_foreign_change = -filled_size;
                 self.free_foreign_change = dec![0.0];
-            }
-            else {
+            } else {
                 self.lock_foreign_change = dec![0.0];
                 self.free_foreign_change = self.foreign_change;
             }
         }
 
         // home_change = home_free_change + home_locked_change - commission_home
-        if self.home_change != (self.free_home_change + self.lock_home_change - self.commission_home) {
+        if self.home_change
+            != (self.free_home_change + self.lock_home_change - self.commission_home)
+        {
             println!(
                 "Order.update_balance_filled: home_change != (free_home_change + lock_home_change). order={:?}",
                 self
             )
         }
 
-        if self.foreign_change != (self.free_foreign_change + self.lock_foreign_change - self.commission_foreign) {
+        if self.foreign_change
+            != (self.free_foreign_change + self.lock_foreign_change - self.commission_foreign)
+        {
             println!(
                 "Order.update_balance_filled: foreign_change != (free_foreign_change + lock_foreign_change). order={:?}",
                 self
diff --git a/src/db/sqlite.rs b/src/db/sqlite.rs
index ffea0361..93ce881c 100644
--- a/src/db/sqlite.rs
+++ b/src/db/sqlite.rs
@@ -8,13 +8,15 @@ use polars_core::prelude::IndexOrder;
 use pyo3::{Py, PyResult, Python};
 use pyo3_polars::PyDataFrame;
 use rusqlite::params_from_iter;
-use rusqlite::{params, Connection, Error, Result, Statement, Transaction};
-use rust_decimal::Decimal;
+use rusqlite::{params, Connection, Error, Result, Transaction};
 use rust_decimal::prelude::FromPrimitive;
 use rust_decimal::prelude::ToPrimitive;
+use rust_decimal::Decimal;
 
+use crate::common::LogStatus;
+use crate::common::OrderSide;
+use crate::common::{time_string, MicroSec, CEIL, DAYS, FLOOR_DAY, FLOOR_SEC, NOW};
 use crate::common::{TimeChunk, Trade};
-use crate::common::{time_string, MicroSec, CEIL, DAYS, FLOOR_SEC, FLOOR_DAY, NOW};
 use crate::db::df::merge_df;
 use crate::db::df::ohlcvv_df;
 use crate::db::df::ohlcvv_from_ohlcvv_df;
@@ -22,37 +24,16 @@ use crate::db::df::select_df;
 use crate::db::df::start_time_df;
 use crate::db::df::TradeBuffer;
 use crate::db::df::{end_time_df, make_empty_ohlcvv, ohlcv_df, ohlcv_from_ohlcvv_df};
-use crate::common::OrderSide;
 
 use crate::db::df::KEY;
 use polars::prelude::Float64Type;
 
-use std::thread;
 use crossbeam_channel::Sender;
+use std::thread;
 
 use super::df::convert_timems_to_datetime;
 use super::df::vap_df;
 
-pub trait TradeTableQuery {
-    fn open(name: &str) -> Result<Self, Error>
-    where
-        Self: Sized;
-    fn create_table_if_not_exists(&self);
-    fn drop_table(&self);
-    fn vaccum(&self);
-    fn recreate_table(&self);
-
-    fn select<F>(&mut self, start_time: MicroSec, end_time: MicroSec, f: F)
-    where
-        F: FnMut(&Trade);
-
-    // fn select<F>(&mut self, start_time: MicroSec, to_time: MicroSec, f: dyn FnMut(&Trade));
-    fn select_all_statement(&self) -> Statement;
-    fn select_statement(&self, start_time: MicroSec, end_time: MicroSec) -> (Statement, Vec<i64>);
-    fn start_time(&self) -> Result<MicroSec, Error>;
-    fn end_time(&self) -> Result<MicroSec, Error>;
-}
-
 #[derive(Debug)]
 pub struct TradeTableDb {
     pub file_name: String,
@@ -61,7 +42,6 @@ pub struct TradeTableDb {
 
 impl TradeTableDb {
     pub fn clone_connection(&self) -> TradeTableDb {
-
         let conn = Connection::open(&self.file_name).unwrap();
         let db = TradeTableDb {
             file_name: self.file_name.clone(),
@@ -75,8 +55,8 @@ impl TradeTableDb {
     pub fn insert_transaction(tx: &Transaction, trades: &Vec<Trade>) -> Result<i64, Error> {
         let mut insert_len = 0;
 
-        let sql = r#"insert or replace into trades (time_stamp, action, price, size, id)
-                                values (?1, ?2, ?3, ?4, ?5) "#;
+        let sql = r#"insert or replace into trades (time_stamp, action, price, size, status, id)
+                                values (?1, ?2, ?3, ?4, ?5, ?6) "#;
 
         for rec in trades {
             let result = tx.execute(
@@ -86,6 +66,7 @@ impl TradeTableDb {
                     rec.order_side.to_string(),
                     rec.price.to_f64().unwrap(),
                     rec.size.to_f64().unwrap(),
+                    rec.status.to_string(),
                     rec.id
                 ],
             );
@@ -95,7 +76,7 @@ impl TradeTableDb {
                     insert_len += size;
                 }
                 Err(e) => {
-                    println!("insert error {}", e);
+                    log::error!("insert error {}", e);
                     return Err(e);
                 }
             }
@@ -106,7 +87,10 @@ impl TradeTableDb {
 
     pub fn insert_records(&mut self, trades: &Vec<Trade>) -> Result<i64, Error> {
         // create transaction with immidate mode
-        let tx = self.connection.transaction_with_behavior(rusqlite::TransactionBehavior::Deferred).unwrap();
+        let tx = self
+            .connection
+            .transaction_with_behavior(rusqlite::TransactionBehavior::Deferred)
+            .unwrap();
 
         let insert_len = Self::insert_transaction(&tx, trades)?;
 
@@ -132,8 +116,7 @@ impl TradeTableDb {
                 if mode == "wal" {
                     log::debug!("wal mode already set");
                     return true;
-                }
-                else {
+                } else {
                     return false;
                 }
             }
@@ -165,7 +148,6 @@ impl TradeTableDb {
         match result {
             Ok(_result) => {
                 println!("set wal mode result success");
-
             }
             Err(e) => {
                 println!("set wal mode error = {}", e);
@@ -173,16 +155,52 @@ impl TradeTableDb {
         }
 
         println!("set wal mode end");
-        
+
         let _ = conn.close();
-        return true;        
+        return true;
     }
+
+    /*
+    pub fn select_start_end_rec(&self, date: MicroSec) -> Result<Vec<Trade>, Error> {
+        let date_start = FLOOR_DAY(date);
+        let date_end = date_start + DAYS(1);
+
+        let sql = r#"select time_stamp, action, price, size, status, id from trades where $1 < time_stamp and time_stamp < $2 and (status = "S" or status = "E")"#;
+
+        let mut trades: Vec<Trade> = vec![];
+
+        let statement = self.connection.prepare(sql).unwrap();
+
+        let result = statement.query_map([date_start, date_end], |row| {
+            let time_stamp: i64 = row.get(0).unwrap();
+            let bs_str: String = row.get(1).unwrap();
+            let bs: OrderSide = OrderSide::from(&bs_str);
+            let price: f64 = row.get(2).unwrap();
+            let price = Decimal::from_f64(price).unwrap();
+            let size: f64 = row.get(3).unwrap();
+            let size = Decimal::from_f64(size).unwrap();
+            let status_str: String = row.get(4).unwrap();
+            let status = LogStatus::from(status_str.as_str());
+            let id: String = row.get(5).unwrap();
+
+            let trade = Trade::new(time_stamp, bs, price, size, status, id);
+                
+            trades.push(trade);
+        });
+
+        if result.is_err() {
+            log::error!("select start end rec error {:?}", result);
+            return Err(result.unwrap_err());
+        }
+
+        Ok(trades)
+    }
+    */
 }
 
-impl TradeTableQuery for TradeTableDb {
+impl TradeTableDb {
     fn open(name: &str) -> Result<Self, Error> {
         log::debug!("open database {}", name);
-        Self::set_wal_mode(name);
 
         let result = Connection::open(name);
         log::debug!("Database open path = {}", name);
@@ -195,7 +213,7 @@ impl TradeTableQuery for TradeTableDb {
                 };
 
                 Ok(db)
-            },
+            }
             Err(e) => {
                 log::debug!("{:?}", e);
                 return Err(e);
@@ -203,47 +221,70 @@ impl TradeTableQuery for TradeTableDb {
         }
     }
 
-
-    fn create_table_if_not_exists(&self) {
+    fn create_table_if_not_exists(&self) -> Result<(), Error> {
         let _r = self.connection.execute(
             "CREATE TABLE IF NOT EXISTS trades (
             time_stamp    INTEGER,
             action  TEXT,
             price   NUMBER,
             size    NUMBER,
+            status  TEXT,
             id      TEXT primary key
         )",
             (),
         );
 
-        let _r = self.connection.execute(
+        let r = self.connection.execute(
             "CREATE index if not exists time_index on trades(time_stamp)",
             (),
         );
+        if r.is_err() {
+            log::error!("create table error {:?}", r);
+            r.unwrap();
+        }
 
-        let _r = self.connection.execute(
-            "PRAGMA  journal_mode=wal",
-            (),
-        );
+        let r = self.connection.pragma_update(None, "journal_mode", "wal"); 
+        if r.is_err() {
+            log::error!("set wal mode error {:?}", r);
+            r.unwrap();
+        }
+
+        Ok(())
     }
 
-    fn drop_table(&self) {
-        let _r = self.connection.execute("drop table trades", ());
+    fn drop_table(&self) -> Result<(), Error> {
+        let r = self.connection.execute("drop table trades", ());
+
+        if r.is_err() {
+            log::error!("drop table error {:?}", r);
+            r.unwrap();
+        }
+
+        Ok(())
     }
 
-    fn vaccum(&self) {
-        let _r = self.connection.execute("VACCUM", ());
+    fn vaccum(&self) -> Result<(), Error> {
+        let r = self.connection.execute("VACCUM", ());
+
+        if r.is_err() {
+            log::error!("vaccum error {:?}", r);
+            r.unwrap();
+        }
+
+        Ok(())
     }
 
-    fn recreate_table(&self) {
-        self.create_table_if_not_exists();
-        self.drop_table();
-        self.create_table_if_not_exists();
+    fn recreate_table(&self) -> Result<(), Error> {
+        self.create_table_if_not_exists()?;
+        self.drop_table()?;
+        self.create_table_if_not_exists()?;
+
+        Ok(())
     }
 
     // 時間選択は左側は含み、右側は含まない。
     // 0をいれたときは全件検索
-    fn select<F>(&mut self, start_time: MicroSec, end_time: MicroSec, mut f: F)
+    pub fn select<F>(&mut self, start_time: MicroSec, end_time: MicroSec, mut f: F)
     where
         F: FnMut(&Trade),
     {
@@ -251,11 +292,11 @@ impl TradeTableQuery for TradeTableDb {
         let param: Vec<i64>;
 
         if 0 < end_time {
-            sql = "select time_stamp, action, price, size, id from trades where $1 <= time_stamp and time_stamp < $2 order by time_stamp";
+            sql = "select time_stamp, action, price, size, status, id from trades where $1 <= time_stamp and time_stamp < $2 order by time_stamp";
             param = vec![start_time, end_time];
         } else {
             //sql = "select time_stamp, action, price, size, liquid, id from trades where $1 <= time_stamp order by time_stamp";
-            sql = "select time_stamp, action, price, size, id from trades where $1 <= time_stamp order by time_stamp";
+            sql = "select time_stamp, action, price, size, status, id from trades where $1 <= time_stamp order by time_stamp";
             param = vec![start_time];
         }
 
@@ -267,13 +308,16 @@ impl TradeTableQuery for TradeTableDb {
             .query_map(params_from_iter(param.iter()), |row| {
                 let bs_str: String = row.get_unwrap(1);
                 let bs: OrderSide = bs_str.as_str().into();
+                let status_str: String = row.get_unwrap(4);
+                let status = LogStatus::from(status_str.as_str());
 
                 Ok(Trade {
                     time: row.get_unwrap(0),
                     price: Decimal::from_f64(row.get_unwrap(2)).unwrap(),
                     size: Decimal::from_f64(row.get_unwrap(3)).unwrap(),
                     order_side: bs,
-                    id: row.get_unwrap(4),
+                    status: status,
+                    id: row.get_unwrap(5),
                 })
             })
             .unwrap();
@@ -290,10 +334,45 @@ impl TradeTableQuery for TradeTableDb {
         }
     }
 
+    pub fn select_query(&mut self, sql: &str, param: Vec<i64>) -> Vec<Trade> {
+        let mut statement = self.connection.prepare(sql).unwrap();
+        let mut trades: Vec<Trade> = vec![];
+
+        let _transaction_iter = statement
+            .query_map(params_from_iter(param.iter()), |row| {
+                let bs_str: String = row.get_unwrap(1);
+                let bs: OrderSide = bs_str.as_str().into();
+                let status_str: String = row.get_unwrap(4);
+                let status = LogStatus::from(status_str.as_str());
+
+                Ok(Trade {
+                    time: row.get_unwrap(0),
+                    price: Decimal::from_f64(row.get_unwrap(2)).unwrap(),
+                    size: Decimal::from_f64(row.get_unwrap(3)).unwrap(),
+                    order_side: bs,
+                    status: status,
+                    id: row.get_unwrap(5),
+                })
+            })
+            .unwrap();
+
+        for trade in _transaction_iter {
+            match trade {
+                Ok(t) => {
+                    trades.push(t);
+                }
+                Err(e) => log::error!("{:?}", e),
+            }
+        }
+
+        return trades;
+    }
+
+    /*
     fn select_all_statement(&self) -> Statement {
         let statement = self
             .connection
-            .prepare("select time_stamp, action, price, size, id from trades order by time_stamp")
+            .prepare("select time_stamp, action, price, size, status, id from trades order by time_stamp")
             .unwrap();
         return statement;
     }
@@ -303,10 +382,10 @@ impl TradeTableQuery for TradeTableDb {
         let param: Vec<i64>;
 
         if 0 < end_time {
-            sql = "select time_stamp, action, price, size, id from trades where $1 <= time_stamp and time_stamp < $2 order by time_stamp";
+            sql = "select time_stamp, action, price, size, status, id from trades where $1 <= time_stamp and time_stamp < $2 order by time_stamp";
             param = vec![start_time, end_time];
         } else {
-            sql = "select time_stamp, action, price, size, id from trades where $1 <= time_stamp order by time_stamp";
+            sql = "select time_stamp, action, price, size, status, id from trades where $1 <= time_stamp order by time_stamp";
             param = vec![start_time];
         }
 
@@ -338,6 +417,7 @@ impl TradeTableQuery for TradeTableDb {
 
         return r;
     }
+    */
 }
 
 #[derive(Debug)]
@@ -360,7 +440,7 @@ impl TradeTable {
             let handle = self.handle.take().unwrap();
 
             if handle.is_finished() == false {
-                println!("thread is already started");
+                log::debug!("thread is already started");
 
                 // check self.tx is valid and return clone of self.tx
                 if self.tx.is_some() {
@@ -376,7 +456,7 @@ impl TradeTable {
         self.tx = Some(tx);
 
         let handle = thread::spawn(move || {
-            let mut db = TradeTableDb::open(file_name.as_str()).unwrap();                                                    
+            let mut db = TradeTableDb::open(file_name.as_str()).unwrap();
             loop {
                 match rx.recv() {
                     Ok(trades) => {
@@ -629,7 +709,6 @@ impl TradeTable {
         return Ok(r);
     }
 
-
     pub fn py_ohlcvv_polars(
         &mut self,
         mut start_time: MicroSec,
@@ -639,13 +718,12 @@ impl TradeTable {
         start_time = TradeTable::ohlcv_start(start_time); // 開始tickは確定足、終了は未確定足もOK.
 
         let mut df = self.ohlcvv_df(start_time, end_time, window_sec);
-        
+
         let df = convert_timems_to_datetime(&mut df).clone();
 
         return Ok(PyDataFrame(df));
     }
 
-
     pub fn ohlcv_df(
         &mut self,
         start_time: MicroSec,
@@ -705,7 +783,6 @@ impl TradeTable {
         return Ok(r);
     }
 
-
     pub fn py_ohlcv_polars(
         &mut self,
         mut start_time: MicroSec,
@@ -715,18 +792,17 @@ impl TradeTable {
         start_time = TradeTable::ohlcv_start(start_time); // 開始tickは確定足、終了は未確定足もOK.
 
         let mut df = self.ohlcv_df(start_time, end_time, window_sec);
-        let df = convert_timems_to_datetime(&mut df).clone();        
+        let df = convert_timems_to_datetime(&mut df).clone();
         let df = PyDataFrame(df);
 
         return Ok(df);
     }
 
     pub fn py_vap(
-
         &mut self,
         start_time: MicroSec,
-        end_time: MicroSec
-        , price_unit: i64
+        end_time: MicroSec,
+        price_unit: i64,
     ) -> PyResult<PyDataFrame> {
         let df = self.vap(start_time, end_time, price_unit);
 
@@ -735,13 +811,8 @@ impl TradeTable {
         Ok(py_df)
     }
 
-    pub fn vap(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-        price_unit: i64
-    ) -> DataFrame {
-        self.update_cache_df(start_time, end_time);                        
+    pub fn vap(&mut self, start_time: MicroSec, end_time: MicroSec, price_unit: i64) -> DataFrame {
+        self.update_cache_df(start_time, end_time);
         let df = vap_df(&self.cache_df, start_time, end_time, price_unit);
 
         df
@@ -752,14 +823,13 @@ impl TradeTable {
         start_time: MicroSec,
         end_time: MicroSec,
     ) -> PyResult<PyDataFrame> {
-        let mut df  = self.select_df_from_db(start_time, end_time);
-        let df = convert_timems_to_datetime(&mut df).clone();        
+        let mut df = self.select_df_from_db(start_time, end_time);
+        let df = convert_timems_to_datetime(&mut df).clone();
         let df = PyDataFrame(df);
 
         return Ok(df);
     }
 
-
     pub fn py_select_trades(
         &mut self,
         start_time: MicroSec,
@@ -776,7 +846,11 @@ impl TradeTable {
         return Ok(r);
     }
 
-    pub fn select_array(&mut self, start_time: MicroSec, end_time: MicroSec) -> ndarray::Array2<f64> {
+    pub fn select_array(
+        &mut self,
+        start_time: MicroSec,
+        end_time: MicroSec,
+    ) -> ndarray::Array2<f64> {
         self.update_cache_df(start_time, end_time);
 
         let trades = self.select_df_from_db(start_time, end_time);
@@ -1042,7 +1116,7 @@ impl TradeTable {
         start_time: MicroSec,
         force: bool,
     ) -> Vec<i64> {
-        let from_time = FLOOR_DAY(NOW()) - DAYS(ndays);  
+        let from_time = FLOOR_DAY(NOW()) - DAYS(ndays);
 
         let time_gap = if force {
             vec![TimeChunk {
@@ -1086,8 +1160,8 @@ impl TradeTable {
     }
 }
 
-impl TradeTableQuery for TradeTable {
-    fn open(name: &str) -> Result<Self, Error> {
+impl TradeTable {
+    pub fn open(name: &str) -> Result<Self, Error> {
         let result = TradeTableDb::open(name);
         log::debug!("Database open path = {}", name);
 
@@ -1114,19 +1188,22 @@ impl TradeTableQuery for TradeTable {
         }
     }
 
-    fn create_table_if_not_exists(&self) {
-        self.connection.create_table_if_not_exists();
+    pub fn create_table_if_not_exists(&self) -> Result<(), Error> {
+        self.connection.create_table_if_not_exists()
     }
-    fn drop_table(&self) {
-        self.connection.drop_table();
+
+    pub fn drop_table(&self) -> Result<(), Error> {
+        self.connection.drop_table()
     }
-    fn vaccum(&self) {
-        self.connection.vaccum();
+
+    pub fn vaccum(&self) -> Result<(), Error> {
+        self.connection.vaccum()
     }
-    fn recreate_table(&self) {
-        self.connection.recreate_table();
+
+    pub fn recreate_table(&self) -> Result<(), Error> {
+        self.connection.recreate_table()
     }
-    
+
     /// Selects trades from the database within the specified time range and applies the given closure to each trade.
     ///
     /// # Arguments
@@ -1158,22 +1235,6 @@ impl TradeTableQuery for TradeTable {
         self.connection.select(start_time, end_time, f);
     }
 
-    // fn select<F>(&mut self, from_time: MicroSec, to_time: MicroSec, f: dyn FnMut(&Trade));
-    fn select_all_statement(&self) -> Statement {
-        return self.connection.select_all_statement();
-    }
-    fn select_statement(&self, start_time: MicroSec, end_time: MicroSec) -> (Statement, Vec<i64>) {
-        return self.connection.select_statement(start_time, end_time);
-    }
-
-    fn start_time(&self) -> Result<MicroSec, Error> {
-        self.connection.start_time()
-    }
-
-    /// select max(end) time_stamp in db
-    fn end_time(&self) -> Result<MicroSec, Error> {
-        self.connection.end_time()
-    }
 }
 
 ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -1184,6 +1245,7 @@ impl TradeTableQuery for TradeTable {
 mod test_transaction_table {
     use rust_decimal_macros::dec;
 
+    use crate::common::init_debug_log;
     use crate::common::init_log;
     use crate::common::time_string;
     use crate::common::DAYS;
@@ -1195,27 +1257,56 @@ mod test_transaction_table {
 
     #[test]
     fn test_open() {
-        let _result = TradeTable::open("test.db");
+        let result = TradeTable::open("test.db");
+
+        assert!(result.is_ok());
     }
 
     #[test]
     fn test_create_table_and_drop() {
         let tr = TradeTable::open("test.db").unwrap();
 
-        tr.create_table_if_not_exists();
-        tr.drop_table();
+        let r = tr.create_table_if_not_exists();
+        assert!(r.is_ok());
+        
+        let r = tr.drop_table();
+        assert!(r.is_ok());
     }
 
     #[test]
     fn test_insert_table() {
+        init_debug_log();
         let mut tr = TradeTable::open("test.db").unwrap();
-        tr.recreate_table();
-
-        let rec1 = Trade::new(1, OrderSide::Buy, dec![10.0], dec![10.0], "abc1".to_string());
-        let rec2 = Trade::new(2, OrderSide::Buy, dec![10.1], dec![10.2], "abc2".to_string());
-        let rec3 = Trade::new(3, OrderSide::Buy, dec![10.2], dec![10.1], "abc3".to_string());
+        let r = tr.recreate_table();
+        assert!(r.is_ok());
+
+        let rec1 = Trade::new(
+            1,
+            OrderSide::Buy,
+            dec![10.0],
+            dec![10.0],
+            LogStatus::UnFix,
+            "abc1".to_string(),
+        );
+        let rec2 = Trade::new(
+            2,
+            OrderSide::Buy,
+            dec![10.1],
+            dec![10.2],
+            LogStatus::UnFix,
+            "abc2".to_string(),
+        );
+        let rec3 = Trade::new(
+            3,
+            OrderSide::Buy,
+            dec![10.2],
+            dec![10.1],
+            LogStatus::UnFix,
+            "abc3".to_string(),
+        );
 
-        let _r = tr.insert_records(&vec![rec1, rec2, rec3]);
+        let r = tr.insert_records(&vec![rec1, rec2, rec3]);
+        assert!(r.is_ok());
     }
 
     #[test]
@@ -1223,16 +1314,17 @@ mod test_transaction_table {
         test_insert_table();
 
         let mut table = TradeTable::open("test.db").unwrap();
-        println!("0-0");
 
         table.select(0, 0, |row| println!("{:?}", row));
     }
 
     #[test]
     fn test_select_array() {
-        let db_name = db_full_path("BN", "SPOT", "BTCBUSD");
+        init_debug_log();
 
-        let mut db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        let db = TradeTable::open("test.db");
+        assert!(db.is_ok());
+        let mut db = db.unwrap();
 
         let array = db.select_array(0, 0);
 
@@ -1241,16 +1333,15 @@ mod test_transaction_table {
 
     #[test]
     fn test_info() {
-        let db_name = db_full_path("BN", "SPOT", "BTCBUSD");
-
-        let mut db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        init_debug_log();
+        let mut db = TradeTable::open("test.db").unwrap();
         println!("{}", db.info());
     }
 
     #[test]
     fn test_start_time() {
-        let db_name = db_full_path("BN", "SPOT", "BTCBUSD");
-        let db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        init_debug_log();
+        let db = TradeTable::open("test.db").unwrap();
 
         let start_time = db.start_time();
 
@@ -1261,8 +1352,8 @@ mod test_transaction_table {
 
     #[test]
     fn test_end_time() {
-        let db_name = db_full_path("BN", "SPOT", "BTCBUSD");
-        let db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        init_debug_log();
+        let db = TradeTable::open("test.db").unwrap();
 
         let end_time = db.end_time();
 
@@ -1426,30 +1517,49 @@ mod test_transaction_table {
 
     #[test]
     fn test_start_thread() {
-        let mut table = TradeTable::open(db_full_path("BN", "SPOT", "BTCBUSD").to_str().unwrap()).unwrap(); 
+        let mut table =
+            TradeTable::open(db_full_path("BN", "SPOT", "BTCBUSD").to_str().unwrap()).unwrap();
         let tx = table.start_thread();
 
-        let v = vec![Trade{ time: 1, order_side: OrderSide::Buy, price: dec![1.0], size: dec![1.0], id: "I".to_string()}];
+        let v = vec![Trade {
+            time: 1,
+            order_side: OrderSide::Buy,
+            price: dec![1.0],
+            size: dec![1.0],
+            status: LogStatus::UnFix,
+            id: "I".to_string(),
+        }];
         tx.send(v).unwrap();
 
-        let v = vec![Trade{ time: 1, order_side: OrderSide::Buy, price: dec![1.0], size: dec![1.0], id: "I".to_string()}];
+        let v = vec![Trade {
+            time: 1,
+            order_side: OrderSide::Buy,
+            price: dec![1.0],
+            size: dec![1.0],
+            status: LogStatus::UnFix,
+            id: "I".to_string(),
+        }];
         tx.send(v).unwrap();
 
-        let tx = table.start_thread();  
+        let tx = table.start_thread();
 
-        let v = vec![Trade{ time: 1, order_side: OrderSide::Buy, price: dec![1.0], size: dec![1.0], id: "B".to_string()}];
+        let v = vec![Trade {
+            time: 1,
+            order_side: OrderSide::Buy,
+            price: dec![1.0],
+            size: dec![1.0],
+            status: LogStatus::UnFix,
+            id: "B".to_string(),
+        }];
         tx.send(v).unwrap();
 
-//        sleep(Duration::from_millis(5000));      
+        //        sleep(Duration::from_millis(5000));
     }
 
     #[test]
     fn test_wal_mode() {
-        //let table = TradeTable::open(db_full_path("BN", "SPOT", "BTCBUSD").to_str().unwrap()).unwrap(); 
+        //let table = TradeTable::open(db_full_path("BN", "SPOT", "BTCBUSD").to_str().unwrap()).unwrap();
 
         TradeTableDb::set_wal_mode(db_full_path("BN", "SPOT", "BTCBUSD").to_str().unwrap());
     }
-
 }
-
-
diff --git a/src/exchange/bb/message.rs b/src/exchange/bb/message.rs
index 8aca5721..e926e1b6 100644
--- a/src/exchange/bb/message.rs
+++ b/src/exchange/bb/message.rs
@@ -7,6 +7,7 @@ use serde_derive::Serialize;
 use serde_derive::Deserialize;
 use serde_json::Value;
 
+use crate::common::LogStatus;
 use crate::common::OrderSide;
 use crate::common::Trade;
 use crate::exchange::string_to_i64;
@@ -39,8 +40,6 @@ impl BybitTradeMessage {
 }
 
 
-
-
 #[derive(Debug, Serialize, Deserialize)]
 pub struct BybitTradeMessageResult {
     category: String,     // "category":""
@@ -74,6 +73,7 @@ impl BybitTrade {
             } else {
                 OrderSide::Sell
             },
+            status: LogStatus::Unknown,
             id: self.exec_id.to_string(),
         };
     }
diff --git a/src/exchange/bb/mod.rs b/src/exchange/bb/mod.rs
index ecd4537a..71fc6e4e 100644
--- a/src/exchange/bb/mod.rs
+++ b/src/exchange/bb/mod.rs
@@ -17,14 +17,14 @@ use pyo3_polars::PyDataFrame;
 use rust_decimal::Decimal;
 use rust_decimal::prelude::FromPrimitive;
 
-use crate::common::{OrderSide, Trade, DAYS};
+use crate::common::{OrderSide, Trade, DAYS, LogStatus};
 use crate::common::NOW;
 use crate::common::{to_naive_datetime, MicroSec};
-use crate::db::sqlite::{TradeTable, TradeTableQuery};
+use crate::db::sqlite::TradeTable;
 use crate::fs::db_full_path;
 
 
-use super::{download_log, make_download_url_list};
+use super::{download_logs, make_download_url_list};
 
 // TODO: implement ByBit Config.
 #[pyclass]
@@ -53,7 +53,7 @@ impl BBMarket {
 
         let db = TradeTable::open(db_name.as_str()).expect("cannot open db");
 
-        db.create_table_if_not_exists();
+        let _ = db.create_table_if_not_exists();
 
         return BBMarket {
             name: market_name.to_string(),
@@ -83,14 +83,23 @@ impl BBMarket {
         self.db.update_cache_all();
     }
 
-    pub fn download(&mut self, ndays: i64, force: bool) -> i64 {
+    pub fn download(&mut self, ndays: i64, force: bool, verbose: bool) -> PyResult<i64> {
         let latest_time = NOW() - DAYS(1);
         let days_gap = self.db.make_time_days_chunk_from_days(ndays, latest_time, force);
         let urls: Vec<String> = make_download_url_list(self.name.as_str(), days_gap, Self::make_historical_data_url_timestamp);
         let tx = self.db.start_thread();
-        let download_rec = download_log(urls, tx, true, BBMarket::rec_to_trade);
-
-        return download_rec;
+        
+        match download_logs(urls, tx, true, verbose, BBMarket::rec_to_trade) {
+            Ok(records) => {
+                return Ok(records);
+            }
+            Err(e) => {
+                return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!(
+                    "download: {:?}",
+                    e
+                )));
+            }
+        }
     }
 
     pub fn select_trades_a(
@@ -157,7 +166,7 @@ impl BBMarket {
     }
 
     fn vaccum(&self) {
-        self.db.vaccum();
+        let _ = self.db.vaccum();
     }
 
     pub fn _repr_html_(&self) -> String {
@@ -220,7 +229,7 @@ impl BBMarket {
 
         let size = Decimal::from_f64(size).unwrap();
 
-        let trade = Trade::new(timestamp, order_side, price, size, id);
+        let trade = Trade::new(timestamp, order_side, price, size, LogStatus::FixArchiveBlock, id);
 
         return trade;
     }
diff --git a/src/exchange/binance/market.rs b/src/exchange/binance/market.rs
index a2e59406..7da25d3c 100644
--- a/src/exchange/binance/market.rs
+++ b/src/exchange/binance/market.rs
@@ -13,7 +13,7 @@ use std::sync::{Arc, Mutex};
 use std::thread::{sleep, JoinHandle, self};
 use std::time::Duration;
 
-use crate::common::{convert_pyresult_vec, MarketMessage, time_string, OrderType, OrderStatus};
+use crate::common::{convert_pyresult_vec, MarketMessage, time_string, OrderType, OrderStatus, LogStatus, FLOOR_DAY};
 use crate::common::DAYS;
 use crate::common::{convert_pyresult, MarketStream};
 use crate::common::{to_naive_datetime, MicroSec};
@@ -21,10 +21,10 @@ use crate::common::{MarketConfig, MultiChannel};
 use crate::common::{Order, OrderSide, Trade};
 use crate::common::{HHMM, NOW, TODAY};
 use crate::db::df::KEY;
-use crate::db::sqlite::{TradeTable, TradeTableQuery};
+use crate::db::sqlite::TradeTable;
 use crate::exchange::binance::message::{BinancePublicWsMessage, BinanceWsRespond};
 
-use super::message::BinanceUserStreamMessage;
+use super::message::{BinanceUserStreamMessage, BinanceMessageId};
 use super::message::{
     BinanceListOrdersResponse, BinanceOrderStatus, BinanceWsBoardUpdate,
     BinanceAccountInformation
@@ -32,11 +32,11 @@ use super::message::{
 use super::rest::{cancel_order, get_balance};
 use super::rest::cancell_all_orders;
 use super::rest::open_orders;
-use super::rest::{insert_trade_db, new_limit_order, new_market_order, order_status, trade_list};
+use super::rest::{new_limit_order, new_market_order, order_status, trade_list};
 use super::ws::listen_userdata_stream;
 
 use crate::exchange::{
-    check_exist, download_log, make_download_url_list, AutoConnectClient, OrderBook, BoardItem};
+    check_exist, AutoConnectClient, OrderBook, BoardItem, download_log};
 
 use crate::exchange::binance::config::BinanceConfig;
 
@@ -183,7 +183,10 @@ impl BinanceMarket {
 
         let db = TradeTable::open(db_name.as_str()).expect("cannot open db");
 
-        db.create_table_if_not_exists();
+        let r = db.create_table_if_not_exists();
+        if r.is_err() {
+            log::error!("Error in create_table_if_not_exists: {:?}", r);
+        }
 
         let symbol = config.trade_symbol.clone();
 
@@ -207,12 +210,42 @@ impl BinanceMarket {
         self.db.reset_cache_duration();
     }
 
-    #[pyo3(signature = (ndays, force = false))]
-    pub fn download(&mut self, ndays: i64, force: bool) -> i64 {
+    pub fn download_log(&mut self, date: MicroSec, verbose: bool) -> PyResult<i64> {
+        let date = FLOOR_DAY(date);
+
+        let url = Self::make_historical_data_url_timestamp(self.symbol.as_str(), date);
+
+        match download_log(&url, &self.db.start_thread(), false, verbose, &BinanceMarket::rec_to_trade) {
+            Ok(download_rec) => {
+                log::info!("downloaded: {}", download_rec);
+                if verbose {
+                    println!("downloaded: {}", download_rec);
+                }
+                Ok(download_rec)
+            }
+            Err(e) => {
+                log::error!("Error in download_logs: {:?}", e);
+                if verbose {
+                    println!("Error in download_logs: {:?}", e);
+                }
+                Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!(
+                    "Error in download_logs: {:?}",
+                    e
+                )))
+            }
+        }
+    }
+
+    #[pyo3(signature = (ndays, force = false, verbose=false))]
+    pub fn download(&mut self, ndays: i64, force: bool, verbose: bool) -> i64 {
         log::info!("log download: {} days", ndays);
+        if verbose {
+            println!("log download: {} days", ndays);
+        }
+
         let latest_date;
 
-        match self.get_latest_archive_timestamp() {
+        match self.get_latest_archive_date() {
             Ok(timestamp) => latest_date = timestamp,
             Err(_) => {
                 latest_date = NOW() - DAYS(2);
@@ -220,44 +253,154 @@ impl BinanceMarket {
         }
 
         log::info!("archive latest_date: {}", time_string(latest_date));
+        if verbose {
+            println!("archive latest_date: {}", time_string(latest_date));
+        }
 
-        // download from archive
-        let days_gap = self
-            .db
-            .make_time_days_chunk_from_days(ndays, latest_date, force);
-        let urls: Vec<String> = make_download_url_list(
-            self.symbol.as_str(),
-            days_gap,
-            Self::make_historical_data_url_timestamp,
-        );
-        let tx = self.db.start_thread();
-        let download_rec = download_log(urls, tx, false, BinanceMarket::rec_to_trade);
+        let mut download_rec: i64 = 0;
 
-        // self.repave_today();
+        for i in 0..ndays {
+            let date = latest_date - i * DAYS(1);
 
-        log::info!("downloaded: {}", download_rec);
+            if ! force && self.validate_db_by_date(date) {
+                log::info!("{} is valid", time_string(date));
+                
+                if verbose {
+                    println!("{} skip download", time_string(date));
+                }
+                continue;
+            }
 
-        return download_rec;
+            match self.download_log(date, verbose) {
+                Ok(rec) => {
+                    log::info!("downloaded: {}", download_rec);
+                    download_rec += rec;
+                },
+                Err(e) => {
+                    log::error!("Error in download_log: {:?}", e);
+                    if verbose {
+                        println!("Error in download_log: {:?}", e);
+                    }
+                }
+            }
+        }
+
+        // download from rest API
+        download_rec += self.download_latest(force, verbose);
+
+        download_rec
     }
 
-    pub fn repave_today(&mut self) {
-        log::info!("repave_today log download");
-        
-        let latest_date;
+    #[pyo3(signature = (force=false, verbose = false))]
+    pub fn download_latest(&mut self, force:bool, verbose: bool) -> i64 {
+        if verbose {
+            println!("start download from rest API");
+        }
 
-        match self.get_latest_archive_timestamp() {
-            Ok(timestamp) => latest_date = timestamp,
-            Err(_) => {
-                latest_date = NOW() - DAYS(1);
-            }
+        let mut start_id: BinanceMessageId = 0;
+
+        if force {
+            let (fix_id, _fix_time) = self.latest_fix_time();
+            start_id = fix_id;
+        }
+        else {
+            let (stable_id, _stable_time)= self.latest_stable_time(verbose);
+            start_id = stable_id;
         }
 
-        let tx = self.db.start_thread();
-        let r = insert_trade_db(&self.config, latest_date, tx.clone());
+        let ch = self.db.start_thread();
 
-        if r.is_err() {
-            log::error!("Error in insert_trade_db: {:?}", r);
+        let record_number = download_historical_trades_from_id(&BinanceConfig::BTCUSDT(), start_id, verbose,&mut |row|
+        {
+            ch.send(row.clone()).unwrap();
+
+            Ok(())
+        }).unwrap();
+
+        if verbose {
+            println!("REST downloaded: {}", record_number);
         }
+
+        return record_number;
+    }
+
+    #[pyo3(signature = (verbose = false))]
+    pub fn latest_stable_time(&mut self, verbose: bool) -> (BinanceMessageId, MicroSec) {
+        let sql = r#"select time_stamp, action, price, size, status, id from trades where $1 < time_stamp and status = "E" or status = "a" order by time_stamp desc"#;
+
+        let r = self.db.connection.select_query(sql, vec![NOW()-DAYS(4)]);
+
+        if r.len() == 0 {
+            log::warn!("no record");
+            return (0, 0);
+        }
+
+        let id: BinanceMessageId = r[0].id.parse().unwrap();
+
+        if verbose {
+            println!("latest_stable_message: {:?}({:?}) / message id={:?}", r[0].time, time_string(r[0].time), r[0].id);
+        }
+
+        return (id, r[0].time);
+    }
+
+    pub fn latest_fix_time(&mut self) -> (BinanceMessageId, MicroSec) {
+        let sql = r#"select time_stamp, action, price, size, status, id from trades where $1 < time_stamp and status = "E" order by time_stamp desc"#;
+
+        let r = self.db.connection.select_query(sql, vec![NOW()-DAYS(2)]);
+
+        if r.len() == 0 {
+            log::warn!("no record");
+            return (0, 0);
+        }
+
+        let id: BinanceMessageId = r[0].id.parse().unwrap();
+
+        return (id, r[0].time);
+    }
+
+    #[pyo3(signature = (allow_gap_rec=50))]
+    pub fn analyze_db(&mut self, allow_gap_rec: u64) -> i64 {
+        let mut first_id: BinanceMessageId = 0;
+        let mut first_time: MicroSec = 0;
+
+        let mut last_id: BinanceMessageId = 0;
+        let mut last_time: MicroSec = 0;
+
+        let mut gap_count: i64 = 0;
+        let mut record_count: i64 = 0;
+
+        self.db.connection.select(0, 0, |trade|{
+            if first_id == 0 {
+                first_id = trade.id.parse::<BinanceMessageId>().unwrap();
+                first_time = trade.time;
+            }
+
+            let id = trade.id.clone();
+            let id = id.parse::<BinanceMessageId>().unwrap();
+
+            let time = trade.time;
+
+            if last_id != 0 && id + allow_gap_rec < last_id {
+                println!("MISSING: FROM: {}({})  -> TO: {}({}), {}[rec]", 
+                    time_string(last_time), last_id,
+                    time_string(time), id,
+                    last_id - id - 1);
+                gap_count += 1;
+            }
+
+            last_id = id;
+            last_time = time;
+        });
+
+        println!("Database analyze / BEGIN: {}({})  -> END: {}({}), {}[rec]  / {}[gap]", 
+            time_string(first_time), first_id,
+            time_string(last_time), last_id,
+            last_id - first_id + 1,
+            gap_count
+        );
+
+        gap_count
     }
 
     pub fn cache_all_data(&mut self) {
@@ -330,7 +473,7 @@ impl BinanceMarket {
     }
 
     pub fn vaccum(&self) {
-        self.db.vaccum();
+        let _ = self.db.vaccum();
     }
 
     pub fn _repr_html_(&self) -> String {
@@ -742,7 +885,7 @@ impl BinanceMarket {
     }
 }
 
-use crate::exchange::binance::rest::get_board_snapshot;
+use crate::exchange::binance::rest::{get_board_snapshot, download_historical_trades_from_id};
 
 const HISTORY_WEB_BASE: &str = "https://data.binance.vision/data/spot/daily/trades";
 
@@ -797,67 +940,90 @@ impl BinanceMarket {
             _ => OrderSide::Unknown,
         };
 
-        let trade = Trade::new(timestamp, order_side, price, size, id);
+        let trade = Trade::new(timestamp, order_side, price, size, LogStatus::FixArchiveBlock, id);
 
         return trade;
     }
 
-    /*
-    Order book management.
-    https://binance-docs.github.io/apidocs/spot/en/#diff-depth-stream
-
-    1. Open a stream to wss://stream.binance.com:9443/ws/bnbbtc@depth.
-    2. Buffer the events you receive from the stream.
-    3. Get a depth snapshot from https://api.binance.com/api/v3/depth?symbol=BNBBTC&limit=1000 .
-    4. Drop any event where u is <= lastUpdateId in the snapshot.
-    5. The first processed event should have U <= lastUpdateId+1 AND u >= lastUpdateId+1.
-    6. While listening to the stream, each new event's U should be equal to the previous event's u+1.
-    7. The data in each event is the absolute quantity for a price level.
-    8. If the quantity is 0, remove the price level.
-    9. Receiving an event that removes a price level that is not in your local order book can happen and is normal.
-    */
-
-    fn get_latest_archive_timestamp(&self) -> Result<MicroSec, String> {
-        match self.get_latest_archive_date() {
-            Ok(date) => {
-                return Ok(date + HHMM(23, 59));
-            }
-            Err(e) => {
-                return Err(e);
-            }
-        }
-    }
-
     fn get_latest_archive_date(&self) -> Result<MicroSec, String> {
         let mut latest = TODAY();
         let mut i = 0;
 
         loop {
-            latest -= i * DAYS(1);
 
-            let url = Self::make_historical_data_url_timestamp(self.symbol.as_str(), latest);
+            let has_archive = self.has_archive(latest);
+
+            if has_archive.is_err() {
+                log::error!("Error in has_archive: {:?}", has_archive);
+                return Err(format!("Error in has_archive: {:?}", has_archive));
+            }            
 
-            if check_exist(url.as_str()) {
-                log::debug!("{} exists", url);
+            let has_archive = has_archive.unwrap();            
+
+            if has_archive {
                 return Ok(latest);
-            } else {
-                log::debug!("{} does not exist", url);
             }
+
+            latest -= DAYS(1);
             i += 1;
 
-            if i > 5 {
-                log::error!("{} does not exist", url);
-                return Err(format!("{} does not exist", url));
+            if 5 < i {
+                return Err(format!("get_latest_archive max retry error"));
             }
         }
     }
+
+    fn has_archive(&self, date: MicroSec) -> Result<bool, String> {
+        let url = Self::make_historical_data_url_timestamp(self.symbol.as_str(), date);
+
+        if check_exist(url.as_str()) {
+            log::debug!("{} exists", url);
+            return Ok(true);
+        } else {
+            log::debug!("{} does not exist", url);
+        }
+        return Ok(false);
+    }
+
+    /// Check if database is valid at the date
+    /// TODO: implement
+    fn validate_db_by_date(&mut self, date: MicroSec) -> bool {
+        let start_time = FLOOR_DAY(date);
+        let end_time = start_time + DAYS(1);
+
+        // startからendまでのレコードにS,Eが1つづつあるかどうかを確認する。
+        let sql = r#"select time_stamp, action, price, size, status, id from trades where $1 <= time_stamp and time_stamp < $2 and (status = "S" or status = "E") order by time_stamp"#;
+        let trades = self.db.connection.select_query(sql, vec![start_time, end_time]);
+
+        if trades.len() != 2 {
+            log::debug!("S,E is not 2 {}", trades.len());
+            return false;
+        }
+
+        let first = trades[0].clone();
+        let last = trades[1].clone();
+
+        // Sから始まりEでおわることを確認
+        if first.status != LogStatus::FixBlockStart && last.status != LogStatus::FixBlockEnd {
+            log::debug!("S,E is not S,E");
+            return false;
+        }
+
+        // S, Eのレコードの間が十分にあること（トラフィックにもよるが２２時間を想定）
+        if last.time - first.time < HHMM(20, 0) {
+            log::debug!("batch is too short");
+            return false;
+        }
+
+        true
+    }
 }
 
 #[cfg(test)]
 mod binance_test {
     use std::{thread::sleep, time::Duration};
 
-    use crate::common::{init_debug_log, init_log, time_string};
+    use crate::{common::{init_debug_log, init_log, time_string}, exchange::binance::rest::{download_historical_trades, download_historical_trades_from_id}};
 
     use super::*;
 
@@ -866,12 +1032,20 @@ mod binance_test {
         init_log();
         println!(
             "{}",
-            BinanceMarket::make_historical_data_url_timestamp("BTCUSD", 1)
+            BinanceMarket::make_historical_data_url_timestamp("BTCUSD", 0)
         );
         assert_eq!(
-            BinanceMarket::make_historical_data_url_timestamp("BTCUSD", 1),
-            "https://data.binance.vision/data/spot/daily/trades/BTCBUSD/BTCBUSD-trades-1970-01-01.zip"            
+            BinanceMarket::make_historical_data_url_timestamp("BTCUSD", 0),
+            "https://data.binance.vision/data/spot/daily/trades/BTCUSD/BTCUSD-trades-1970-01-01.zip"            
+        );
+
+        println!(
+            "{}",
+            BinanceMarket::make_historical_data_url_timestamp("BTCUSD", NOW())
         );
+
+        println!("{} / {}", TODAY(), time_string(TODAY()));
+        println!("{} / {}", DAYS(1), time_string(DAYS(1)));        
     }
 
     #[test]
@@ -882,7 +1056,7 @@ mod binance_test {
         println!("{}", time_string(market.db.start_time().unwrap_or(0)));
         println!("{}", time_string(market.db.end_time().unwrap_or(0)));
         println!("Let's donwload");
-        market.download(2, false);
+        market.download(2, false, true);
     }
 
     #[test]
@@ -945,20 +1119,101 @@ mod binance_test {
 
     #[test]
     fn test_latest_archive_date() {
-        let market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
+        let market = BinanceMarket::new(&BinanceConfig::TEST_BTCUSDT());
         //let market = BinanceMarket::new("BTCBUSD", true);
 
+        init_debug_log();
+
         println!("{}", time_string(market.get_latest_archive_date().unwrap()));
     }
 
     #[test]
-    fn test_latest_archive_timestamp() {
-        let market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
+    fn test_validate_db_by_days() {
+        let mut market = BinanceMarket::new(&BinanceConfig::TEST_BTCUSDT());
         //let market = BinanceMarket::new("BTCBUSD", true);
 
-        println!(
-            "{}",
-            time_string(market.get_latest_archive_timestamp().unwrap())
-        );
+        let date = market.get_latest_archive_date().unwrap();
+
+        println!("{}", time_string(date));
+        // market.download(1, false, true);
+
+        let valid = market.validate_db_by_date(date);
+
+        println!("valid: {}", valid);
+    }
+
+    #[test]
+    fn test_latest_fixed_db_time() {
+        let mut market = BinanceMarket::new(&BinanceConfig::TEST_BTCUSDT());
+        //let market = BinanceMarket::new("BTCBUSD", true);
+
+        let r = market.latest_fix_time();
+        println!("FIXTIME: {:?}", time_string(r.1));
+
+        let r = market.latest_stable_time(true);
+        println!("STABLETIME: {:?}", time_string(r.1));
+    }
+
+    #[test]
+    fn test_download_historical_data() {
+        init_debug_log();
+        let mut market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
+        //let market = BinanceMarket::new("BTCBUSD", true);
+
+        let r = market.latest_stable_time(true);
+        println!("STABLETIME: {:?}", time_string(r.1));
+
+        /*
+        let ((fid, ftime), (eid, etime)) = download_historical_trades(&BinanceConfig::BTCUSDT(), r.0, &mut |row|Ok({
+            println!("{:?}", row);
+        })).unwrap();
+        */
+
+        let ((fid, ftime), (eid, etime), rec_no) = download_historical_trades(&BinanceConfig::BTCUSDT(), 1, &mut |row|Ok({
+            println!("{:?}", row);
+        })).unwrap();
+
+        println!("{} / {}", fid, time_string(ftime));
+        println!("{} / {}", eid, time_string(etime));
+        println!("{}", rec_no);
     }
+
+    #[test]
+    fn test_download_historical_data2() {
+//        init_debug_log();
+let mut market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
+//let market = BinanceMarket::new("BTCBUSD", true);
+
+        let (stable_id, stable_time)= market.latest_stable_time(true);
+        println!("STABLETIME: {:?}", time_string(stable_time));
+
+        let rec_no = download_historical_trades_from_id(&BinanceConfig::BTCUSDT(), stable_id, true,&mut |row|Ok({
+            // println!("{:?}", row);
+        })).unwrap();
+
+        println!("{}", rec_no);
+    }
+
+    #[test]
+    fn test_analyze_db() {
+        let mut market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
+
+        let mut last_id: BinanceMessageId = 0;
+        let mut last_time: MicroSec = 0;
+
+        market.db.connection.select(0, 0, |trade|{
+            let id = trade.id.clone();
+            let id = id.parse::<BinanceMessageId>().unwrap();
+
+            let time = trade.time;
+
+            if last_id != 0 && id != last_id + 1{
+                println!("{} / {}", time, time_string(time));
+            }
+
+            last_id = id;
+            last_time = time;
+        });
+    }
+
 }
diff --git a/src/exchange/binance/message.rs b/src/exchange/binance/message.rs
index c82ee1bb..3c0c639c 100644
--- a/src/exchange/binance/message.rs
+++ b/src/exchange/binance/message.rs
@@ -11,7 +11,7 @@ use serde_derive::{Deserialize, Serialize};
 use crate::{
     common::{
         MarketMessage, 
-        Order, OrderSide, OrderStatus, OrderType, Trade, AccountStatus, orderside_deserialize, ordertype_deserialize, orderstatus_deserialize,
+        Order, OrderSide, OrderStatus, OrderType, Trade, AccountStatus, orderside_deserialize, ordertype_deserialize, orderstatus_deserialize, LogStatus,
     },
     exchange::{string_to_decimal, BoardItem},
 };
@@ -93,6 +93,7 @@ impl BinanceTradeMessage {
             } else {
                 OrderSide::Sell
             },
+            status: LogStatus::UnFix,
             id: self.id.to_string(),
         };
     }
@@ -138,6 +139,7 @@ impl BinanceWsTradeMessage {
             } else {
                 OrderSide::Sell
             },
+            status: LogStatus::UnFix,
             id: self.t.to_string(),
         };
     }
diff --git a/src/exchange/binance/rest.rs b/src/exchange/binance/rest.rs
index b4104da2..967395ed 100644
--- a/src/exchange/binance/rest.rs
+++ b/src/exchange/binance/rest.rs
@@ -1,11 +1,11 @@
 // Copyright(c) 2022-2023. yasstake. All rights reserved.
 
+use std::io;
+use std::io::Write;
+use std::io::stdout;
 use std::thread::sleep;
 use std::time::Duration;
 
-// use crossbeam_channel::Receiver;
-use crossbeam_channel::Sender;
-
 use super::message::BinanceAccountInformation;
 use super::message::BinanceCancelOrderResponse;
 use super::message::BinanceMessageId;
@@ -14,17 +14,18 @@ use super::message::BinanceOrderStatus;
 use super::message::BinanceRestBoard;
 use super::message::BinanceTradeMessage;
 use super::BinanceConfig;
-use crate::common::time_string;
+use crate::common::LogStatus;
 use crate::common::MicroSec;
 use crate::common::OrderSide;
 use crate::common::Trade;
-use crate::common::HHMM;
 use crate::common::NOW;
+use crate::common::time_string;
 use crate::exchange::rest_delete;
 use crate::exchange::rest_get;
 use crate::exchange::rest_post;
 use crate::exchange::rest_put;
 
+/*
 /// Processes recent trades for a given symbol and applies a function to the resulting vector of trades.
 ///
 /// # Arguments
@@ -53,8 +54,9 @@ where
         Err(e) => Err(e),
     }
 }
+*/
 
-/// Processes a Binance trade message and returns the first trade ID and time.
+/// Processes a Binance trade message and returns the first (trade ID and time) and last.
 ///
 /// # Arguments
 ///
@@ -93,32 +95,37 @@ where
 pub fn process_binance_trade_message<F>(
     message: &str,
     f: &mut F,
-) -> Result<(BinanceMessageId, MicroSec), String>
+) -> Result<((BinanceMessageId, MicroSec), (BinanceMessageId, MicroSec), i64), String>
 where
-    F: FnMut(Vec<Trade>) -> Result<(), String>,
+    F: FnMut(&Vec<Trade>) -> Result<(), String>,
 {
     let result = serde_json::from_str::<Vec<BinanceTradeMessage>>(message);
 
     match result {
         Ok(message) => {
             let mut trades: Vec<Trade> = vec![];
-            let mut first_id: u64 = 0;
-            let mut first_time: MicroSec = 0;
 
             for m in message {
-                let t = m.to_trade();
+                let mut t = m.to_trade();
+                t.status = LogStatus::FixRestApiBlock;
+                trades.push(t);
+            }
 
-                if m.id < first_id || first_id == 0 {
-                    first_id = m.id;
-                    first_time = t.time;
-                }
+            let l = trades.len();
 
-                trades.push(t);
+            if l == 0 {
+                return Ok(((0, 0), (0, 0), 0));
             }
 
-            (*f)(trades)?;
+            (*f)(&trades)?;
+
+            let first_id = trades[0].id.parse::<BinanceMessageId>().unwrap();
+            let first_time = trades[0].time;
 
-            return Ok((first_id, first_time));
+            let last_id: BinanceMessageId = trades[l - 1].id.parse::<BinanceMessageId>().unwrap();
+            let last_time: MicroSec = trades[l - 1].time;
+
+            return Ok(((first_id, first_time), (last_id, last_time), l as i64));
         }
         Err(e) => {
             println!("Error: {:?}", e);
@@ -139,48 +146,20 @@ where
 ///
 /// A `Result` containing a tuple of `BinanceMessageId` and `MicroSec` structs on success, or an error message on failure.
 ///
-/// # Example
-///
-/// ```
-/// use rbot::exchange::binance::rest::process_old_trade;
-///
-/// let symbol = "BTCUSDT";
-/// let from_id = 0;        // 0 means the latest trade
-///
-/// let result = process_old_trade(symbol, from_id, |trades| {
-///     Ok(())
-/// });
-///
-/// match result {
-///     Ok((message_id, microsec)) => {
-///        println!("message_id: {}, microsec: {}", message_id, microsec);
-///     }
-///     Err(e) => {
-///         // Handle error
-///     }
-/// }
-/// ```
-pub fn process_old_trade<F>(
+pub fn download_historical_trades<F>(
     config: &BinanceConfig,
-    from_id: u64,
+    from_id: BinanceMessageId,
     f: &mut F,
-) -> Result<(BinanceMessageId, MicroSec), String>
+) -> Result<((BinanceMessageId, MicroSec), (BinanceMessageId, MicroSec), i64), String>
 where
-    F: FnMut(Vec<Trade>) -> Result<(), String>,
+    F: FnMut(&Vec<Trade>) -> Result<(), String>,
 {
     let path: String;
 
-    if from_id == 0 {
-        path = format!(
-            "/api/v3/historicalTrades?symbol={}&limit=1000",
-            config.trade_symbol
-        );
-    } else {
-        path = format!(
-            "/api/v3/historicalTrades?symbol={}&fromId={}&limit=1000",
-            config.trade_symbol, from_id
-        );
-    }
+    path = format!(
+        "/api/v3/historicalTrades?symbol={}&fromId={}&limit=1000",
+        config.trade_symbol, from_id
+    );
 
     let result = rest_get(&config.rest_endpoint, path.as_str(), vec![], None, None);
 
@@ -194,7 +173,7 @@ where
 
 const PAGE_SIZE: u64 = 1000;
 const API_INTERVAL_LIMIT: i64 = 100 * 1000;
-
+/*
 pub fn insert_trade_db(
     config: &BinanceConfig,
     start_time: MicroSec,
@@ -205,10 +184,52 @@ pub fn insert_trade_db(
         Ok(())
     });
 }
+*/
+
+pub fn download_historical_trades_from_id<F>(
+    config: &BinanceConfig,
+    mut start_id: BinanceMessageId,
+    verbose: bool,
+    f: &mut F,    
+) -> Result<i64, String> 
+where
+    F: FnMut(&Vec<Trade>) -> Result<(), String>,
+{
+    let s_id : BinanceMessageId = 0;
+    let s_time: MicroSec = 0;
+    let e_id: BinanceMessageId = 0;
+    let e_time: MicroSec = 0;    
+
+    let duration = Duration::from_millis(100);
+
+    let mut records: i64 = 0;
+
+    loop {
+        let ((s_id, s_time), (e_id, e_time), count) = download_historical_trades(config, start_id, f)?;
+        log::debug!("s_id: {} / s_time: {} / e_id: {} / e_time: {}", s_id, s_time, e_id, e_time);
+        if s_id == e_id {
+            break;
+        }
+
+        records += count;
+
+        start_id = e_id + 1;
+
+        if verbose {
+            print!("Download Historical API {} - {}\r", time_string(s_time), time_string(e_time));
+            io::stdout().flush().unwrap();
+        }
+
+        sleep(duration);
+    }
+
+    Ok(records)
+}
 
+/*
 pub fn process_old_trade_from<F>(
     config: &BinanceConfig,
-    from_time: MicroSec,
+    from_: MicroSec,
     f: &mut F,
 ) -> Result<(BinanceMessageId, MicroSec), String>
 where
@@ -234,7 +255,8 @@ where
         }
     }
 }
-
+*/
+/*
 fn _process_old_trade_from<F>(
     config: &BinanceConfig,
     from_time: MicroSec,
@@ -247,14 +269,14 @@ where
     let mut count = 0;
 
     // TODO: check implementation
-    let start_time = server_time(config);
+    // let start_time = server_time(config);
 
     loop {
         let start_timer = NOW();
 
         log::debug!("from_id: {}", from_id);
 
-        let result = process_old_trade(&config, from_id, f);
+        let result = download_historical_trades(&config, from_id, f);
 
         match result {
             Ok((trade_id, trade_time)) => {
@@ -291,6 +313,7 @@ where
         }
     }
 }
+*/
 
 pub fn get_board_snapshot(config: &BinanceConfig) -> Result<BinanceRestBoard, String> {
     let path = format!("/api/v3/depth?symbol={}&limit=1000", config.trade_symbol);
@@ -309,7 +332,6 @@ pub fn get_board_snapshot(config: &BinanceConfig) -> Result<BinanceRestBoard, St
     }
 }
 
-
 use hmac::{Hmac, Mac};
 use rust_decimal::Decimal;
 use serde_json::Value;
@@ -534,7 +556,6 @@ where
     }
 }
 
-
 /// Creates a new limit order on Binance exchange.
 ///
 /// # Arguments
@@ -549,10 +570,10 @@ where
 ///
 /// A `Result` containing a `BinanceOrderResponse` object if the order was successfully placed,
 /// or an error message as a `String` if the order failed.
-/// 
+///
 /// For SPOT:
 /// https://binance-docs.github.io/apidocs/spot/en/#new-order-trade
-/// 
+///
 /// For MARGIN:
 /// https://binance-docs.github.io/apidocs/spot/en/#margin-account-new-order-trade
 pub fn new_limit_order(
@@ -566,7 +587,7 @@ pub fn new_limit_order(
     let side = order_side_string(side);
     let mut body = format!(
         "symbol={}&side={}&type=LIMIT&timeInForce=GTC&quantity={}&price={}",
-        config.trade_symbol, side, size, price 
+        config.trade_symbol, side, size, price
     );
 
     if cliend_order_id.is_some() {
@@ -577,7 +598,6 @@ pub fn new_limit_order(
     parse_response::<BinanceOrderResponse>(binance_post_sign(&config, path, body.as_str()))
 }
 
-
 /// Creates a new market order on Binance exchange.
 ///
 /// # Arguments
@@ -593,7 +613,7 @@ pub fn new_limit_order(
 ///
 /// For SPOT:
 /// https://binance-docs.github.io/apidocs/spot/en/#new-order-trade
-/// 
+///
 /// For MARGIN:
 /// https://binance-docs.github.io/apidocs/spot/en/#margin-account-new-order-trade
 pub fn new_market_order(
@@ -652,11 +672,10 @@ pub fn cancell_all_orders(
     ))
 }
 
-
 /// Get Balance from user account
 /// for SPOT
-/// https://binance-docs.github.io/apidocs/spot/en/#account-information-user_data/// 
-/// 
+/// https://binance-docs.github.io/apidocs/spot/en/#account-information-user_data///
+///
 /// for MARGIN
 /// https://binance-docs.github.io/apidocs/spot/en/#margin-account-balance-user_data
 pub fn get_balance(config: &BinanceConfig) -> Result<BinanceAccountInformation, String> {
@@ -701,8 +720,7 @@ pub fn order_status(config: &BinanceConfig) -> Result<Vec<BinanceOrderStatus>, S
 
 use crate::exchange::binance::message::BinanceListOrdersResponse;
 
-
-/// https://binance-docs.github.io/apidocs/spot/en/#current-open-orders-user_data 
+/// https://binance-docs.github.io/apidocs/spot/en/#current-open-orders-user_data
 pub fn open_orders(config: &BinanceConfig) -> Result<Vec<BinanceOrderStatus>, String> {
     let path = "/api/v3/openOrders";
     let query = format!("symbol={}", config.trade_symbol);
@@ -710,7 +728,6 @@ pub fn open_orders(config: &BinanceConfig) -> Result<Vec<BinanceOrderStatus>, St
     parse_response::<Vec<BinanceOrderStatus>>(binance_get_sign(&config, path, Some(query.as_str())))
 }
 
-
 pub fn trade_list(config: &BinanceConfig) -> Result<Vec<BinanceListOrdersResponse>, String> {
     let path = "/api/v3/myTrades";
     let query = format!("symbol={}&limit=1000", config.trade_symbol);
@@ -728,7 +745,7 @@ mod tests {
     use rust_decimal::prelude::FromPrimitive;
 
     use super::*;
-    use crate::common::{init_debug_log, time_string, HHMM, Order};
+    use crate::common::{init_debug_log, time_string, Order, HHMM};
 
     #[test]
     fn test_trade_list() {
@@ -762,6 +779,7 @@ mod tests {
         assert_eq!(trade.is_ok(), true);
     }
 
+    /*
     #[test]
     fn test_process_recent_trade() {
         let config = BinanceConfig::BTCUSDT();
@@ -793,7 +811,7 @@ mod tests {
 
         let mut latest_time = 0;
 
-        let result = process_old_trade(&config, 0, &mut |trades| {
+        let result = download_historical_trades(&config, 0, &mut |trades| {
             println!("trades: {:?}", trades);
             latest_time = trades.last().unwrap().time;
 
@@ -817,7 +835,7 @@ mod tests {
 
         let mut latest_time = 0;
 
-        let result = process_old_trade(&config, 0, &mut |trades| {
+        let result = download_historical_trades(&config, 0, &mut |trades| {
             latest_time = trades.last().unwrap().time;
 
             Ok(())
@@ -878,6 +896,8 @@ mod tests {
         println!("latest_time: {}", time_string(latest_time));
     }
 
+    */
+
     #[test]
     fn test_server_time() {
         let config = BinanceConfig::BTCUSDT();
@@ -1015,7 +1035,6 @@ mod tests {
         println!("message: {:?}", message);
     }
 
-
     #[test]
     fn test_open_orders() {
         init_debug_log();
@@ -1048,7 +1067,6 @@ mod tests {
         let balance = get_balance(&config).unwrap();
         println!("{:?}", balance);
 
-
         // make limit order
         let order = new_limit_order(
             &config,
@@ -1056,7 +1074,8 @@ mod tests {
             Decimal::from_f64(24_000.0).unwrap(),
             Decimal::from_f64(0.001).unwrap(),
             Some(&"LimitOrder-test"),
-        ).unwrap();
+        )
+        .unwrap();
 
         // cancel
 
@@ -1069,6 +1088,5 @@ mod tests {
             let result = cancel_order(&config, &order.order_id);
             println!("result: {:?}", result);
         }
-
     }
 }
diff --git a/src/exchange/rest.rs b/src/exchange/rest.rs
index 47a9c124..703e28be 100644
--- a/src/exchange/rest.rs
+++ b/src/exchange/rest.rs
@@ -3,18 +3,17 @@
 
 use std::{
     fs::File,
-    io::{copy, BufReader, Cursor, Write},
+    io::{copy, BufReader, Cursor, Write, stdout},
     path::Path,
 };
 
+use crate::common::{LogStatus, Trade};
+use crossbeam_channel::Sender;
 use csv::{self, StringRecord};
 use flate2::bufread::GzDecoder;
 use reqwest::Method;
 use tempfile::tempdir;
 use zip::ZipArchive;
-use crate::common::Trade;
-use crossbeam_channel::Sender;
-
 
 pub fn log_download_tmp(url: &str, tmp_dir: &Path) -> Result<String, String> {
     let client = reqwest::blocking::Client::new();
@@ -35,9 +34,13 @@ pub fn log_download_tmp(url: &str, tmp_dir: &Path) -> Result<String, String> {
     log::debug!(
         "Response code = {} / download size {}",
         response.status().as_str(),
-        response.content_length().unwrap_or_default()   // if error, return 0
+        response.content_length().unwrap_or_default() // if error, return 0
     );
 
+    if ! response.status().is_success() {
+        return Err(format!("Err: response code {}", response.status().as_str()));
+    }
+
     let fname = response
         .url()
         .path_segments()
@@ -91,15 +94,15 @@ where
 
     let result = log_download_tmp(url, tmp_dir.path());
 
-    let file_path: String;
-    match result {
+    let file_path = match result {
         Ok(path) => {
-            file_path = path;
+            path
         }
         Err(e) => {
+            log::error!("download error {}", e.to_string());
             return Err(e);
         }
-    }
+    };
 
     log::debug!("let's extract = {}", file_path);
 
@@ -290,71 +293,121 @@ where
     urls
 }
 
-pub fn download_log<F>(urls: Vec<String>, tx: Sender<Vec<Trade>>, has_header: bool, f: F) -> i64
+pub fn download_logs<F>(
+    urls: Vec<String>,
+    tx: Sender<Vec<Trade>>,
+    has_header: bool,
+    verbose: bool,
+    f: F,
+) -> Result<i64, String>
 where
     F: Fn(&StringRecord) -> Trade,
 {
     let mut download_rec = 0;
 
     for url in urls {
-        log::debug!("download url = {}", url);
+        match download_log(&url, &tx, has_header, verbose, &f) {
+            Ok(count) => {
+                download_rec += count;
+            }
+            Err(e) => {
+                log::error!("download error {}", e);
+                if verbose {
+                    println!("skip log url={} / {}", url, e);
+                }
+            }
+        }
+    }
 
-        let mut buffer: Vec<Trade> = vec![];
+    return Ok(download_rec);
+}
+
+pub fn download_log<F>(
+    url: &String,
+    tx: &Sender<Vec<Trade>>,
+    has_header: bool,
+    verbose: bool,
+    f: &F,
+) -> Result<i64, String>
+where
+    F: Fn(&StringRecord) -> Trade,
+{
+    if verbose {
+        print!("log download (url = {})", url);
+    }
+    let mut download_rec = 0;
 
-        let result = log_download(url.as_str(), has_header, |rec| {
-            let trade = f(&rec);
+    let mut buffer: Vec<Trade> = vec![];
+    let mut is_first_record = true;
 
-            buffer.push(trade);
+    let result = log_download(url.as_str(), has_header, |rec| {
+        let mut trade = f(&rec);
+        trade.status = LogStatus::FixArchiveBlock;
 
-            if 2000 < buffer.len() {
-                let result = tx.send(buffer.to_vec());
+        buffer.push(trade);
 
-                match result {
-                    Ok(_) => {}
-                    Err(e) => {
-                        log::error!("{:?}", e);
-                    }
-                }
-                buffer.clear();
+        if 2000 < buffer.len() {
+            if is_first_record {
+                buffer[0].status = LogStatus::FixBlockStart;
+                is_first_record = false;
             }
-        });
 
-        if buffer.len() != 0 {
             let result = tx.send(buffer.to_vec());
+
             match result {
                 Ok(_) => {}
                 Err(e) => {
                     log::error!("{:?}", e);
                 }
             }
-
             buffer.clear();
         }
+    });
+
+    let buffer_len = buffer.len();
+
+    if buffer_len != 0 {
+        buffer[buffer_len - 1].status = LogStatus::FixBlockEnd;
 
+        let result = tx.send(buffer.to_vec());
         match result {
-            Ok(count) => {
-                log::debug!("Downloaded rec = {} ", count);
-                download_rec += count;
-            }
+            Ok(_) => {}
             Err(e) => {
-                log::error!("extract err = {}", e.as_str());
+                log::error!("{:?}", e);
             }
         }
+
+        buffer.clear();
+    }
+
+    match result {
+        Ok(count) => {
+            log::debug!("Downloaded rec = {} ", count);
+            download_rec += count;
+        }
+        Err(e) => {
+            log::error!("extract err = {}", e.as_str());
+            return Err(format!("extract err = {}", e.as_str()));
+        }
     }
 
     log::debug!("download rec = {}", download_rec);
+    if verbose {
+        println!(" download complete rec = {}", download_rec);
+        let _ = stdout().flush();
+    }
 
-    return download_rec;
+    return Ok(download_rec);
 }
 
 pub fn do_rest_request(
     method: Method,
     url: &str,
     headers: Vec<(&str, &str)>,
-    body: &str
+    body: &str,
 ) -> Result<String, String> {
     let client = reqwest::blocking::Client::new();
-    
+
     let mut request_builder = client.request(method.clone(), url);
 
     // make request builder as a common function.
@@ -362,7 +415,7 @@ pub fn do_rest_request(
         request_builder = request_builder.header(key, value);
     }
 
-    if body !=  "" {
+    if body != "" {
         request_builder = request_builder.body(body.to_string());
     }
 
@@ -382,7 +435,7 @@ pub fn do_rest_request(
         "Response code = {} / download size {:?} / method({:?}) / URL = {} / path{}",
         response.status().as_str(),
         response.content_length(),
-        method, 
+        method,
         url,
         body
     );
@@ -395,7 +448,7 @@ pub fn rest_get(
     path: &str,
     headers: Vec<(&str, &str)>,
     param: Option<&str>,
-    body: Option<&str>
+    body: Option<&str>,
 ) -> Result<String, String> {
     let mut url = format!("{}{}", server, path);
     if param.is_some() {
@@ -443,7 +496,6 @@ pub fn rest_put(
     do_rest_request(Method::PUT, &url, headers, body)
 }
 
-
 pub fn restapi<F>(server: &str, path: &str, f: F) -> Result<(), String>
 where
     F: Fn(String) -> Result<(), String>,
@@ -556,5 +608,4 @@ mod test_exchange {
         .unwrap();
         println!("{}", s);
     }
-
 }
diff --git a/src/session/orderlist.rs b/src/session/orderlist.rs
index b4ff6ee2..1d4305db 100644
--- a/src/session/orderlist.rs
+++ b/src/session/orderlist.rs
@@ -191,7 +191,6 @@ impl OrderList {
                 self.list[0].execute_price = trade.price;
                 self.list[0].quote_vol = self.list[0].execute_price * self.list[0].execute_size;
 
-                remain_size = 0.into();
                 filled_orders.push(self.list[0].clone());
 
                 // TODO: calc fills and profit
diff --git a/src/session/runner.rs b/src/session/runner.rs
index 2adf06e5..8becf37e 100644
--- a/src/session/runner.rs
+++ b/src/session/runner.rs
@@ -1,9 +1,6 @@
 // Copyright(c) 2022-2023. yasstake. All rights reserved.
 
-use std::{
-    io::{stdout, Write},
-    os::unix::process::CommandExt,
-};
+use std::io::{stdout, Write};
 
 use pyo3::{pyclass, pymethods, Py, PyAny, PyErr, PyObject, Python};
 use rust_decimal::prelude::ToPrimitive;


Test patch:
diff --git a/src/session/mod_test.rs b/src/session/mod_test.rs
index b60ea9f8..28c08074 100644
--- a/src/session/mod_test.rs
+++ b/src/session/mod_test.rs
@@ -5,7 +5,7 @@ mod tests {
     use orderlist::OrderList;
     use rust_decimal_macros::dec;
 
-    use crate::{session::orderlist, common::{OrderSide, Order, NOW, OrderType, OrderStatus, Trade, init_debug_log}};
+    use crate::{session::orderlist, common::{OrderSide, Order, NOW, OrderType, OrderStatus, Trade, init_debug_log, LogStatus}};
 
     #[test]
     fn test_order_list() {
@@ -146,7 +146,9 @@ mod tests {
         // if buy trades comes in the buy trades list, its araises error log, and returns empty list.
         let trade = Trade::new(
             NOW(), OrderSide::Buy,
-            dec![150.0], dec![10.0], "ordr1".to_string());
+            dec![150.0], dec![10.0], 
+            LogStatus::UnFix,            
+            "ordr1".to_string());
 
             let filled_orders = order_list.consume_trade(&trade);
 
@@ -155,7 +157,9 @@ mod tests {
         // if sell trades and equal to the highest buy order price, it will do nothing.
         let trade2 = Trade::new(
             NOW(), OrderSide::Sell,
-            dec![250.0], dec![250.0], "ordr2".to_string());
+            dec![250.0], dec![250.0], 
+            LogStatus::UnFix,            
+            "ordr2".to_string());
 
 
         let filled_orders = order_list.consume_trade(&trade2);
@@ -164,7 +168,9 @@ mod tests {
         // Partially filled.
         let trade3 = Trade::new(
             NOW(), OrderSide::Sell,
-            dec![249.9], dec![10.0], "ordr3".to_string());
+            dec![249.9], dec![10.0],
+            LogStatus::UnFix,
+            "ordr3".to_string());
 
         let filled_orders = order_list.consume_trade(&trade3);
         assert_eq!(filled_orders.len(), 1);
@@ -176,7 +182,9 @@ mod tests {
         // Filled filled.
         let trade3 = Trade::new(
             NOW(), OrderSide::Sell,
-            dec![199.0], dec![100.0], "ordr3".to_string());
+            dec![199.0], dec![100.0], 
+            LogStatus::UnFix,
+            "ordr3".to_string());
 
         let filled_orders = order_list.consume_trade(&trade3);
         assert_eq!(filled_orders.len(), 2);
@@ -189,9 +197,6 @@ mod tests {
 
         assert_eq!(order_list.remain_size(), dec![25.0]);
 
-
-
-
 /*
 
         assert_eq!(filled_orders.len(), 2);
diff --git a/src/test/binance_test/market.ipynb b/src/test/binance_test/market.ipynb
index 122b31a2..9d2659b2 100644
--- a/src/test/binance_test/market.ipynb
+++ b/src/test/binance_test/market.ipynb
@@ -11,7 +11,7 @@
     "from IPython.display import HTML\n",
     "\n",
     "from rbot import *\n",
-    "config = BinanceConfig.TESTSPOT(\"BTC\", \"USDT\")\n",
+    "config = BinanceConfig.SPOT(\"BTC\", \"USDT\")\n",
     "\n",
     "HTML(json2html.convert(config.__str__()))"
    ]
@@ -26,7 +26,54 @@
     "from rbot import BinanceMarket\n",
     "\n",
     "          #BinanceMarket\n",
-    "binance = BinanceMarket(config)\n"
+    "binance = BinanceMarket(config)\n",
+    "\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "binance.analyze_db()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "binance.latest_fix_time()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "binance.latest_stable_time(True)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "binance.download_latest(True)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "\n",
+    "binance.download(1, True, True)"
    ]
   },
   {
diff --git a/src/test/binance_test/nullagent.py b/src/test/binance_test/nullagent.py
index 79602dec..b4aefeef 100644
--- a/src/test/binance_test/nullagent.py
+++ b/src/test/binance_test/nullagent.py
@@ -40,13 +40,15 @@ def on_update(self, session, updated_order):
     def on_account_update(self, session, account):
         print("account update: ", session.current_time, account)
         pass
-    
-market = BinanceMarket(BinanceConfig.TEST_BTCUSDT)
-market.download(30)
-
-print(BinanceConfig.TEST_BTCUSDT)
 
+   
+market = BinanceMarket(BinanceConfig.BTCUSDT)
 market.start_market_stream()
+
+
+market.download(5, force=False, verbose=True)
+
+market.analyze_db()
 #market.start_user_stream()
     
 agent = MyAgent()