Repository: yasstake/rbot (language: python)

Issue:
VACCUMとDROP＆DELETEDBを実装する
スレッドが動いている場合はエラーを表示して止まるようにする。
Market作成時に、設定の概要を印刷する。MarketConfigにSIMPLEFormatでプロパティを追加して印刷



Discussion / hints:

Fix patch:
diff --git a/python/rbot/__init__.py b/python/rbot/__init__.py
index a4c4e71f..83b2b9be 100644
--- a/python/rbot/__init__.py
+++ b/python/rbot/__init__.py
@@ -1,7 +1,4 @@
 from .rbot import *
-import pandas as pd
-import numpy as np
-import time
 import signal
 
 if hasattr(rbot, "__all__"):
@@ -12,376 +9,10 @@ def terminate(_a, _b):
     print("terminate", _a, _b)
     exit(0)
     
-signal.signal(signal.SIGTERM, terminate)
 signal.signal(signal.SIGINT, terminate)
 
-
-def decode_order_side(bs):
-    if bs == 0:
-        return "Sell"
-    elif bs == 1:
-        return "Buy"
-    else:
-        return "ERROR"
-
-
-def trades_to_df(array):
-    df = pd.DataFrame(
-        array, columns=["timestamp", "price", "size", "side"])
-    df['timestamp'] = pd.to_datetime(
-        (df["timestamp"]), utc=True, unit='us')
-    df = df.set_index('timestamp')
-
-    df['side'] = df['side'].map(decode_order_side)
-
-    return df
-
-
-def ohlcvv_to_df(array):
-    df = pd.DataFrame(
-        array, columns=["timestamp", "side", "open", "high", "low", "close", "volume",
-                        #                        "count",
-                        "start_time", "end_time"])
-    df['timestamp'] = pd.to_datetime(
-        (df["timestamp"]), utc=True, unit='us')
-    df = df.set_index('timestamp')
-
-    df['start_time'] = pd.to_datetime(
-        (df["start_time"]), utc=True, unit='us')
-
-    df['end_time'] = pd.to_datetime(
-        (df["end_time"]), utc=True, unit='us')
-
-    df['side'] = df['side'].map(decode_order_side)
-
-    return df
-
-
-def ohlcv_to_df(array):
-    df = pd.DataFrame(
-        array, columns=["timestamp", "open", "high", "low", "close", "volume",
-                        #                "count"
-                        ])
-
-    df['timestamp'] = pd.to_datetime(
-        (df["timestamp"]), utc=True, unit='us')
-    df = df.set_index('timestamp')
-
-    return df
-
-
-def result_to_df(result_list):
-    update_time = []
-    order_id = []
-    order_sub_id = []
-    order_side = []
-    post_only = []
-    create_time = []
-    status = []
-    open_price = []
-    open_home_size = []
-    open_foreign_size = []
-    close_price = []
-    close_home_size = []
-    close_foreign_size = []
-    order_price = []
-    order_home_size = []
-    order_foreign_size = []
-    profit = []
-    fee = []
-    total_profit = []
-    position_change = []
-    message = []
-
-    for item in result_list:
-        update_time.append(item.update_time)
-        order_id.append(item.order_id)
-        order_sub_id.append(item.order_sub_id)
-        order_side.append(str(item.order_side))
-        post_only.append(item.post_only)
-        create_time.append(item.create_time)
-        status.append(str(item.status))
-        open_price.append(item.open_price)
-        open_home_size.append(item.open_home_size)
-        open_foreign_size.append(item.open_foreign_size)
-        close_price.append(item.close_price)
-        close_home_size.append(item.close_home_size)
-        close_foreign_size.append(item.close_foreign_size)
-        order_price.append(item.order_price)
-        order_home_size.append(item.order_home_size)
-        order_foreign_size.append(item.order_foreign_size)
-        profit.append(item.profit)
-        fee.append(item.fee)
-        total_profit.append(item.total_profit)
-        position_change.append(item.position_change)
-        message.append(item.message)
-
-    df = pd.DataFrame(
-        data={"update_time": update_time, "order_id": order_id, "sub_id": order_sub_id,
-              "order_side": order_side,
-              # "post_only": post_only,
-              "create_time": create_time,
-              "status":  status,
-              "open_price": open_price, "open_size": open_home_size, "open_volume": open_foreign_size,
-              "close_price": close_price, "close_size": close_home_size, "close_volume": close_foreign_size,
-              "order_price": order_price, "order_size": order_home_size, "order_volume": order_foreign_size,
-              "profit": profit, "fee": fee,
-              "total_profit": total_profit, "position_change": position_change, "message": message},
-        columns=["update_time", "order_id", "sub_id", "order_side",
-                 # "post_only",
-                 "create_time", "status",
-                 "open_price", "open_size", "open_volume",
-                 "close_price", "close_size", "close_volume",
-                 "order_price", "order_size", "order_volume",
-                 "profit", "fee", "total_profit", "position_change", "message"])
-    df["update_time"] = pd.to_datetime(
-        (df["update_time"]), utc=True, unit="us")
-    df["create_time"] = pd.to_datetime(
-        (df["create_time"]), utc=True, unit="us")
-    df["sum_profit"] = df["total_profit"].cumsum()
-    df["position"] = df["position_change"].cumsum()
-    # df["sum_pos"] = df["pos_change"].cumsum()
-    df = df.set_index("create_time", drop=True)
-
-    return df
-
-
-class BaseAgent:
-    def __init__(self):
-        self._indicators = {}
-
-    def key_in_indicators(self, key):
-        if key in self._indicators:
-            return True
-        else:
-            return False
-
-    def indicator(self, key):
-        if not self.key_in_indicators(key):
-            return None
-
-        indicator_array = np.array(self._indicators[key]).T
-        s = pd.Series(data=indicator_array[1], index=indicator_array[0])
-        
-        return s
-
-    def log_indicator(self, key, time, val):
-        if not self.key_in_indicators(key):
-            self._indicators[key] = []
-
-        self._indicators[key].append([time, val])
-
-    def get_indicator_names(self):
-        names = []
-        for k in self._indicators:
-            names.append(k)
-        return names
-
-    def clock_interval(self):
-        return 60
-
-    def _on_tick(self, time, session, price, side, size):
-        self.on_tick(time, Session(session), price, side, size)
-
-    def _on_clock(self, time, session):
-        self.on_clock(time, Session(session))
-
-    def _on_update(self, time, session, result):
-        self.on_update(time, Session(session), result)
-
-"""
-class Session:
-    def __init__(self, session):
-        self.session = session
-
-    def __getattr__(self, func):
-        return getattr(self.session, func)
-
-    @property
-    def best_sell_price(self):
-        return self.session.sell_board_edge_price
-
-    @property
-    def best_buy_price(self):
-        return self.session.buy_board_edge_price
-
-    def ohlcv(self, time_window, num_of_bars, exchange_name=None, market_name=None):
-        if not exchange_name:
-            exchange_name = self.session.exchange_name
-            market_name = self.session.market_name
-
-        market = Market.open(exchange_name, market_name)
-
-        now = self.session.current_timestamp
-
-        return market.ohlcvv(now - time_window * num_of_bars * 1_000_000, now, time_window)
-
-    def _repr_html_(self):
-        html = "<table><tbody>"
-        html += "<tr><td colspan='3', align='left'>市況</td></tr>"
-        html += "<tr><td></td><td width='120'>内容</td><td>関数</td><td>出力</td></tr>"
-        html += "<tr><td></td><td>現在時刻</td><td>Session.current_timestamp</td><td>{}({})</td></tr>".format(
-            time_string(self.current_timestamp), self.current_timestamp)
-        html += "<tr><td></td><td>最終売Take価格/最終買Make価格</td><td>Session.best_buy_price</td><td>{}</td></tr>".format(
-            self.best_buy_price)
-        html += "<tr><td></td><td>最終買Take価格/最終売Make価格</td><td>Session.best_sell_price</td><td>{}</td></tr>".format(
-            self.best_sell_price)
-        html += "<tr><td colspan='3', align='left'>処理中オーダー</td></tr>"
-        html += "<tr><td>Long</td><td>買注文数量</td><td>Session.long_order_len</td><td>{}</td></tr>".format(
-            self.long_order_len)
-        html += "<tr><td></td><td>買注文一覧</td><td>Session.long_orders</td><td></td></tr>"
-        html += "<tr><td></td><td colspan='3'>{}</td></tr>".format(
-            self.long_orders)
-        html += "<tr><td>Short</td><td>売注文数量</td><td>Session.short_order_len</td><td>{}</td></tr>".format(
-            self.short_order_len)
-        html += "<tr><td></td><td>売注文一覧</td><td>Session.short_orders</td><td></td></tr>"
-        html += "<tr><td></td><<td colspan='3'>{}</td></tr>".format(
-            self.short_orders)
-        html += "<tr><td></td></tr>"
-        html += "<tr><td colspan='3', align='left'>ポジション</td></tr>"
-        html += "<tr><td>Long</td><td>平均単価</td><td>session.long_position_price</td><td>{}</td></tr>".format(
-            self.long_position_price)
-        html += "<tr><td></td><td>サイズ</td><td>session.long_position_size</td><td>{}</td></tr>".format(
-            self.long_position_size)
-        html += "<tr><td>Short</td><td>平均単価</td><td>session.short_position_price</td><td>{}</td></tr>".format(
-            self.short_position_price)
-        html += "<tr><td></td><td>サイズ</td><td>session.short_position_size</td><td>{}</td></tr>".format(
-            self.short_position_size)
-        html += "</tbody></table>"
-
-        return html
-
-"""
-
-class Market:
-    MARKET = {}
-    DUMMY_MODE = True
-
-    @classmethod
-    def dummy_mode(cls, dummy=True):
-        cls.DUMMY_MODE = dummy
-
-    @classmethod
-    def open(cls, exchange: str, market):
-        exchange = exchange.upper()
-        key = Market.key(exchange, market)
-
-        if key in cls.MARKET:
-            return cls.MARKET[key]
-
-        if exchange == "FTX":
-            m = FtxMarket(market, cls.DUMMY_MODE)
-            cls.MARKET[key] = m
-            return m
-        elif exchange == "BN":
-            m = BinanceMarket(market, cls.DUMMY_MODE)
-            cls.MARKET[key] = m
-            return m
-        elif exchange == "BB":
-            m = ByBitMarket(market, cls.DUMMY_MODE)
-            cls.MARKET[key] = m
-            return m
-        else:
-            print("unknown market ", market)
-
-    @classmethod
-    def download(cls, ndays, force=False):
-        for m in cls.MARKET:
-            cls.MARKET[m].download(ndays, force)
-            cls.MARKET[m].vaccum()
-
-    @classmethod
-    def vaccum():
-        for m in cls.MARKET:
-            cls.MARKET[m].vaccum()
-
-    @classmethod
-    def _cache_data(cls):
-        for m in cls.MARKET:
-            cls.MARKET[m].cache_all_data()
-
-    @staticmethod
-    def key(exchange, market):
-        return exchange.upper() + "/" + market.upper()
-
-
-class BackRunner:
-    def __init__(self, exchange_name, market_name, size_in_price_currency):
-        self.exchange_name = exchange_name
-        self.market_name = market_name
-        self.backtester = _BackTester(
-            self.exchange_name, self.market_name, size_in_price_currency)
-        self.agent_name = ""
-        self.last_exec_time = 0
-        self.clocl_interval = 0
-        self.result = None
-
-    def run(self, agent, start_time=0, end_time=0):
-        Market.open(self.exchange_name,
-                    self.market_name).reset_cache_duration()
-
-        counter_s = time.perf_counter()
-        self.agent_name = agent.__class__.__name__
-        self.clock_interval = agent.clock_interval()
-
-        r = self.backtester.run(agent, start_time, end_time)
-
-        counter_e = time.perf_counter()
-        self.last_exec_time = counter_e - counter_s
-
-        self.result = result_to_df(r)
-
-        return self.result
-
-    @property
-    def last_run_duration(self):
-        return self.last_run_end - self.last_run_start
-
-    def __getattr__(self, func):
-        return getattr(self.backtester, func)
-
-    def __str__(self):
-        return "start={}({}) end={}({}) {}[us] records={}".format(
-            rbot.time_string(self.last_run_start),
-            self.last_run_start,
-            rbot.time_string(self.last_run_end),
-            self.last_run_end,
-            self.last_run_duration,
-            self.last_run_count,
-        )
-
-    def _repr_html_(self):
-        table = "<table><caption>Backtest outline</caption>"
-        table += "<tr><td>Exchange name</td><td>{}</td></tr>".format(
-            self.exchange_name)
-        table += "<tr><td>Market name</td><td>{}</td></tr>".format(
-            self.market_name)
-        table += "<tr><td>Size in price currency</td><td>{}</td></tr>".format(
-            self.size_in_price_currency)
-        table += "<tr><td>maker fee rate</td><td>{} [%]</td></tr>".format(
-            self.maker_fee_rate*100)
-        table += "<tr><td>Agent class name</td><td>{}</td></tr>".format(
-            self.agent_name)
-        table += "<tr><td> enable: on_tick</td><td>{}</td><td>{:,}</td></tr>".format(
-            self.agent_on_tick, self.on_tick_count)
-        table += "<tr><td> enable: on_clock</td><td>{}</td><td>{:,}</td></tr>".format(
-            self.agent_on_clock, self.on_clock_count)
-        table += "<tr><td> enable: on_update</td><td>{}</td><td>{:,}</td></tr>".format(
-            self.agent_on_update, self.on_update_count)
-        table += "<tr><td> clock interval:  </td><td>{} [sec]</td>".format(
-            self.clock_interval)
-        table += "<tr><td>start</td><td>{} ({:,})</td></tr>".format(
-            rbot.time_string(self.last_run_start), self.last_run_start)
-        table += "<tr><td>end</td><td>{} ({:,})</td></tr>".format(
-            rbot.time_string(self.last_run_end), self.last_run_end)
-        table += "<tr><td>duration</td><td>{:,.0f} [sec] / {:.2f} [days]</td></tr>".format(
-            self.last_run_duration/1_000_000, self.last_run_duration / rbot.DAYS(1))
-        table += "<tr><td># of records</td><td>{:,} [rec]</td></tr>".format(
-            self.last_run_record)
-        table += "<tr><td># of result record</td><td>{:,.0f} [rec]</td></tr>".format(
-            len(self.result))
-        table += "<tr><td>Simulation time</td><td>{:,.0f} [sec]</td></tr></table>".format(
-            self.last_exec_time)
-
-        return table
-
+print("rbot version: ", rbot.__version__)
+print("!!! ABSOLUTELY NO WARRANTY !!!")
+print("!!!  USE AT YOUR OWN RISK  !!!")
+print("See document at https://github.com/yasstake/rusty-bot")
+print(" All rights reserved. (c) 2022-2023 rbot developers / yasstake")
diff --git a/src/db/sqlite.rs b/src/db/sqlite.rs
index 93ce881c..0eee2c2f 100644
--- a/src/db/sqlite.rs
+++ b/src/db/sqlite.rs
@@ -15,6 +15,7 @@ use rust_decimal::Decimal;
 
 use crate::common::LogStatus;
 use crate::common::OrderSide;
+use crate::common::SEC;
 use crate::common::{time_string, MicroSec, CEIL, DAYS, FLOOR_DAY, FLOOR_SEC, NOW};
 use crate::common::{TimeChunk, Trade};
 use crate::db::df::merge_df;
@@ -51,6 +52,22 @@ impl TradeTableDb {
         return db;
     }
 
+    pub fn delete_unstable_data(tx: &Transaction, start_time: MicroSec, end_time: MicroSec) {
+        let sql = r#"delete from trades where $1 <= time_stamp and time_stamp < $2 and status = "U""#;
+
+        let result = tx.execute(sql, params![start_time, end_time]);
+
+        match result {
+            Ok(rec_size) => {
+                log::debug!("delete unstable data [{}]\n", rec_size);
+            }
+            Err(e) => {
+                log::error!("delete unstable data error {}", e);
+            }
+        }
+    }
+
+    // TODO: delete before insert data.
     // insert records with param transaction and trades
     pub fn insert_transaction(tx: &Transaction, trades: &Vec<Trade>) -> Result<i64, Error> {
         let mut insert_len = 0;
@@ -86,12 +103,22 @@ impl TradeTableDb {
     }
 
     pub fn insert_records(&mut self, trades: &Vec<Trade>) -> Result<i64, Error> {
+        let trades_len = trades.len();
+        let start_time = trades[0].time - SEC(5);
+        let end_time = trades[trades_len - 1].time;        
+        
         // create transaction with immidate mode
         let tx = self
             .connection
             .transaction_with_behavior(rusqlite::TransactionBehavior::Deferred)
             .unwrap();
 
+        // when fix data comes, delete unstable data first
+        if trades_len!= 0 && trades[0].status != LogStatus::UnFix {
+            Self::delete_unstable_data(&tx, start_time, end_time);
+        }
+
+        // then insert data
         let insert_len = Self::insert_transaction(&tx, trades)?;
 
         let result = tx.commit();
@@ -257,18 +284,22 @@ impl TradeTableDb {
 
         if r.is_err() {
             log::error!("drop table error {:?}", r);
-            r.unwrap();
         }
 
+        print!("database is dropped. To use database, restart the program.");
+        print!("to delete completely, delete the database file from OS.");
+        print!("DB file = {}", self.file_name);
+
+        self.vacuum()?;
+
         Ok(())
     }
 
-    fn vaccum(&self) -> Result<(), Error> {
-        let r = self.connection.execute("VACCUM", ());
+    fn vacuum(&self) -> Result<(), Error> {
+        let r = self.connection.execute("VACUUM", ());
 
         if r.is_err() {
-            log::error!("vaccum error {:?}", r);
-            r.unwrap();
+            log::error!("vacuum error {:?}", r);
         }
 
         Ok(())
@@ -480,14 +511,14 @@ impl TradeTable {
     pub fn is_thread_running(&self) -> bool {
         if let Some(handler) = self.handle.as_ref() {
             if handler.is_finished() {
-                println!("thread is finished");
+                log::debug!("thread is finished");
                 return false;
             } else {
-                println!("thread is running");
+                log::debug!("thread is running");
                 return true;
             }
         } else {
-            println!("thread is not started");
+            log::debug!("thread is not started");
             return false;
         }
     }
@@ -1193,11 +1224,15 @@ impl TradeTable {
     }
 
     pub fn drop_table(&self) -> Result<(), Error> {
+        if self.is_thread_running() {
+            println!("WARNING: thread is running, may cause dead lock. Execute drop_table() before start_market_stream().");
+        }
+
         self.connection.drop_table()
     }
 
-    pub fn vaccum(&self) -> Result<(), Error> {
-        self.connection.vaccum()
+    pub fn vacuum(&self) -> Result<(), Error> {
+        self.connection.vacuum()
     }
 
     pub fn recreate_table(&self) -> Result<(), Error> {
diff --git a/src/exchange/bb/mod.rs b/src/exchange/bb/mod.rs
index 71fc6e4e..a112113c 100644
--- a/src/exchange/bb/mod.rs
+++ b/src/exchange/bb/mod.rs
@@ -165,8 +165,8 @@ impl BBMarket {
         return self.db.get_file_name();
     }
 
-    fn vaccum(&self) {
-        let _ = self.db.vaccum();
+    fn vacuum(&self) {
+        let _ = self.db.vacuum();
     }
 
     pub fn _repr_html_(&self) -> String {
diff --git a/src/exchange/binance/config.rs b/src/exchange/binance/config.rs
index d74b3391..ca31afed 100644
--- a/src/exchange/binance/config.rs
+++ b/src/exchange/binance/config.rs
@@ -10,6 +10,8 @@ use crate::{fs::db_full_path, common::MarketConfig};
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[pyclass]
 pub struct BinanceConfig {
+    pub test_net: bool,
+
     pub exchange_name: String,
     pub trade_category: String,
     pub trade_symbol: String,
@@ -65,6 +67,7 @@ impl BinanceConfig {
     pub fn TESTSPOT(foreign_symbol: &str, home_symbol: &str) -> Self {
         let mut config = BinanceConfig::SPOT(foreign_symbol, home_symbol);
 
+        config.test_net = true;
         config.trade_category = "TESTSPOT".to_string();
         config.rest_endpoint = "https://testnet.binance.vision".to_string();
         config.public_ws_endpoint = "wss://testnet.binance.vision/ws".to_string();
@@ -109,6 +112,7 @@ impl BinanceConfig {
 
 
         return BinanceConfig {
+            test_net: false,
             exchange_name: "BN".to_string(),
             trade_category: "SPOT".to_string(),
             trade_symbol: upper_symbol,
@@ -177,4 +181,13 @@ impl BinanceConfig {
 
         serde_json::to_string(&printobj).unwrap()
     }
+
+    pub fn short_info(&self) -> String {
+        if self.test_net {
+            return format!("---TEST NET--- {}", self.trade_symbol);
+        }
+        else {
+            return format!("*** LIVE NET *** {}", self.trade_symbol);
+        }
+    }
 }
diff --git a/src/exchange/binance/market.rs b/src/exchange/binance/market.rs
index 7da25d3c..2bfe8dda 100644
--- a/src/exchange/binance/market.rs
+++ b/src/exchange/binance/market.rs
@@ -177,6 +177,8 @@ pub struct BinanceMarket {
 impl BinanceMarket {
     #[new]
     pub fn new(config: &BinanceConfig) -> Self {
+        println!("\nBinance {:?}\n", config.short_info());
+
         let db_name = Self::db_path(&config).unwrap();
 
         log::debug!("create TradeTable: {}", db_name);
@@ -201,6 +203,16 @@ impl BinanceMarket {
         };
     }
 
+    pub fn drop_table(&mut self) -> PyResult<()> {
+        match self.db.drop_table() {
+            Ok(_) => Ok(()),
+            Err(e) => Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!(
+                "Error in drop_table: {:?}",
+                e
+            ))),
+        }
+    }
+    
     #[getter]
     pub fn get_cache_duration(&self) -> MicroSec {
         return self.db.get_cache_duration();
@@ -301,11 +313,11 @@ impl BinanceMarket {
 
         if force {
             let (fix_id, _fix_time) = self.latest_fix_time();
-            start_id = fix_id;
+            start_id = fix_id + 1;
         }
         else {
             let (stable_id, _stable_time)= self.latest_stable_time(verbose);
-            start_id = stable_id;
+            start_id = stable_id + 1;
         }
 
         let ch = self.db.start_thread();
@@ -326,7 +338,7 @@ impl BinanceMarket {
 
     #[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 sql = r#"select time_stamp, action, price, size, status, id from trades where $1 < time_stamp and status = "E" or status = "e" order by time_stamp desc"#;
 
         let r = self.db.connection.select_query(sql, vec![NOW()-DAYS(4)]);
 
@@ -391,15 +403,21 @@ impl BinanceMarket {
 
             last_id = id;
             last_time = time;
+            record_count += 1;
         });
 
-        println!("Database analyze / BEGIN: {}({})  -> END: {}({}), {}[rec]  / {}[gap]", 
+        println!("Database analyze / BEGIN: {}({})  -> END: {}({}), {}[rec]  / {}[gap] / total rec {}", 
             time_string(first_time), first_id,
             time_string(last_time), last_id,
             last_id - first_id + 1,
-            gap_count
+            gap_count,
+            record_count
         );
 
+        if 1 < gap_count {
+            println!("WARNING database has {} gaps. Download with force option, or drop-and-create database.", gap_count);
+        }
+
         gap_count
     }
 
@@ -472,8 +490,8 @@ impl BinanceMarket {
         return self.config.market_config.clone();
     }
 
-    pub fn vaccum(&self) {
-        let _ = self.db.vaccum();
+    pub fn vacuum(&self) {
+        let _ = self.db.vacuum();
     }
 
     pub fn _repr_html_(&self) -> String {
diff --git a/src/exchange/binance/rest.rs b/src/exchange/binance/rest.rs
index 967395ed..5a0df85f 100644
--- a/src/exchange/binance/rest.rs
+++ b/src/exchange/binance/rest.rs
@@ -103,8 +103,8 @@ where
 
     match result {
         Ok(message) => {
-            let mut trades: Vec<Trade> = vec![];
-
+            let mut trades: Vec<Trade> = vec![];            
+            
             for m in message {
                 let mut t = m.to_trade();
                 t.status = LogStatus::FixRestApiBlock;
@@ -117,13 +117,16 @@ where
                 return Ok(((0, 0), (0, 0), 0));
             }
 
-            (*f)(&trades)?;
-
             let first_id = trades[0].id.parse::<BinanceMessageId>().unwrap();
             let first_time = trades[0].time;
+            trades[0].status = LogStatus::FixRestApiStart;
 
             let last_id: BinanceMessageId = trades[l - 1].id.parse::<BinanceMessageId>().unwrap();
             let last_time: MicroSec = trades[l - 1].time;
+            trades[l - 1].status = LogStatus::FixRestApiEnd;
+
+
+            (*f)(&trades)?;
 
             return Ok(((first_id, first_time), (last_id, last_time), l as i64));
         }
diff --git a/src/session/runner.rs b/src/session/runner.rs
index 8becf37e..f11922f2 100644
--- a/src/session/runner.rs
+++ b/src/session/runner.rs
@@ -51,7 +51,7 @@ impl Runner {
             has_account_update: false,
             start_timestamp: 0,
             loop_duration: -1, // -1 means infinite loop
-            print_interval: SEC(10),
+            print_interval: SEC(5),
             execute_mode: ExecuteMode::BackTest,
             current_timestamp: 0,
             current_clock: 0,


Test patch:
diff --git a/src/test/binance_test/nullagent.py b/src/test/binance_test/nullagent.py
index b4aefeef..e56c063f 100644
--- a/src/test/binance_test/nullagent.py
+++ b/src/test/binance_test/nullagent.py
@@ -41,21 +41,29 @@ def on_account_update(self, session, account):
         print("account update: ", session.current_time, account)
         pass
 
+
    
 market = BinanceMarket(BinanceConfig.BTCUSDT)
-market.start_market_stream()
+#market.start_market_stream()
 
+#init_debug_log()
+
+market.analyze_db()
 
-market.download(5, force=False, verbose=True)
+market.download(2, force=False, verbose=True)
 
 market.analyze_db()
-#market.start_user_stream()
+market.start_market_stream()
+
+from time import sleep
+
+sleep(10)
     
-agent = MyAgent()
-runner = Runner()
+#agent = MyAgent()
+#runner = Runner()
 
 #runner.back_test(market, agent, interval_sec=60, start_time=0, end_time=0, verbose=True)
-runner.dry_run(market, agent, interval_sec=60, verbose=True)
+#runner.dry_run(market, agent, interval_sec=60, verbose=True)
 
 
     
\ No newline at end of file
diff --git a/src/test/market_drop_table.ipynb b/src/test/market_drop_table.ipynb
new file mode 100644
index 00000000..74efd47f
--- /dev/null
+++ b/src/test/market_drop_table.ipynb
@@ -0,0 +1,77 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from rbot import BinanceConfig, BinanceMarket"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "binance = BinanceMarket(BinanceConfig.BTCUSDT)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "binance.drop_table()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "binance"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "binance.file_name"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "venv",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.11.6"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}