Repository: yasstake/rbot (language: python)

Issue:
Staging


<!-- This is an auto-generated comment: release notes by coderabbit.ai -->

## Summary by CodeRabbit

- **新機能**
	- BinanceとBybitの取引所での市場データと取引操作を扱う新機能を追加しました。
	- BinanceとBybitのWebSocket APIを通じて公開データストリームとプライベートデータストリームを管理する機能を追加しました。
- **バグ修正**
	- Bybitの`config.rs`でのAPIキーとシークレットの扱いを改善しました。
	- 複数の取引所モジュールでのエラーハンドリングとデータキャッシュ処理を改善しました。
- **リファクタリング**
	- BinanceとBybitのモジュール構成を再編成し、コードの可読性を向上させました。
	- メッセージ処理方法を改善し、データ構造を更新しました。
- **ドキュメント**
	- 新しい機能と変更点に関するドキュメントを更新しました。

<!-- end of auto-generated comment: release notes by coderabbit.ai -->


Discussion / hints:

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index e280b186..2d67983e 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -39,8 +39,12 @@ serde = { version = "1.0"}
 serde_json = { version = "1.0" }
 serde_derive = { version = "1.0" }
 serde_with = { version = "3.6.1" }
+
 log = { version = "0.4" }
 env_logger = { version = "0.11.0" }
+tracing = { version = "0.1.40" }
+tracing-subscriber = { version = "0.3.18" }
+
 directories = "5.0.1"
 
 crossbeam-channel = { version = "0.5.11" }
diff --git a/exchanges/binance.bak/Cargo.toml b/exchanges/binance.bak/Cargo.toml
deleted file mode 100644
index 38514caa..00000000
--- a/exchanges/binance.bak/Cargo.toml
+++ /dev/null
@@ -1,42 +0,0 @@
-[package]
-name = "binance"
-version = "0.1.0"
-edition = "2021"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[dependencies]
-rbot_lib = {workspace = true}
-rbot_blockon = {workspace = true}
-rbot_market = {workspace = true}
-
-anyhow = {workspace = true}
-
-chrono = {workspace = true}
-log = {workspace = true}
-
-features = {workspace = true}
-
-rust_decimal = {workspace = true}
-rust_decimal_macros = {workspace = true}
-serde = {workspace = true}
-serde_derive = {workspace = true}
-serde_json = {workspace = true}
-
-crossbeam-channel = {workspace = true}
-csv = {workspace = true}
-polars-core = {workspace = true}
-pyo3-polars = {workspace = true}
-
-tokio = {workspace = true}
-futures = {workspace=true}
-async-stream = {workspace = true}
-
-
-# https://pyo3.rs/v0.13.2/faq
-[dependencies.pyo3]
-version = "0.20"
-features = ["rust_decimal", "auto-initialize", "abi3-py38"]
-
-[features]
-extension-module = ["pyo3/extension-module"]
diff --git a/exchanges/binance.bak/market.rs b/exchanges/binance.bak/market.rs
deleted file mode 100644
index 0e90bf5d..00000000
--- a/exchanges/binance.bak/market.rs
+++ /dev/null
@@ -1,1273 +0,0 @@
-// Copyright(c) 2022-2023. yasstake. All rights reserved.
-
-use chrono::Datelike;
-use crossbeam_channel::Sender;
-use csv::StringRecord;
-use pyo3::prelude::*;
-use pyo3_polars::PyDataFrame;
-use rust_decimal::prelude::FromPrimitive;
-use rust_decimal::Decimal;
-use rust_decimal_macros::dec;
-use serde_json::Value;
-use std::borrow::BorrowMut;
-use std::sync::{Arc, Mutex, RwLock};
-use std::thread::sleep;
-use std::time::Duration;
-use tokio::runtime::Runtime;
-use tokio::task::JoinHandle;
-
-use crate::common::DAYS;
-use crate::common::NOW;
-use crate::common::{convert_pyresult, MarketStream};
-use crate::common::{
-    convert_pyresult_vec, flush_log, time_string, LogStatus, MarketMessage, OrderStatus, OrderType,
-    FLOOR_DAY,
-};
-use crate::common::{to_naive_datetime, MicroSec};
-use crate::common::{MarketConfig, MultiChannel};
-use crate::common::{Order, OrderSide, Trade};
-use crate::db::df::KEY;
-use crate::db::sqlite::TradeTable;
-use crate::exchange::binance::message::{BinancePublicWsMessage, BinanceWsRespond};
-use crate::exchange::binance::ws::PING_INTERVAL;
-
-use super::message::{
-    BinanceAccountInformation, BinanceListOrdersResponse, BinanceOrderStatus, BinanceWsBoardUpdate,
-};
-use super::message::{BinanceMessageId, BinanceUserStreamMessage};
-use super::rest::cancell_all_orders;
-use super::rest::open_orders;
-use super::rest::{cancel_order, get_balance};
-use super::rest::{new_limit_order, new_market_order, order_status, trade_list};
-use super::ws::listen_userdata_stream;
-
-use crate::exchange::{
-    download_log, latest_archive_date, AutoConnectClient, BinanceWsOpMessage, BoardItem, OrderBook,
-    WsOpMessage,
-};
-
-use crate::exchange::binance::config::BinanceConfig;
-
-use super::ws::SWITCH_INTERVAL;
-
-#[pyclass]
-pub struct BinanceAccount {
-    pub api_key: String,
-    pub secret_key: String,
-    pub subaccount: String,
-}
-
-#[derive(Debug)]
-pub struct BinanceOrderBook {
-    config: BinanceConfig,
-    last_update_id: u64,
-    board: OrderBook,
-}
-
-impl BinanceOrderBook {
-    pub fn new(config: &BinanceConfig) -> Self {
-        return BinanceOrderBook {
-            config: config.clone(),
-            last_update_id: 0,
-            board: OrderBook::new(&config.market_config),
-        };
-    }
-
-    pub fn update(&mut self, update_data: &BinanceWsBoardUpdate) {
-        if self.last_update_id == 0 {
-            log::debug!(
-                "reflesh board {} / {}->{}",
-                self.last_update_id,
-                update_data.u,
-                update_data.U
-            );
-            sleep(Duration::from_millis(150)); // 100ms毎に更新されるので、150ms待つ。
-            self.reflesh_board();
-        }
-
-        // 4. Drop any event where u is <= lastUpdateId in the snapshot.
-        if update_data.u <= self.last_update_id {
-            log::debug!(
-                "Drop any event where u({}) is <= lastUpdateId({}) in the snapshot.",
-                update_data.u,
-                self.last_update_id
-            );
-
-            return;
-        }
-
-        // 5. The first processed event should have U <= lastUpdateId+1 AND u >= lastUpdateId+1.
-        if update_data.U <= self.last_update_id + 1 && update_data.u >= self.last_update_id + 1 {
-            log::debug!(
-                "lastupdate({}) / U({}) / u({})",
-                self.last_update_id,
-                update_data.U,
-                update_data.u
-            );
-            self.board
-                .update(&update_data.bids, &update_data.asks, false);
-        }
-
-        // 6. While listening to the stream, each new event's U should be equal to the previous event's u+1.
-        if update_data.U != self.last_update_id + 1 {
-            log::warn!(
-                "U is not equal to the previous event's u+1 {} {}",
-                update_data.U,
-                self.last_update_id + 1
-            );
-        }
-
-        self.last_update_id = update_data.u;
-    }
-
-    fn reflesh_board(&mut self) {
-        let snapshot = get_board_snapshot(&self.config).unwrap();
-        self.last_update_id = snapshot.last_update_id;
-
-        self.board.update(&snapshot.bids, &snapshot.asks, true);
-    }
-
-    fn get_board_vec(&self) -> Result<(Vec<BoardItem>, Vec<BoardItem>), ()> {
-        let (bids, asks) = self.board.get_board_vec().unwrap();
-
-        Ok((bids, asks))
-    }
-
-    fn get_board(&mut self) -> PyResult<(PyDataFrame, PyDataFrame)> {
-        let r = self.board.get_board();
-        if r.is_err() {
-            return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!(
-                "Error in get_board: {:?}",
-                r
-            )));
-        }
-
-        let (mut bids, mut asks) = r.unwrap();
-
-        if bids.is_empty() || asks.is_empty() {
-            return Ok((PyDataFrame(bids), PyDataFrame(asks)));
-        }
-
-        let bids_edge: f64 = bids.column(KEY::price).unwrap().max().unwrap();
-        let asks_edge: f64 = asks.column(KEY::price).unwrap().min().unwrap();
-
-        if asks_edge < bids_edge {
-            log::warn!("bids_edge({}) < asks_edge({})", bids_edge, asks_edge);
-
-            self.reflesh_board();
-
-            (bids, asks) = self.board.get_board().unwrap();
-        }
-
-        return Ok((PyDataFrame(bids), PyDataFrame(asks)));
-    }
-
-    fn get_edge_price(&self) -> PyResult<(Decimal, Decimal)> {
-        Ok(self.board.get_edge_price())
-    }
-}
-
-#[pyclass]
-pub struct Binance {
-    pub test_net: bool,
-}
-
-#[pymethods]
-impl Binance {
-    #[new]
-    pub fn new(test_net: bool) -> Self {
-        return Binance { test_net };
-    }
-
-    pub fn open_market(&self, config: &BinanceConfig) -> BinanceMarket {
-        return BinanceMarket::create(config);
-    }
-}
-
-const SYNC_RECORDS_FOR_PUBLIC: i64 = 3;
-
-
-#[derive(Debug)]
-#[pyclass(name = "BinanceMarket")]
-pub struct BinanceMarket {
-    pub config: BinanceConfig,
-    pub db: TradeTable,
-    pub board: Arc<Mutex<BinanceOrderBook>>,
-    pub public_handler: Option<JoinHandle<()>>,
-    pub user_handler: Option<JoinHandle<()>>,
-    pub channel: Arc<Mutex<MultiChannel<MarketMessage>>>,
-}
-
-
-#[pymethods]
-impl BinanceMarket {
-    #[staticmethod]
-    pub fn create(config: &BinanceConfig) -> Self {
-        println!("\nBinance {:?}\n", config.short_info());
-
-        let db_name = Self::db_path(&config).unwrap();
-        log::debug!("db_path: {}", db_name);
-
-        let db = TradeTable::open(db_name.as_str()).expect("cannot open db");
-        log::debug!("db is opend. db_path= {}", db_name);
-
-        log::info!("db is opend success. db_path= {}", db_name);
-
-        return BinanceMarket {
-            config: config.clone(),
-            db,
-            board: Arc::new(Mutex::new(BinanceOrderBook::new(config))),
-            public_handler: None,
-            user_handler: None,
-            channel: Arc::new(Mutex::new(MultiChannel::new())),
-        };
-    }
-
-    #[new]
-    pub fn new(config: &BinanceConfig) -> Self {
-        println!("Warning: depricated use open_market as below");
-        println!("> binance = Binance()");
-        println!("> market = binance.open_market(config)");
-
-        Self::create(config)
-    }
-
-    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_market_name(&self) -> (String, String, String) {
-        /*
-        self.server_config.exchange_name.clone(),
-        self.config.trade_category.clone(),
-        self.config.trade_symbol.clone(),
-        */
-        (
-            self.config.exchange_name.clone(),
-            self.config.market_config.trade_category.clone(),
-            self.config.market_config.trade_symbol.clone(),
-        )
-    }
-
-    #[getter]
-    pub fn get_cache_duration(&self) -> MicroSec {
-        return self.db.get_cache_duration();
-    }
-
-    pub fn reset_cache_duration(&mut self) {
-        self.db.reset_cache_duration();
-    }
-
-    #[pyo3(signature = (*, ndays, force = false, low_priority=false, verbose=true, archive_only=false))]
-    pub fn download(
-        &mut self,
-        ndays: i64,
-        force: bool,
-        low_priority: bool,
-        verbose: bool,
-        archive_only: bool,
-    ) -> i64 {
-        log::info!("log download: {} days", ndays);
-        if verbose {
-            println!("log download: {} days", ndays);
-            flush_log();
-        }
-
-        let latest_date;
-
-        match self.get_latest_archive_date() {
-            Ok(timestamp) => latest_date = timestamp,
-            Err(_) => {
-                latest_date = NOW() - DAYS(2);
-            }
-        }
-
-        log::info!("archive latest_date: {}", time_string(latest_date));
-        if verbose {
-            println!("archive latest_date: {}", time_string(latest_date));
-            flush_log();
-        }
-
-        let mut download_rec: i64 = 0;
-
-        let tx = &self.db.start_thread();
-
-        for i in 0..ndays {
-            let date = latest_date - i * DAYS(1);
-
-            if !force && self.validate_db_by_date(date) {
-                log::info!("{} is valid", time_string(date));
-
-                if verbose {
-                    println!("{} skip download", time_string(date));
-                    flush_log();
-                }
-                continue;
-            }
-
-            match self.download_log(tx, date, low_priority, 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);
-                    }
-                }
-            }
-        }
-
-        if archive_only {
-            return download_rec;
-        }
-
-        // download from rest API
-        download_rec += self.download_latest(force, verbose);
-
-        if verbose {
-            println!("\nREST downloaded: {}[rec]", download_rec);
-            print!("Waiting for Insert DB...");
-            flush_log();
-        }
-
-        // self.wait_for_settlement(tx);
-
-        if verbose {
-            println!("Done");
-            flush_log();
-        }
-
-        download_rec
-    }
-
-    pub fn stop_db_thread(&mut self) {
-        self.db.stop_thread();
-    }
-
-    #[pyo3(signature = (force=false, verbose = true))]
-    pub fn download_latest(&mut self, force: bool, verbose: bool) -> i64 {
-        if verbose {
-            println!("start download from rest API");
-            flush_log();
-        }
-
-        let start_id: u64;
-
-        if force {
-            let (fix_id, _fix_time) = self.latest_fix_time();
-            start_id = fix_id + 1;
-        } else {
-            let (stable_id, _stable_time) = self.latest_stable_time(verbose);
-            start_id = stable_id + 1;
-        }
-
-        if start_id == 1 {
-            println!(
-                "ERROR: no record in database path= {}",
-                self.get_file_name()
-            );
-            flush_log();
-
-            return 0;
-        }
-
-        let ch = self.db.start_thread();
-
-        let record_number = download_historical_trades_from_id(
-            &BinanceConfig::BTCUSDT(),
-            start_id,
-            verbose,
-            &mut |row| {
-                //TODO:
-                while 100 < ch.len() {
-                    sleep(Duration::from_millis(100));
-                }
-
-                ch.send(row.clone()).unwrap();
-
-                Ok(())
-            },
-        )
-        .unwrap();
-
-        if verbose {
-            println!("\nREST downloaded: {}[rec]", record_number);
-            flush_log();
-        }
-
-        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 = "e") 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;
-            record_count += 1;
-        });
-
-        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,
-            record_count
-        );
-
-        if 1 < gap_count {
-            println!("WARNING database has {} gaps. Download with force option, or drop-and-create database.", gap_count);
-        }
-
-        gap_count
-    }
-
-    pub fn cache_all_data(&mut self) {
-        self.db.update_cache_all();
-    }
-
-    pub fn select_trades(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-    ) -> PyResult<PyDataFrame> {
-        return self.db.py_select_trades_polars(start_time, end_time);
-    }
-
-    pub fn ohlcvv(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-        window_sec: i64,
-    ) -> PyResult<PyDataFrame> {
-        return self.db.py_ohlcvv_polars(start_time, end_time, window_sec);
-    }
-
-    pub fn ohlcv(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-        window_sec: i64,
-    ) -> PyResult<PyDataFrame> {
-        return self.db.py_ohlcv_polars(start_time, end_time, window_sec);
-    }
-
-    pub fn vap(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-        price_unit: i64,
-    ) -> PyResult<PyDataFrame> {
-        return self.db.py_vap(start_time, end_time, price_unit);
-    }
-
-    pub fn info(&mut self) -> String {
-        return self.db.info();
-    }
-
-    #[getter]
-    pub fn get_board(&self) -> PyResult<(PyDataFrame, PyDataFrame)> {
-        self.board.lock().unwrap().get_board()
-    }
-
-    #[getter]
-    pub fn get_board_vec(&self) -> PyResult<(Vec<BoardItem>, Vec<BoardItem>)> {
-        Ok(self.board.lock().unwrap().get_board_vec().unwrap())
-    }
-
-    #[getter]
-    pub fn get_edge_price(&self) -> PyResult<(Decimal, Decimal)> {
-        self.board.lock().unwrap().get_edge_price()
-    }
-
-    #[getter]
-    pub fn get_file_name(&self) -> String {
-        return self.db.get_file_name();
-    }
-
-    #[getter]
-    pub fn get_market_config(&self) -> MarketConfig {
-        return self.config.market_config.clone();
-    }
-
-    #[getter]
-    pub fn get_running(&self) -> bool {
-        return self.db.is_running();
-    }
-
-    pub fn vacuum(&self) {
-        let _ = self.db.vacuum();
-    }
-
-    pub fn _repr_html_(&self) -> String {
-        return format!(
-            "<b>Binance DB ({})</b>{}",
-            self.config.market_config.trade_symbol,
-            self.db._repr_html_()
-        );
-    }
-
-
-    pub fn start_market_stream(&mut self) {
-        let endpoint = &self.config.public_ws_endpoint;
-
-        let mut agent_channel = self.channel.clone();
-
-        // TODO: parameterize
-        let mut websocket: AutoConnectClient<BinanceConfig, BinanceWsOpMessage> =
-            AutoConnectClient::new(
-                &self.config,
-                endpoint,
-                PING_INTERVAL,
-                SWITCH_INTERVAL,
-                SYNC_RECORDS_FOR_PUBLIC,
-                None,
-            );
-
-        websocket.subscribe(&self.config.public_subscribe_channel);
-
-        websocket.connect();
-        
-        let db_channel = self.db.start_thread();
-        let board = self.board.clone();
-
-        let handler = tokio::task::spawn(async move {
-                loop {
-                    let message = websocket.receive_text().await;
-                    if message.is_err() {
-                        log::warn!("Error in websocket.receive_message: {:?}", message);
-                        continue;
-                    }
-                    let m = message.unwrap();
-
-                    let message_value = serde_json::from_str::<Value>(&m);
-
-                    if message_value.is_err() {
-                        log::warn!("Error in serde_json::from_str: {:?}", message_value);
-                        continue;
-                    }
-                    let message_value: Value = message_value.unwrap();
-
-                    if message_value.is_object() {
-                        let o = message_value.as_object().unwrap();
-
-                        if o.contains_key("e") {
-                            log::debug!("Message: {:?}", &m);
-
-                            let message: BinancePublicWsMessage = serde_json::from_str(&m).unwrap();
-
-                            match message.clone() {
-                                BinancePublicWsMessage::Trade(trade) => {
-                                    log::debug!("Trade: {:?}", trade);
-                                    let r = db_channel.send(vec![trade.to_trade()]);
-
-                                    if r.is_err() {
-                                        log::error!(
-                                            "Error in db_channel.send: {:?} {:?}",
-                                            trade,
-                                            r
-                                        );
-                                    }
-
-                                    let multi_agent_channel = agent_channel.borrow_mut();
-
-                                    let r =
-                                        multi_agent_channel.lock().unwrap().send(message.into());
-
-                                    if r.is_err() {
-                                        log::error!(
-                                            "Error in agent_channel.send: {:?} {:?}",
-                                            trade,
-                                            r
-                                        );
-                                    }
-                                }
-                                BinancePublicWsMessage::BoardUpdate(board_update) => {
-                                    board.lock().unwrap().update(&board_update);
-                                }
-                            }
-                        } else if o.contains_key("result") {
-                            let message: BinanceWsRespond = serde_json::from_str(&m).unwrap();
-                            log::debug!("Result: {:?}", message);
-                        } else {
-                            continue;
-                        }
-                    }
-                }
-        });
-        self.public_handler = Some(handler);
-
-        log::info!("start_market_stream");
-    }
-
-
-    pub fn start_user_stream(&mut self) {
-        let mut agent_channel = self.channel.clone();
-
-        let cfg = self.config.clone();
-
-        self.user_handler = Some(listen_userdata_stream(
-            &self.config,
-            move |message: BinanceUserStreamMessage| {
-                log::debug!("UserStream: {:?}", message);
-                let mutl_agent_channel = agent_channel.borrow_mut();
-                let m = message.convert_to_market_message(&cfg);
-                let _ = mutl_agent_channel.lock().unwrap().send(m);
-            },
-        ));
-
-        log::info!("start_user_stream");
-    }
-
-    /*
-    pub fn stop_user_stream(&mut self) {
-        match self.user_handler.take() {
-            Some(h) => {
-                h.join().unwrap();
-            }
-            None => {}
-        }
-    }
-    */
-
-    pub fn is_user_stream_running(&self) -> bool {
-        if let Some(handler) = &self.user_handler {
-            return !handler.is_finished();
-        }
-        return false;
-    }
-
-    pub fn is_market_stream_running(&self) -> bool {
-        if let Some(handler) = &self.public_handler {
-            return !handler.is_finished();
-        }
-        return false;
-    }
-
-    pub fn is_db_thread_running(&self) -> bool {
-        return self.db.is_thread_running();
-    }
-
-    #[getter]
-    pub fn get_channel(&mut self) -> MarketStream {
-
-        // TODO: confirm buffer size
-        let ch = self.channel.lock().unwrap().open_channel(1000);
-
-        return MarketStream { reciver: ch };
-    }
-
-    pub fn open_backtest_channel(
-        &mut self,
-        time_from: MicroSec,
-        time_to: MicroSec,
-    ) -> MarketStream {
-        let channel = self.channel.lock().unwrap().open_channel(1_000);
-        let sender = self.channel.clone();
-
-        let mut table_db = self.db.connection.clone_connection();
-
-        // TODO: change to tokio
-        tokio::task::spawn(async move {
-            let mut channel = sender.lock().unwrap();
-            table_db.select(time_from, time_to, |trade| {
-                let message: MarketMessage = trade.into();
-                let r = channel.send(message);
-
-                if r.is_err() {
-                    log::error!("Error in channel.send: {:?}", r);
-                }
-            });
-            channel.close();
-        });
-
-        return MarketStream { reciver: channel };
-    }
-
-    /*
-    #[pyo3(signature = (side, price, size, client_order_id=None))]
-    pub fn new_limit_order_raw(
-        &self,
-        side: &str,
-        price: Decimal,
-        size: Decimal,
-        client_order_id: Option<&str>,
-    ) -> PyResult<BinanceOrderResponse> {
-        let price_scale = self.config.market_config.price_scale;
-        let size_scale = self.config.market_config.size_scale;
-        let order_side = OrderSide::from(side);
-
-        let price = price.round_dp(price_scale);
-        let size = size.round_dp(size_scale);
-
-        let response = new_limit_order(&self.config, order_side, price, size, client_order_id);
-
-        convert_pyresult(response)
-    }
-    */
-
-    #[pyo3(signature = (side, price, size, client_order_id=None))]
-    pub fn limit_order(
-        &self,
-        side: &str,
-        price: Decimal,
-        size: Decimal,
-        client_order_id: Option<&str>,
-    ) -> PyResult<Vec<Order>> {
-        let price_scale = self.config.market_config.price_scale;
-        let price_dp = price.round_dp(price_scale);
-
-        let size_scale = self.config.market_config.size_scale;
-        let size_dp = size.round_dp(size_scale);
-        let order_side = OrderSide::from(side);
-
-        let response =
-            new_limit_order(&self.config, order_side, price_dp, size_dp, client_order_id);
-
-        if response.is_err() {
-            log::error!(
-                "limit_order: side = {:?}, price = {:?}/{:?}, size = {:?}/{:?}, id = {:?}, result={:?}",
-                side,
-                price,
-                price_dp,
-                size,
-                size_dp,
-                client_order_id,
-                response
-            );
-
-            let err = format!(
-                "limit_order({:?}, {:?}/{:?}, {:?}/{:?}, {:?}) -> {:?}",
-                side,
-                price,
-                price_dp,
-                size,
-                size_dp,
-                client_order_id,
-                response.unwrap_err()
-            );
-            return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(err));
-        }
-
-        convert_pyresult(response)
-    }
-
-    /*
-    pub fn new_market_order_raw(
-        &self,
-        side: &str,
-        size: Decimal,
-        client_order_id: Option<&str>,
-    ) -> PyResult<BinanceOrderResponse> {
-        let size_scale = self.config.market_config.size_scale;
-        let size = size.round_dp(size_scale);
-
-        let order_side = OrderSide::from(side);
-
-        let response = new_market_order(&self.config, order_side, size, client_order_id);
-
-        convert_pyresult(response)
-    }
-    */
-
-    pub fn market_order(
-        &self,
-        side: &str,
-        size: Decimal,
-        client_order_id: Option<&str>,
-    ) -> PyResult<Vec<Order>> {
-        let size_scale = self.config.market_config.size_scale;
-        let size = size.round_dp(size_scale);
-
-        let order_side = OrderSide::from(side);
-
-        let response = new_market_order(&self.config, order_side, size, client_order_id);
-
-        if response.is_err() {
-            log::error!(
-                "market_order: side = {:?}, size = {:?}, id = {:?}, result={:?}",
-                side,
-                size,
-                client_order_id,
-                response
-            );
-
-            let err = format!(
-                "market_order({:?}, {:?}, {:?}) -> {:?}",
-                side,
-                size,
-                client_order_id,
-                response.unwrap_err()
-            );
-
-            return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(err));
-        }
-
-        convert_pyresult(response)
-    }
-
-    pub fn dry_market_order(
-        &self,
-        create_time: MicroSec,
-        order_id: &str,
-        client_order_id: &str,
-        side: OrderSide,
-        size: Decimal,
-        transaction_id: &str,
-    ) -> Vec<Order> {
-        let (bids, asks) = self.board.lock().unwrap().get_board_vec().unwrap();
-
-        let board = if side == OrderSide::Buy { asks } else { bids };
-
-        let mut orders: Vec<Order> = vec![];
-        let mut split_index = 0;
-
-        let mut remain_size = size;
-
-        // TODO: consume boards
-        for item in board {
-            if remain_size <= dec![0.0] {
-                break;
-            }
-
-            let execute_size;
-            let order_status;
-            split_index += 1;
-
-            if remain_size <= item.size {
-                order_status = OrderStatus::Filled;
-                execute_size = remain_size;
-                remain_size = dec![0.0];
-            } else {
-                order_status = OrderStatus::PartiallyFilled;
-                execute_size = item.size;
-                remain_size -= item.size;
-            }
-
-            let mut order = Order::new(
-                &self.config.market_config.trade_symbol,
-                create_time,
-                &order_id,
-                &client_order_id,
-                side,
-                OrderType::Market,
-                order_status,
-                dec![0.0],
-                size,
-            );
-
-            order.transaction_id = format!("{}-{}", transaction_id, split_index);
-            order.update_time = create_time;
-            order.is_maker = false;
-            order.execute_price = item.price;
-            order.execute_size = execute_size;
-            order.remain_size = remain_size;
-            order.quote_vol = order.execute_price * order.execute_size;
-
-            orders.push(order);
-        }
-
-        if remain_size > dec![0.0] {
-            log::error!("remain_size > 0.0: {:?}", remain_size);
-        }
-
-        return orders;
-    }
-
-    pub fn cancel_order(&self, order_id: &str) -> PyResult<Order> {
-        let response = cancel_order(&self.config, order_id);
-
-        return convert_pyresult(response);
-    }
-
-    pub fn cancel_all_orders(&self) -> PyResult<Vec<Order>> {
-        let response = cancell_all_orders(&self.config);
-
-        if response.is_ok() {
-            return convert_pyresult_vec(response);
-        }
-
-        return PyResult::Ok(vec![]);
-    }
-
-    #[getter]
-    pub fn get_order_status(&self) -> PyResult<Vec<BinanceOrderStatus>> {
-        let status = order_status(&self.config);
-
-        convert_pyresult(status)
-    }
-
-    #[getter]
-    pub fn get_open_orders(&self) -> PyResult<Vec<Order>> {
-        let status = open_orders(&self.config);
-
-        log::debug!("OpenOrder: {:?}", status);
-
-        convert_pyresult_vec(status)
-    }
-
-    #[getter]
-    pub fn get_trade_list(&self) -> PyResult<Vec<BinanceListOrdersResponse>> {
-        let status = trade_list(&self.config);
-
-        convert_pyresult(status)
-    }
-
-    #[getter]
-    pub fn get_account(&self) -> PyResult<BinanceAccountInformation> {
-        let status = get_balance(&self.config);
-
-        convert_pyresult(status)
-    }
-}
-
-use crate::exchange::binance::rest::{download_historical_trades_from_id, get_board_snapshot};
-
-impl BinanceMarket {
-    pub fn db_path(config: &BinanceConfig) -> PyResult<String> {
-        Ok(config.get_db_path())
-    }
-
-    pub fn wait_for_settlement(&mut self, tx: &Sender<Vec<Trade>>) {
-        while 5 < tx.len() {
-            sleep(Duration::from_millis(1 * 100));
-        }
-    }
-
-
-    pub fn download_log(
-        &mut self,
-        tx: &Sender<Vec<Trade>>,
-        date: MicroSec,
-        low_priority: bool,
-        verbose: bool,
-    ) -> PyResult<i64> {
-        let date = FLOOR_DAY(date);
-        let url = Self::make_historical_data_url_timestamp(&self.config, date);
-
-        match download_log(
-            &url,
-            tx,
-            false,
-            low_priority,
-            verbose,
-            &BinanceMarket::rec_to_trade,
-        ) {
-            Ok(download_rec) => Ok(download_rec),
-            Err(e) => Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!(
-                "Error in download_logs: {:?}",
-                e
-            ))),
-        }
-    }
-
-
-    fn get_latest_archive_date(&self) -> Result<MicroSec, String> {
-        let f = |date: MicroSec| -> String {
-            Self::make_historical_data_url_timestamp(&self.config, date)
-        };
-
-        latest_archive_date(&f)
-    }
-
-    /// Check if database is valid at the date
-    fn validate_db_by_date(&mut self, date: MicroSec) -> bool {
-        self.db.connection.validate_by_date(date)
-    }
-}
-
-#[cfg(test)]
-mod binance_test {
-    use std::{thread::sleep, time::Duration};
-
-    use crate::{
-        common::{init_debug_log, init_log, time_string, DAYS, TODAY},
-        exchange::binance::rest::{download_historical_trades, download_historical_trades_from_id},
-    };
-
-    use super::*;
-
-    #[test]
-    fn test_make_historical_data_url_timestamp() {
-        let config = BinanceConfig::BTCUSDT();
-        init_log();
-        println!(
-            "{}",
-            BinanceMarket::make_historical_data_url_timestamp(&config, 0)
-        );
-        assert_eq!(
-            BinanceMarket::make_historical_data_url_timestamp(&config, 0),
-            "https://data.binance.vision/data/spot/daily/trades/BTCUSDT/BTCUSDT-trades-1970-01-01.zip"            
-        );
-
-        println!(
-            "{}",
-            BinanceMarket::make_historical_data_url_timestamp(&config, NOW())
-        );
-
-        println!("{} / {}", TODAY(), time_string(TODAY()));
-        println!("{} / {}", DAYS(1), time_string(DAYS(1)));
-    }
-
-    #[test]
-    fn test_download() {
-        init_debug_log();
-        let mut market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
-        //let mut market = BinanceMarket::new("BTCBUSD", true);
-        println!("{}", time_string(market.db.start_time().unwrap_or(0)));
-        println!("{}", time_string(market.db.end_time(0).unwrap_or(0)));
-        println!("Let's donwload");
-        market.download(2, false, false, true, false);
-    }
-
-    #[test]
-    fn test_db_info() {
-        let mut market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
-        //let mut market = BinanceMarket::new("BTCBUSD", true);
-
-        println!("{:?}", market.db.info());
-    }
-
-    #[test]
-    fn test_ohlcv() {
-        let mut market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
-        //let mut market = BinanceMarket::new("BTCBUSD", true);
-
-        let _ = market.ohlcv(0, 0, 3600);
-
-        println!("{:?}", market.db.ohlcv_df(0, 0, 3600));
-    }
-
-    #[test]
-    fn bench_ohlcv() {
-        let mut market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
-        //let mut market = BinanceMarket::new("BTCBUSD", true);
-
-        let _ = market.ohlcv(0, 0, 3600);
-    }
-
-    #[test]
-    fn test_start_ws() {
-        let mut market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
-        //let mut market = BinanceMarket::new("BTCBUSD", true);
-
-        market.start_market_stream();
-        sleep(Duration::from_secs(20));
-    }
-
-    #[test]
-    fn test_reflesh_board() {
-        let config = BinanceConfig::BTCUSDT();
-        let _market = BinanceMarket::new(&config);
-        //let mut market = BinanceMarket::new("BTCBUSD", true);
-
-        let update_data = get_board_snapshot(&config).unwrap();
-
-        // TODO: test
-
-        println!("{:?}", update_data);
-    }
-
-    #[test]
-    fn test_ws_start() {
-        let mut market = BinanceMarket::new(&BinanceConfig::BTCUSDT());
-        //let mut market = BinanceMarket::new("BTCBUSD", true);
-
-        market.start_market_stream();
-
-        sleep(Duration::from_secs(10));
-    }
-
-    #[test]
-    fn test_latest_archive_date() {
-        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_validate_db_by_days() {
-        let mut market = BinanceMarket::new(&BinanceConfig::TEST_BTCUSDT());
-        //let market = BinanceMarket::new("BTCBUSD", true);
-
-        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/exchanges/binance.bak/src/lib.rs b/exchanges/binance.bak/src/lib.rs
deleted file mode 100644
index 798b158c..00000000
--- a/exchanges/binance.bak/src/lib.rs
+++ /dev/null
@@ -1,16 +0,0 @@
-mod config;
-mod message;
-mod rest;
-mod publicws;
-// mod privatews;
-
-pub use message::*;
-pub use config::*;
-pub use rest::*;
-pub use publicws::*;
-//pub use privatews::*;
-
-
-
-
-
diff --git a/exchanges/binance.bak/src/privatews.rs b/exchanges/binance.bak/src/privatews.rs
deleted file mode 100644
index ae146e50..00000000
--- a/exchanges/binance.bak/src/privatews.rs
+++ /dev/null
@@ -1,90 +0,0 @@
-
-pub struct BinancePrivateWsClient {
-    ws: AutoConnectClient<BinanceServerConfig, BinanceWsOpMessage>,
-    handler: Option<JoinHandle<()>>,
-    listen_key: String,
-}
-
-
-impl BinancePrivateWsClient {
-    pub async fn new(server: &BinanceServerConfig) -> Self {
-        let dummy_config = MarketConfig::new("dummy", "dummy", "dummy", 0, 0, 0); 
-
-        let mut private_ws = AutoConnectClient::new(
-            server,
-            &dummy_config,
-            &server.get_user_ws_server(),
-            PING_INTERVAL_SEC,
-            SWITCH_INTERVAL_SEC,
-            SYNC_WAIT_RECORDS,
-            None,
-            None,
-        );
-
-        private_ws.subscribe(&server..config.private_subscribe_channel).await;
-
-        Self {
-            ws: private_ws,
-            handler: None,
-        }
-    }
-
-
-    pub async fn connect(&mut self) {
-        self.ws.connect().await
-    }
-
-    pub async fn open_stream<'a>(
-        &'a mut self,
-    ) -> impl Stream<Item = Result<MultiMarketMessage, String>> + 'a {
-        let mut s = Box::pin(self.ws.open_stream().await);
-
-        stream! {
-            while let Some(message) = s.next().await {
-                match message {
-                    Ok(m) => {
-                        match Self::parse_message(m) {
-                            Err(e) => {
-                                println!("Parse Error: {:?}", e);
-                                continue;
-                            }
-                            Ok(m) => {
-                                let market_message = Self::convert_ws_message(m);
-
-                                match market_message
-                                {
-                                    Err(e) => {
-                                        println!("Convert Error: {:?}", e);
-                                        continue;
-                                    }
-                                    Ok(m) => {
-                                        yield Ok(m);
-                                    }
-                                }
-                            }
-                        }
-                    }
-                    Err(e) => {
-                        println!("Receive Error: {:?}", e);
-                    }
-                }
-            }
-        }
-    }
-
-    fn parse_message(message: String) -> anyhow::Result<BinanceUserWsMessage> {
-        let m = serde_json::from_str::<BinanceUserWsMessage>(&message);
-
-        if m.is_err() {
-            log::warn!("Error in serde_json::from_str: {:?}", message);
-            return Err(anyhow!("Error in serde_json::from_str: {:?}", message));
-        }
-
-        Ok(m.unwrap())
-    }
-
-    fn convert_ws_message(message: BinanceUserWsMessage) -> anyhow::Result<MultiMarketMessage> {
-        //Ok(message.into())
-        todo!();
-    }
-}
diff --git a/exchanges/binance.bak/src/publicws.rs b/exchanges/binance.bak/src/publicws.rs
deleted file mode 100644
index 53a5cefe..00000000
--- a/exchanges/binance.bak/src/publicws.rs
+++ /dev/null
@@ -1,181 +0,0 @@
-
-use futures::Stream;
-use futures::StreamExt;
-use serde::{Deserialize, Serialize};
-use tokio::task::JoinHandle;
-
-
-use async_stream::stream;
-
-use rbot_lib::{common::{
-    MarketConfig, 
-    MicroSec, MultiMarketMessage, ServerConfig, MICRO_SECOND, NOW}, net::{AutoConnectClient, WsOpMessage}};
-
-use crate::{BinancePublicWsMessage, BinanceServerConfig};
-
-use anyhow::anyhow;
-
-/// https://binance-docs.github.io/apidocs/spot/en/#listen-key-spot
-/// Ping/Keep-alive a ListenKey (USER_STREAM)
-/// PUT /api/v3/userDataStream
-/// Keepalive a user data stream to prevent a time out.
-/// User data streams will close after 60 minutes.
-/// It's recommended to send a ping about every 30 minutes.
-
-const KEY_EXTEND_INTERVAL: MicroSec = 5 * 60 * MICRO_SECOND; // 30 min
-
-pub const PING_INTERVAL_SEC: i64 = 60 * 3; // every 3 min
-pub const SWITCH_INTERVAL_SEC: i64 = 60 * 60 * 12; // 12 hours
-const SYNC_WAIT_RECORDS: i64 = 3; // no overlap
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct BinanceWsOpMessage {
-    method: String,
-    params: Vec<String>,
-    id: i64,
-}
-
-impl WsOpMessage for BinanceWsOpMessage {
-    fn new() -> Self {
-        BinanceWsOpMessage {
-            method: "SUBSCRIBE".to_string(),
-            params: vec![],
-            id: NOW() % 1000,
-        }
-    }
-
-    fn add_params(&mut self, params: &Vec<String>) {
-        log::debug!("add_params: {:?} / {:?}", self.params, params);
-        self.params.extend(params.clone());
-    }
-
-    fn make_message(&self) -> Vec<String> {
-        vec![self.to_string()]
-    }
-
-    fn to_string(&self) -> String {
-        if self.params.len() == 0 {
-            return "".to_string();
-        } else {
-            return serde_json::to_string(&self).unwrap();
-        }
-    }
-}
-
-pub struct BinancePublicWsClient {
-    ws: AutoConnectClient<BinanceServerConfig, BinanceWsOpMessage>,
-    handler: Option<JoinHandle<()>>,
-}
-
-impl BinancePublicWsClient {
-    pub async fn new(server: &BinanceServerConfig, config: &MarketConfig) -> Self {
-        let mut public_ws = AutoConnectClient::new(
-            server,
-            config,
-            &server.get_public_ws_server(),
-            PING_INTERVAL_SEC,
-            SWITCH_INTERVAL_SEC,
-            SYNC_WAIT_RECORDS,
-            None,
-            None,
-        );
-
-        public_ws.subscribe(&config.public_subscribe_channel).await;
-
-        Self {
-            ws: public_ws,
-            handler: None,
-        }
-    }
-
-    pub async fn connect(&mut self) {
-        self.ws.connect().await
-    }
-
-    pub async fn open_stream<'a>(
-        &'a mut self,
-    ) -> impl Stream<Item = Result<MultiMarketMessage, String>> + 'a {
-        let mut s = Box::pin(self.ws.open_stream().await);
-
-        stream! {
-            while let Some(message) = s.next().await {
-                match message {
-                    Ok(m) => {
-                        match Self::parse_message(m) {
-                            Err(e) => {
-                                println!("Parse Error: {:?}", e);
-                                continue;
-                            }
-                            Ok(m) => {
-                                let market_message = Self::convert_ws_message(m);
-
-                                match market_message
-                                {
-                                    Err(e) => {
-                                        println!("Convert Error: {:?}", e);
-                                        continue;
-                                    }
-                                    Ok(m) => {
-                                        yield Ok(m);
-                                    }
-                                }
-                            }
-                        }
-                    }
-                    Err(e) => {
-                        println!("Receive Error: {:?}", e);
-                    }
-                }
-            }
-        }
-    }
-
-    fn parse_message(message: String) -> anyhow::Result<BinancePublicWsMessage> {
-        let m = serde_json::from_str::<BinancePublicWsMessage>(&message);
-
-        if m.is_err() {
-            log::warn!("Error in serde_json::from_str: {:?}", message);
-            return Err(anyhow!("Error in serde_json::from_str: {:?}", message));
-        }
-
-        Ok(m.unwrap())
-    }
-
-    // TODO: implement
-    fn convert_ws_message(message: BinancePublicWsMessage) -> anyhow::Result<MultiMarketMessage> {
-        Ok(message.into())
-    }
-}
-
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use rbot_lib::common::MarketConfig;
-    use crate::BinanceConfig;
-
-    #[tokio::test]
-    async fn test_binance_public_ws_client() {
-        let server = BinanceServerConfig::new(true);
-        let config = BinanceConfig::BTCUSDT();
-
-        let mut client = BinancePublicWsClient::new(&server, &config).await;
-
-        client.connect().await;
-
-        let stream = client.open_stream().await;
-
-        let mut stream = Box::pin(stream);
-
-        while let Some(message) = stream.next().await {
-            match message {
-                Ok(m) => {
-                    println!("Message: {:?}", m);
-                }
-                Err(e) => {
-                    println!("Error: {:?}", e);
-                }
-            }
-        }
-    }
-}
diff --git a/exchanges/binance/src/config.rs b/exchanges/binance/src/config.rs
index a048a5c3..7653085a 100644
--- a/exchanges/binance/src/config.rs
+++ b/exchanges/binance/src/config.rs
@@ -19,7 +19,6 @@ pub struct BinanceServerConfig {
     pub rest_server: String,
     pub public_ws: String,
     pub private_ws: String,
-    pub db_base_dir: String,
     pub history_web_base: String,
     api_key: SecretString,
     api_secret: SecretString,
@@ -59,7 +58,6 @@ impl BinanceServerConfig {
             rest_server,
             public_ws: public_ws_server,
             private_ws: private_ws_server,
-            db_base_dir: "".to_string(),
             history_web_base: "https://data.binance.vision".to_string(),
             api_key: SecretString::new(&api_key),
             api_secret: SecretString::new(&api_secret),
@@ -137,198 +135,3 @@ impl BinanceConfig {
     }
 }
 
-
-/*
-#[derive(Clone, Debug, Serialize, Deserialize)]
-#[pyclass]
-pub struct BinanceConfig {
-    #[pyo3(set)]    
-    pub test_net: bool,
-    #[pyo3(set)]
-    pub exchange_name: String,
-    #[pyo3(set)]    
-    pub home_currency: String,
-    #[pyo3(set)]    
-    pub foreign_currency: String,
-    #[pyo3(set)]
-    pub testnet: bool,
-    // server config
-    #[pyo3(set)]    
-    pub rest_endpoint: String,
-    #[pyo3(set)]    
-    pub public_ws_endpoint: String,
-    #[pyo3(set)]    
-    pub private_ws_endpoint: String,
-    #[pyo3(set)]    
-    pub history_web_base: String,
-    #[pyo3(set)]    
-    pub new_order_path: String,
-    #[pyo3(set)]    
-    pub cancel_order_path: String,
-    #[pyo3(set)]    
-    pub open_orders_path: String,
-    #[pyo3(set)]    
-    pub account_path: String,
-    #[pyo3(set)]    
-    pub public_subscribe_channel: Vec<String>,
-    #[pyo3(set)]    
-    pub user_data_stream_path: String,
-
-    // key & secret
-    #[pyo3(set)]
-    pub api_key: String,
-    #[pyo3(set)]
-    pub api_secret: String,
-
-    #[pyo3(get)]
-    pub market_config: MarketConfig,
-
-    #[pyo3(get, set)]
-    pub db_base_dir: String,
-}
-
-#[pymethods]
-impl BinanceConfig {
-    #[allow(non_snake_case)]
-    #[classattr]
-    pub fn BTCUSDT() -> Self {
-        return BinanceConfig::SPOT("BTC", "USDT");
-    }
-
-    #[classattr]
-    #[allow(non_snake_case)]    
-    pub fn TEST_BTCUSDT() -> Self {
-        let mut config = BinanceConfig::TESTSPOT("BTC", "USDT");
-
-        config.home_currency = "USDT".to_string();
-        config.foreign_currency = "BTC".to_string();
-
-        return config;
-    }
-
-    #[allow(non_snake_case)]    
-    #[staticmethod]
-    pub fn TESTSPOT(foreign_symbol: &str, home_symbol: &str) -> Self {
-        let mut config = BinanceConfig::SPOT(foreign_symbol, home_symbol);
-
-        config.test_net = true;
-        config.rest_endpoint = "https://testnet.binance.vision".to_string();
-        config.public_ws_endpoint = "wss://testnet.binance.vision/ws".to_string();
-        config.private_ws_endpoint = "wss://testnet.binance.vision/ws".to_string();
-        config.testnet = true;
-        config.market_config.market_order_price_slip = dec![0.5];
-
-        return config;
-    }
-
-    #[allow(non_snake_case)]    
-    #[staticmethod]
-    pub fn SPOT(foreign_symbol: &str, home_symbol: &str) -> Self {
-
-        let symbol = format!("{}{}", foreign_symbol, home_symbol);
-
-        let upper_symbol = symbol.to_uppercase();
-        let lower_symbol = symbol.to_lowercase();
-
-        let api_key = if let Ok(key) = std::env::var("BINANCE_API_KEY") {
-            key
-        }
-        else {
-            log::error!("no key found in env[BINANCE_API_KEY]");
-            "".to_string()
-        };
-
-        let api_secret = if let Ok(secret) = std::env::var("BINANCE_API_SECRET") {
-            secret
-        }
-        else {
-            log::error!("no secret found in env[BINANCE_API_SECRET]");
-            "".to_string()
-        };
-
-        let mut market_config = MarketConfig::new(
-            "SPOT",
-            home_symbol,
-            foreign_symbol,
-            2,
-            4,
-        );
-
-        market_config.taker_fee = dec![0.0001];
-        market_config.maker_fee = dec![0.0001];
-
-        market_config.trade_category = "SPOT".to_string();
-        market_config.trade_symbol =  upper_symbol;
-
-        return BinanceConfig {
-            test_net: false,
-            exchange_name: "BINANCE".to_string(),
-
-            home_currency: home_symbol.to_string(),
-            foreign_currency: foreign_symbol.to_string(),
-
-            rest_endpoint: "https://api.binance.com".to_string(),
-            public_ws_endpoint: "wss://stream.binance.com:9443/ws".to_string(),
-            private_ws_endpoint: "wss://stream.binance.com:9443/ws".to_string(),
-            history_web_base: "https://data.binance.vision/data/spot/daily/trades".to_string(),
-            new_order_path: "/api/v3/order".to_string(),
-            cancel_order_path: "/api/v3/order".to_string(),
-            open_orders_path: "/api/v3/openOrders".to_string(),
-            account_path: "/api/v3/account".to_string(),
-            user_data_stream_path: "/api/v3/userDataStream".to_string(),
-
-            public_subscribe_channel: vec![
-                        format!("{}@trade", lower_symbol),
-                        format!("{}@depth@100ms", lower_symbol)],
-            testnet: false,
-            api_key,
-            api_secret,
-            market_config,
-            db_base_dir: "".to_string(),
-        };
-    }
-
-    #[getter]
-    pub fn get_db_path(&self) -> String {
-        let mut exchange_name = self.exchange_name.clone();
-
-        if self.testnet {
-            exchange_name = format!("{}-TESTNET", exchange_name);
-        }
-
-        let db_path = db_full_path(&exchange_name, &self.market_config.trade_category, &self.market_config.trade_symbol, &self.db_base_dir);
-
-        return db_path.to_str().unwrap().to_string();
-    }
-
-    pub fn __repr__(&self) -> String {
-        let mut printobj = self.clone();
-
-        if printobj.api_key.len() > 2 {
-            printobj.api_key = format!("{}*******************", printobj.api_key[0..2].to_string());
-        } else {
-            printobj.api_key = "!! NO KEY !!".to_string();
-        }
-
-        if printobj.api_secret.len() > 2 {
-            printobj.api_secret = format!(
-                "{}*******************",
-                printobj.api_secret[0..2].to_string()
-            );
-        } else {
-            printobj.api_secret = "!! NO SECRET !!".to_string();
-        }
-
-        serde_json::to_string(&printobj).unwrap()
-    }
-
-    pub fn short_info(&self) -> String {
-        if self.test_net {
-            return format!("---TEST NET--- {}", self.market_config.trade_symbol);
-        }
-        else {
-            return format!("*** LIVE NET *** {}", self.market_config.trade_symbol);
-        }
-    }
-}
-*/
diff --git a/exchanges/binance/src/market.rs b/exchanges/binance/src/market.rs
index d5afec60..9c4d6659 100644
--- a/exchanges/binance/src/market.rs
+++ b/exchanges/binance/src/market.rs
@@ -1,31 +1,34 @@
 // Copyright(c) 2022-2024. yasstake. All rights reserved.
 
 use std::sync::{Arc, Mutex, RwLock};
+use std::thread::sleep;
 
 use anyhow::Context;
 use futures::StreamExt;
 use pyo3_polars::PyDataFrame;
 use rbot_blockon::BLOCK_ON;
-use rbot_lib::common::flush_log;
 use rbot_lib::common::AccountCoins;
 use rbot_lib::common::BoardItem;
+use rbot_lib::common::BoardTransfer;
 use rbot_lib::common::MarketConfig;
 use rbot_lib::common::MarketMessage;
 use rbot_lib::common::MarketStream;
+use rbot_lib::common::MicroSec;
 use rbot_lib::common::MultiMarketMessage;
 use rbot_lib::common::Order;
 use rbot_lib::common::OrderBook;
 use rbot_lib::common::MARKET_HUB;
-use rbot_lib::common::MicroSec;
+use rbot_lib::common::{flush_log, LogStatus};
+use rbot_lib::common::{time_string, NOW};
 use rbot_lib::db::TradeTable;
-use rbot_lib::net::BroadcastMessage;
+use rbot_lib::net::{BroadcastMessage, RestApi};
 use rust_decimal::Decimal;
 // Copyright(c) 2022-2024. yasstake. All rights reserved.
 use tokio::task::JoinHandle;
 
 // use rbot_market::OrderInterface;
-use rbot_market::OrderInterfaceImpl;
 use rbot_market::MarketImpl;
+use rbot_market::OrderInterfaceImpl;
 // use rbot_market::MarketInterface;
 
 use crate::BinancePrivateWsClient;
@@ -35,7 +38,6 @@ use crate::BinanceServerConfig;
 
 use pyo3::prelude::*;
 
-
 #[pyclass]
 pub struct Binance {
     enable_order: bool,
@@ -44,9 +46,7 @@ pub struct Binance {
 }
 
 #[pymethods]
-impl 
-// OrderInterface for
-Binance {
+impl Binance {
     #[new]
     pub fn new(production: bool) -> Self {
         let server_config = BinanceServerConfig::new(production);
@@ -58,17 +58,20 @@ Binance {
         }
     }
 
-
     #[getter]
     pub fn get_exchange_name(&self) -> String {
         self.server_config.exchange_name.clone()
     }
 
+    #[getter]
+    fn get_production(&self) -> bool {
+        self.server_config.production
+    }
+
     pub fn open_market(&self, config: &MarketConfig) -> BinanceMarket {
         return BinanceMarket::new(&self.server_config, config);
     }
 
-
     //--- OrderInterfaceImpl ----
     #[setter]
     pub fn set_enable_order_with_my_own_risk(&mut self, enable_order: bool) {
@@ -127,7 +130,6 @@ Binance {
     }
 }
 
-
 impl OrderInterfaceImpl<BinanceRestApi, BinanceServerConfig> for Binance {
     fn set_enable_order_feature(&mut self, enable_order: bool) {
         self.enable_order = enable_order;
@@ -142,7 +144,6 @@ impl OrderInterfaceImpl<BinanceRestApi, BinanceServerConfig> for Binance {
     }
 
     async fn async_start_user_stream(&mut self) -> anyhow::Result<()> {
-
         let exchange_name = self.server_config.exchange_name.clone();
         let server_config = self.server_config.clone();
 
@@ -150,7 +151,7 @@ impl OrderInterfaceImpl<BinanceRestApi, BinanceServerConfig> for Binance {
             let mut ws = BinancePrivateWsClient::new(&server_config).await;
             ws.connect().await;
 
-            let mut market_channel = MARKET_HUB.open_channel();
+            let market_channel = MARKET_HUB.open_channel();
             let mut ws_stream = Box::pin(ws.open_stream().await);
 
             while let Some(message) = ws_stream.next().await {
@@ -163,7 +164,7 @@ impl OrderInterfaceImpl<BinanceRestApi, BinanceServerConfig> for Binance {
                 match message {
                     MultiMarketMessage::Order(order) => {
                         for o in order {
-                            let _ =  market_channel.send(BroadcastMessage {
+                            let _ = market_channel.send(BroadcastMessage {
                                 exchange: exchange_name.clone(),
                                 category: o.category.clone(),
                                 symbol: o.symbol.clone(),
@@ -188,7 +189,6 @@ impl OrderInterfaceImpl<BinanceRestApi, BinanceServerConfig> for Binance {
         }));
 
         Ok(())
-
     }
 }
 
@@ -206,7 +206,9 @@ impl BinanceMarket {
     #[new]
     pub fn new(server_config: &BinanceServerConfig, config: &MarketConfig) -> Self {
         log::debug!("open market BybitMarket::new");
-        BLOCK_ON(async { Self::async_new(server_config, config).await.unwrap() })
+        BLOCK_ON(async { 
+            Self::async_new(server_config, config, ! server_config.production).await.unwrap() 
+        })
     }
     #[getter]
     fn get_config(&self) -> MarketConfig {
@@ -239,9 +241,11 @@ impl BinanceMarket {
         MarketImpl::reset_cache_duration(self)
     }
 
+    /*
     fn stop_db_thread(&mut self) {
         MarketImpl::stop_db_thread(self)
     }
+    */
 
     fn cache_all_data(&mut self) -> anyhow::Result<()> {
         MarketImpl::cache_all_data(self)
@@ -323,7 +327,7 @@ impl BinanceMarket {
         MarketImpl::_repr_html_(self)
     }
 
-    #[pyo3(signature = (ndays, force=false, verbose=true, low_priority=true))]
+    #[pyo3(signature = (ndays, force=false, verbose=false, low_priority=true))]
     fn download_archive(
         &mut self,
         ndays: i64,
@@ -331,36 +335,40 @@ impl BinanceMarket {
         verbose: bool,
         low_priority: bool,
     ) -> anyhow::Result<i64> {
-        BLOCK_ON(async {
-            MarketImpl::async_download_archives(self, ndays, force, verbose, low_priority).await
-        })
+        MarketImpl::download_archives(self, ndays, force, verbose, low_priority)
     }
 
-    #[pyo3(signature = (verbose=true))]
+    #[pyo3(signature = (verbose=false))]
     fn download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        BLOCK_ON(async { self.async_download_lastest(verbose).await })
+        if verbose {
+            println!(
+                "download_latest: {} {}",
+                self.config.trade_category, self.config.trade_symbol
+            );
+            flush_log();
+        }
+
+        MarketImpl::download_latest(self, verbose)
     }
 
-    fn expire_unfix_data(&mut self) -> anyhow::Result<()> {
-        MarketImpl::expire_unfix_data(self)
+    #[pyo3(signature = (force=false))]
+    fn expire_unfix_data(&mut self, force: bool) -> anyhow::Result<()> {
+        BLOCK_ON(async { self.async_expire_unfix_data(force).await })
     }
 
-    fn find_latest_gap(&self) -> anyhow::Result<(MicroSec, MicroSec)> {
-        MarketImpl::find_latest_gap(self)
+    fn find_latest_gap(&self, force: bool) -> anyhow::Result<(MicroSec, MicroSec)> {
+        MarketImpl::find_latest_gap(self, force)
     }
 
-    fn download_gap(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        MarketImpl::download_gap(self, verbose)
+    #[pyo3(signature = (force=false, verbose=false))]
+    fn download_gap(&mut self, force: bool, verbose: bool) -> anyhow::Result<i64> {
+        MarketImpl::download_gap(self, force, verbose)
     }
 
     fn start_market_stream(&mut self) -> anyhow::Result<()> {
         MarketImpl::start_market_stream(self)
     }
 
-    fn open_realtime_channel(&mut self) -> anyhow::Result<MarketStream> {
-        MarketImpl::open_realtime_channel(self)
-    }
-
     fn open_backtest_channel(
         &mut self,
         time_from: MicroSec,
@@ -370,7 +378,6 @@ impl BinanceMarket {
     }
 }
 
-
 impl MarketImpl<BinanceRestApi, BinanceServerConfig> for BinanceMarket {
     fn get_server_config(&self) -> BinanceServerConfig {
         self.server_config.clone()
@@ -393,11 +400,7 @@ impl MarketImpl<BinanceRestApi, BinanceServerConfig> for BinanceMarket {
     }
 
     fn download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        todo!()
-    }
-
-    fn download_gap(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        todo!()
+        BLOCK_ON(async { self.async_download_latest(verbose).await })
     }
 
     fn get_db(&self) -> Arc<Mutex<TradeTable>> {
@@ -405,73 +408,431 @@ impl MarketImpl<BinanceRestApi, BinanceServerConfig> for BinanceMarket {
     }
 
     fn get_history_web_base_url(&self) -> String {
-        todo!()
-    }
-
-    fn get_latest_archive_date(&self) -> anyhow::Result<rbot_lib::common::MicroSec> {
-        todo!()
+        self.server_config.history_web_base.clone()
     }
 
     fn get_order_book(&self) -> Arc<RwLock<OrderBook>> {
-        todo!()
+        self.board.clone()
     }
 
-    fn reflesh_order_book(&mut self) {
-        todo!()
+    fn reflesh_order_book(&mut self) -> anyhow::Result<()> {
+        BLOCK_ON(async { self.async_reflesh_order_book_self().await })
     }
 
     fn start_market_stream(&mut self) -> anyhow::Result<()> {
-        todo!()
+        BLOCK_ON(async { self.async_start_market_stream().await })
     }
 
-    fn open_realtime_channel(&mut self) -> anyhow::Result<rbot_lib::common::MarketStream> {
-        todo!()
+    fn download_gap(&mut self, force: bool, verbose: bool) -> anyhow::Result<i64> {
+        BLOCK_ON(async {
+            log::debug!("download_gap");
+            self.async_download_gap(force, verbose).await
+        })
     }
 
-    fn open_backtest_channel(
+    fn download_archives(
         &mut self,
-        time_from: rbot_lib::common::MicroSec,
-        time_to: rbot_lib::common::MicroSec,
-    ) -> anyhow::Result<rbot_lib::common::MarketStream> {
-        todo!()
+        ndays: i64,
+        force: bool,
+        verbose: bool,
+        low_priority: bool,
+    ) -> anyhow::Result<i64> {
+        BLOCK_ON(async {
+            self.async_download_archives(ndays, force, verbose, low_priority)
+                .await
+        })
     }
 }
 
-
 impl BinanceMarket {
-    async fn async_new(server_config: &BinanceServerConfig, config: &MarketConfig) -> anyhow::Result<Self> {
-        let db_path = Self::make_db_path(
+    async fn async_new(
+        server_config: &BinanceServerConfig,
+        config: &MarketConfig,
+        test_mode: bool,
+    ) -> anyhow::Result<Self> {
+        let db_path = TradeTable::make_db_path(
             &server_config.exchange_name,
             &config.trade_category,
             &config.trade_symbol,
-            &server_config.db_base_dir,
+            test_mode
         );
 
         let db = TradeTable::open(&db_path)
-              .with_context(|| format!("Error in TradeTable::open: {:?}", db_path))?;
+            .with_context(|| format!("Error in TradeTable::open: {:?}", db_path))?;
 
-        let public_ws = BinancePublicWsClient::new(&server_config, &config).await;
+        // let public_ws = BinancePublicWsClient::new(&server_config, &config).await;
 
-        let mut market = BinanceMarket {
+        let market = BinanceMarket {
             server_config: server_config.clone(),
             config: config.clone(),
             db: Arc::new(Mutex::new(db)),
-            board: Arc::new(RwLock::new(OrderBook::new(server_config, &config))),            
+            board: Arc::new(RwLock::new(OrderBook::new(server_config, &config))),
             public_handler: None,
         };
 
-        market.cache_all_data().with_context(|| format!("Error in cache_all_data"))?;
-
         Ok(market)
     }
 
-    async fn async_download_lastest(&mut self, verbose: bool) -> anyhow::Result<i64> {
+    async fn async_start_market_stream(&mut self) -> anyhow::Result<()> {
+        if self.public_handler.is_some() {
+            log::info!("market stream is already running.");
+            return Ok(());
+        }
+
+        let db_channel = {
+            let mut lock = self.db.lock().unwrap();
+            lock.start_thread().await
+        };
+
+        let orderbook = self.board.clone();
+
+        let server_config = self.server_config.clone();
+        let config = self.config.clone();
+
+        let hub_channel = MARKET_HUB.open_channel();
+
+        self.public_handler = Some(tokio::task::spawn(async move {
+            let mut public_ws = BinancePublicWsClient::new(&server_config, &config).await;
+
+            let exchange_name = server_config.exchange_name.clone();
+            let trade_category = config.trade_category.clone();
+            let trade_symbol = config.trade_symbol.clone();
+
+            public_ws.connect().await;
+
+            let _ = Self::async_reflesh_order_book(&orderbook, &server_config, &config).await;
+
+            sleep(std::time::Duration::from_secs(1));
+
+            let ws_stream = public_ws.open_stream().await;
+            let mut ws_stream = Box::pin(ws_stream);
+
+            loop {
+                let message = ws_stream.next().await;
+                if message.is_none() {
+                    log::error!("Error in ws_stream.recv: {:?}", message);
+                    continue;
+                }
+
+                let message = message.unwrap();
+
+                if message.is_err() {
+                    log::error!("Error in ws_stream.recv: {:?}", message);
+                    continue;
+                }
+
+                let messages = message.unwrap();
+
+                match messages {
+                    MultiMarketMessage::Trade(trade) => {
+                        log::debug!("Trade: {:?}", trade);
+                        let r = db_channel.send(trade.clone());
+
+                        if r.is_err() {
+                            log::error!("Error in db_channel.send: {:?}", r);
+                        }
+
+                        for message in trade {
+                            let r = hub_channel.send(BroadcastMessage {
+                                exchange: exchange_name.clone(),
+                                category: trade_category.clone(),
+                                symbol: trade_symbol.clone(),
+                                msg: MarketMessage::Trade(message),
+                            });
+                            if r.is_err() {
+                                log::error!("Error in hub_channel.send: {:?}", r);
+                            }
+                        }
+                    }
+                    MultiMarketMessage::Orderbook(board) => {
+                        Self::update_orderbook(&orderbook, &board);
+                    }
+                    MultiMarketMessage::Control(control) => {
+                        if control.status == false {
+                            log::error!("Control message: {:?}", control);
+                        }
+                    }
+                    _ => {
+                        log::info!("Market stream message: {:?}", messages);
+                    }
+                }
+            }
+        }));
+
+        Ok(())
+    }
+
+    async fn async_reflesh_order_book_self(&mut self) -> anyhow::Result<()> {
+        let transfer =
+            BinanceRestApi::get_board_snapshot(&self.server_config, &self.config).await?;
+
+        Self::update_orderbook(&self.board, &transfer);
+
+        log::debug!("reflesh order book done");
+
+        Ok(())
+    }
+
+    async fn async_reflesh_order_book(
+        orderbook: &Arc<RwLock<OrderBook>>,
+        server: &BinanceServerConfig,
+        config: &MarketConfig,
+    ) -> anyhow::Result<()> {
+        let transfer = BinanceRestApi::get_board_snapshot(&server, &config).await?;
+
+        let orderbook = orderbook.clone();
+        Self::update_orderbook(&orderbook, &transfer);
+
+        log::debug!("reflesh order book done");
+
+        Ok(())
+    }
+
+    fn update_orderbook(orderbook: &Arc<RwLock<OrderBook>>, transfer: &BoardTransfer) {
+        let mut b = orderbook.write().unwrap();
+
+        //        let first_update_id = b.get_first_update_id();
+        let last_update_id = b.get_last_update_id();
+
+        // 4. Drop any event where u is <= lastUpdateId in the snapshot.
+        if last_update_id != 0 {
+            if transfer.last_update_id <= last_update_id {
+                log::warn!(
+                    "transfer update_id is too small: {} <= {}",
+                    transfer.last_update_id,
+                    b.get_last_update_id()
+                );
+                return;
+            }
+
+            if last_update_id + 1 != transfer.first_update_id {
+                log::warn!(
+                    "last_update_id is not continuous: {} + 1 != {}",
+                    last_update_id,
+                    transfer.last_update_id
+                );
+            }
+        }
+
+        b.update(&transfer);
+    }
+
+    async fn async_download_gap(&mut self, force: bool, verbose: bool) -> anyhow::Result<i64> {
+        log::debug!("[start] download_gap ");
+
+        let gap_result = self.find_latest_gap_trade(force);
+        if gap_result.is_err() {
+            log::warn!("no gap found: {:?}", gap_result);
+            return Ok(0);
+        }
+
+        let start_time = NOW();
+        let (unfix_start, unfix_end) = gap_result.unwrap();
+
         if verbose {
-            println!("async_download_lastest");
-            flush_log();
+            if let Some(t) = unfix_start.clone() {
+                println!("unfix_start: {:?}", t.__str__());
+            }
+            if let Some(t) = unfix_end.clone() {
+                println!("unfix_end  : {:?}", t.__str__());
+            }
+        }
+
+        let unfix_start_id = if let Some(trade) = unfix_start {
+            trade.id.parse::<i64>().unwrap()
+        } else {
+            log::warn!("no gap start found: {:?}", unfix_start);
+            return Ok(0);
+        };
+
+        let unfix_end_id = if let Some(trade) = unfix_end {
+            trade.id.parse::<i64>().unwrap_or_default()
+        } else {
+            0
+        };
+
+        let mut from_id = unfix_start_id + 1;
+        let mut insert_len: i64 = 0;
+
+        let tx = { self.db.lock().unwrap().start_thread().await };
+
+        loop {
+            if from_id == unfix_end_id {
+                log::debug!("from_id == unfix_end_id: {:?}", from_id);
+                break;
+            }
+
+            log::debug!("download recent from_id: {:?}", from_id);
+            let mut trades = BinanceRestApi::get_historical_trades(
+                &self.server_config,
+                &self.config,
+                from_id,
+                0,
+            )
+            .await?;
+
+            let l = trades.len();
+
+            if l == 0 {
+                if verbose {
+                    println!("end of data");
+                }
+                break;
+            }
+
+            if verbose {
+                println!(
+                    "Downloaded: from:{}({})[{}]  to:{}({})[{}]",
+                    time_string(trades[0].time),
+                    trades[0].time,
+                    trades[0].id,
+                    time_string(trades[l - 1].time),
+                    trades[l - 1].time,
+                    trades[l - 1].id
+                );
+            }
+
+            let trade_id = trades[l - 1].id.parse::<i64>()?;
+            from_id = trade_id + 1;
+
+            if unfix_end_id == 0 {
+                trades.retain(|t| t.time <= start_time);
+            } else {
+                trades.retain(|t| {
+                    (t.id.parse::<i64>().unwrap() <= unfix_end_id) && (t.time <= start_time)
+                });
+            }
+
+            let l = trades.len();
+            log::debug!("Trim downloaded: {:?}", l);
+
+            if l == 0 {
+                break;
+            }
+
+            trades[0].status = LogStatus::FixRestApiStart;
+            trades[l - 1].status = LogStatus::FixRestApiEnd;
+
+            /*
+            Binanceではklineから作ったデータは存在しないので削除不要
+            let expire_message =
+                TradeTableDb::expire_control_message(trades[0].time, trades[l - 1].time);
+
+            tx.send(expire_message)?;
+            */
+            tx.send(trades)?;
+
+            insert_len += l as i64;
+            log::debug!("inserted: {}", l);
+            std::thread::sleep(std::time::Duration::from_millis(200));
         }
+        Ok(insert_len)
+    }
+}
+
+#[cfg(test)]
+mod binance_market_test {
+    use rbot_lib::common::init_debug_log;
+
+    use crate::BinanceConfig;
+
+    #[test]
+    fn test_dowlad_latest() {
+        init_debug_log();
+        use super::*;
+        let server_config = BinanceServerConfig::new(true);
+        let config = BinanceConfig::BTCUSDT();
+
+        let mut market = BinanceMarket::new(&server_config, &config);
+
+        log::debug!("market build complete");
+
+        let r = market.download_latest(true);
+
+        assert!(r.is_ok());
+        log::debug!("download {:?}", r);
+
+        sleep(std::time::Duration::from_secs(5));
+    }
+
+    #[test]
+    fn test_dowlad_gap() {
+        // init_debug_log();
+        use super::*;
+        let server_config = BinanceServerConfig::new(true);
+        let config = BinanceConfig::BTCUSDT();
+
+        let mut market = BinanceMarket::new(&server_config, &config);
+
+        //market.download_archive(2, true, true, false);
+
+        log::debug!("market build complete");
+        //let _ = market.expire_unfix_data();
 
-        Ok(0)
+        let r = market.download_gap(true, true);
+
+        assert!(r.is_ok());
+
+        sleep(std::time::Duration::from_secs(60));
+    }
+}
+
+#[cfg(test)]
+mod test_market_impl {
+    use rbot_lib::common::init_debug_log;
+
+    use crate::BinanceConfig;
+
+    #[tokio::test]
+    async fn test_async_download_latest() -> anyhow::Result<()> {
+        use super::*;
+        init_debug_log();
+
+        let server = BinanceServerConfig::new(true);
+        let market_config = BinanceConfig::BTCUSDT();
+
+        //let binance = Binance::new(true);
+
+        let mut market = BinanceMarket::async_new(&server, &market_config, true).await?;
+        //let mut market = binance.open_market(&market_config);
+
+        let rec = market.async_download_latest(true).await.unwrap();
+
+        assert!(rec > 0);
+
+        Ok(())
     }
 
+    #[test]
+    fn test_download_latest() {
+        init_debug_log();
+        use super::*;
+        let server = BinanceServerConfig::new(true);
+        let market_config = BinanceConfig::BTCUSDT();
+
+        let mut market = BinanceMarket::new(&server, &market_config);
+
+        let rec = market.download_latest(true).unwrap();
+        assert!(rec > 0);
+    }
+
+    use rust_decimal_macros::dec;
+
+    #[test]
+    fn test_market_order() {
+        init_debug_log();
+        use super::*;
+        // let server = BinanceServerConfig::new(true);
+        let market_config = BinanceConfig::BTCUSDT();
+
+        let mut binance = Binance::new(true);
+        binance.enable_order = true;
+
+        let rec = binance.market_order
+            (&market_config, 
+            "Buy", 
+            dec![0.001],
+            None);
+        assert!(rec.is_ok());
+    }
 }
diff --git a/exchanges/binance/src/message.rs b/exchanges/binance/src/message.rs
index 73b53f85..746f85ee 100644
--- a/exchanges/binance/src/message.rs
+++ b/exchanges/binance/src/message.rs
@@ -1,20 +1,49 @@
-// Copyright(c) 2022-2023. yasstakself#![allow(non_snake_case)]
+// Copyright(c) 2022-2024. yasstake
+
+#![allow(non_snake_case)]
 #![allow(non_camel_case_types)]
 
 use pyo3::{pyclass, pymethods};
-use rbot_lib::common::{msec_to_microsec, orderside_deserialize, orderstatus_deserialize, ordertype_deserialize, string_to_decimal, string_to_f64, AccountCoins, BoardItem, BoardTransfer, Coin, LogStatus, MarketConfig, MarketMessage, MultiMarketMessage, Order, OrderBookRaw, OrderSide, OrderStatus, OrderType, Trade};
+use rbot_lib::common::{
+    msec_to_microsec, orderside_deserialize, orderstatus_deserialize, ordertype_deserialize,
+    string_to_decimal, string_to_f64, AccountCoins, BoardItem, BoardTransfer, Coin, ControlMessage,
+    LogStatus, MarketConfig, MultiMarketMessage, Order, OrderSide,
+    OrderStatus, OrderType, Trade,
+};
 use rust_decimal::Decimal;
 use rust_decimal_macros::dec;
 use serde_derive::{Deserialize, Serialize};
-use std::{ops::Mul, str::FromStr};
-
-use crate::BinanceConfig;
+use std::str::FromStr;
 
 
 pub type BinanceMessageId = u64;
 
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum BinanceWsRawMessage {
+    message(BinancePublicWsMessage),
+    reply(BinanceSubscriptionReply),
+}
+
+impl Into<BinancePublicWsMessage> for BinanceWsRawMessage {
+    fn into(self) -> BinancePublicWsMessage {
+        match self {
+            BinanceWsRawMessage::message(m) => m,
+            BinanceWsRawMessage::reply(r) => BinancePublicWsMessage::Control(
+                if let Some(msg) = r.result {
+                    msg
+                }
+                else {
+                    "None".to_string()
+                }
+            )
+        }
+    }
+}
+
+
 #[pyclass]
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct BinanceSubscriptionReply {
     pub result: Option<String>,
     pub id: u64,
@@ -27,6 +56,8 @@ pub enum BinancePublicWsMessage {
     Trade(BinanceWsTradeMessage),
     #[serde(rename = "depthUpdate")]
     BoardUpdate(BinanceWsBoardUpdate),
+    #[serde(rename = "control")]
+    Control(String),
 }
 
 impl BinancePublicWsMessage {
@@ -37,11 +68,8 @@ impl BinancePublicWsMessage {
     pub fn __repr__(&self) -> String {
         serde_json::to_string(&self).unwrap()
     }
-
 }
 
-
-
 impl Into<MultiMarketMessage> for BinancePublicWsMessage {
     fn into(self) -> MultiMarketMessage {
         match self {
@@ -50,22 +78,22 @@ impl Into<MultiMarketMessage> for BinancePublicWsMessage {
 
                 let t = trade.to_trade();
                 trades.push(t);
-                MultiMarketMessage::Trade (trades)
-            },
+                MultiMarketMessage::Trade(trades)
+            }
             BinancePublicWsMessage::BoardUpdate(board_update) => {
-                let mut board: BoardTransfer = board_update.into();
-                // TODO: implment
-                log::warn!("BinancePublicWsMessage::BoardUpdate is not implemented yet");
+                let board: BoardTransfer = board_update.into();
 
                 MultiMarketMessage::Orderbook(board)
             }
+            BinancePublicWsMessage::Control(m) => MultiMarketMessage::Control(ControlMessage {
+                status: true,
+                operation: "".to_string(),
+                message: m,
+            }),
         }
     }
 }
 
-
-
-
 #[pyclass]
 //  {"result":null,"id":1}
 #[derive(Debug, Serialize, Deserialize)]
@@ -214,6 +242,7 @@ impl Into<BoardTransfer> for BinanceWsBoardUpdate {
     fn into(self) -> BoardTransfer {
         let mut board = BoardTransfer::new();
 
+        board.first_update_id = self.U;
         board.last_update_id = self.u;
         board.last_update_time = msec_to_microsec(self.event_time as i64);
 
@@ -223,7 +252,6 @@ impl Into<BoardTransfer> for BinanceWsBoardUpdate {
     }
 }
 
-
 #[derive(Debug, Serialize, Deserialize)]
 // ["26127.87000000","20.79393000"]
 #[pyclass]
@@ -333,12 +361,8 @@ impl BinanceOrderFill {
     pub fn __repr__(&self) -> String {
         serde_json::to_string(&self).unwrap()
     }
-
-
-        
 }
 
-
 #[allow(non_snake_case)]
 #[pyclass]
 #[derive(Debug, Serialize, Deserialize)]
@@ -423,7 +447,6 @@ impl BinanceOrderResponse {
     }
 }
 
-
 pub fn binance_order_response_vec_to_orders(
     config: &MarketConfig,
     order_response: &Vec<BinanceOrderResponse>,
@@ -512,11 +535,8 @@ impl BinanceCancelOrderResponse {
     }
 }
 
-
 /*
 BiannceListOrderResponse will parse json below
-
-
 [
   {
     "symbol": "BNBBTC",
@@ -537,6 +557,7 @@ BiannceListOrderResponse will parse json below
 
 */
 
+/*
 #[allow(non_snake_case)]
 #[pyclass]
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -566,7 +587,6 @@ impl BinanceListOrdersResponse {
         serde_json::to_string(&self).unwrap()
     }
 }
-/*
 impl Into<Order> for BinanceListOrdersResponse {
     fn into(self) -> Order {
         let order_side: OrderSide = if self.isBuyer {
@@ -598,6 +618,8 @@ impl Into<Order> for BinanceListOrdersResponse {
 }
 */
 
+/*------------ Private WS ------------------------*/
+
 /*
 BinanceAccountUpdate is parse json as blow
 
@@ -622,6 +644,31 @@ pub struct BinanceAccountUpdate {
     u: u64,
     B: Vec<BinanceBalance>,
 }
+
+impl BinanceAccountUpdate {
+    pub fn __str__(&self) -> String {
+        self.__repr__()
+    }
+
+    pub fn __repr__(&self) -> String {
+        serde_json::to_string(&self).unwrap()
+    }
+
+    pub fn to_coins(&self) -> AccountCoins {
+        let mut coins: Vec<Coin> = vec![];
+
+        for balance in self.B.iter() {
+            let coin: Coin = balance.to_coin();
+            coins.push(coin);
+        }
+
+        AccountCoins {
+            coins: coins,
+        }
+    }
+}
+
+
 #[pyclass]
 #[derive(Debug, Serialize, Deserialize)]
 pub struct BinanceBalance {
@@ -633,30 +680,16 @@ pub struct BinanceBalance {
     l: Decimal,
 }
 
-/*
-pub fn binance_account_update_to_account_status(
-    config: &BinanceConfig,
-    account_update: &BinanceAccountUpdate,
-) -> AccountStatus {
-    let mut account_status = AccountStatus::default();
-
-    let l = account_update.B.len();
-
-    for i in 0..l {
-        if account_update.B[i].a == config.foreign_currency {
-            account_status.foreign_free = account_update.B[i].f;
-            account_status.foreign_locked = account_update.B[i].l;
-            account_status.foreign = account_status.foreign_free + account_status.foreign_locked;
-        } else if account_update.B[i].a == config.home_currency {
-            account_status.home_free = account_update.B[i].f;
-            account_status.home_locked = account_update.B[i].l;
-            account_status.home = account_status.home_free + account_status.home_locked;
+impl BinanceBalance {
+    pub fn to_coin(&self) -> Coin {
+        Coin {
+            symbol: self.a.clone(),
+            volume: self.f + self.l,
+            free: self.f,
+            locked: self.l,
         }
     }
-
-    return account_status;
 }
-*/
 
 /// BinanceBalancdUpdate
 /// Represents a balance update message received from the Binance exchange.
@@ -802,180 +835,40 @@ impl BinanceExecutionReport {
         serde_json::to_string(&self).unwrap()
     }
 
-        fn to_order(&self, config: &MarketConfig) -> Order {
-            let mut order = Order::new(
-                &config.trade_category,
-                &self.symbol,
-                msec_to_microsec(self.time),
-                &self.order_id.to_string(),
-                &self.client_order_id,
-                self.order_side,
-                self.order_type,
-                self.current_order_status,
-                self.order_price,
-                self.order_quantity,
-            );
-    
-            order.status = self.current_order_status;
-            order.transaction_id = self.trade_id.to_string();
-            order.update_time = msec_to_microsec(self.transaction_time);
-            order.execute_price = self.last_executed_price;
-            order.execute_size = self.last_executed_quantity;
-            order.remain_size = self.order_quantity - self.cumulative_filled_quantity;
-            order.quote_vol = self.last_quote_quantity;
-            order.commission = self.commission_amount;
-            order.commission_asset = self.commission_asset.clone().unwrap_or_default();
-            order.is_maker = self.is_maker;
-    
-            if self.order_reject_reason != "NONE" {
-                order.message = self.order_reject_reason.clone();
-            }
-    
-            log::debug!("order: {:?}", order);
-    
-            order
-        }
-
-}
-
-
-    /*
-    fn from(order: &BinanceExecutionReport) -> Self {
-        let order_id = order.order_id.to_string();
-        let client_order_id = order.client_order_id.clone();
-
-        let order_side:OrderSide = order.order_side;
-        let order_price: Decimal = order.order_price;
-        let order_size:Decimal = order.order_quantity;
-        let order_type: OrderType = order.order_type;
-        let order_status = order.current_order_status;
-        let order_quote = order_price * order_size;
-
-        let trade_id = order.trade_id.to_string();
-        //let create_time = binance_to_microsec(order.O);
-        let update_time = binance_to_microsec(order.time);
-        let execute_size = order.last_executed_quantity;
-        let execute_price = order.last_executed_price;
-        let execute_total = order.cumulative_filled_quantity;
-        let commission_amount = order.commission_amount;
-        let commition_asset = order.commission_asset.clone().unwrap_or_default();
-        let ismaker = order.is_maker;
-
-        let remain_size =
-            if order_status == OrderStatus::Filled || order_status == OrderStatus::Canceled {
-                Decimal::from(0)
-            } else {
-                order.order_quantity - execute_total
-            };
-
-        let fills = OrderFill {
-            transaction_id: trade_id,
-            update_time: update_time,
-            price: execute_price,
-            filled_size: execute_size,
-            quote_vol: execute_price * execute_size,
-            commission: order.commission_amount,
-            commission_asset: commition_asset,
-            maker: ismaker,
-        };
-
-        let mut account_change = AccountChange::new();
-
-        match order_status {
-            OrderStatus::New => {
-                // in new order, asset is locked.
-                if order_side == OrderSide::Buy {
-                    let lock_size = order_size * order.order_price;
-                    account_change.lock_home_change = lock_size;
-                    account_change.free_home_change = - lock_size;
-                    // TODO: may be need to add fee
-                }
-                else if order_side == OrderSide::Sell {
-                    let lock_size = order_size;
-                    account_change.lock_foreign_change = lock_size;
-                    account_change.free_foreign_change = - lock_size;
-                    // TODO: may be need to add fee
-                }
-            }
-
-            OrderStatus::PartiallyFilled | OrderStatus::Filled => {
-                // lock is free, and foreing sizie is change.
-                if order_side == OrderSide::Buy {
-                    let unlock_size = execute_size * execute_price;
-                    if order.is_maker {
-                        account_change.lock_home_change = - unlock_size;
-                    }
-                    account_change.home_change = - unlock_size;
-
-                    account_change.foreign_change = execute_size;
-                }
-                else if order_side == OrderSide::Sell {
-                    let unlock_size = execute_size;
-
-                    if order.is_maker {
-                        account_change.lock_foreign_change = - unlock_size;
-                    }
-                    account_change.foreign_change = - unlock_size;
-
-                    account_change.home_change = execute_size * execute_price;
+    fn to_order(&self, category: &str) -> Order {
+        let mut order = Order::new(
+            category,
+            &self.symbol,
+            msec_to_microsec(self.time),
+            &self.order_id.to_string(),
+            &self.client_order_id,
+            self.order_side,
+            self.order_type,
+            self.current_order_status,
+            self.order_price,
+            self.order_quantity,
+        );
 
-                }
-            }
-            OrderStatus::Canceled => {
-                // lock is free
-                if order_side == OrderSide::Buy {
-                    if order.on_order_book {
-                        let unlock_size = order_size * order_price;
-                        account_change.lock_home_change = - unlock_size;
-                        account_change.free_home_change = unlock_size;
-                    }
-                }
-                else if order_side == OrderSide::Sell {
-                    if order.on_order_book {
-                        let unlock_size = order_size;
-                        account_change.lock_foreign_change = - unlock_size;
-                        account_change.free_foreign_change = unlock_size;
-                    }
-                }
-            }
-            OrderStatus::Rejected => {log::error!("Rejected: not implemented {:?}", order);},
-            OrderStatus::Expired => {log::error!("Expired: not implemented {:?}", order);},
-            OrderStatus::Error => {log::error!("Error: not implemented {:?}", order);},
+        order.status = self.current_order_status;
+        order.transaction_id = self.trade_id.to_string();
+        order.update_time = msec_to_microsec(self.transaction_time);
+        order.execute_price = self.last_executed_price;
+        order.execute_size = self.last_executed_quantity;
+        order.remain_size = self.order_quantity - self.cumulative_filled_quantity;
+        order.quote_vol = self.last_quote_quantity;
+        order.commission = self.commission_amount;
+        order.commission_asset = self.commission_asset.clone().unwrap_or_default();
+        order.is_maker = self.is_maker;
+
+        if self.order_reject_reason != "NONE" {
+            order.message = self.order_reject_reason.clone();
         }
 
-        let order = Order::new(
-            order.symbol,
-            update_time,
-            order_id,
-            client_order_id,
-            order_side,
-            order_type,
-            order_status,
-            order_price,
-            order_size,
-        );
-        )
-        let r = Order {
-            symbol: order.symbol.clone(),
-            create_time: binance_to_microsec(order.transaction_time),
-            order_id: order.order_id.to_string(),
-            order_list_index: order.order_list_id,
-            client_order_id: order.client_order_id.clone(),
-            order_side: order_side,
-            order_type: order_type,
-            price: order.order_price,
-            size: order.order_quantity,
-            remain_size: remain_size,
-            status: order_status,
-            account_change: account_change,
-            message: "".to_string(),
-            fills: fills,
-        };
+        log::debug!("order: {:?}", order);
 
-        return r;
+        order
     }
-    */
-
+}
 
 
 #[allow(non_snake_case)]
@@ -988,62 +881,29 @@ pub enum BinanceUserWsMessage {
 }
 
 
-/*
-impl BinanceUserStreamMessage {
-    pub fn convert_to_market_message(&self, config: &BinanceConfig) -> MarketMessage {
-        let mut message = MarketMessage::new();
-
+impl BinanceUserWsMessage {
+    pub fn convert_multimarketmessage(&self, category: &str) -> MultiMarketMessage {
         log::debug!("RAW user stream:\n{:?}\n", self);
 
-        match self {
-            BinanceUserStreamMessage::outboundAccountPosition(account) => {
-                let status = binance_account_update_to_account_status(config, account);
-                message.account = Some(status);
+        let message = match self {
+            BinanceUserWsMessage::outboundAccountPosition(account) => {
+                let coins: AccountCoins = account.to_coins();
+                MultiMarketMessage::Account(coins)
             }
-            BinanceUserStreamMessage::balanceUpdate(_balance) => {
+            BinanceUserWsMessage::balanceUpdate(_balance) => {
                 log::error!("not implemented");
+                MultiMarketMessage::Message("not implemented".to_string())
             }
-            BinanceUserStreamMessage::executionReport(order) => {
-                let mut order: Order = order.into();
-                order.update_balance(&config.market_config);
-                message.order = Some(order);
+            BinanceUserWsMessage::executionReport(report) => {
+                let order: Order = report.to_order(category);
+                MultiMarketMessage::Order(vec![order])
             }
         };
 
         message
     }
 }
-*/
 
-/*
-impl Into<Order> for BinanceUserStreamMessage {
-    fn into(self) -> Order {
-        match self {
-            BinanceUserStreamMessage::executionReport(order) => order.into(),
-            _ => {
-                log::error!("not supported");
-                Order {
-                    symbol: "".to_string(),
-                    create_time: 0,
-                    order_id: "".to_string(),
-                    order_list_index: 0,
-                    client_order_id: "".to_string(),
-                    order_side: OrderSide::Buy,
-                    order_type: OrderType::Limit,
-                    price: Decimal::from(0),
-                    size: Decimal::from(0),
-                    remain_size: Decimal::from(0),
-                    status: OrderStatus::New,
-                    account_change: AccountChange::new(),
-                    message: "".to_string(),
-                    fills: None,
-                    profit: None,
-                }
-            }
-        }
-    }
-}
-*/
 
 /*
 https://binance-docs.github.io/apidocs/spot/en/#account-information-user_data
@@ -1126,11 +986,8 @@ pub struct BinanceAccountInformation {
     pub uid: i64,
 }
 
-
-
 #[pymethods]
 impl BinanceAccountInformation {
-
     pub fn into_coins(&self) -> AccountCoins {
         let mut coins = AccountCoins::new();
 
@@ -1147,10 +1004,8 @@ impl BinanceAccountInformation {
 
         coins
     }
-    
 }
 
-
 #[pyclass]
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct BinanceCommissionRates {
@@ -1244,8 +1099,6 @@ pub struct BinanceOrderStatus {
     selfTradePreventionMode: String,
 }
 
-
-
 pub fn binance_order_status_vec_to_orders(
     config: &MarketConfig,
     order_response: &Vec<BinanceOrderStatus>,
@@ -1300,26 +1153,34 @@ impl BinanceOrderStatus {
 }
 
 
-/*
+
+
 #[cfg(test)]
 mod binance_message_test {
+    use crate::config::BinanceConfig;
+    // use crate::config::BinanceServerConfig;
     use super::*;
-    use crate::exchange::binance::message::{
-        BinanceCancelOrderResponse, BinanceOrderResponse, BinancePublicWsMessage,
-        BinanceTradeMessage, BinanceWsBoardUpdate, BinanceWsTradeMessage,
-    };
 
-    const HISTORY: &str = r#"[{"id":990877266,"price":"26092.63000000","qty":"0.00046000","quoteQty":"12.00260980","time":1692935644243,"isBuyerMaker":false,"isBestMatch":true},{"id":990877267,"price":"26093.10000000","qty":"0.00189000","quoteQty":"49.31595900","time":1692935644243,"isBuyerMaker":false,"isBestMatch":true},{"id":990877268,"price":"26093.75000000","qty":"0.00209000","quoteQty":"54.53593750","time":1692935644276,"isBuyerMaker":false,"isBestMatch":true},{"id":990877269,"price":"26094.00000000","qty":"0.00306000","quoteQty":"79.84764000","time":1692935647281,"isBuyerMaker":false,"isBestMatch":true},{"id":990877270,"price":"26094.00000000","qty":"0.00075000","quoteQty":"19.57050000","time":1692935647283,"isBuyerMaker":false,"isBestMatch":true}]"#;
+    #[test]
+    fn test_binance_public_ws_message() {
+        let message: BinancePublicWsMessage = serde_json::from_str(TRADE_WS).unwrap();
+        println!("{:?}", message);
+
+        let message: BinancePublicWsMessage = serde_json::from_str(BOARD_UPDATE).unwrap();
+        println!("{:?}", message);
+    }
 
+    const TRADE_WS: &str = r#"{"e":"trade","E":1693226465430,"s":"BTCUSDT","t":3200243634,"p":"26132.02000000","q":"0.00244000","b":22161265544,"a":22161265465,"T":1693226465429,"m":false,"M":true}"#;        
     #[test]
-    fn test_binance_message() {
-        let message: Vec<BinanceTradeMessage> = serde_json::from_str(HISTORY).unwrap();
+    fn test_binance_trade_message() {
+
+        let message: BinanceWsTradeMessage = serde_json::from_str(TRADE_WS).unwrap();
 
         println!("{:?}", message);
     }
 
-    const BOARD_UPDATE: &str = r#"{"e":"depthUpdate","E":1693266904308,"s":"BTCUSDT","U":38531387766,"u":38531387832,"b":[["26127.87000000","20.79393000"],["26126.82000000","0.02674000"],["26125.95000000","0.00000000"],["26125.78000000","0.38302000"],["26125.68000000","0.00000000"],["26125.10000000","0.00000000"],["26125.05000000","0.00000000"],["26124.76000000","0.00000000"],["26124.75000000","0.21458000"],["26114.84000000","1.14830000"],["26114.15000000","0.00000000"],["26090.85000000","0.00000000"],["26090.84000000","0.00000000"],["26090.32000000","2.29642000"],["26090.31000000","3.82738000"],["26087.99000000","0.03733000"],["26084.34000000","0.00000000"],["25553.07000000","0.13647000"],["25500.81000000","0.14160000"],["25496.85000000","0.00000000"],["25284.00000000","0.03996000"],["24827.83000000","0.00000000"],["24300.17000000","0.00000000"],["23772.50000000","0.00047000"],["23515.08000000","0.00000000"],["18289.50000000","0.00000000"],["13063.93000000","0.00091000"]],"a":[["26127.88000000","5.58099000"],["26128.39000000","0.20072000"],["26128.79000000","0.21483000"],["26129.26000000","0.38297000"],["26129.52000000","0.00000000"],["26129.53000000","0.00000000"],["26134.50000000","0.06000000"],["26134.99000000","1.07771000"],["26135.10000000","0.00700000"],["26155.27000000","0.00050000"],["26155.28000000","0.00000000"],["27027.87000000","0.00200000"],["27290.25000000","0.00000000"],["27817.92000000","0.00000000"],["28345.58000000","0.00000000"]]}"#;
 
+    const BOARD_UPDATE: &str = r#"{"e":"depthUpdate","E":1693266904308,"s":"BTCUSDT","U":38531387766,"u":38531387832,"b":[["26127.87000000","20.79393000"],["26126.82000000","0.02674000"],["26125.95000000","0.00000000"],["26125.78000000","0.38302000"],["26125.68000000","0.00000000"],["26125.10000000","0.00000000"],["26125.05000000","0.00000000"],["26124.76000000","0.00000000"],["26124.75000000","0.21458000"],["26114.84000000","1.14830000"],["26114.15000000","0.00000000"],["26090.85000000","0.00000000"],["26090.84000000","0.00000000"],["26090.32000000","2.29642000"],["26090.31000000","3.82738000"],["26087.99000000","0.03733000"],["26084.34000000","0.00000000"],["25553.07000000","0.13647000"],["25500.81000000","0.14160000"],["25496.85000000","0.00000000"],["25284.00000000","0.03996000"],["24827.83000000","0.00000000"],["24300.17000000","0.00000000"],["23772.50000000","0.00047000"],["23515.08000000","0.00000000"],["18289.50000000","0.00000000"],["13063.93000000","0.00091000"]],"a":[["26127.88000000","5.58099000"],["26128.39000000","0.20072000"],["26128.79000000","0.21483000"],["26129.26000000","0.38297000"],["26129.52000000","0.00000000"],["26129.53000000","0.00000000"],["26134.50000000","0.06000000"],["26134.99000000","1.07771000"],["26135.10000000","0.00700000"],["26155.27000000","0.00050000"],["26155.28000000","0.00000000"],["27027.87000000","0.00200000"],["27290.25000000","0.00000000"],["27817.92000000","0.00000000"],["28345.58000000","0.00000000"]]}"#;
     #[test]
     fn test_binance_board_update() {
         let message: BinanceWsBoardUpdate = serde_json::from_str(BOARD_UPDATE).unwrap();
@@ -1327,15 +1188,55 @@ mod binance_message_test {
         println!("{:?}", message);
     }
 
-    const TRADE_WS: &str = r#"{"e":"trade","E":1693226465430,"s":"BTCUSDT","t":3200243634,"p":"26132.02000000","q":"0.00244000","b":22161265544,"a":22161265465,"T":1693226465429,"m":false,"M":true}"#;
+    #[test]
+    fn test_multi_market_message() {
+        let trade: BinanceWsTradeMessage = serde_json::from_str(TRADE_WS).unwrap();
+        let trade = BinancePublicWsMessage::Trade(trade);        
+        let message: MultiMarketMessage = trade.into();
+        println!("{:?}", message);
+
+        let board: BinanceWsBoardUpdate = serde_json::from_str(BOARD_UPDATE).unwrap();
+        let board = BinancePublicWsMessage::BoardUpdate(board);
+        let message: MultiMarketMessage = board.into();
+        println!("{:?}", message);
+    }
+
+
+
+    const HISTORY: &str = r#"[{"id":990877266,"price":"26092.63000000","qty":"0.00046000","quoteQty":"12.00260980","time":1692935644243,"isBuyerMaker":false,"isBestMatch":true},{"id":990877267,"price":"26093.10000000","qty":"0.00189000","quoteQty":"49.31595900","time":1692935644243,"isBuyerMaker":false,"isBestMatch":true},{"id":990877268,"price":"26093.75000000","qty":"0.00209000","quoteQty":"54.53593750","time":1692935644276,"isBuyerMaker":false,"isBestMatch":true},{"id":990877269,"price":"26094.00000000","qty":"0.00306000","quoteQty":"79.84764000","time":1692935647281,"isBuyerMaker":false,"isBestMatch":true},{"id":990877270,"price":"26094.00000000","qty":"0.00075000","quoteQty":"19.57050000","time":1692935647283,"isBuyerMaker":false,"isBestMatch":true}]"#;
 
     #[test]
-    fn test_binance_trade_message() {
-        let message: BinanceWsTradeMessage = serde_json::from_str(TRADE_WS).unwrap();
+    fn test_binance_trade_history_message() {
+        let message: Vec<BinanceTradeMessage> = serde_json::from_str(HISTORY).unwrap();
 
         println!("{:?}", message);
     }
 
+    const REST_BOARD: &str = r#"{
+        "lastUpdateId": 1027024,
+        "bids": [
+          [
+            "4.00000000",
+            "431.00000000"
+          ]
+        ],
+        "asks": [
+          [
+            "4.00000200",
+            "12.00000000"
+          ]
+        ]
+      }"#;
+
+    
+    #[test]
+    fn test_binance_rest_board_update_message() {
+        let message: BinanceRestBoard = serde_json::from_str(REST_BOARD).unwrap();
+
+        println!("{:?}", message);
+    }
+    
+
     #[test]
     fn test_binance_ws_message() {
         let message: BinancePublicWsMessage = serde_json::from_str(TRADE_WS).unwrap();
@@ -1346,10 +1247,18 @@ mod binance_message_test {
     }
 
     #[test]
-    fn test_binance_order_response() {
-        let order_response: BinanceOrderResponse = serde_json::from_str(r#"{"symbol":"BTCUSDT","orderId":28,"orderListId":-1,"clientOrderId":"6gCrw2kRUAF9CvJDGP16IP","transactTime":1507725176595,"price":"0.00000000","origQty":"10.00000000","executedQty":"10.00000000","cummulativeQuoteQty":"10.00000000","status":"FILLED","timeInForce":"GTC","type":"MARKET","side":"SELL","workingTime":1507725176595,"selfTradePreventionMode":"NONE","fills":[{"price":"4000.00000000","qty":"1.00000000","commission":"4.00000000","commissionAsset":"USDT","tradeId":56},{"price":"3999.00000000","qty":"5.00000000","commission":"19.99500000","commissionAsset":"USDT","tradeId":57},{"price":"3998.00000000","qty":"2.00000000","commission":"7.99600000","commissionAsset":"USDT","tradeId":58},{"price":"3997.00000000","qty":"1.00000000","commission":"3.99700000","commissionAsset":"USDT","tradeId":59},{"price":"3995.00000000","qty":"1.00000000","commission":"3.99500000","commissionAsset":"USDT","tradeId":60}]}"#).unwrap();
+    fn test_binance_order_response() -> anyhow::Result<()>{
+        let order: BinanceOrderResponse = serde_json::from_str(r#"{"symbol":"BTCUSDT","orderId":28,"orderListId":-1,"clientOrderId":"6gCrw2kRUAF9CvJDGP16IP","transactTime":1507725176595,"price":"0.00000000","origQty":"10.00000000","executedQty":"10.00000000","cummulativeQuoteQty":"10.00000000","status":"FILLED","timeInForce":"GTC","type":"MARKET","side":"SELL","workingTime":1507725176595,"selfTradePreventionMode":"NONE","fills":[{"price":"4000.00000000","qty":"1.00000000","commission":"4.00000000","commissionAsset":"USDT","tradeId":56},{"price":"3999.00000000","qty":"5.00000000","commission":"19.99500000","commissionAsset":"USDT","tradeId":57},{"price":"3998.00000000","qty":"2.00000000","commission":"7.99600000","commissionAsset":"USDT","tradeId":58},{"price":"3997.00000000","qty":"1.00000000","commission":"3.99700000","commissionAsset":"USDT","tradeId":59},{"price":"3995.00000000","qty":"1.00000000","commission":"3.99500000","commissionAsset":"USDT","tradeId":60}]}"#)?;
+        println!("{:?}", order);
 
-        println!("{:?}", order_response);
+        let config = BinanceConfig::BTCUSDT();
+        let order_vec = order.to_order_vec(&config);
+        println!("{:?}", order_vec);
+
+        let order_vec2 = binance_order_response_vec_to_orders(&config, &vec![order]);
+        assert_eq!(order_vec, order_vec2);
+
+        Ok(())
     }
 
     #[test]
@@ -1361,7 +1270,7 @@ mod binance_message_test {
 
     // TODO: test cancel all orders response
     #[test]
-    fn test_binance_cancel_all_orders() {
+    fn test_binance_cancel_multiple_orders() {
         let order_response: Vec<BinanceCancelOrderResponse> = serde_json::from_str(
             r#"
             [
@@ -1407,12 +1316,6 @@ mod binance_message_test {
         println!("{:?}", order_response);
     }
 
-    #[test]
-    fn test_binance_exution_report() {
-        let order_response: BinanceUserStreamMessage = serde_json::from_str(r#"{"e":"executionReport","E":1499405658658,"s":"ETHBTC","c":"mUvoqJxFIILMdfAW5iGSOW","S":"BUY","o":"LIMIT","f":"GTC","q":"1.00000000","p":"0.10264410","P":"0.00000000","F":"0.00000000","g":-1,"C":"","x":"NEW","X":"NEW","r":"NONE","i":4293153,"l":"0.00000000","z":"0.00000000","L":"0.00000000","n":"0","N":null,"T":1499405658657,"t":-1,"I":8641984,"w":true,"m":false,"M":false,"O":1499405658657,"Z":"0.00000000","Y":"0.00000000","Q":"0.00000000","W":1499405658657,"V":"NONE"}"#).unwrap();
-
-        println!("{:?}", order_response);
-    }
 
     #[test]
     fn test_binance_account_inforamtion() {
@@ -1431,37 +1334,54 @@ mod binance_message_test {
             "permissions":["SPOT"],"uid":354937868}"#).unwrap();
 
         println!("{:?}", order_response);
+
+        let coins = order_response.into_coins();
+        println!("{:?}", coins);
     }
 
+    
     #[test]
     fn test_binance_order_status() {
         let order_response: BinanceOrderStatus = serde_json::from_str(
             r#"{"symbol":"LTCBTC","orderId":1,"orderListId":-1,"clientOrderId":"myOrder1","price":"0.1","origQty":"1.0","executedQty":"0.0","cummulativeQuoteQty":"0.0","status":"NEW","timeInForce":"GTC","type":"LIMIT","side":"BUY","stopPrice":"0.0","icebergQty":"0.0","time":1499827319559,"updateTime":1499827319559,"isWorking":true,"workingTime":1499827319559,"origQuoteOrderQty":"0.000000","selfTradePreventionMode":"NONE"}"#).unwrap();
 
         println!("{:?}", order_response);
+
+        let config = BinanceConfig::BTCUSDT();
+        let order = order_response.to_order(&config);
+        println!("{:?}", order);
     }
 
     #[test]
-    fn test_binance_order_status_cap_NEW() {
+    fn test_binance_order_status_cap_new() {
         let order_response: BinanceOrderStatus = serde_json::from_str(
             r#"{"symbol":"LTCBTC","orderId":1,"orderListId":-1,"clientOrderId":"myOrder1","price":"0.1","origQty":"1.0","executedQty":"0.0","cummulativeQuoteQty":"0.0","status":"NEW","timeInForce":"GTC","type":"LIMIT","side":"BUY","stopPrice":"0.0","icebergQty":"0.0","time":1499827319559,"updateTime":1499827319559,"isWorking":true,"workingTime":1499827319559,"origQuoteOrderQty":"0.000000","selfTradePreventionMode":"NONE"}"#).unwrap();
 
         println!("{:?}", order_response);
     }
 
+
+    /*------------ Private WS ------------------------*/
     #[test]
-    fn test_binance_execution_report() {
-        let _execution: BinanceUserStreamMessage = serde_json::from_str(
+    fn test_account_update() {
+        let account_update: BinanceAccountUpdate = serde_json::from_str(
             r#"
-            {"e":"outboundAccountPosition","E":1694430911241,"u":1694430911240,"B":[{"a":"BTC","f":"1.16500000","l":"0.00000000"},{"a":"BUSD","f":"9599.07923000","l":"25.00000000"}]}
+            {
+                "e": "outboundAccountPosition",
+                "E": 1564034571105,            
+                "u": 1564034571073,            
+                "B": [                         
+                  {
+                    "a": "ETH",                
+                    "f": "10000.000000",       
+                    "l": "0.000000"            
+                  }
+                ]
+              }
             "#).unwrap();
-    }
 
-    #[test]
-    fn test_binance_list_orders_response() {
-        let list = r#"[{"symbol":"BNBBTC","id":28457,"orderId":100234,"orderListId":-1,"price":"4.00000100","qty":"12.00000000","quoteQty":"48.000012","commission":"10.10000000","commissionAsset":"BNB","time":1499865549590,"isBuyer":true,"isMaker":false,"isBestMatch":true}]"#;
-
-        let _list: Vec<BinanceListOrdersResponse> = serde_json::from_str(list).unwrap();
+        println!("{:?}", account_update);
     }
+
 }
-*/
+
diff --git a/exchanges/binance/src/rest.rs b/exchanges/binance/src/rest.rs
index 2d20d99c..3ac5d257 100644
--- a/exchanges/binance/src/rest.rs
+++ b/exchanges/binance/src/rest.rs
@@ -1,14 +1,22 @@
 // Copyright(c) 2022-2024. yasstake. All rights reserved.
 
-use crate::{binance_order_response_vec_to_orders, binance_order_status_vec_to_orders, BinanceAccountInformation, BinanceCancelOrderResponse, BinanceOrderResponse, BinanceOrderStatus, BinanceRestBoard, BinanceServerConfig, BinanceTradeMessage};
+use crate::{
+    binance_order_status_vec_to_orders,
+    BinanceAccountInformation, BinanceCancelOrderResponse, BinanceOrderResponse,
+    BinanceOrderStatus, BinanceRestBoard, BinanceServerConfig, BinanceTradeMessage,
+};
 
 use anyhow::anyhow;
 use csv::StringRecord;
-use rbot_lib::{common::{hmac_sign, AccountCoins, BoardTransfer, Kline, LogStatus, MarketConfig, MicroSec, Order, OrderSide, OrderType, ServerConfig, Trade, NOW}, net::{rest_delete, rest_get, rest_post, rest_put, RestApi}};
+use rbot_lib::{
+    common::{
+        flush_log, hmac_sign, AccountCoins, BoardTransfer, Kline, LogStatus, MarketConfig, MicroSec, Order, OrderSide, OrderType, ServerConfig, Trade, NOW
+    },
+    net::{rest_delete, rest_get, rest_post, rest_put, RestApi},
+};
 use rust_decimal::{prelude::FromPrimitive, Decimal};
 use serde_json::Value;
 
-use anyhow::Result;
 use anyhow::Context;
 
 pub struct BinanceRestApi {}
@@ -35,25 +43,39 @@ impl RestApi<BinanceServerConfig> for BinanceRestApi {
     /// https://binance-docs.github.io/apidocs/spot/en/#recent-trades-list
     async fn get_recent_trades(
         server: &BinanceServerConfig,
-        config: &rbot_lib::common::MarketConfig,
+        config: &MarketConfig,
     ) -> anyhow::Result<Vec<Trade>> {
+        log::debug!("get_recent_trades: {:?}", &config.trade_symbol);
+
         let path = "/api/v3/trades";
-        let params = format!("symbol={}&limit=1000", config.trade_symbol);
+        let params = format!("symbol={}&limit=1000", &config.trade_symbol);
 
-        let messasge = Self::get(&server, path, &params)
+        let messasge = Self::get(server, path, &params)
             .await
             .with_context(|| format!("get_recent_trades error"))?;
 
+        log::debug!("get_recent_trades: {:?}", messasge);
+
         let trades: Vec<BinanceTradeMessage> = serde_json::from_value(messasge)?;
 
-        Ok(trades.into_iter().map(|t| t.to_trade()).collect())
+        log::debug!("get_recent_trades: {:?}", trades.len());
+
+        let mut result: Vec<Trade> = vec![];
+
+        for t in trades {
+            result.push(t.to_trade());
+
+            log::debug!("trade: {:?}", t.to_trade());
+        }
+
+        Ok(result)
     }
 
     async fn get_trade_klines(
-        server: &BinanceServerConfig,
-        config: &MarketConfig,
-        start_time: MicroSec,
-        end_time: MicroSec,
+        _server: &BinanceServerConfig,
+        _config: &MarketConfig,
+        _start_time: MicroSec,
+        _end_time: MicroSec,
     ) -> anyhow::Result<Vec<Kline>> {
         todo!();
     }
@@ -69,7 +91,6 @@ impl RestApi<BinanceServerConfig> for BinanceRestApi {
     ) -> anyhow::Result<Vec<Order>> {
         let path = "/api/v3/order";
         let side = Self::order_side_string(side);
-        //let type = "LIMIT";
 
         let order_type_str: &str = match order_type {
             OrderType::Limit => "LIMIT",
@@ -78,12 +99,12 @@ impl RestApi<BinanceServerConfig> for BinanceRestApi {
         };
 
         let mut body = format!(
-            "symbol={}&side={}&type={}&timeInForce=GTC&quantity={}",
+            "symbol={}&side={}&type={}&quantity={}",
             config.trade_symbol, side, order_type_str, size
         );
 
         if order_type == OrderType::Limit {
-            body = format!("{}&price={}", body, price);
+            body = format!("{}&price={}&timeInForce=GTC", body, price);
         }
 
         if client_order_id.is_some() {
@@ -134,8 +155,7 @@ impl RestApi<BinanceServerConfig> for BinanceRestApi {
 
         let orders: Vec<BinanceOrderStatus> = serde_json::from_value(message)?;
 
-        let orders 
-            = binance_order_status_vec_to_orders(config, &orders);
+        let orders = binance_order_status_vec_to_orders(config, &orders);
 
         Ok(orders)
     }
@@ -144,7 +164,8 @@ impl RestApi<BinanceServerConfig> for BinanceRestApi {
         let path = "/api/v3/account";
 
         let message = Self::get_sign(server, path, None)
-            .await.with_context(||format!("get_account error"))?;
+            .await
+            .with_context(|| format!("get_account error"))?;
 
         let account: BinanceAccountInformation = serde_json::from_value(message)?;
 
@@ -159,15 +180,14 @@ impl RestApi<BinanceServerConfig> for BinanceRestApi {
         mm: i64,
         dd: i64,
     ) -> String {
-            // https://data.binance.vision/data/spot/daily/trades/BTCBUSD/BTCBUSD-trades-2022-11-19.zip
-            let category = category.to_lowercase();
-            if category == "spot" {
+        // https://data.binance.vision/data/spot/daily/trades/BTCBUSD/BTCBUSD-trades-2022-11-19.zip
+        let category = category.to_lowercase();
+        if category == "spot" {
             return format!(
-                "/data/{}/daily/trades/{}/{}-trades-{:04}-{:02}-{:02}.zip",
-                category, symbol, symbol, yyyy, mm, dd
+                "{}/data/{}/daily/trades/{}/{}-trades-{:04}-{:02}-{:02}.zip",
+                history_web_base, category, symbol, symbol, yyyy, mm, dd
             );
-        }
-        else {
+        } else {
             log::error!("Unknown category {}", category);
             return "".to_string();
         }
@@ -232,11 +252,16 @@ impl BinanceRestApi {
     async fn get(server: &BinanceServerConfig, path: &str, params: &str) -> anyhow::Result<Value> {
         let query = format!("{}?{}", path, params);
 
+        log::debug!("path{} / body: {}", path, query);
+        flush_log();
+
         let response = rest_get(&server.get_rest_server(), &query, vec![], None, None)
             .await
             .with_context(|| format!("rest_get error: {}/{}", &server.get_rest_server(), &query))?;
 
-        Ok(Self::parse_binance_result(&response)?)
+        log::debug!("path{} / body: {}", path, response);
+
+        Self::parse_binance_result(response)
     }
 
     async fn get_sign(
@@ -270,7 +295,7 @@ impl BinanceRestApi {
 
         log::debug!("AUTH GET: path{} / body: {} / {:?}", path, q, message);
 
-        Ok(Self::parse_binance_result(&message)?)
+        Self::parse_binance_result(message)
     }
 
     async fn post_sign(
@@ -291,7 +316,7 @@ impl BinanceRestApi {
             .await
             .with_context(|| format!("post_sign error {}/{}", server.get_rest_server(), path))?;
 
-        Ok(Self::parse_binance_result(&message)?)
+        Self::parse_binance_result(message)
     }
 
     fn sign_with_timestamp(secret_key: &str, message: &str) -> String {
@@ -317,7 +342,7 @@ impl BinanceRestApi {
             .await
             .with_context(|| format!("post_key error {}/{}", server.get_rest_server(), path))?;
 
-        Ok(Self::parse_binance_result(&result)?)
+        Self::parse_binance_result(result)
     }
 
     async fn put_key(
@@ -333,7 +358,7 @@ impl BinanceRestApi {
             .await
             .with_context(|| format!("post_key error {}/{}", server.get_rest_server(), path))?;
 
-        Ok(Self::parse_binance_result(&result)?)
+        Self::parse_binance_result(result)
     }
 
     pub async fn delete_sign(
@@ -354,7 +379,7 @@ impl BinanceRestApi {
             .await
             .with_context(|| format!("delete_sign error {}/{}", server.get_rest_server(), path))?;
 
-        Ok(Self::parse_binance_result(&result)?)
+        Self::parse_binance_result(result)
     }
 
     fn order_side_string(side: OrderSide) -> String {
@@ -367,8 +392,8 @@ impl BinanceRestApi {
         }
     }
 
-    pub fn parse_binance_result(message: &String) -> anyhow::Result<Value> {
-        let v = serde_json::from_str::<Value>(message.as_str())
+    pub fn parse_binance_result(message: String) -> anyhow::Result<Value> {
+        let v = serde_json::from_str::<Value>(&message)
             .with_context(|| format!("json format error {:?}", message))?;
 
         let code = v.get("code");
@@ -384,37 +409,73 @@ impl BinanceRestApi {
     }
 
     pub async fn create_listen_key(server: &BinanceServerConfig) -> anyhow::Result<String> {
-        let message = Self::post_key(server,  "/api/v3/userDataStream", "")
-            .await.with_context(||format!("create_listen_key error"))?;
-    
-            if message.get("listenKey").is_some() {
-                let listen_key = message.get("listenKey").unwrap().as_str().unwrap();
-                return Ok(listen_key.to_string());
-            } else {
-                let err_message = format!("listenKey is not found {}", message);
-                return Err(anyhow!(err_message));
-            }
+        let message = Self::post_key(server, "/api/v3/userDataStream", "")
+            .await
+            .with_context(|| format!("create_listen_key error"))?;
+
+        if message.get("listenKey").is_some() {
+            let listen_key = message.get("listenKey").unwrap().as_str().unwrap();
+            return Ok(listen_key.to_string());
+        } else {
+            let err_message = format!("listenKey is not found {}", message);
+            return Err(anyhow!(err_message));
         }
-    
+    }
+
     pub async fn extend_listen_key(server: &BinanceServerConfig, key: &str) -> anyhow::Result<()> {
         let path = format!("/api/v3/userDataStream?listenKey={}", key);
         let _message = Self::put_key(server, path.as_str(), "")
-            .await.with_context(||format!("extend_listen_key error"))?;
+            .await
+            .with_context(|| format!("extend_listen_key error"))?;
 
         Ok(())
     }
-    
-}
 
+    pub async fn get_historical_trades(
+        server: &BinanceServerConfig,
+        config: &MarketConfig,
+        from_id: i64,
+        from_time: MicroSec,
+    ) -> anyhow::Result<Vec<Trade>> {
+        let path = "/api/v3/historicalTrades";
+
+        let params = if from_id == 0 {
+            format!("symbol={}&limit=1000", config.trade_symbol)
+        } else {
+            format!(
+                "symbol={}&fromId={}&limit=1000",
+                config.trade_symbol, from_id
+            )
+        };
+
+        let result = Self::get(&server, path, &params).await?;
+
+        let binance_trades: Vec<BinanceTradeMessage> = serde_json::from_value(result)?;
+
+        let mut trades: Vec<Trade> = vec![];
+
+        for t in binance_trades {
+            let mut trade = t.to_trade();
+
+            if from_time != 0 && trade.time < from_time {
+                continue;
+            }
+
+            trade.status = LogStatus::FixArchiveBlock;
+
+            trades.push(trade);
+        }
+
+        Ok(trades)
+    }
+}
 
 #[cfg(test)]
 mod binance_api_test {
+    use super::*;
     use crate::BinanceConfig;
     use rbot_lib::common::init_debug_log;
     use rust_decimal_macros::dec;
-    use tokio::*;
-    use super::*;
-
     
     #[tokio::test]
     async fn test_board_snapshot() -> anyhow::Result<()> {
@@ -428,7 +489,7 @@ mod binance_api_test {
     }
 
     #[tokio::test]
-    async fn test_recent_trades() -> anyhow::Result<()>{
+    async fn test_recent_trades() -> anyhow::Result<()> {
         let server = BinanceServerConfig::new(false);
         let config = BinanceConfig::BTCUSDT();
 
@@ -453,14 +514,41 @@ mod binance_api_test {
         let config = BinanceConfig::BTCUSDT();
 
         let result = BinanceRestApi::new_order(
-                &server, &config, OrderSide::Buy, 
-                dec![50000], dec![0.001], 
-                OrderType::Limit,                
-                None).await;
+            &server,
+            &config,
+            OrderSide::Buy,
+            dec![50000],
+            dec![0.001],
+            OrderType::Limit,
+            None,
+        )
+        .await;
         println!("result: {:?}", result);
     }
 
 
+    #[tokio::test]
+    async fn test_new_order_market() {
+        let server = BinanceServerConfig::new(false);
+        let config = BinanceConfig::BTCUSDT();
+
+        init_debug_log();
+
+        let result = BinanceRestApi::new_order(
+            &server,
+            &config,
+            OrderSide::Buy,
+            dec![0],
+            dec![0.001],
+            OrderType::Market,
+            None,
+        )
+        .await;
+        println!("result: {:?}", result);
+    }
+
+
+
     #[tokio::test]
     async fn test_open_orders() {
         init_debug_log();
@@ -472,7 +560,7 @@ mod binance_api_test {
     }
 
     #[tokio::test]
-    async fn test_get_account() -> anyhow::Result<()>{
+    async fn test_get_account() -> anyhow::Result<()> {
         init_debug_log();
         let server = BinanceServerConfig::new(false);
 
@@ -492,4 +580,26 @@ mod binance_api_test {
 
         Ok(())
     }
+
+    #[tokio::test]
+    async fn test_get_historical_trades() {
+        init_debug_log();
+        let server = BinanceServerConfig::new(false);
+        let config = BinanceConfig::BTCUSDT();
+
+        let result = BinanceRestApi::get_historical_trades(&server, &config, 0, 0).await;
+        println!("result: {:?}", result);
+    }
+
+    #[tokio::test]
+    async fn test_get_historical_trades_from_id_10000() {
+        init_debug_log();
+        let server = BinanceServerConfig::new(false);
+        let config = BinanceConfig::BTCUSDT();
+
+        let result = BinanceRestApi::get_historical_trades(&server, &config, 10000, 0).await;
+        println!("result: {:?}", result);
+    }
+
+
 }
diff --git a/exchanges/binance/src/ws.rs b/exchanges/binance/src/ws.rs
index 979541e3..5e5acea4 100644
--- a/exchanges/binance/src/ws.rs
+++ b/exchanges/binance/src/ws.rs
@@ -1,18 +1,19 @@
 use futures::Stream;
 use futures::StreamExt;
 use rbot_lib::net::ReceiveMessage;
-use serde::{Deserialize as _, Serialize as _};
+// use serde::{Deserialize as _, Serialize as _};
 use tokio::task::JoinHandle;
 
 use async_stream::stream;
 
 use rbot_lib::{
-    common::{MarketConfig, MicroSec, MultiMarketMessage, ServerConfig, MICRO_SECOND, NOW},
+    common::{MarketConfig, MultiMarketMessage, ServerConfig, NOW},
     net::{AutoConnectClient, WsOpMessage},
 };
 
 use crate::BinanceRestApi;
 use crate::BinanceUserWsMessage;
+use crate::BinanceWsRawMessage;
 use crate::{BinancePublicWsMessage, BinanceServerConfig};
 
 use serde_derive::{Deserialize, Serialize};
@@ -26,7 +27,7 @@ use anyhow::anyhow;
 /// User data streams will close after 60 minutes.
 /// It's recommended to send a ping about every 30 minutes.
 
-const KEY_EXTEND_INTERVAL: MicroSec = 5 * 60 * MICRO_SECOND; // 30 min
+// const KEY_EXTEND_INTERVAL: MicroSec = 5 * 60 * MICRO_SECOND; // 30 min
 
 pub const PING_INTERVAL_SEC: i64 = 60 * 3; // every 3 min
 pub const SWITCH_INTERVAL_SEC: i64 = 60 * 60 * 12; // 12 hours
@@ -68,7 +69,7 @@ impl WsOpMessage for BinanceWsOpMessage {
 
 pub struct BinancePublicWsClient {
     ws: AutoConnectClient<BinanceServerConfig, BinanceWsOpMessage>,
-    handler: Option<JoinHandle<()>>,
+    _handler: Option<JoinHandle<()>>,
 }
 
 impl BinancePublicWsClient {
@@ -88,7 +89,7 @@ impl BinancePublicWsClient {
 
         Self {
             ws: public_ws,
-            handler: None,
+            _handler: None,
         }
     }
 
@@ -137,14 +138,16 @@ impl BinancePublicWsClient {
     }
 
     fn parse_message(message: String) -> anyhow::Result<BinancePublicWsMessage> {
-        let m = serde_json::from_str::<BinancePublicWsMessage>(&message);
+        let m = serde_json::from_str::<BinanceWsRawMessage>(&message);
 
         if m.is_err() {
             log::warn!("Error in serde_json::from_str: {:?}", message);
             return Err(anyhow!("Error in serde_json::from_str: {:?}", message));
         }
 
-        Ok(m.unwrap())
+        let m = m.unwrap();
+
+        Ok(m.into())
     }
 
     // TODO: implement
@@ -155,8 +158,8 @@ impl BinancePublicWsClient {
 
 pub struct BinancePrivateWsClient {
     ws: AutoConnectClient<BinanceServerConfig, BinanceWsOpMessage>,
-    handler: Option<JoinHandle<()>>,
-    listen_key: String,
+    _handler: Option<JoinHandle<()>>,
+    _listen_key: String,
 }
 
 impl BinancePrivateWsClient {
@@ -178,8 +181,8 @@ impl BinancePrivateWsClient {
 
         Self {
             ws: private_ws,
-            handler: None,
-            listen_key: "".to_string(),
+            _handler: None,
+            _listen_key: "".to_string(),
         }
     }
 
@@ -239,8 +242,7 @@ impl BinancePrivateWsClient {
     }
 
     fn convert_ws_message(message: BinanceUserWsMessage) -> anyhow::Result<MultiMarketMessage> {
-        //Ok(message.into())
-        todo!();
+        Ok(message.convert_multimarketmessage("SPOT"))
     }
 
     async fn make_connect_url(server: &BinanceServerConfig) -> anyhow::Result<String> {
@@ -256,7 +258,7 @@ impl BinancePrivateWsClient {
 mod tests {
     use super::*;
     use crate::BinanceConfig;
-    use rbot_lib::common::{init_debug_log, MarketConfig};
+    use rbot_lib::common::init_debug_log;
 
     #[tokio::test]
     async fn test_binance_public_ws_client() {
@@ -293,7 +295,7 @@ mod tests {
     #[tokio::test]
     async fn test_make_connect_url() {
         let server = BinanceServerConfig::new(false);
-        let config = BinanceConfig::BTCUSDT();
+        let _config = BinanceConfig::BTCUSDT();
 
         let url = BinancePrivateWsClient::make_connect_url(&server).await;
 
diff --git a/exchanges/bybit/Cargo.toml b/exchanges/bybit/Cargo.toml
index 332c59c0..ab735039 100644
--- a/exchanges/bybit/Cargo.toml
+++ b/exchanges/bybit/Cargo.toml
@@ -15,6 +15,7 @@ anyhow = {workspace = true}
 
 chrono = {workspace = true}
 log = {workspace = true}
+tracing = {workspace = true}
 
 features = {workspace = true}
 
@@ -36,6 +37,7 @@ async-stream = {workspace = true}
 
 
 
+
 # https://pyo3.rs/v0.13.2/faq
 [dependencies.pyo3]
 version = "0.20"
diff --git a/exchanges/bybit/src/config.rs b/exchanges/bybit/src/config.rs
index 9a00a0df..4618f585 100644
--- a/exchanges/bybit/src/config.rs
+++ b/exchanges/bybit/src/config.rs
@@ -15,7 +15,6 @@ pub struct BybitServerConfig {
     pub rest_server: String,
     pub public_ws: String,
     pub private_ws: String,
-    pub db_base_dir: String,
     pub history_web_base: String,
     api_key: SecretString,
     api_secret: SecretString,
@@ -55,7 +54,6 @@ impl BybitServerConfig {
             rest_server,
             public_ws: public_ws_server,
             private_ws: private_ws_server,
-            db_base_dir: "".to_string(),
             history_web_base: "https://public.bybit.com".to_string(),
             api_key: SecretString::new(&api_key),
             api_secret: SecretString::new(&api_secret)
diff --git a/exchanges/bybit/src/market.rs b/exchanges/bybit/src/market.rs
index 763190e9..228c7a11 100644
--- a/exchanges/bybit/src/market.rs
+++ b/exchanges/bybit/src/market.rs
@@ -79,6 +79,11 @@ impl Bybit {
         self.server_config.exchange_name.clone()
     }
 
+    #[getter]
+    fn get_production(&self) -> bool {
+        self.server_config.production
+    }
+
     pub fn open_market(&self, config: &MarketConfig) -> BybitMarket {
         return BybitMarket::new(&self.server_config, config);
     }
@@ -223,7 +228,9 @@ impl BybitMarket {
     #[new]
     pub fn new(server_config: &BybitServerConfig, config: &MarketConfig) -> Self {
         log::debug!("open market BybitMarket::new");
-        BLOCK_ON(async { Self::async_new(server_config, config).await.unwrap() })
+        BLOCK_ON(async { 
+            Self::async_new(server_config, config, ! server_config.production).await.unwrap() 
+        })
     }
     #[getter]
     fn get_config(&self) -> MarketConfig {
@@ -256,9 +263,11 @@ impl BybitMarket {
         MarketImpl::reset_cache_duration(self)
     }
 
+    /*
     fn stop_db_thread(&mut self) {
         MarketImpl::stop_db_thread(self)
     }
+    */
 
     fn cache_all_data(&mut self) -> anyhow::Result<()> {
         MarketImpl::cache_all_data(self)
@@ -340,7 +349,7 @@ impl BybitMarket {
         MarketImpl::_repr_html_(self)
     }
 
-    #[pyo3(signature = (ndays, force=false, verbose=true, low_priority=true))]
+    #[pyo3(signature = (ndays, force=false, verbose=false, low_priority=true))]    
     fn download_archive(
         &mut self,
         ndays: i64,
@@ -348,36 +357,34 @@ impl BybitMarket {
         verbose: bool,
         low_priority: bool,
     ) -> anyhow::Result<i64> {
-        BLOCK_ON(async {
-            MarketImpl::async_download_archives(self, ndays, force, verbose, low_priority).await
-        })
+            MarketImpl::download_archives(self, ndays, force, verbose, low_priority)
     }
 
-    #[pyo3(signature = (verbose=true))]
+    #[pyo3(signature = (verbose=false))]    
     fn download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        BLOCK_ON(async { self.async_download_lastest(verbose).await })
+        MarketImpl::download_latest(self, verbose)
     }
 
-    fn expire_unfix_data(&mut self) -> anyhow::Result<()> {
-        MarketImpl::expire_unfix_data(self)
+    #[pyo3(signature = (force=false))]
+    fn expire_unfix_data(&mut self, force: bool) -> anyhow::Result<()> {
+        BLOCK_ON(async {
+            self.async_expire_unfix_data(force).await
+        })
     }
 
-    fn find_latest_gap(&self) -> anyhow::Result<(MicroSec, MicroSec)> {
-        MarketImpl::find_latest_gap(self)
+    fn find_latest_gap(&self, force: bool) -> anyhow::Result<(MicroSec, MicroSec)> {
+        MarketImpl::find_latest_gap(self, force)
     }
 
-    fn download_gap(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        MarketImpl::download_gap(self, verbose)
+    #[pyo3(signature = (force=false, verbose=false))]    
+    fn download_gap(&mut self, force: bool, verbose: bool) -> anyhow::Result<i64> {
+        MarketImpl::download_gap(self, force, verbose)
     }
 
     fn start_market_stream(&mut self) -> anyhow::Result<()> {
         MarketImpl::start_market_stream(self)
     }
 
-    fn open_realtime_channel(&mut self) -> anyhow::Result<MarketStream> {
-        MarketImpl::open_realtime_channel(self)
-    }
-
     fn open_backtest_channel(
         &mut self,
         time_from: MicroSec,
@@ -388,20 +395,24 @@ impl BybitMarket {
 }
 
 impl BybitMarket {
-    pub async fn async_new(server_config: &BybitServerConfig, config: &MarketConfig) -> anyhow::Result<Self> {
-        let db_path = Self::make_db_path(
+    pub async fn async_new(
+        server_config: &BybitServerConfig,
+        config: &MarketConfig,
+        test_mode: bool,
+    ) -> anyhow::Result<Self> {
+        let db_path = TradeTable::make_db_path(
             &server_config.exchange_name,
             &config.trade_category,
             &config.trade_symbol,
-            &server_config.db_base_dir,
+            test_mode    
         );
 
         let db = TradeTable::open(&db_path)
-              .with_context(|| format!("Error in TradeTable::open: {:?}", db_path))?;
+            .with_context(|| format!("Error in TradeTable::open: {:?}", db_path))?;
 
         let public_ws = BybitPublicWsClient::new(&server_config, &config).await;
 
-        let mut market =BybitMarket {
+        let mut market = BybitMarket {
             server_config: server_config.clone(),
             config: config.clone(),
             db: Arc::new(Mutex::new(db)),
@@ -409,8 +420,6 @@ impl BybitMarket {
             public_handler: None,
         };
 
-        market.cache_all_data().with_context(|| format!("Error in cache_all_data"))?;
-
         Ok(market)
     }
 }
@@ -437,49 +446,14 @@ impl MarketImpl<BybitRestApi, BybitServerConfig> for BybitMarket {
     }
 
     fn download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        BLOCK_ON(async { self.async_download_lastest(verbose).await })
+        BLOCK_ON(async { self.async_download_latest(verbose).await })
     }
 
-    fn download_gap(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        log::debug!("[start] download_gap ");
-        let gap_result = self.find_latest_gap();
-        if gap_result.is_err() {
-            log::warn!("no gap found: {:?}", gap_result);
-            return Ok(0);
-        }
-
-        let (unfix_start, unfix_end) = gap_result.unwrap();
-        //let (unfix_start, unfix_end) = self.find_latest_gap();
-        log::debug!("unfix_start: {:?}, unfix_end: {:?}", unfix_start, unfix_end);
-
-        if unfix_end - unfix_start <= HHMM(0, 1) {
-            log::info!("no need to download");
-            return Ok(0);
-        }
-
-        let klines = BLOCK_ON(BybitRestApi::get_trade_klines(
-            &self.server_config,
-            &self.get_config(),
-            unfix_start,
-            unfix_end,
-        ))?;
-
-        let trades: Vec<Trade> = convert_klines_to_trades(klines);
-        let rec = trades.len() as i64;
-
-        let expire_message = TradeTableDb::expire_control_message(unfix_start, unfix_end);
-
-        let tx = {
-            let mut lock = self.db.lock().unwrap();
-            lock.start_thread()
-        };
-
-        tx.send(expire_message)?;
-        tx.send(trades)?;
-
-        Ok(rec)
+    fn download_gap(&mut self, force: bool, verbose: bool) -> anyhow::Result<i64> {
+        BLOCK_ON(async { self.async_download_gap(force, verbose).await })
     }
 
+
     fn get_db(&self) -> Arc<Mutex<TradeTable>> {
         self.db.clone()
     }
@@ -492,32 +466,32 @@ impl MarketImpl<BybitRestApi, BybitServerConfig> for BybitMarket {
         self.board.clone()
     }
 
-    fn reflesh_order_book(&mut self) {
+    fn reflesh_order_book(&mut self) -> anyhow::Result<()>{
         todo!()
     }
 
-    fn open_realtime_channel(&mut self) -> anyhow::Result<MarketStream> {
-        todo!()
+    fn start_market_stream(&mut self) -> anyhow::Result<()> {
+        BLOCK_ON(async {
+            self.async_start_market_stream().await
+        })
     }
-
-    fn open_backtest_channel(
+    
+    fn download_archives(
         &mut self,
-        time_from: MicroSec,
-        time_to: MicroSec,
-    ) -> anyhow::Result<MarketStream> {
-        todo!()
-    }
-
-    fn start_market_stream(&mut self) -> anyhow::Result<()> {
-        BLOCK_ON(async { self.async_start_market_stream().await })
+        ndays: i64,
+        force: bool,
+        verbose: bool,
+        low_priority: bool,
+    ) -> anyhow::Result<i64> {
+        BLOCK_ON(async {
+            self.async_download_archives(ndays, force, verbose, low_priority).await
+        })
     }
 
-    fn get_latest_archive_date(&self) -> anyhow::Result<MicroSec> {
-        todo!()
-    }
 }
 
 impl BybitMarket {
+    /*
     async fn async_download_lastest(&mut self, verbose: bool) -> anyhow::Result<i64> {
         if verbose {
             print!("async_download_lastest");
@@ -550,6 +524,7 @@ impl BybitMarket {
 
         Ok(rec)
     }
+    */
 
     async fn async_start_market_stream(&mut self) -> anyhow::Result<()> {
         if self.public_handler.is_some() {
@@ -559,10 +534,10 @@ impl BybitMarket {
 
         let db_channel = {
             let mut lock = self.db.lock().unwrap();
-            lock.start_thread()
+            lock.start_thread().await
         };
 
-        let mut orderbook = self.board.clone();
+        let orderbook = self.board.clone();
 
         let server_config = self.server_config.clone();
         let config = self.config.clone();
@@ -577,7 +552,7 @@ impl BybitMarket {
             let trade_symbol = config.trade_symbol.clone();
 
             public_ws.connect().await;
-            let mut ws_stream = public_ws.open_stream().await;
+            let ws_stream = public_ws.open_stream().await;
             let mut ws_stream = Box::pin(ws_stream);
 
             loop {
@@ -629,17 +604,55 @@ impl BybitMarket {
                     }
                     _ => {
                         log::info!("Market stream message: {:?}", messages);
-                    } /*
-                      MultiMarketMessage::Order(_) => todo!(),
-                      MultiMarketMessage::Account(_) => todo!(),
-                      MultiMarketMessage::Message(_) => todo!(),
-                      */
+                    } 
                 }
             }
         }));
 
         Ok(())
     }
+
+    async fn async_download_gap(&mut self, force: bool, verbose: bool) -> anyhow::Result<i64> {
+        log::debug!("[start] download_gap ");
+        let gap_result = self.find_latest_gap(force);
+        if gap_result.is_err() {
+            log::warn!("no gap found: {:?}", gap_result);
+            return Ok(0);
+        }
+
+        let (unfix_start, unfix_end) = gap_result.unwrap();
+        //let (unfix_start, unfix_end) = self.find_latest_gap();
+        log::debug!("unfix_start: {:?}, unfix_end: {:?}", unfix_start, unfix_end);
+
+        if unfix_end - unfix_start <= HHMM(0, 1) {
+            log::info!("no need to download");
+            return Ok(0);
+        }
+
+        let klines = BLOCK_ON(BybitRestApi::get_trade_klines(
+            &self.server_config,
+            &self.get_config(),
+            unfix_start,
+            unfix_end,
+        ))?;
+
+        let trades: Vec<Trade> = convert_klines_to_trades(klines);
+        let rec = trades.len() as i64;
+
+        let expire_message = TradeTableDb::expire_control_message(unfix_start, unfix_end, false);
+
+        let tx = {
+            let mut lock = self.db.lock().unwrap();
+            lock.start_thread().await
+        };
+
+        tx.send(expire_message)?;
+        tx.send(trades)?;
+
+        Ok(rec)
+    }
+
+
 }
 
 #[cfg(test)]
@@ -769,708 +782,3 @@ mod market_test {
     }
 }
 
-/*
-//----------- FOR BACKUP --------------------------------------------
-//#[derive(Debug)]
-#[pyclass]
-pub struct _BybitMarket {
-    pub server_config: BybitServerConfig,
-    pub config: MarketConfig,
-    pub db: TradeTable,
-    pub board: Arc<RwLock<BybitOrderBook>>,
-    pub public_ws: WebSocketClient<BybitServerConfig, BybitWsOpMessage>,
-    pub public_handler: Option<tokio::task::JoinHandle<()>>,
-    pub user_handler: Option<tokio::task::JoinHandle<()>>,
-    pub agent_channel: Arc<RwLock<MultiChannel<MarketMessage>>>,
-    pub broadcast_message: bool,
-    pub udp_sender: Option<Arc<Mutex<UdpSender>>>,
-}
-
-#[pymethods]
-impl _BybitMarket {
-    #[setter]
-    pub fn set_broadcast_message(&mut self, broadcast_message: bool) {
-        self.broadcast_message = broadcast_message;
-        self.open_udp();
-    }
-
-
-
-    pub fn start_market_stream(&mut self) {
-        RUNTIME.block_on(async move { self._start_market_stream().await })
-    }
-
-    pub fn start_user_stream(&mut self) {
-        let agent_channel = self.agent_channel.clone();
-
-        let server_config = self.server_config.clone();
-        let market_config = self.config.clone();
-
-        let udp_sender = self.udp_sender.clone();
-
-        self.user_handler = Some(listen_userdata_stream(
-            &server_config,
-            move |message: BybitUserStreamMessage| {
-                log::debug!("UserStream: {:?}", message);
-
-                let ms = message.convert_to_market_message(&market_config);
-
-                for m in ms {
-                    if udp_sender.is_some() {
-                        let sender = udp_sender.as_ref().unwrap().as_ref();
-                        sender.lock().unwrap().send_market_message(&m);
-                    }
-
-                    let mut mutl_agent_channel = agent_channel.write().unwrap();
-                    let _ = mutl_agent_channel.send(m);
-                    drop(mutl_agent_channel);
-                }
-            },
-        ));
-
-        log::info!("start_user_stream");
-    }
-
-    /* TODO: implment */
-
-    #[getter]
-    pub fn get_channel(&mut self) -> MarketStream {
-        let ch = self.agent_channel.write().unwrap().open_channel(0);
-
-        MarketStream { reciver: ch }
-    }
-
-    pub fn open_backtest_channel(
-        &mut self,
-        time_from: MicroSec,
-        time_to: MicroSec,
-    ) -> MarketStream {
-        self.db.connection.select_stream(time_from, time_to)
-    }
-
-    #[pyo3(signature = (side, price, size, client_order_id=None))]
-    pub fn limit_order(
-        &self,
-        side: &str,
-        price: Decimal,
-        size: Decimal,
-        client_order_id: Option<&str>,
-    ) -> PyResult<Vec<Order>> {
-        let price_scale = self.config.price_scale;
-        let price_dp = price.round_dp(price_scale);
-
-        let size_scale = self.config.size_scale;
-        let size_dp = size.round_dp(size_scale);
-        let order_side = OrderSide::from(side);
-
-        let response = new_limit_order(
-            &self.server_config,
-            &self.config,
-            order_side,
-            price_dp,
-            size_dp,
-            client_order_id.clone(),
-        );
-
-        if response.is_err() {
-            log::error!(
-                "limit_order: side = {:?}, price = {:?}/{:?}, size = {:?}/{:?}, id = {:?}, result={:?}",
-                side,
-                price,
-                price_dp,
-                size,
-                size_dp,
-                client_order_id,
-                response
-            );
-
-            let err = format!(
-                "limit_order({:?}, {:?}, {:?}, {:?}) -> {:?}",
-                side,
-                price_dp,
-                size_dp,
-                client_order_id,
-                response.unwrap_err()
-            );
-            return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(err));
-        }
-
-        Ok(vec![response.unwrap()])
-    }
-
-    /*
-    pub fn new_market_order_raw(
-        &self,
-        side: &str,
-        size: Decimal,
-        client_order_id: Option<&str>,
-    ) -> PyResult<BinanceOrderResponse> {
-        let size_scale = self.config.market_config.size_scale;
-        let size = size.round_dp(size_scale);
-
-        let order_side = OrderSide::from(side);
-
-        let response = new_market_order(&self.config, order_side, size, client_order_id);
-
-        convert_pyresult(response)
-    }
-    */
-
-    pub fn market_order(
-        &self,
-        side: &str,
-        size: Decimal,
-        client_order_id: Option<&str>,
-    ) -> PyResult<Vec<Order>> {
-        let size_scale = self.config.size_scale;
-        let size = size.round_dp(size_scale);
-
-        let order_side = OrderSide::from(side);
-
-        let response = new_market_order(
-            &self.server_config,
-            &self.config,
-            order_side,
-            size,
-            client_order_id,
-        );
-
-        if response.is_err() {
-            log::error!(
-                "market_order: side = {:?}, size = {:?}, id = {:?}, result={:?}",
-                side,
-                size,
-                client_order_id,
-                response
-            );
-
-            let err = format!(
-                "market_order({:?}, {:?}, {:?}) -> {:?}",
-                side,
-                size,
-                client_order_id,
-                response.unwrap_err()
-            );
-
-            return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(err));
-        }
-
-        Ok(vec![response.unwrap()])
-    }
-
-    pub fn dry_market_order(
-        &self,
-        create_time: MicroSec,
-        order_id: &str,
-        client_order_id: &str,
-        side: OrderSide,
-        size: Decimal,
-        transaction_id: &str,
-    ) -> Vec<Order> {
-        let (bids, asks) = self.board.read().unwrap().get_board_vec().unwrap();
-
-        let board = if side == OrderSide::Buy { asks } else { bids };
-
-        let mut orders: Vec<Order> = vec![];
-        let mut split_index = 0;
-
-        let mut remain_size = size;
-
-        // TODO: consume boards
-        for item in board {
-            if remain_size <= dec![0.0] {
-                break;
-            }
-
-            let execute_size;
-            let order_status;
-            split_index += 1;
-
-            if remain_size <= item.size {
-                order_status = OrderStatus::Filled;
-                execute_size = remain_size;
-                remain_size = dec![0.0];
-            } else {
-                order_status = OrderStatus::PartiallyFilled;
-                execute_size = item.size;
-                remain_size -= item.size;
-            }
-
-            let mut order = Order::new(
-                &self.config.trade_symbol,
-                create_time,
-                &order_id,
-                &client_order_id.to_string(),
-                side,
-                OrderType::Market,
-                order_status,
-                dec![0.0],
-                size,
-            );
-
-            order.transaction_id = format!("{}-{}", transaction_id, split_index);
-            order.update_time = create_time;
-            order.is_maker = false;
-            order.execute_price = item.price;
-            order.execute_size = execute_size;
-            order.remain_size = remain_size;
-            order.quote_vol = order.execute_price * order.execute_size;
-
-            orders.push(order);
-        }
-
-        if remain_size > dec![0.0] {
-            log::error!("remain_size > 0.0: {:?}", remain_size);
-        }
-
-        return orders;
-    }
-
-    pub fn cancel_order(&self, order_id: &str) -> PyResult<Order> {
-        let response = cancel_order(&self.server_config, &self.config, order_id);
-
-        if response.is_err() {
-            log::error!("Error in cancel_order: {:?}", response);
-            return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!(
-                "Error in cancel_order {:?}",
-                response
-            )));
-        }
-
-        let order = response.unwrap();
-
-        Ok(order)
-    }
-
-    pub fn cancel_all_orders(&self) -> PyResult<Vec<Order>> {
-        let response = cancell_all_orders(&self.server_config, &self.config);
-
-        if response.is_err() {
-            log::error!("Error in cancel_all_orders: {:?}", response);
-            return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!(
-                "Error in cancel_all_orders {:?}",
-                response
-            )));
-        }
-
-        return Ok(response.unwrap());
-    }
-
-    #[getter]
-    pub fn get_order_status(&self) -> PyResult<Vec<BybitOrderStatus>> {
-        let status = order_status(&self.server_config.rest_server, &self.config);
-
-        // TODO: IMPLEMENT convert_pyresult(status)
-        return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
-            "Not implemented",
-        ));
-    }
-
-    #[getter]
-    pub fn get_open_orders(&self) -> PyResult<Vec<Order>> {
-        let orders = open_orders(&self.server_config, &self.config);
-
-        log::debug!("OpenOrder: {:?}", orders);
-        Ok(orders.unwrap())
-    }
-
-    #[getter]
-    pub fn get_trade_list(&self) -> PyResult<Vec<BybitOrderStatus>> {
-        let status = trade_list(&self.server_config.rest_server, &self.config);
-
-        //         convert_pyresult(status)
-        return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
-            "Not implemented",
-        ));
-    }
-
-    #[getter]
-    pub fn get_account(&self) -> PyResult<BybitAccountInformation> {
-        let status = get_balance(&self.server_config.rest_server, &self.config);
-
-        //convert_pyresult(status)
-        return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
-            "Not implemented",
-        ));
-    }
-
-    #[getter]
-    pub fn get_recent_trades(&self) -> Vec<Trade> {
-        let trades = get_recent_trade(&self.server_config.rest_server, &self.config);
-
-        if trades.is_err() {
-            log::error!("Error in get_recent_trade: {:?}", trades);
-            return vec![];
-        }
-
-        trades.unwrap().into()
-    }
-}
-
-
-impl _BybitMarket {
-    pub fn new(server_config: &BybitServerConfig, config: &MarketConfig) -> Self {
-        let db = TradeTable::open(&Self::make_db_path(
-            &server_config.exchange_name,
-            &config.trade_category,
-            &config.trade_symbol,
-            &server_config.db_base_dir,
-        ));
-        if db.is_err() {
-            log::error!("Error in TradeTable::open: {:?}", db);
-        }
-
-        return _BybitMarket {
-            server_config: server_config.clone(),
-            config: config.clone(),
-            db: db.unwrap(),
-            board: Arc::new(RwLock::new(BybitOrderBook::new(config))),
-            public_ws: WebSocketClient::new(
-                &server_config,
-                &format!("{}/{}", &server_config.public_ws, config.trade_category),
-                config.public_subscribe_channel.clone(),
-                PING_INTERVAL_SEC,
-                SWITCH_INTERVAL_SEC,
-                SYNC_RECORDS,
-                None,
-            ),
-            public_handler: None,
-            user_handler: None,
-            agent_channel: Arc::new(RwLock::new(MultiChannel::new())),
-            broadcast_message: false,
-            udp_sender: None,
-        };
-    }
-
-    pub fn open_udp(&mut self) {
-        if !self.broadcast_message {
-            return;
-        }
-
-        self.udp_sender = Some(Arc::new(Mutex::new(UdpSender::open(
-            &self.server_config.exchange_name,
-            &self.config.trade_category,
-            &self.config.trade_symbol,
-        ))));
-    }
-
-    pub fn make_db_path(
-        exchange_name: &str,
-        trade_category: &str,
-        trade_symbol: &str,
-        db_base_dir: &str,
-    ) -> String {
-        let db_path = db_full_path(&exchange_name, trade_category, trade_symbol, db_base_dir);
-
-        return db_path.to_str().unwrap().to_string();
-    }
-
-    pub async fn _start_market_stream(&mut self) {
-        // if thread is working, do nothing.
-        if self.public_handler.is_some() {
-            println!("market stream is already running.");
-            return;
-        }
-
-        // delete unstable data
-        let db_channel = self.db.start_thread();
-
-        // if the latest data is overrap with REST minute, do not delete data.
-        let trades = self.get_recent_trades();
-        let l = trades.len();
-        if l != 0 {
-            let rest_start_time = trades[l - 1].time;
-            let last_time = self.db.end_time(rest_start_time);
-            if last_time.is_err() {
-                let now = NOW();
-
-                log::debug!(
-                    "db has gap, so delete after FIX data now={:?} / db_end={:?}",
-                    time_string(now),
-                    time_string(rest_start_time)
-                );
-
-                let delete_message = self.db.connection.make_expire_control_message(now);
-
-                log::debug!("delete_message: {:?}", delete_message);
-
-                db_channel.send(delete_message).unwrap();
-            } else {
-                log::debug!(
-                    "db has no gap, so continue to receive data {}",
-                    time_string(rest_start_time)
-                );
-            }
-
-            db_channel.send(trades).unwrap();
-        }
-
-        let agent_channel = self.agent_channel.clone();
-        let ws_channel = self.public_ws._open_channel().await;
-
-        self.public_ws
-            ._connect(|message| {
-                let m = serde_json::from_str::<BybitWsMessage>(&message);
-
-                if m.is_err() {
-                    log::warn!("Error in serde_json::from_str: {:?}", message);
-                    println!("ERR: {:?}", message);
-                }
-
-                let m = m.unwrap();
-
-                return m.into();
-            })
-            .await;
-
-        let board = self.board.clone();
-        let db_channel_for_after = db_channel.clone();
-
-        let udp_sender = self.udp_sender.clone();
-
-        let handler = tokio::task::spawn(async move {
-            loop {
-                let message = ws_channel.recv();
-                let message = message.unwrap();
-
-                if message.trade.len() != 0 {
-                    log::debug!("Trade: {:?}", message.trade);
-                    let r = db_channel.send(message.trade.clone());
-
-                    if r.is_err() {
-                        log::error!("Error in db_channel.send: {:?}", r);
-                    }
-                }
-
-                if message.orderbook.is_some() {
-                    let orderbook = message.orderbook.clone().unwrap();
-                    let mut b = board.write().unwrap();
-                    b.update(&orderbook);
-                    drop(b);
-                }
-
-                let messages = message.extract();
-
-                // update board
-
-                // send message to agent
-                for m in messages {
-                    if m.trade.is_some() {
-                        // broadcast only trade message
-                        if udp_sender.is_some() {
-                            let sender = udp_sender.as_ref().unwrap().as_ref();
-                            sender.lock().unwrap().send_market_message(&m);
-                        }
-
-                        let mut ch = agent_channel.write().unwrap();
-                        let r = ch.send(m.clone());
-                        drop(ch);
-
-                        if r.is_err() {
-                            log::error!("Error in db_channel.send: {:?}", r);
-                        }
-                    }
-                }
-            }
-        });
-
-        self.public_handler = Some(handler);
-
-        // update recent trade
-        // wait for channel open
-        sleep(Duration::from_millis(100)); // TODO: fix to wait for channel open
-        let trade = self.get_recent_trades();
-        db_channel_for_after.send(trade).unwrap();
-
-        // TODO: store recent trade timestamp.
-
-        log::info!("start_market_stream");
-    }
-
-    fn history_web_url(&self, date: MicroSec) -> String {
-        Self::make_historical_data_url_timestamp(
-            &self.server_config.history_web_base,
-            &self.config.trade_symbol,
-            date,
-        )
-    }
-
-
-    fn get_latest_archive_date(&self) -> Result<MicroSec, String> {
-        let f = |date: MicroSec| -> String {
-            Self::make_historical_data_url_timestamp(
-                &self.server_config.history_web_base,
-                &self.config.trade_symbol,
-                date,
-            )
-        };
-
-        latest_archive_date(&f)
-    }
-
-    pub fn download_log(
-        &mut self,
-        tx: &Sender<Vec<Trade>>,
-        date: MicroSec,
-        low_priority: bool,
-        verbose: bool,
-    ) -> PyResult<i64> {
-        let date = FLOOR_DAY(date);
-        let url = self.history_web_url(date);
-
-        match download_log(&url, tx, true, low_priority, verbose, &Self::rec_to_trade) {
-            Ok(download_rec) => Ok(download_rec),
-            Err(e) => Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!(
-                "Error in download_logs: {:?}",
-                e
-            ))),
-        }
-    }
-
-    /// timestamp,      symbol,side,size,price,  tickDirection,trdMatchID,                          grossValue,  homeNotional,foreignNotional
-    /// 1620086396.8268,BTCUSDT,Buy,0.02,57199.5,ZeroMinusTick,224061a0-e105-508c-9696-b53ab4b5bb03,114399000000.0,0.02,1143.99
-    fn rec_to_trade(rec: &StringRecord) -> Trade {
-        let timestamp = rec
-            .get(0)
-            .unwrap_or_default()
-            .parse::<f64>()
-            .unwrap_or_default()
-            * 1_000_000.0;
-
-        let timestamp = timestamp as MicroSec;
-
-        let order_side = match rec.get(2).unwrap_or_default() {
-            "Buy" => OrderSide::Buy,
-            "Sell" => OrderSide::Sell,
-            _ => OrderSide::Unknown,
-        };
-
-        let id = rec.get(6).unwrap_or_default().to_string();
-
-        let price = rec
-            .get(4)
-            .unwrap_or_default()
-            .parse::<f64>()
-            .unwrap_or_default();
-
-        let price: Decimal = Decimal::from_f64(price).unwrap();
-
-        let size = rec
-            .get(3)
-            .unwrap_or_default()
-            .parse::<f64>()
-            .unwrap_or_default();
-
-        let size = Decimal::from_f64(size).unwrap();
-
-        let trade = Trade::new(
-            timestamp,
-            order_side,
-            price,
-            size,
-            LogStatus::FixArchiveBlock,
-            &id,
-        );
-
-        return trade;
-    }
-
-    /// Check if database is valid at the date
-    fn validate_db_by_date(&mut self, date: MicroSec) -> bool {
-        self.db.connection.validate_by_date(date)
-    }
-}
-
-#[cfg(test)]
-mod test_bybit_market {
-    use csv::StringRecord;
-    use rust_decimal_macros::dec;
-
-    use crate::{
-        common::{time_string, DAYS, NOW},
-        exchange::bybit::config::{BybitConfig, BybitServerConfig},
-    };
-
-    #[test]
-    fn test_make_historical_data_url_timestamp() {
-        let config = BybitConfig::BTCUSDT();
-
-        let url = super::_BybitMarket::make_historical_data_url_timestamp(
-            "https://public.bybit.com/trading/",
-            &config.trade_symbol,
-            1620000000000_000,
-        );
-
-        assert_eq!(
-            url,
-            "https://public.bybit.com/trading/BTCUSDT/BTCUSDT2021-05-03.csv.gz"
-        );
-
-        let url = super::_BybitMarket::make_historical_data_url_timestamp(
-            "https://public.bybit.com/trading/",
-            &config.trade_symbol,
-            1234,
-        );
-
-        assert_eq!(
-            url,
-            "https://public.bybit.com/trading/BTCUSDT/BTCUSDT1970-01-01.csv.gz"
-        );
-    }
-
-    #[test]
-    fn test_get_latest_archive_date() {
-        let server_config = BybitServerConfig::new(false);
-        let config = BybitConfig::BTCUSDT();
-
-        let market = super::_BybitMarket::new(&server_config, &config);
-
-        let date = market.get_latest_archive_date().unwrap();
-        println!("date: {}", date);
-        println!("date: {}", time_string(date));
-    }
-
-    #[test]
-    fn test_rec_to_trade() {
-        let rec = "1692748800.279,BTCUSD,Sell,641,26027.00,ZeroMinusTick,80253109-efbb-58ca-9adc-d458b66201e9,2.462827064202559e+06,641,0.02462827064202559".to_string();
-        let rec = rec.split(',').collect::<Vec<&str>>();
-        let rec = StringRecord::from(rec);
-        let trade = super::_BybitMarket::rec_to_trade(&rec);
-
-        assert_eq!(trade.time, 1692748800279000);
-        assert_eq!(trade.order_side, super::OrderSide::Sell);
-        assert_eq!(trade.price, dec![26027.0]);
-        assert_eq!(trade.size, dec![641.0]);
-        assert_eq!(trade.id, "80253109-efbb-58ca-9adc-d458b66201e9");
-    }
-
-    #[test]
-    fn test_make_expire_message_control() {
-        let server_config = BybitServerConfig::new(false);
-        let config = BybitConfig::BTCUSDT();
-
-        let mut market = super::BybitMarket::new(&server_config, &config);
-
-        let message = market.db.connection.make_expire_control_message(NOW());
-
-        println!("{:?}", message);
-    }
-
-    #[test]
-    fn test_lastest_fix_time() {
-        let server_config = BybitServerConfig::new(false);
-        let config = BybitConfig::BTCUSDT();
-
-        let mut market = super::BybitMarket::new(&server_config, &config);
-
-        let start_time = NOW() - DAYS(2);
-
-        let fix_time = market.db.connection.latest_fix_time(start_time);
-
-        println!("fix_time: {:?}/{:?}", time_string(fix_time), fix_time);
-
-        let unfix_time = market.db.connection.first_unfix_time(fix_time);
-        println!("unfix_time: {:?}/{:?}", time_string(unfix_time), unfix_time);
-    }
-}
-*/
diff --git a/exchanges/bybit/src/rest.rs b/exchanges/bybit/src/rest.rs
index f98c706b..5cbeabf6 100644
--- a/exchanges/bybit/src/rest.rs
+++ b/exchanges/bybit/src/rest.rs
@@ -123,8 +123,8 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
 
         let params = format!(
             "category={}&symbol={}&limit={}",
-            config.trade_category.as_str(),
-            config.trade_symbol.as_str(),
+            &config.trade_category,
+            &config.trade_symbol,
             1000 // max records.
         );
 
@@ -479,6 +479,8 @@ config: &MarketConfig,
     }
 }
 
+use tracing::instrument;    
+
 impl BybitRestApi {
     async fn get(
         server: &BybitServerConfig,
diff --git a/rbot/Cargo.toml b/rbot/Cargo.toml
index 347ecb58..238131dc 100644
--- a/rbot/Cargo.toml
+++ b/rbot/Cargo.toml
@@ -4,7 +4,8 @@ version = "0.3.1"
 edition = "2021"
 
 [buildflags]
-rustflags = ["--cfg", "tokio_unstable"]
+rustflags = ["--cfg tokio_unstable", 
+            "--cfg tracing_unstable"]
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 [lib]
@@ -23,6 +24,8 @@ once_cell = {workspace = true}
 
 console-subscriber = {version="0.2.0"}
 
+tracing = {workspace= true}
+tracing-subscriber = {workspace= true}
 
 # https://pyo3.rs/v0.13.2/faq
 [dependencies.pyo3]
diff --git a/rbot/src/lib.rs b/rbot/src/lib.rs
index 1ed3fccc..fde59c23 100644
--- a/rbot/src/lib.rs
+++ b/rbot/src/lib.rs
@@ -6,11 +6,9 @@ mod test;
 
 use binance::{Binance, BinanceConfig};
 use pyo3::{pymodule, types::PyModule, wrap_pyfunction, PyResult, Python};
-use rbot_lib::common::{
-    get_orderbook_list,
-    init_debug_log, init_log, time_string, 
-    AccountPair, BoardItem, MarketConfig, Order, OrderSide, OrderStatus, OrderType, Trade, DAYS, DAYS_BEFORE, FLOOR_SEC, HHMM, MIN, NOW, SEC
-};
+use rbot_lib::{common::{
+    get_orderbook, get_orderbook_list, init_debug_log, init_log, time_string, AccountPair, BoardItem, MarketConfig, Order, OrderSide, OrderStatus, OrderType, Trade, DAYS, DAYS_BEFORE, FLOOR_SEC, HHMM, MIN, NOW, SEC
+}, db::get_db_root, db::set_db_root};
 
 
 use rbot_session::{Logger, Session, Runner, ExecuteMode};
@@ -18,41 +16,22 @@ use rbot_session::{Logger, Session, Runner, ExecuteMode};
 use bybit::{Bybit, BybitConfig};
 
 
-
-
-/*
-use exchange::BoardItem;
-use exchange::bybit::{BybitMarket, Bybit};
-use exchange::bybit::config::{BybitConfig, BybitServerConfig};
-use pyo3::prelude::*;
-// use exchange::ftx::FtxMarket;
-use exchange::binance::{BinanceMarket, BinanceConfig, Binance};
-// use exchange::bb::BBMarket;
-
-use common::*;
-use session::{Session, ExecuteMode, Logger};
-use session::Runner;
-*/
-
-
-// use net::{Broadcast, BroadcastMessage};
-//use sim::back::BackTester;
-//use sim::session::DummySession;
-
-/// A Python module implemented in Rust.
-
-
 // use console_subscriber;
 #[pymodule]
 fn rbot(_py: Python, m: &PyModule) -> PyResult<()> {
     // console_subscriber::init();  // for tokio thread debug
+    //tracing_subscriber::fmt::init();
 
     m.add("__version__", env!("CARGO_PKG_VERSION"))?;
 
+    m.add_function(wrap_pyfunction!(get_db_root, m)?)?;
+    m.add_function(wrap_pyfunction!(set_db_root, m)?)?;
+
     m.add_function(wrap_pyfunction!(init_log, m)?)?;
     m.add_function(wrap_pyfunction!(init_debug_log, m)?)?;
 
     m.add_function(wrap_pyfunction!(get_orderbook_list, m)?)?;
+    m.add_function(wrap_pyfunction!(get_orderbook, m)?)?;
 
     // time util
     m.add_function(wrap_pyfunction!(time_string, m)?)?;
diff --git a/rbot_lib/src/common/account.rs b/rbot_lib/src/common/account.rs
deleted file mode 100644
index e69de29b..00000000
diff --git a/rbot_lib/src/common/ch.rs b/rbot_lib/src/common/ch.rs
index bc7930d3..af03cb48 100644
--- a/rbot_lib/src/common/ch.rs
+++ b/rbot_lib/src/common/ch.rs
@@ -13,6 +13,7 @@ use super::order::Trade;
 use super::AccountCoins;
 use super::AccountPair;
 use super::BoardTransfer;
+use super::MarketConfig;
 use super::OrderBookRaw;
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -34,6 +35,24 @@ pub enum MarketMessage {
 }
 
 impl MarketMessage {
+    pub fn update_config(&mut self, config: &MarketConfig) {
+        match self {
+            MarketMessage::Trade(_trade) => {
+                //                
+            }
+            MarketMessage::Order(order) => {
+                order.update_balance(config);
+            }
+            MarketMessage::Account(_account) => {
+                //
+            }
+            MarketMessage::Orderbook(_orderbook) => {
+                //
+            }
+            _ => {}
+        }
+    }
+
     pub fn from_trade(trade: Trade) -> Self {
         MarketMessage::Trade(trade)
     }
diff --git a/rbot_lib/src/common/mod.rs b/rbot_lib/src/common/mod.rs
index a781647a..65ce62a0 100644
--- a/rbot_lib/src/common/mod.rs
+++ b/rbot_lib/src/common/mod.rs
@@ -8,7 +8,6 @@ mod order;
 mod ch;
 mod logger;
 mod config;
-mod account;
 mod env;
 mod orderbook;
 mod util;
@@ -20,7 +19,6 @@ pub use ch::*;
 pub use logger::*;
 pub use order::*;
 pub use config::*;
-pub use account::*;
 pub use env::*;
 pub use orderbook::*;
 pub use util::*;
diff --git a/rbot_lib/src/common/order.rs b/rbot_lib/src/common/order.rs
index 1ad66377..064f87bc 100644
--- a/rbot_lib/src/common/order.rs
+++ b/rbot_lib/src/common/order.rs
@@ -1,6 +1,8 @@
 // Copyright(c) 2022-4. yasstake. All rights reserved.
 // ABUSOLUTELY NO WARRANTY.
 
+use std::path::Display;
+
 use crate::common::time::time_string;
 
 use super::time::MicroSec;
@@ -228,7 +230,8 @@ pub enum LogStatus {
     FixRestApiStart,
     FixRestApiBlock, // データが確定(アーカイブ）し、ブロックの中間を表す（REST API）
     FixRestApiEnd,
-    ExpireControl, // 削除指示
+    ExpireControlForce, // 削除指示（アーカイブ意外は強制削除）
+    ExpireControl, // 削除指示(通常：WSデータのみ削除)
     Unknown,       // 未知のステータス / 未確定のステータス
 }
 
@@ -248,7 +251,8 @@ impl From<&str> for LogStatus {
             "s" => LogStatus::FixRestApiStart,
             "a" => LogStatus::FixRestApiBlock,
             "e" => LogStatus::FixRestApiEnd,
-            "X" => LogStatus::ExpireControl,
+            "X" => LogStatus::ExpireControlForce,
+            "x" => LogStatus::ExpireControl,            
             _ => {
                 log::error!("Unknown log status: {:?}", status);
                 LogStatus::Unknown
@@ -267,8 +271,9 @@ impl LogStatus {
             LogStatus::FixRestApiStart => "s".to_string(),
             LogStatus::FixRestApiBlock => "a".to_string(),
             LogStatus::FixRestApiEnd => "e".to_string(),
-            LogStatus::ExpireControl => "X".to_string(),
-            LogStatus::Unknown => "x".to_string(),
+            LogStatus::ExpireControlForce => "X".to_string(),            
+            LogStatus::ExpireControl => "x".to_string(),
+            LogStatus::Unknown => "?".to_string(),
         }
     }
 }
@@ -320,32 +325,54 @@ impl Trade {
         )
     }
 
+
     pub fn __str__(&self) -> String {
         format!(
-            "{{timestamp:{}({:?}), order_side:{:?}, price:{:?}, size:{:?}, id:{:?}}}",
+            "{{timestamp:{}({:?}), order_side:{:?}, price:{:?}, size:{:?}, id:{:?}, status{:?}}}",
             time_string(self.time),
             self.time,
             self.order_side,
             self.price,
             self.size,
-            self.id
+            self.id,
+            self.status
         )
     }
 
     pub fn __repr__(&self) -> String {
         format!(
-            "{{timestamp:{:?}, order_side:{:?}, price:{:?}, size:{:?}, id:{:?}}}",
-            self.time, self.order_side, self.price, self.size, self.id
+            "{{timestamp:{:?}, order_side:{:?}, price:{:?}, size:{:?}, id:{:?}, status:{:?}}}",
+            self.time, self.order_side, self.price, self.size, self.id, self.status
         )
     }
 }
 
+impl Into<String> for &Trade {
+    fn into(self) -> String {
+        self.__str__()
+    }
+}
+
 impl Into<MarketMessage> for &Trade {
     fn into(self) -> MarketMessage {
         MarketMessage::Trade(self.clone())
     }
 }
 
+impl Default for Trade {
+    fn default() -> Self {
+        return Trade {
+            time: 0,
+            order_side: OrderSide::Buy,
+            price: dec![0.0],
+            size: dec![0.0],
+            status: LogStatus::UnFix,
+            id: "".to_string(),
+        };
+    }
+}
+
+
 #[pyclass]
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct AccountChange {
@@ -407,9 +434,15 @@ pub struct AccountCoins {
     pub coins: Vec<Coin>,
 }
 
+impl Default for AccountCoins {
+    fn default() -> Self {
+        AccountCoins::new()
+    }    
+}
+
 impl AccountCoins {
     pub fn new() -> Self {
-        return AccountCoins { coins: Vec::new() };
+        AccountCoins { coins: Vec::new() }
     }
 
     pub fn push(&mut self, coin: Coin) {
@@ -420,6 +453,24 @@ impl AccountCoins {
         self.coins.append(&mut coins.coins);
     }
 
+    pub fn update(&mut self, coins: &AccountCoins) {
+        for coin in coins.coins.iter() {
+            let mut found = false;
+            for my_coin in self.coins.iter_mut() {
+                if my_coin.symbol == coin.symbol {
+                    my_coin.volume = coin.volume;
+                    my_coin.free = coin.free;
+                    my_coin.locked = coin.locked;
+                    found = true;
+                    break;
+                }
+            }
+            if !found {
+                self.coins.push(coin.clone());
+            }
+        }
+    }
+
     pub fn extract_pair(&self, config: &MarketConfig) -> AccountPair {
         let mut home = Coin::default();
         let mut foreign = Coin::default();
@@ -434,6 +485,44 @@ impl AccountCoins {
 
         return AccountPair { home, foreign };
     }
+
+    pub fn diff_update(&mut self, symbol: &str, volume: Decimal, free: Decimal, locked: Decimal) {
+        for coin in self.coins.iter_mut() {
+            if coin.symbol == symbol {
+                coin.volume += volume;
+                coin.free += free;
+                coin.locked += locked;
+                return;
+            }
+        }
+        let coin = Coin {
+            symbol: symbol.to_string(),
+            volume,
+            free,
+            locked,
+        };
+        self.coins.push(coin);
+    }
+
+    pub fn apply_order(&mut self, config: &MarketConfig, order: &Order) {
+        self.diff_update(
+            &config.home_currency, 
+            order.home_change,
+            order.free_home_change,
+            order.lock_home_change,
+        );
+
+        self.diff_update(
+            &config.foreign_currency,
+            order.foreign_change,
+            order.free_foreign_change,
+            order.lock_foreign_change,
+        );
+    }
+
+    pub fn __repr__(&self) -> String {
+        serde_json::to_string(&self).unwrap()
+    }
 }
 
 
diff --git a/rbot_lib/src/common/orderbook.rs b/rbot_lib/src/common/orderbook.rs
index c05d6d40..d2459b4a 100644
--- a/rbot_lib/src/common/orderbook.rs
+++ b/rbot_lib/src/common/orderbook.rs
@@ -127,7 +127,7 @@ pub fn get_orderbook_bin(path: &str) -> anyhow::Result<Vec<u8>> {
     Ok(transfer.to_vec())
 }
 
-
+#[pyfunction]
 pub fn get_orderbook(path: &str) -> anyhow::Result<OrderBook> {
     ALL_BOARD
         .lock()
@@ -151,6 +151,7 @@ pub fn get_orderbook_df(path: &str) -> anyhow::Result<(DataFrame, DataFrame)> {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct BoardTransfer {
     pub last_update_time: MicroSec,
+    pub first_update_id: u64,
     pub last_update_id: u64,
     pub bids: Vec<BoardItem>,
     pub asks: Vec<BoardItem>,
@@ -161,6 +162,7 @@ impl BoardTransfer {
     pub fn new() -> Self {
         BoardTransfer {
             last_update_time: 0,
+            first_update_id: 0,
             last_update_id: 0,
             bids: vec![],
             asks: vec![],
@@ -170,6 +172,7 @@ impl BoardTransfer {
 
     pub fn from_orderbook(order_book: &OrderBookRaw) -> Self {
         BoardTransfer {
+            first_update_id: order_book.first_update_id,
             last_update_time: order_book.last_update_time,
             last_update_id: order_book.last_update_id,
             bids: order_book.bids.get(),
@@ -348,6 +351,7 @@ impl Board {
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct OrderBookRaw {
     pub last_update_time: MicroSec,
+    pub first_update_id: u64,    
     pub last_update_id: u64,
     pub bids: Board,
     pub asks: Board,
@@ -356,6 +360,7 @@ pub struct OrderBookRaw {
 impl OrderBookRaw {
     pub fn new(max_depth: u32) -> Self {
         OrderBookRaw {
+            first_update_id: 0,
             last_update_id: 0,
             last_update_time: 0,
             bids: Board::new(max_depth, false),
@@ -396,6 +401,7 @@ impl OrderBookRaw {
 
     pub fn update(&mut self, board_transfer: &BoardTransfer) {
         self.last_update_time = board_transfer.last_update_time;
+        self.first_update_id = board_transfer.first_update_id;
         self.last_update_id = board_transfer.last_update_id;
 
         if board_transfer.snapshot {
@@ -443,6 +449,18 @@ impl OrderBook {
         }
     }
 
+    pub fn get_first_update_id(&self) -> u64 {
+        self.board.lock().unwrap().first_update_id
+    }
+
+    pub fn get_last_update_time(&self) -> MicroSec {
+        self.board.lock().unwrap().last_update_time
+    }
+
+    pub fn get_last_update_id(&self) -> u64 {
+        self.board.lock().unwrap().last_update_id
+    }
+
     pub fn from_bin(
         exchange_name: &str,
         config: &MarketConfig,
@@ -672,6 +690,7 @@ mod board_test {
         let mut b = OrderBookRaw::new(0);
 
         let board_transfer = BoardTransfer{
+            first_update_id: 0,
             last_update_time: 0,
             last_update_id: 0,
             bids: vec![
diff --git a/rbot_lib/src/common/util.rs b/rbot_lib/src/common/util.rs
index 89aee008..98ac7332 100644
--- a/rbot_lib/src/common/util.rs
+++ b/rbot_lib/src/common/util.rs
@@ -18,13 +18,6 @@ use anyhow::anyhow;
 use super::env_rbot_db_root;
 
 
-pub static DB_ROOT: Lazy<String> = Lazy::new(|| {
-    if let Ok(path) = env_rbot_db_root() {
-        return path;
-    } else {
-        "".to_string()
-    }
-});
 
 #[derive(Clone, Deserialize)]
 pub struct SecretString {
diff --git a/rbot_lib/src/db/fs.rs b/rbot_lib/src/db/fs.rs
index b2f7d3ca..5df7ebf8 100644
--- a/rbot_lib/src/db/fs.rs
+++ b/rbot_lib/src/db/fs.rs
@@ -2,36 +2,55 @@
 
 use std::fs;
 use std::path::PathBuf;
+use std::sync::Mutex;
 use directories::ProjectDirs;
+use once_cell::sync::Lazy;
+use pyo3::pyfunction;
 
-use crate::common::DB_ROOT;
+use crate::common::env_rbot_db_root;
 
 
-pub fn project_dir() -> PathBuf {
-    let proj_dir = ProjectDirs::from("net", "takibi", "rbot").unwrap();
-
-    return proj_dir.data_dir().to_owned();
+#[pyfunction]
+pub fn get_db_root() -> String {
+    DB_ROOT.lock().unwrap().to_string()
 }
 
+#[pyfunction]
+pub fn set_db_root(path: &str) {
+    let mut root_path = DB_ROOT.lock().unwrap();
 
-pub fn db_full_path(exchange_name: &str, category: &str, symbol: &str, base_dir: &str) -> PathBuf {
-    let path = DB_ROOT.to_string();    
-    
-    let project_dir = if base_dir != "" {
-        PathBuf::from(base_dir)
-    }
-    else if path != "" {
-        PathBuf::from(path)
+    *root_path = path.to_string();
+}
+
+pub static DB_ROOT: Lazy<Mutex<String>> = Lazy::new(|| Mutex::new({
+    if let Ok(path) = env_rbot_db_root() {
+        path
+    } else {
+        project_dir()        
     }
-    else {
-        project_dir()
-    };
+}));
 
+
+pub fn project_dir() -> String {
+    let proj_dir = ProjectDirs::from("net", "takibi", "rbot").unwrap();
+
+    return proj_dir.data_dir().to_str().unwrap().to_string();
+}
+
+pub fn db_full_path(exchange_name: &str, category: &str, symbol: &str, base_dir: &str, test_net: bool) -> PathBuf {
+    let project_dir = get_db_root();
+    let project_dir = PathBuf::from(project_dir);
+    
     let db_dir = project_dir.join("DB");
     let exchange_dir = db_dir.join(exchange_name);
     let _ = fs::create_dir_all(&exchange_dir);
 
-    let db_name = format!("{}-{}.db", category, symbol);
+    let mut db_name = format!("{}-{}.db", category, symbol);
+
+    if test_net {
+        db_name = format!("TEST-{}", db_name);
+    }
+    
     let db_path = exchange_dir.join(db_name);
 
     return db_path;
@@ -44,6 +63,7 @@ mod test_fs {
     #[test]
     fn test_project_dir() {
         let path = project_dir();
+        let path = PathBuf::from(path);
 
         let db_name = path.join(".db");
 
@@ -52,13 +72,13 @@ mod test_fs {
 
     #[test]
     fn test_db_full_path() {
-        let db = db_full_path("FTX", "SPOT", "BTC-PERP", "/tmp");
+        let db = db_full_path("FTX", "SPOT", "BTC-PERP", "/tmp", false);
         println!("{:?}", db);
 
-        let db = db_full_path("FTX", "SPOT", "BTC-PERP", "/tmp/");
+        let db = db_full_path("FTX", "SPOT", "BTC-PERP", "/tmp/", true);
         println!("{:?}", db);
 
-        let db = db_full_path("FTX", "SPOT", "BTC-PERP", "./");
+        let db = db_full_path("FTX", "SPOT", "BTC-PERP", "./", false);
         println!("{:?}", db);
 
     }
diff --git a/rbot_lib/src/db/sqlite.rs b/rbot_lib/src/db/sqlite.rs
index bfd3c86c..6e879d16 100644
--- a/rbot_lib/src/db/sqlite.rs
+++ b/rbot_lib/src/db/sqlite.rs
@@ -7,6 +7,7 @@ use anyhow::Context;
 
 use numpy::IntoPyArray;
 use numpy::PyArray2;
+use once_cell::sync::Lazy;
 use polars::prelude::DataFrame;
 use polars::prelude::Float64Type;
 use polars_core::prelude::IndexOrder;
@@ -22,6 +23,10 @@ use std::time::Duration;
 use tokio::task::spawn;
 use tokio::task::JoinHandle;
 
+//use std::thread::JoinHandle;
+//use std::thread::spawn;
+
+use crate::common::env_rbot_db_root;
 use crate::common::MarketStream;
 
 //use crossbeam_channel::Receiver;
@@ -46,9 +51,13 @@ use crate::db::df::{end_time_df, make_empty_ohlcvv, ohlcv_df, ohlcv_from_ohlcvv_
 
 use crate::db::df::KEY;
 
+use super::db_full_path;
 use super::df::convert_timems_to_datetime;
 use super::df::vap_df;
 
+
+
+
 #[derive(Debug)]
 pub struct TradeTableDb {
     pub file_name: String,
@@ -68,21 +77,29 @@ impl TradeTableDb {
 
     /// delete unstable data, include both edge.
     /// start_time <= (time_stamp) <= end_time
-    pub fn delete_unstable_data(tx: &Transaction, start_time: MicroSec, end_time: MicroSec) {
+    pub fn delete_unstable_data(
+        tx: &Transaction,
+        start_time: MicroSec,
+        end_time: MicroSec,
+    ) -> anyhow::Result<i64> {
         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]);
+        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) => {
-                println!("delete unstable data error {}", e);
-                log::error!("delete unstable data error {}", e);
-            }
-        }
+        Ok(result as i64)
+    }
+
+    pub fn delete_unarchived_data(
+        tx: &Transaction,
+        start_time: MicroSec,
+        end_time: MicroSec,
+    ) -> anyhow::Result<i64> {
+        let sql = r#"delete from trades where $1 <= time_stamp and time_stamp <= $2 and (status = "U" or status = "s" or status = "a" or status = "e")"#;
+
+        let result = tx.execute(sql, params![start_time, end_time])?;
+
+        Ok(result as i64)
     }
 
     /// insert trades into database
@@ -122,36 +139,78 @@ impl TradeTableDb {
         Ok(tx)
     }
 
-    pub fn latest_fix_time(&mut self, start_time: MicroSec) -> anyhow::Result<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 limit 1"#;
+    pub fn latest_fix_trade(
+        &mut self,
+        start_time: MicroSec,
+        force: bool,
+    ) -> anyhow::Result<Option<Trade>> {
+        let sql = if force {
+            r#"select time_stamp, action, price, size, status, id from trades where ($1 < time_stamp) and (status = "E") order by time_stamp desc limit 1"#
+        } else {
+            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 limit 1"#
+        };
+
         let trades = self.select_query(sql, vec![start_time])?;
 
         if trades.len() == 0 {
+            return Ok(None);
+        }
+
+        log::debug!("latest_fix_trade {}", trades[0].__str__());
+
+        Ok(Some(trades[0].clone()))
+    }
+
+    pub fn latest_fix_time(
+        &mut self,
+        start_time: MicroSec,
+        force: bool,
+    ) -> anyhow::Result<MicroSec> {
+        let trade = self.latest_fix_trade(start_time, force)?;
+
+        if trade.is_none() {
             return Ok(0);
         }
 
-        Ok(TradeTable::ohlcv_end(trades[0].time) + 1) // plus 1 microsecond(1us) to avoid duplicate data
+        let trade = trade.unwrap();
+
+        Ok(TradeTable::ohlcv_end(trade.time) + 1) // plus 1 microsecond(1us) to avoid duplicate data
     }
 
-    pub fn first_unfix_time(&mut self, start_time: MicroSec) -> anyhow::Result<MicroSec> {
-        let sql = r#"select time_stamp, action, price, size, status, id from trades where $1 < time_stamp order by time_stamp limit 1"#;
+    pub fn first_unfix_trade(&mut self, start_time: MicroSec) -> anyhow::Result<Option<Trade>> {
+        let sql = r#"select time_stamp, action, price, size, status, id from trades where ($1 < time_stamp) and (status = "U") order by time_stamp limit 1"#;
         let trades = self.select_query(sql, vec![start_time])?;
 
         if trades.len() == 0 {
+            return Ok(None);
+        }
+
+        Ok(Some(trades[0].clone()))
+    }
+
+    pub fn first_unfix_time(&mut self, start_time: MicroSec) -> anyhow::Result<MicroSec> {
+        let trade = self.first_unfix_trade(start_time)?;
+        if trade.is_none() {
             return Ok(0);
         }
 
-        Ok(TradeTable::ohlcv_end(trades[0].time) - 1) // minus 1 microsecond(1us) to avoid duplicate data
+        let trade = trade.unwrap();
+
+        Ok(TradeTable::ohlcv_end(trade.time) - 1) // minus 1 microsecond(1us) to avoid duplicate data
     }
 
     /// 2日以内のUnstableデータを削除するメッセージを作成する。
     /// ２日以内にデータがない場合はエラーを返す。
-    pub fn make_expire_control_message(&mut self, now: MicroSec) -> anyhow::Result<Vec<Trade>> {
+    pub fn make_expire_control_message(
+        &mut self,
+        now: MicroSec,
+        force: bool,
+    ) -> anyhow::Result<Vec<Trade>> {
         log::debug!("make_expire_control_message from {}", time_string(now));
 
         let start_time = now - DAYS(2);
 
-        let fix_time = self.latest_fix_time(start_time)?;
+        let fix_time = self.latest_fix_time(start_time, force)?;
         ensure!(fix_time != 0, "no fix data in 2 days");
 
         log::debug!(
@@ -160,17 +219,27 @@ impl TradeTableDb {
             time_string(now)
         );
 
-        Ok(Self::expire_control_message(fix_time, now))
+        Ok(Self::expire_control_message(fix_time, now, force))
     }
 
-    pub fn expire_control_message(start_time: MicroSec, end_time: MicroSec) -> Vec<Trade> {
+    pub fn expire_control_message(
+        start_time: MicroSec,
+        end_time: MicroSec,
+        force: bool,
+    ) -> Vec<Trade> {
+        let status = if force {
+            LogStatus::ExpireControlForce
+        } else {
+            LogStatus::ExpireControl
+        };
+
         let mut trades: Vec<Trade> = vec![];
         let t = Trade::new(
             start_time,
             OrderSide::Unknown,
             dec![0.0],
             dec![0.0],
-            LogStatus::ExpireControl,
+            status,
             "",
         );
         trades.push(t);
@@ -180,7 +249,7 @@ impl TradeTableDb {
             OrderSide::Unknown,
             dec![0.0],
             dec![0.0],
-            LogStatus::ExpireControl,
+            status,
             "",
         );
         trades.push(t);
@@ -198,25 +267,31 @@ impl TradeTableDb {
         let end_time = trades[trades_len - 1].time;
         let log_status = trades[0].status;
 
-        // when fix data comes, delete unstable data first
-        if log_status != LogStatus::UnFix {
-            // create transaction with immidate mode
+        log::debug!(
+            "insert_records {} {} {} {}",
+            trades[0].status,
+            time_string(start_time),
+            time_string(end_time),
+            trades_len
+        );
+
+        if log_status == LogStatus::ExpireControl && trades_len == 2 {
+            log::debug!("delete unarchived data");
             let tx = self.begin_transaction()?;
-            Self::delete_unstable_data(&tx, start_time, end_time);
+            let rec = Self::delete_unstable_data(&tx, trades[0].time, trades[1].time)?;
             tx.commit()?;
-
-            if log_status == LogStatus::ExpireControl {
-                return Ok(0);
-            }
-        } else if log_status == LogStatus::ExpireControl {
+            return Ok(rec);
+        } else if log_status == LogStatus::ExpireControlForce && trades_len == 2 {
+            log::debug!("delete unarchived data(force");
             let tx = self.begin_transaction()?;
-            Self::delete_unstable_data(&tx, trades[0].time, trades[1].time);
+            let rec = Self::delete_unarchived_data(&tx, trades[0].time, trades[1].time)?;
             tx.commit()?;
+            return Ok(rec);
         }
 
         // create transaction with immidate mode
         let tx = self.begin_transaction()?;
-
+        let _ = Self::delete_unstable_data(&tx, start_time, end_time);
         // then insert data
         let insert_len = Self::insert_transaction(&tx, trades)?;
         tx.commit()?;
@@ -380,7 +455,6 @@ impl TradeTableDb {
     }
 
     fn create_table_if_not_exists(&mut self) -> anyhow::Result<()> {
-
         self.connection.execute(
             "CREATE TABLE IF NOT EXISTS trades (
             time_stamp    INTEGER,
@@ -407,8 +481,9 @@ impl TradeTableDb {
         println!("DB file = {}", self.file_name);
         flush_log();
 
-        self.connection.execute("drop table trades", ())
-            .with_context(||format!("database drop error"))?;
+        self.connection
+            .execute("drop table trades", ())
+            .with_context(|| format!("database drop error"))?;
 
         self.vacuum()?;
 
@@ -416,7 +491,8 @@ impl TradeTableDb {
     }
 
     fn vacuum(&self) -> anyhow::Result<()> {
-        self.connection.execute("VACUUM", ())
+        self.connection
+            .execute("VACUUM", ())
             .with_context(|| format!("database VACUUM error"))?;
 
         Ok(())
@@ -432,9 +508,14 @@ impl TradeTableDb {
 
     // 時間選択は左側は含み、右側は含まない。
     // 0をいれたときは全件検索
-    pub fn select<F>(&mut self, start_time: MicroSec, end_time: MicroSec, mut f: F) -> anyhow::Result<()>
+    pub fn select<F>(
+        &mut self,
+        start_time: MicroSec,
+        end_time: MicroSec,
+        mut f: F,
+    ) -> anyhow::Result<()>
     where
-        F: FnMut(&Trade),
+        F: FnMut(&Trade) -> anyhow::Result<()>,
     {
         let sql: &str;
         let param: Vec<i64>;
@@ -468,14 +549,17 @@ impl TradeTableDb {
                     id: row.get_unwrap(5),
                 })
             })
-            .with_context(||format!("select trade error"))?;
+            .with_context(|| format!("select trade error"))?;
 
         log::debug!("create iter {} microsec", NOW() - start_time);
 
         for trade in _transaction_iter {
             match trade {
                 Ok(t) => {
-                    f(&t);
+                    let r = f(&t);
+                    if r.is_err() {
+                        return Err(r.err().unwrap());
+                    }
                 }
                 Err(e) => {
                     return Err(anyhow!("select query: SQL error {}", e));
@@ -532,11 +616,9 @@ impl TradeTableDb {
             // let mut channel = sender.lock().unwrap();
             let r = table_db.select(time_from, time_to, |trade| {
                 let message: MarketMessage = trade.into();
-                let r = sender.send(message);
+                let r = sender.send(message)?;
 
-                if r.is_err() {
-                    log::error!("Error in channel.send: {:?}", r);
-                }
+                Ok(())
             });
 
             if r.is_err() {
@@ -562,6 +644,18 @@ pub struct TradeTable {
 impl TradeTable {
     const OHLCV_WINDOW_SEC: i64 = 60; // min
 
+    pub fn make_db_path(
+        exchange_name: &str,
+        trade_category: &str,
+        trade_symbol: &str,
+        test_net: bool,
+    ) -> String {
+
+        let db_path = db_full_path(&exchange_name, &trade_category, trade_symbol, "", test_net);
+
+        return db_path.to_str().unwrap().to_string();
+    }
+
     /// check if db thread is running.
     pub fn is_running(&self) -> bool {
         if self.handle.is_none() {
@@ -577,8 +671,12 @@ impl TradeTable {
 
     /// create new expire control message(from latest fix time to now)
     /// if there is not fix record in 2 days, return error.
-    pub fn make_expire_control_message(&mut self, now: MicroSec) -> anyhow::Result<Vec<Trade>> {
-        self.connection.make_expire_control_message(now)
+    pub fn make_expire_control_message(
+        &mut self,
+        now: MicroSec,
+        force: bool,
+    ) -> anyhow::Result<Vec<Trade>> {
+        self.connection.make_expire_control_message(now, force)
     }
 
     pub fn set_cache_ohlcvv(&mut self, df: DataFrame) {
@@ -605,7 +703,7 @@ impl TradeTable {
         self.cache_ohlcvv = df;
     }
 
-    pub fn start_thread(&mut self) -> Sender<Vec<Trade>> {
+    pub async fn start_thread(&mut self) -> Sender<Vec<Trade>> {
         // check if the thread is already started
         // check self.tx is valid and return clone of self.tx
         log::debug!("start_thread");
@@ -622,6 +720,50 @@ impl TradeTable {
 
         let handle = spawn(async move {
             let mut db = TradeTableDb::open(file_name.as_str()).unwrap();
+            let rx = rx; // Move rx into the closure's environment
+            loop {
+                match rx.recv() {
+                    Ok(trades) => {
+                        let result = db.insert_records(&trades);
+
+                        if result.is_err() {
+                            log::error!("insert error {:?}", result);
+                            continue;
+                        }
+                        log::debug!("recv trades: {}", trades.len());
+                    }
+                    Err(e) => {
+                        log::error!("recv error(sender program died?) {:?}", e);
+                        break;
+                    }
+                }
+            }
+        });
+
+        self.handle = Some(handle);
+
+        return self.tx.clone().unwrap();
+    }
+
+    /*
+    pub fn start_thread(&mut self) -> Sender<Vec<Trade>> {
+        // check if the thread is already started
+        // check self.tx is valid and return clone of self.tx
+        log::debug!("start_thread");
+        if self.is_running() {
+            log::info!("DB Thread is already started, reuse tx");
+            return self.tx.clone().unwrap();
+        }
+
+        let (tx, rx) = unbounded();
+
+        let file_name = self.file_name.clone();
+
+        self.tx = Some(tx);
+
+        let handle = spawn(move || {
+            let mut db = TradeTableDb::open(file_name.as_str()).unwrap();
+            let rx = rx; // Move rx into the closure's environment
             loop {
                 match rx.recv() {
                     Ok(trades) => {
@@ -645,7 +787,9 @@ impl TradeTable {
 
         return self.tx.clone().unwrap();
     }
+    */
 
+    /*
     pub fn stop_thread(&mut self) {
         if self.handle.is_some() {
             let handle = self.handle.take().unwrap();
@@ -657,6 +801,7 @@ impl TradeTable {
             self.tx = None;
         }
     }
+    */
 
     pub fn validate_by_date(&mut self, date: MicroSec) -> anyhow::Result<bool> {
         self.connection.validate_by_date(date)
@@ -666,6 +811,10 @@ impl TradeTable {
         return self.cache_duration;
     }
 
+    pub fn ohlcv_floor_fix_time(t: MicroSec, unit_sec: i64) -> MicroSec {
+        return FLOOR_SEC(t, unit_sec)
+    }
+
     pub fn ohlcv_start(t: MicroSec) -> MicroSec {
         return FLOOR_SEC(t, TradeTable::OHLCV_WINDOW_SEC);
     }
@@ -678,11 +827,17 @@ impl TradeTable {
         self.cache_duration = 0;
     }
 
-    pub fn select_df_from_db(&mut self, start_time: MicroSec, end_time: MicroSec) -> anyhow::Result<DataFrame> {
+    pub fn select_df_from_db(
+        &mut self,
+        start_time: MicroSec,
+        end_time: MicroSec,
+    ) -> anyhow::Result<DataFrame> {
         let mut buffer = TradeBuffer::new();
 
         self.select(start_time, end_time, |trade| {
             buffer.push_trade(trade);
+
+            Ok(())
         })?;
 
         return Ok(buffer.to_dataframe());
@@ -694,7 +849,7 @@ impl TradeTable {
         Ok(())
     }
 
-    pub fn update_cache_all(&mut self) -> anyhow::Result<()>{
+    pub fn update_cache_all(&mut self) -> anyhow::Result<()> {
         self.update_cache_df(0, 0)
     }
 
@@ -705,7 +860,11 @@ impl TradeTable {
         self.cache_ohlcvv = select_df(&self.cache_ohlcvv, cache_timing, 0);
     }
 
-    pub fn update_cache_df(&mut self, start_time: MicroSec, mut end_time: MicroSec) -> anyhow::Result<()> {
+    pub fn update_cache_df(
+        &mut self,
+        start_time: MicroSec,
+        mut end_time: MicroSec,
+    ) -> anyhow::Result<()> {
         log::debug!("update_cache_df {} -> {}", start_time, end_time);
 
         let df_start_time: i64;
@@ -733,8 +892,17 @@ impl TradeTable {
                 // no cache / update all
                 self.load_df(start_time, end_time)?;
 
-                let df_start_time = start_time_df(&self.cache_df).unwrap();
-                let df_end_time = end_time_df(&self.cache_df).unwrap();
+                let df_start_time = if let Some(t) = start_time_df(&self.cache_df) {
+                    t
+                } else {
+                    return Ok(());
+                };
+
+                let df_end_time = if let Some(t) = end_time_df(&self.cache_df) {
+                    t
+                } else {
+                    return Ok(());
+                };
 
                 // update ohlcv
                 self.cache_ohlcvv = ohlcvv_df(
@@ -743,7 +911,7 @@ impl TradeTable {
                     df_end_time,
                     TradeTable::OHLCV_WINDOW_SEC,
                 )?;
-                return Ok(());  // update cache all.
+                return Ok(()); // update cache all.
             }
         }
 
@@ -835,11 +1003,12 @@ impl TradeTable {
 
     pub fn ohlcvv_df(
         &mut self,
-        start_time: MicroSec,
+        mut start_time: MicroSec,
         end_time: MicroSec,
         time_window_sec: i64,
-    ) -> anyhow::Result<DataFrame> 
-    {
+    ) -> anyhow::Result<DataFrame> {
+        start_time = TradeTable::ohlcv_floor_fix_time(start_time, time_window_sec); // 開始tickは確定足、終了は未確定足もOK.
+
         self.update_cache_df(start_time, end_time)?;
 
         if time_window_sec % TradeTable::OHLCV_WINDOW_SEC == 0 {
@@ -902,7 +1071,7 @@ impl TradeTable {
         end_time: MicroSec,
         window_sec: i64,
     ) -> anyhow::Result<PyDataFrame> {
-        start_time = TradeTable::ohlcv_start(start_time); // 開始tickは確定足、終了は未確定足もOK.
+        start_time = TradeTable::ohlcv_floor_fix_time(start_time, window_sec); // 開始tickは確定足、終了は未確定足もOK.
 
         let mut df = self.ohlcvv_df(start_time, end_time, window_sec)?;
 
@@ -913,10 +1082,12 @@ impl TradeTable {
 
     pub fn ohlcv_df(
         &mut self,
-        start_time: MicroSec,
+        mut start_time: MicroSec,
         end_time: MicroSec,
         time_window_sec: i64,
     ) -> anyhow::Result<DataFrame> {
+        start_time = TradeTable::ohlcv_start(start_time); // 開始tickは確定足、終了は未確定足もOK.
+
         self.update_cache_df(start_time, end_time)?;
 
         if time_window_sec % TradeTable::OHLCV_WINDOW_SEC == 0 {
@@ -998,7 +1169,12 @@ impl TradeTable {
         Ok(py_df)
     }
 
-    pub fn vap(&mut self, start_time: MicroSec, end_time: MicroSec, price_unit: i64) -> anyhow::Result<DataFrame> {
+    pub fn vap(
+        &mut self,
+        start_time: MicroSec,
+        end_time: MicroSec,
+        price_unit: i64,
+    ) -> anyhow::Result<DataFrame> {
         self.update_cache_df(start_time, end_time)?;
         let df = vap_df(&self.cache_df, start_time, end_time, price_unit);
 
@@ -1045,8 +1221,7 @@ impl TradeTable {
         let array: ndarray::Array2<f64> = trades
             .select(&[KEY::time_stamp, KEY::price, KEY::size, KEY::order_side])
             .unwrap()
-            .to_ndarray::<Float64Type>(IndexOrder::C)
-            ?;
+            .to_ndarray::<Float64Type>(IndexOrder::C)?;
 
         Ok(array)
     }
@@ -1088,16 +1263,19 @@ impl TradeTable {
         );
     }
 
-
     /// Retrieves the earliest time stamp from the trades table in the SQLite database.
     /// Returns a Result containing the earliest time stamp as a MicroSec value, or an Error if the query fails.
     pub fn start_time(&self) -> anyhow::Result<MicroSec> {
         let sql = "select time_stamp from trades order by time_stamp asc limit 1";
 
-        let r = self.connection.connection.query_row(sql, [], |row| {
-            let min: i64 = row.get(0)?;
-            Ok(min)
-        }).with_context(||format!("start_time select error"))?;
+        let r = self
+            .connection
+            .connection
+            .query_row(sql, [], |row| {
+                let min: i64 = row.get(0)?;
+                Ok(min)
+            })
+            .with_context(|| format!("start_time select error"))?;
 
         return Ok(r);
     }
@@ -1115,7 +1293,7 @@ impl TradeTable {
                 let max: i64 = row.get(0)?;
                 Ok(max)
             })
-            .with_context(||format!("end_time select error"))?;
+            .with_context(|| format!("end_time select error"))?;
 
         return Ok(r);
     }
@@ -1252,8 +1430,7 @@ impl TradeTable {
                     chunks.push(c);
                 }
                 index += 1;
-            }
-            else {
+            } else {
                 log::error!("select_time_chunks_in_db error {}", chunk.err().unwrap());
             }
         }
@@ -1341,17 +1518,33 @@ impl TradeTable {
         return self.connection.insert_records(trades);
     }
 
-    pub fn find_latest_gap(&mut self) -> anyhow::Result<(MicroSec, MicroSec)> {
+    pub fn find_latest_gap(&mut self, force: bool) -> anyhow::Result<(MicroSec, MicroSec)> {
         let start_time = NOW() - DAYS(2);
 
-        let fix_time = self.latest_fix_time(start_time)?;
+        let fix_time = self.latest_fix_time(start_time, force)?;
         let unfix_time = self.first_unfix_time(fix_time)?;
 
         Ok((fix_time, unfix_time))
     }
 
-    pub fn latest_fix_time(&mut self, start_time: MicroSec) -> anyhow::Result<MicroSec> {
-        self.connection.latest_fix_time(start_time)
+    pub fn latest_fix_trade(
+        &mut self,
+        start_time: MicroSec,
+        force: bool,
+    ) -> anyhow::Result<Option<Trade>> {
+        self.connection.latest_fix_trade(start_time, force)
+    }
+
+    pub fn latest_fix_time(
+        &mut self,
+        start_time: MicroSec,
+        force: bool,
+    ) -> anyhow::Result<MicroSec> {
+        self.connection.latest_fix_time(start_time, force)
+    }
+
+    pub fn first_unfix_trade(&mut self, start_time: MicroSec) -> anyhow::Result<Option<Trade>> {
+        self.connection.first_unfix_trade(start_time)
     }
 
     pub fn first_unfix_time(&mut self, start_time: MicroSec) -> anyhow::Result<MicroSec> {
@@ -1424,9 +1617,14 @@ impl TradeTable {
     /// assert_eq!(trades.len(), 1);
     /// assert_eq!(trades[0].timestamp, from_time);
     /// ```
-    fn select<F>(&mut self, start_time: MicroSec, end_time: MicroSec, f: F) -> anyhow::Result<()>
+    pub fn select<F>(
+        &mut self,
+        start_time: MicroSec,
+        end_time: MicroSec,
+        f: F,
+    ) -> anyhow::Result<()>
     where
-        F: FnMut(&Trade),
+        F: FnMut(&Trade) -> anyhow::Result<()>,
     {
         self.connection.select(start_time, end_time, f)
     }
@@ -1504,16 +1702,6 @@ mod test_transaction_table {
         assert!(r.is_ok());
     }
 
-    #[test]
-    fn test_select_fn() -> anyhow::Result<()> {
-        test_insert_table();
-
-        let mut table = TradeTable::open("test.db").unwrap();
-
-        table.select(0, 0, |row| println!("{:?}", row))?;
-        Ok(())
-    }
-
     #[test]
     fn test_select_array() {
         init_debug_log();
@@ -1560,7 +1748,7 @@ mod test_transaction_table {
 
     #[test]
     fn test_select_gap_chunks() -> anyhow::Result<()> {
-        let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", "");
+        let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", "", false);
         let db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
 
         let chunks = db.select_gap_chunks(NOW() - DAYS(1), NOW(), 1_000_000 * 13)?;
@@ -1581,7 +1769,7 @@ mod test_transaction_table {
 
     #[test]
     fn test_select_time_chunk_from() {
-        let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", "");
+        let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", "", false);
         let db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
 
         let chunks = db.find_time_chunk_from(NOW() - DAYS(1), NOW(), 1_000_000 * 10);
@@ -1595,7 +1783,7 @@ mod test_transaction_table {
 
     #[test]
     fn test_select_time_chunk_to() {
-        let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", "/tmp");
+        let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", "/tmp", false);
         let db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
 
         let chunks = db.find_time_chunk_to(NOW() - DAYS(1), NOW(), 1_000_000 * 120);
@@ -1608,8 +1796,8 @@ mod test_transaction_table {
     }
 
     #[test]
-    fn test_select_time_chunks() -> anyhow::Result<()>{
-        let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", "/tmp");
+    fn test_select_time_chunks() -> anyhow::Result<()> {
+        let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", "/tmp", false);
         let db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
 
         let chunks = db.select_time_chunks_in_db(NOW() - DAYS(1), NOW(), 1_000_000 * 10)?;
@@ -1645,7 +1833,7 @@ mod test_transaction_table {
     #[test]
     fn test_select_ohlcv_df() -> anyhow::Result<()> {
         init_log();
-        let db_name = db_full_path("BN", "SPOT", "BTCBUSD", "/tmp");
+        let db_name = db_full_path("BN", "SPOT", "BTCBUSD", "/tmp", false);
 
         let mut db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
 
@@ -1687,18 +1875,18 @@ mod test_transaction_table {
     fn test_select_print() {
         init_log();
 
-        let db_name = db_full_path("BN", "SPOT", "BTCBUSD", "/tmp");
+        let db_name = db_full_path("BN", "SPOT", "BTCBUSD", "/tmp", false);
         let mut db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
 
         let start = NOW();
-        let ohlcv = db.select(NOW() - DAYS(2), NOW(), |_trade| {});
+        let ohlcv = db.select(NOW() - DAYS(2), NOW(), |_trade| {Ok(())});
 
         println!("{:?} / {} microsec", ohlcv, NOW() - start);
     }
 
     #[test]
     fn test_select_df() {
-        let db_name = db_full_path("BN", "SPOT", "BTCBUSD", "/tmp");
+        let db_name = db_full_path("BN", "SPOT", "BTCBUSD", "/tmp", false);
         let mut db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
 
         let df = db.select_df_from_db(NOW() - DAYS(2), NOW());
@@ -1707,9 +1895,9 @@ mod test_transaction_table {
     }
 
     #[test]
-    fn test_update_cache() -> anyhow::Result<()>{
+    fn test_update_cache() -> anyhow::Result<()> {
         init_log();
-        let db_name = db_full_path("BN", "SPOT", "BTCBUSD", "/tmp");
+        let db_name = db_full_path("BN", "SPOT", "BTCBUSD", "/tmp", false);
         let mut db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
 
         db.update_cache_df(NOW() - DAYS(2), NOW())?;
@@ -1717,15 +1905,15 @@ mod test_transaction_table {
         Ok(())
     }
 
-    #[test]
-    fn test_start_thread() {
+    #[tokio::test]
+    async fn test_start_thread() {
         let mut table = TradeTable::open(
-            db_full_path("BN", "SPOT", "BTCBUSD", "/tmp")
+            db_full_path("BN", "SPOT", "BTCBUSD", "/tmp", false)
                 .to_str()
                 .unwrap(),
         )
         .unwrap();
-        let tx = table.start_thread();
+        let tx = table.start_thread().await;
 
         let v = vec![Trade {
             time: 1,
@@ -1747,7 +1935,7 @@ mod test_transaction_table {
         }];
         tx.send(v).unwrap();
 
-        let tx = table.start_thread();
+        let tx = table.start_thread().await;
 
         let v = vec![Trade {
             time: 1,
@@ -1767,7 +1955,7 @@ mod test_transaction_table {
         //let table = TradeTable::open(db_full_path("BN", "SPOT", "BTCBUSD").to_str().unwrap()).unwrap();
 
         TradeTableDb::set_wal_mode(
-            db_full_path("BN", "SPOT", "BTCBUSD", "/tmp")
+            db_full_path("BN", "SPOT", "BTCBUSD", "/tmp", false)
                 .to_str()
                 .unwrap(),
         )?;
diff --git a/rbot_lib/src/net/rest.rs b/rbot_lib/src/net/rest.rs
index a842ffe7..e6d7bc2d 100644
--- a/rbot_lib/src/net/rest.rs
+++ b/rbot_lib/src/net/rest.rs
@@ -1,11 +1,10 @@
 // Copyright(c) 2023-4. yasstake. All rights reserved.
 // Abloultely no warranty.
 
-use anyhow::Context;
-use chrono::Datelike;
 use anyhow::anyhow;
 use anyhow::ensure;
-
+use anyhow::Context;
+use chrono::Datelike;
 
 // use crossbeam_channel::Receiver;
 use crossbeam_channel::Sender;
@@ -24,10 +23,11 @@ use tempfile::tempdir;
 //use tokio::spawn;
 use zip::ZipArchive;
 
+use crate::common::time_string;
 use crate::common::AccountCoins;
 use crate::common::{
-    flush_log, to_naive_datetime, BoardTransfer, Kline, LogStatus, MarketConfig,
-    MicroSec, Order, OrderSide, OrderType, ServerConfig, Trade, DAYS, FLOOR_DAY, TODAY,
+    flush_log, to_naive_datetime, BoardTransfer, Kline, LogStatus, MarketConfig, MicroSec, Order,
+    OrderSide, OrderType, ServerConfig, Trade, DAYS, FLOOR_DAY, TODAY,
 };
 //use crate::db::KEY::low;
 
@@ -35,15 +35,14 @@ pub trait RestApi<T>
 where
     T: ServerConfig,
 {
-    fn get_board_snapshot(
-        server: &T,
-        config: &MarketConfig,
-    ) -> impl std::future::Future<Output = anyhow::Result<BoardTransfer>> + Send;
+    async fn get_board_snapshot(server: &T, config: &MarketConfig)
+        -> anyhow::Result<BoardTransfer>;
 
-    fn get_recent_trades(
+    async fn get_recent_trades(
         server: &T,
         config: &MarketConfig,
-    ) -> impl std::future::Future<Output = anyhow::Result<Vec<Trade>>> + Send;
+    ) -> anyhow::Result<Vec<Trade>>;
+    
     fn get_trade_klines(
         server: &T,
         config: &MarketConfig,
@@ -66,7 +65,7 @@ where
     ) -> impl std::future::Future<Output = anyhow::Result<Order>> + Send;
     fn open_orders(
         server: &T,
-config: &MarketConfig,
+        config: &MarketConfig,
     ) -> impl std::future::Future<Output = anyhow::Result<Vec<Order>>> + Send;
 
     fn get_account(
@@ -102,25 +101,28 @@ config: &MarketConfig,
     fn rec_to_trade(rec: &StringRecord) -> Trade;
     fn archive_has_header() -> bool;
 
-    fn latest_archive_date(server: &T, config: &MarketConfig) -> impl std::future::Future<Output = anyhow::Result<MicroSec>> + Send {
-        async move {
-            let f = |date: MicroSec| -> String { Self::history_web_url(server, config, date) };
-            let mut latest = TODAY();
-            let mut i = 0;
+    async fn latest_archive_date(server: &T, config: &MarketConfig) -> anyhow::Result<MicroSec> {
+        let f = |date: MicroSec| -> String { Self::history_web_url(server, config, date) };
+        let mut latest = TODAY();
+        let mut i = 0;
 
-            loop {
-                let has_archive = has_archive(latest, &f).await?;
+        loop {
+            log::debug!("check log exist = {}({})", time_string(latest), latest);
 
-                if has_archive {
-                    return Ok(latest);
-                }
+            if has_archive(latest, &f).await {
+                return Ok(latest);
+            }
 
-                latest -= DAYS(1);
-                i += 1;
+            latest -= DAYS(1);
+            i += 1;
 
-                if 5 < i {
-                    return Err(anyhow!("Find archive retry over {}/{}/{}", i, latest, f(latest)));
-                }
+            if 5 < i {
+                return Err(anyhow!(
+                    "Find archive retry over {}/{}/{}",
+                    i,
+                    latest,
+                    f(latest)
+                ));
             }
         }
     }
@@ -159,7 +161,8 @@ pub async fn log_download_tmp(url: &str, tmp_dir: &Path) -> anyhow::Result<Strin
         .header("User-Agent", "Mozilla/5.0")
         .header("Accept", "text/html")
         .send()
-        .await.with_context(|| format!("URL get error {}", url))?;
+        .await
+        .with_context(|| format!("URL get error {}", url))?;
 
     log::debug!(
         "Response code = {} / download size {}",
@@ -179,14 +182,19 @@ pub async fn log_download_tmp(url: &str, tmp_dir: &Path) -> anyhow::Result<Strin
         .unwrap_or("tmp.bin");
 
     let fname = tmp_dir.join(fname);
-    let file_name = fname.to_str().unwrap();    
+    let file_name = fname.to_str().unwrap();
 
-    let mut target = File::create(&fname).with_context(|| format!("file create error {}", fname.to_str().unwrap()))?;
+    let mut target = File::create(&fname)
+        .with_context(|| format!("file create error {}", fname.to_str().unwrap()))?;
 
-    let content = response.bytes().await.with_context(|| format!("response bytes error {}", url))?;
+    let content = response
+        .bytes()
+        .await
+        .with_context(|| format!("response bytes error {}", url))?;
     let mut cursor = Cursor::new(content);
 
-    let size = copy(&mut cursor, &mut target).with_context(||format!("write error {}", file_name))?;
+    let size =
+        copy(&mut cursor, &mut target).with_context(|| format!("write error {}", file_name))?;
     ensure!(size > 0, "file is empty {}", file_name);
 
     let _r = target.flush();
@@ -254,7 +262,8 @@ where
     let tmp_dir = tempdir().with_context(|| "create tmp dir error")?;
 
     let file_path = log_download_tmp(url, tmp_dir.path())
-            .await.with_context(|| format!("log_download_tmp error {}->{:?}", url, tmp_dir))?;
+        .await
+        .with_context(|| format!("log_download_tmp error {}->{:?}", url, tmp_dir))?;
 
     let mut buffer: Vec<Trade> = vec![];
     let mut is_first_record = true;
@@ -293,7 +302,8 @@ where
 
     buffer[buffer_len - 1].status = LogStatus::FixBlockEnd;
 
-    tx.send(buffer.to_vec()).with_context(|| format!("channel send error"))?;
+    tx.send(buffer.to_vec())
+        .with_context(|| format!("channel send error"))?;
 
     buffer.clear();
 
@@ -306,8 +316,6 @@ where
     Ok(download_rec)
 }
 
-
-
 fn read_csv_archive<F>(file_path: &str, has_header: bool, mut f: F)
 where
     F: FnMut(&StringRecord),
@@ -591,7 +599,10 @@ pub async fn do_rest_request(
         .header("User-Agent", "Mozilla/5.0")
         .header("Accept", "text/html");
 
-    let response = request_builder.send().await.with_context(|| format!("URL get error {url:}"))?;
+    let response = request_builder
+        .send()
+        .await
+        .with_context(|| format!("URL get error {url:}"))?;
 
     log::debug!(
         "Response code = {} / content size {:?} / method({:?}) / URL = {} ",
@@ -603,7 +614,7 @@ pub async fn do_rest_request(
 
     if response.status().as_str() != "200" {
         return Err(anyhow!(
-            "Response code = {} / download size {:?} / method({:?}) / URL = {} / path{}",
+            "Response code = {} / download size {:?} / method({:?}) / URL = {} / path = {}",
             response.status().as_str(),
             response.content_length(),
             method,
@@ -612,7 +623,10 @@ pub async fn do_rest_request(
         ));
     }
 
-    let body = response.text().await.with_context(|| format!("response text error"))?;
+    let body = response
+        .text()
+        .await
+        .with_context(|| format!("response text error"))?;
 
     log::debug!("body{}", body);
 
@@ -672,7 +686,6 @@ pub async fn rest_put(
     do_rest_request(Method::PUT, &url, headers, body).await
 }
 
-
 pub async fn check_exist(url: &str) -> anyhow::Result<bool> {
     let client = reqwest::Client::new();
 
@@ -681,7 +694,8 @@ pub async fn check_exist(url: &str) -> anyhow::Result<bool> {
         .header("User-Agent", "Mozilla/5.0")
         .header("Accept", "text/html")
         .send()
-        .await.with_context(|| format!("URL get error {}", url))?;
+        .await
+        .with_context(|| format!("URL get error {}", url))?;
 
     log::debug!(
         "Response code = {} / download size {}",
@@ -689,18 +703,29 @@ pub async fn check_exist(url: &str) -> anyhow::Result<bool> {
         response.content_length().unwrap()
     );
 
-
-    anyhow::ensure!(response.status().as_str() == "200", "URL get response error {}/code={}", url, response.status());
+    anyhow::ensure!(
+        response.status().as_str() == "200",
+        "URL get response error {}/code={}",
+        url,
+        response.status()
+    );
 
     Ok(true)
 }
 
-async fn has_archive<F>(date: MicroSec, f: &F) -> anyhow::Result<bool>
+async fn has_archive<F>(date: MicroSec, f: &F) -> bool
 where
     F: Fn(MicroSec) -> String,
 {
     let url = f(date);
-    Ok(check_exist(url.as_str()).await?)
+
+    let result = check_exist(url.as_str()).await;
+
+    if result.is_err() {
+        return false;
+    }
+
+    result.unwrap()
 }
 
 pub async fn latest_archive_date<F>(f: &F) -> Result<MicroSec, String>
@@ -713,14 +738,8 @@ where
     loop {
         let has_archive = has_archive(latest, f).await;
 
-        if has_archive.is_err() {
-            log::error!("Error in has_archive: {:?}", has_archive);
-            return Err(format!("Error in has_archive: {:?}", has_archive));
-        }
-
-        let has_archive = has_archive.unwrap();
-
         if has_archive {
+            log::debug!("latest archive date = {}({})", time_string(latest), latest);
             return Ok(latest);
         }
 
@@ -739,10 +758,10 @@ mod test_exchange {
 
     use super::*;
     use crate::common::init_debug_log;
-    // use crossbeam_channel::bounded;    
+    // use crossbeam_channel::bounded;
 
     #[tokio::test]
-    async fn log_download_temp_test() -> anyhow::Result<()>{
+    async fn log_download_temp_test() -> anyhow::Result<()> {
         init_debug_log();
         let url = "https://data.binance.vision/data/spot/daily/trades/BTCBUSD/BTCBUSD-trades-2022-11-19.zip";
         let tmp_dir = tempdir().unwrap();
@@ -805,8 +824,6 @@ mod test_exchange {
         Ok(())
     }
 
-
-
     /*
     use crate::exchange::binance::BinanceMarket;
 
diff --git a/rbot_market/src/market.rs b/rbot_market/src/market.rs
index ab3e5249..271ea2e8 100644
--- a/rbot_market/src/market.rs
+++ b/rbot_market/src/market.rs
@@ -1,7 +1,13 @@
+// Copyright(c) 2024. yasstake. All rights reserved.
+
 use crossbeam_channel::Sender;
 use rbot_lib::common::AccountCoins;
+use rbot_lib::common::MarketMessage;
 use rbot_lib::db::db_full_path;
 use rust_decimal_macros::dec;
+use tokio_stream::Stream;
+use tokio_stream::StreamExt;
+use std::iter::Filter;
 use std::sync::{Arc, Mutex, RwLock};
 
 use pyo3_polars::PyDataFrame;
@@ -10,6 +16,8 @@ use rbot_lib::common::OrderBook;
 use rbot_lib::net::RestApi;
 use rust_decimal::Decimal;
 
+use tokio::task::spawn;
+
 use anyhow::anyhow;
 #[allow(unused_imports)]
 use anyhow::Context;
@@ -199,6 +207,8 @@ where
 
 pub trait MarketInterface {
     // --- GET CONFIG INFO ----
+    fn get_production(&self) -> bool;
+
     fn get_config(&self) -> MarketConfig;
     fn get_exchange_name(&self) -> String;
     fn get_trade_category(&self) -> String;
@@ -287,27 +297,22 @@ where
     /// Download historical log from REST API
     fn download_latest(&mut self, verbose: bool) -> anyhow::Result<i64>;
 
+    fn download_archives(
+        &mut self,
+        ndays: i64,
+        force: bool,
+        verbose: bool,
+        low_priority: bool,
+    ) -> anyhow::Result<i64>;
+
     /// Download historical log from REST API, between the latest data and the latest FIX data.
     /// If the latest FIX data is not found, generate psudo data from klines.
-    fn download_gap(&mut self, verbose: bool) -> anyhow::Result<i64>;
+    fn download_gap(&mut self, force: bool, verbose: bool) -> anyhow::Result<i64>;
 
     fn get_db(&self) -> Arc<Mutex<TradeTable>>;
 
     fn get_history_web_base_url(&self) -> String;
 
-    fn make_db_path(
-        exchange_name: &str,
-        trade_category: &str,
-        trade_symbol: &str,
-        db_base_dir: &str,
-    ) -> String {
-        let db_path = db_full_path(&exchange_name, trade_category, trade_symbol, db_base_dir);
-
-        return db_path.to_str().unwrap().to_string();
-    }
-    
-    fn get_latest_archive_date(&self) -> anyhow::Result<MicroSec>;
-
 
     /// Check if database is valid at the date
     fn validate_db_by_date(&mut self, date: MicroSec) -> anyhow::Result<bool> {
@@ -317,15 +322,15 @@ where
         lock.validate_by_date(date)
     }
 
-    fn find_latest_gap(&self) -> anyhow::Result<(MicroSec, MicroSec)> {
+    fn find_latest_gap(&self, force: bool) -> anyhow::Result<(MicroSec, MicroSec)> {
         log::debug!("[start] find_latest_gap");
         let start_time = NOW() - DAYS(2);
 
         let db = self.get_db();
-
+        
         let (fix_time, unfix_time) = {
             let mut lock = db.lock().unwrap();
-            let fix_time = lock.latest_fix_time(start_time)?;
+            let fix_time = lock.latest_fix_time(start_time, force)?;
             if fix_time == 0 {
                 return Err(anyhow!("No data found"));
             }
@@ -338,129 +343,37 @@ where
         Ok((fix_time, unfix_time))
     }
 
-    /*
-        async fn _start_market_stream(&mut self) {
-            // if thread is working, do nothing.
-            if self.public_handler.is_some() {
-                println!("market stream is already running.");
-                return;
-            }
-
-            // delete unstable data
-            let db_channel = self.db.start_thread();
-
-            // if the latest data is overrap with REST minute, do not delete data.
-            let trades = self.get_recent_trades();
-            let l = trades.len();
-            if l != 0 {
-                let rest_start_time = trades[l - 1].time;
-                let last_time = self.db.end_time(rest_start_time);
-                if last_time.is_err() {
-                    let now = NOW();
-
-                    log::debug!(
-                        "db has gap, so delete after FIX data now={:?} / db_end={:?}",
-                        time_string(now),
-                        time_string(rest_start_time)
-                    );
-
-                    let delete_message = self.db.connection.make_expire_control_message(now);
-
-                    log::debug!("delete_message: {:?}", delete_message);
-
-                    db_channel.send(delete_message).unwrap();
-                } else {
-                    log::debug!(
-                        "db has no gap, so continue to receive data {}",
-                        time_string(rest_start_time)
-                    );
-                }
-
-                db_channel.send(trades).unwrap();
-            }
-
-            let agent_channel = self.agent_channel.clone();
-            let ws_channel = self.public_ws._open_channel().await;
-
-            self.public_ws
-                ._connect(|message| {
-                    let m = serde_json::from_str::<BybitWsMessage>(&message);
-
-                    if m.is_err() {
-                        log::warn!("Error in serde_json::from_str: {:?}", message);
-                        println!("ERR: {:?}", message);
-                    }
-
-                    let m = m.unwrap();
-
-                    return m.into();
-                })
-                .await;
-
-            let board = self.board.clone();
-            let db_channel_for_after = db_channel.clone();
-
-            let udp_sender = self.udp_sender.clone();
-
-            let handler = tokio::task::spawn(async move {
-                loop {
-                    let message = ws_channel.recv();
-                    let message = message.unwrap();
-
-                    if message.trade.len() != 0 {
-                        log::debug!("Trade: {:?}", message.trade);
-                        let r = db_channel.send(message.trade.clone());
-
-                        if r.is_err() {
-                            log::error!("Error in db_channel.send: {:?}", r);
-                        }
-                    }
+    fn find_latest_gap_trade(&self, force: bool) -> anyhow::Result<(Option<Trade>, Option<Trade>)> {
+        log::debug!("[start] find_latest_gap_trade force={}", force);
+        let start_time = NOW() - DAYS(2);
 
-                    if message.orderbook.is_some() {
-                        let orderbook = message.orderbook.clone().unwrap();
-                        let mut b = board.write().unwrap();
-                        b.update(&orderbook);
-                        drop(b);
-                    }
+        let db = self.get_db();
 
-                    let messages = message.extract();
+        let mut lock = db.lock().unwrap();
+        let latest_trade = lock.latest_fix_trade(start_time, force)?;
 
-                    // update board
+        if latest_trade.is_none() {
+            log::debug!("No data found");
+            return Ok((None, None));
+        }
 
-                    // send message to agent
-                    for m in messages {
-                        if m.trade.is_some() {
-                            // broadcast only trade message
-                            if udp_sender.is_some() {
-                                let sender = udp_sender.as_ref().unwrap().as_ref();
-                                sender.lock().unwrap().send_market_message(&m);
-                            }
+        let latest_trade_id = latest_trade.as_ref().unwrap().id.clone();
+        let fix_time = latest_trade.as_ref().unwrap().time;
 
-                            let mut ch = agent_channel.write().unwrap();
-                            let r = ch.send(m.clone());
-                            drop(ch);
+        let first_unfix_trade = lock.first_unfix_trade(fix_time)?;
 
-                            if r.is_err() {
-                                log::error!("Error in db_channel.send: {:?}", r);
-                            }
-                        }
-                    }
-                }
-            });
+        if first_unfix_trade.is_some() {
+            let first_unfix_trade_id = first_unfix_trade.as_ref().unwrap().id.clone();
 
-            self.public_handler = Some(handler);
-
-            // update recent trade
-            // wait for channel open
-            sleep(Duration::from_millis(100)); // TODO: fix to wait for channel open
-            let trade = self.get_recent_trades();
-            db_channel_for_after.send(trade).unwrap();
+            if first_unfix_trade_id == latest_trade_id {
+                log::debug!("first_unfix_trade_id <= latest_trade_id");
+                return Ok((None, None));
+            }
+        }
 
-            // TODO: store recent trade timestamp.
+        Ok((latest_trade, first_unfix_trade))
+    }
 
-            log::info!("start_market_stream");
-        }
-    */
     // --- DB ----
     fn drop_table(&mut self) -> anyhow::Result<()> {
         let db = self.get_db();
@@ -483,11 +396,13 @@ where
         lock.reset_cache_duration();
     }
 
+    /*
     fn stop_db_thread(&mut self) {
         let db = self.get_db();
         let mut lock = db.lock().unwrap();
         lock.stop_thread()
     }
+    */
 
     fn cache_all_data(&mut self) -> anyhow::Result<()> {
         let db = self.get_db();
@@ -575,7 +490,7 @@ where
     ///
     fn get_order_book(&self) -> Arc<RwLock<OrderBook>>;
 
-    fn reflesh_order_book(&mut self);
+    fn reflesh_order_book(&mut self)-> anyhow::Result<()>;
 
     fn get_board(&mut self) -> anyhow::Result<(PyDataFrame, PyDataFrame)> {
         let orderbook = self.get_order_book();
@@ -615,7 +530,7 @@ where
 
         let json = {
             let lock = orderbook.read().unwrap();
-            lock.get_json(size)?            
+            lock.get_json(size)?
         };
 
         Ok(json)
@@ -623,9 +538,9 @@ where
 
     fn get_board_vec(&self) -> anyhow::Result<(Vec<BoardItem>, Vec<BoardItem>)> {
         let orderbook = self.get_order_book();
-        
+
         let (bids, asks) = {
-            let lock = orderbook.read().unwrap();            
+            let lock = orderbook.read().unwrap();
             lock.get_board_vec()?
         };
 
@@ -643,30 +558,37 @@ where
         Ok(edge_price)
     }
 
-    fn start_db_thread(&mut self) -> Sender<Vec<Trade>> {
+    async fn start_db_thread(&mut self) -> Sender<Vec<Trade>> {
         let db = self.get_db();
         let mut lock = db.lock().unwrap();
 
-        lock.start_thread()
+        lock.start_thread().await
     }
 
-    fn download_recent_trades(
+    async fn async_download_recent_trades(
         &self,
         market_config: &MarketConfig,
-    ) -> impl std::future::Future<Output = anyhow::Result<Vec<Trade>>> + Send where Self: Sync {async {
+    ) -> anyhow::Result<Vec<Trade>>
+    {
         T::get_recent_trades(&self.get_server_config(), market_config).await
-    } }
+    }
 
-    fn expire_unfix_data(&mut self) -> anyhow::Result<()> {
+    async fn async_expire_unfix_data(&mut self, force: bool) -> anyhow::Result<()> {
         let db = self.get_db();
         let now = NOW();
 
         let expire_message = {
             let mut lock = db.lock().unwrap();
-            lock.make_expire_control_message(now)?
+            lock.make_expire_control_message(now, force)
         };
 
-        let tx = self.start_db_thread();
+        if expire_message.is_err() {
+            return Ok(());
+        }
+
+        let expire_message = expire_message.unwrap();
+
+        let tx = self.start_db_thread().await;
         let r = tx.send(expire_message);
         if r.is_err() {
             log::error!("Error in tx.send: {:?}", r.err().unwrap());
@@ -678,15 +600,34 @@ where
     /// Download latest data from REST API
     // fn download_latest(&mut self, verbose: bool) -> i64;
     fn start_market_stream(&mut self) -> anyhow::Result<()>;
-    // fn start_user_stream(&mut self); -> move to OrderInterface
 
-    fn open_realtime_channel(&mut self) -> anyhow::Result<MarketStream>;
     fn open_backtest_channel(
         &mut self,
         time_from: MicroSec,
         time_to: MicroSec,
-    ) -> anyhow::Result<MarketStream>;
+    ) -> anyhow::Result<MarketStream>{
+        let (sender, market_stream) = MarketStream::open();
+        
+        let db = self.get_db();
+
+        std::thread::spawn(move ||{
+            let mut table_db = db.lock().unwrap();
 
+            let result = table_db.select(time_from, time_to, |trade| {
+                let message: MarketMessage = trade.into();
+                sender.send(message)?;
+
+                Ok(())
+            });
+
+            if result.is_err() {
+                log::error!("Error in select: {:?}", result.err().unwrap());
+            }
+        });
+
+        return Ok(market_stream);
+
+    }
 
     /*------------   async ----------------*/
     async fn async_get_latest_archive_date(&self) -> anyhow::Result<MicroSec> {
@@ -716,66 +657,92 @@ where
         .with_context(|| format!("Error in download_archive: {:?}", date))
     }
 
-        /// Download historical data archive and store to database.
-        async fn async_download_archives(
-            &mut self,
-            ndays: i64,
-            force: bool,
-            verbose: bool,
-            low_priority: bool,
-        ) -> anyhow::Result<i64> {
-            log::info!("log download: {} days", ndays);
-            if verbose {
-                println!("log download: {} days", ndays);
-                flush_log();
-            }
-    
-            let latest_date = match self.async_get_latest_archive_date().await {
-                Ok(timestamp) => timestamp,
-                Err(_) => NOW() - DAYS(2),
-            };
-    
-            log::info!("archive latest_date: {}", time_string(latest_date));
-            if verbose {
-                println!("archive latest_date: {}", time_string(latest_date));
-                flush_log();
+    /// Download historical data archive and store to database.
+    async fn async_download_archives(
+        &mut self,
+        ndays: i64,
+        force: bool,
+        verbose: bool,
+        low_priority: bool,
+    ) -> anyhow::Result<i64> {
+        log::info!("log download: {} days", ndays);
+        if verbose {
+            println!("log download: {} days", ndays);
+            flush_log();
+        }
+
+        let latest_date = match self.async_get_latest_archive_date().await {
+            Ok(timestamp) => timestamp,
+            Err(_) => NOW() - DAYS(2),
+        };
+
+        log::info!("archive latest_date: {}", time_string(latest_date));
+        if verbose {
+            println!("archive latest_date: {}", time_string(latest_date));
+            flush_log();
+        }
+
+        let mut download_rec: i64 = 0;
+
+        let tx = self.start_db_thread().await;
+
+        for i in 0..ndays {
+            let date = latest_date - i * DAYS(1);
+
+            if !force && self.validate_db_by_date(date)? {
+                log::info!("{} is valid", time_string(date));
+
+                if verbose {
+                    println!("{} skip download", time_string(date));
+                    flush_log();
+                }
+                continue;
             }
-    
-            let mut download_rec: i64 = 0;
-    
-            let tx = self.start_db_thread();
-    
-            for i in 0..ndays {
-                let date = latest_date - i * DAYS(1);
-    
-                if !force && self.validate_db_by_date(date)? {
-                    log::info!("{} is valid", time_string(date));
-    
-                    if verbose {
-                        println!("{} skip download", time_string(date));
-                        flush_log();
-                    }
-                    continue;
+
+            match self
+                .async_download_archive(&tx, date, low_priority, verbose)
+                .await
+            {
+                Ok(rec) => {
+                    log::info!("downloaded: {}", download_rec);
+                    download_rec += rec;
                 }
-    
-                match self
-                    .async_download_archive(&tx, date, low_priority, verbose)
-                    .await
-                {
-                    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);
-                        }
+                Err(e) => {
+                    log::error!("Error in download_log: {:?}", e);
+                    if verbose {
+                        println!("Error in download_log: {:?}", e);
                     }
                 }
             }
-    
-            Ok(download_rec)
         }
-    
+
+        Ok(download_rec)
+    }
+
+    async fn async_download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
+        if verbose {
+            println!("async_download_lastest");
+            flush_log();
+        }
+
+        let server_config = self.get_server_config();
+        let config = self.get_config().clone();
+
+        let trades = T::get_recent_trades(&server_config, &config).await?;
+        let rec = trades.len() as i64;
+
+        log::debug!("rec: {}", rec);
+
+
+        if verbose {
+            println!("rec: {}", rec);
+            flush_log();
+        }
+
+        let tx = self.start_db_thread().await;
+
+        tx.send(trades)?;        
+
+        Ok(rec)
+    }
 }
diff --git a/rbot_session/src/runner.rs b/rbot_session/src/runner.rs
index fbaf7434..eb8b1f93 100644
--- a/rbot_session/src/runner.rs
+++ b/rbot_session/src/runner.rs
@@ -10,8 +10,7 @@ use futures::{Stream, StreamExt};
 
 use rbot_lib::{
     common::{
-        flush_log, time_string, AccountPair, MarketConfig, MarketMessage, MarketStream, MicroSec,
-        Order, Trade, FLOOR_SEC, MARKET_HUB, NOW, SEC,
+        flush_log, time_string, AccountCoins, MarketConfig, MarketMessage, MarketStream, MicroSec, Order, Trade, FLOOR_SEC, MARKET_HUB, NOW, SEC
     },
     net::{UdpReceiver, UdpSender},
 };
@@ -166,69 +165,6 @@ impl Runner {
         }
     }
 
-    /*
-    #[pyo3(signature = (*,exchange,  market, agent, log_memory=false, execute_time=0, verbose=false, log_file=None, client=false))]
-    pub fn real_run_bak(
-        &mut self,
-        exchange: PyObject,
-        market: PyObject,
-        agent: &PyAny,
-        log_memory: bool,
-        execute_time: i64,
-        verbose: bool,
-        log_file: Option<String>,
-        client: bool,
-    ) -> anyhow::Result<Py<Session>> {
-        self.update_market_info(&market)?;
-        self.update_agent_info(agent)?;
-
-        self.execute_time = execute_time;
-        self.verbose = verbose;
-        self.execute_mode = ExecuteMode::Real;
-
-        let exchange_name = self.exchange_name.clone();
-        let category = self.category.clone();
-        let symbol = self.symbol.clone();
-        let agent_id = self.agent_id.clone();
-
-        if client {
-            //            let rt = tokio::runtime::Runtime::new().unwrap();
-            let rt = tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .build()
-                .unwrap();
-
-            rt.block_on(async {
-                let mut udp = UdpReceiver::open();
-
-                let receiver = udp
-                    .receive_stream(&exchange_name, &category, &symbol, &agent_id)
-                    .await;
-
-                self.async_run(exchange, market, receiver, agent, log_memory, log_file)
-                    .await
-            })
-        } else {
-            self.prepare_data(&exchange, &market)?;
-
-            //            let rt = tokio::runtime::Runtime::new().unwrap();
-            let rt = tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .build()
-                .unwrap();
-
-            rt.block_on(async {
-                let receiver = MARKET_HUB
-                    .subscribe_stream(&exchange_name, &category, &symbol, &agent_id)
-                    .await;
-
-                self.async_run(exchange, market, receiver, agent, log_memory, log_file)
-                    .await
-            })
-        }
-    }
-    */
-
     #[pyo3(signature = (*,exchange,  market, agent, log_memory=false, execute_time=0, verbose=false, log_file=None, client=false))]
     pub fn real_run(
         &mut self,
@@ -345,13 +281,10 @@ impl Runner {
 
     pub fn update_market_info(&mut self, market: &PyObject) -> Result<(), PyErr> {
         let r = Python::with_gil(|py| {
-            let config = market.getattr(py, "market_config").unwrap();
-
             let exchange_name = market.getattr(py, "exchange_name").unwrap();
             let category = market.getattr(py, "trade_category").unwrap();
             let symbol = market.getattr(py, "trade_symbol").unwrap();
 
-            self.config = config.extract::<MarketConfig>(py).unwrap();
             self.exchange_name = exchange_name.extract::<String>(py).unwrap();
             self.category = category.extract::<String>(py).unwrap();
             self.symbol = symbol.extract::<String>(py).unwrap();
@@ -416,7 +349,8 @@ impl Runner {
 
         let time = NOW() / 1_000_000;
 
-        self.agent_id = format!("{}{}", agent_name, Runner::int_to_base64(time));
+        //self.agent_id = format!("{}{}", agent_name, Runner::int_to_base64(time));
+        self.agent_id = format!("{}", agent_name);
 
         Ok(())
     }
@@ -449,7 +383,7 @@ impl Runner {
                     println!("Failed to open log file: {}", &log_file);
                 };
             } else if !log_memory {
-                log::error!(
+                log::warn!(
                     "log_memory and log_file are both false. Please set one of them to true."
                 );
                 println!("WARNING: log_memory and log_file are both false. Please set one of them to true.");
@@ -524,7 +458,9 @@ impl Runner {
         // warm up loop
         let mut warm_up_loop: i64 = WARMUP_STEPS;
         while let Some(message) = rec.next().await {
-            let message = message?;
+            let mut message = message?;
+            message.update_config(&self.config);
+
             self.execute_message_update_session(&py_session, &message)?;
 
             if let MarketMessage::Trade(_trade) = &message {
@@ -539,6 +475,7 @@ impl Runner {
 
         let loop_start_time = NOW();
         while let Some(message) = rec.next().await {
+
             let message = message?;
             self.execute_message(&py_session, agent, &message, interval_sec)?;
             self.loop_count += 1;
@@ -589,6 +526,7 @@ impl Runner {
             flush_log();
         }
 
+        // TODO: retrive session id.
         let py_session = self.create_session(exchange, market, client_mode, log_memory, log_file);
 
         self.call_agent_on_init(&agent, &py_session)?;
@@ -976,7 +914,7 @@ impl Runner {
         message: &MarketMessage,
         interval_sec: i64,
     ) -> anyhow::Result<()> {
-        let config = self.config.clone();
+        let _config = self.config.clone();
 
         // on clockはSession更新前に呼ぶ
         // こうすることでsession.curent_timestampより先の値でon_clockが呼ばれる.
@@ -1015,9 +953,7 @@ impl Runner {
                 // IN Real run, account message is from user stream.
                 // AccountUpdateはFilledかPartiallyFilledのみ発生。
                 if self.has_account_update {
-                    let account_pair = account.extract_pair(&config);
-
-                    self.call_agent_on_account_update(py, agent, py_session, &account_pair)?;
+                    self.call_agent_on_account_update(py, agent, py_session, &account)?;
                 }
             }
             _ => {
@@ -1168,15 +1104,17 @@ impl Runner {
         py: &Python,
         agent: &PyAny,
         py_session: &Py<Session>,
-        account: &AccountPair,
+        account: &AccountCoins,
     ) -> Result<(), PyErr> {
         let mut session = py_session.borrow_mut(*py);
-        session.set_real_account(account);
-        if session.log_account(account).is_err() {
+
+        let account_pair = account.extract_pair(&self.config);
+
+        if session.log_account(&account_pair).is_err() {
             log::error!("call_agent_on_account_update: log_account failed");
         }
 
-        log::debug!("call_agent_on_account_update: {:?}", &account);
+        log::debug!("call_agent_on_account_update: {:?}", &account_pair);
         let py_account = Py::new(*py, account.clone()).unwrap();
 
         agent.call_method1("on_account_update", (session, py_account))?;
@@ -1194,7 +1132,10 @@ impl Runner {
     ) -> Result<(), PyErr> {
         let mut session = py_session.borrow_mut(*py);
         let account = session.get_account();
-        if session.log_account(&account).is_err() {
+
+        let account_pair = account.extract_pair(&self.config);
+       
+        if session.log_account(&account_pair).is_err() {
             log::error!("call_agent_on_account_update_dummy: log_account failed");
         }
         log::debug!("call_agent_on_account_update_dummy: {:?}", &account);
@@ -1252,43 +1193,4 @@ impl Runner {
         }
     }
 
-    // base 64 generated by co pilot!!
-    pub fn int_to_base64(num: i64) -> String {
-        let mut num = num;
-        let mut result = String::new();
-
-        loop {
-            let r = num % 64;
-            num = num / 64;
-
-            let c = match r {
-                0..=25 => (r + 65) as u8 as char,
-                26..=51 => (r + 71) as u8 as char,
-                52..=61 => (r - 4) as u8 as char,
-                62 => '+',
-                63 => '/',
-                _ => panic!("Invalid number"),
-            };
-
-            result.push(c);
-
-            if num == 0 {
-                break;
-            }
-        }
-
-        result
-    }
-}
-
-#[cfg(test)]
-mod runner_test {
-    #[test]
-    fn test_int_to_base64() {
-        let b64 = super::Runner::int_to_base64(0);
-        println!("b64: {}", b64);
-
-        let b64 = super::Runner::int_to_base64(1000001);
-        println!("b64: {}", b64);
-    }
 }
diff --git a/rbot_session/src/session.rs b/rbot_session/src/session.rs
index 61996421..2dc87f4f 100644
--- a/rbot_session/src/session.rs
+++ b/rbot_session/src/session.rs
@@ -12,11 +12,9 @@ use rust_decimal_macros::dec;
 
 use super::{Logger, OrderList};
 use pyo3::prelude::*;
-use rbot_lib::common::{
-    date_string, get_orderbook, hour_string, min_string, time_string, AccountPair, MarketConfig,
-    MarketMessage, MicroSec, Order, OrderBookList, OrderSide, OrderStatus, OrderType, Trade, NOW,
-    SEC,
-};
+use rbot_lib::{common::{
+    date_string, get_orderbook, hour_string, min_string, time_string, AccountCoins, AccountPair, MarketConfig, MarketMessage, MicroSec, Order, OrderBookList, OrderSide, OrderStatus, OrderType, Trade, NOW, SEC
+}, db::TradeTable};
 
 use anyhow;
 
@@ -55,11 +53,13 @@ impl ExecuteMode {
 #[pyclass(name = "Session")]
 #[derive(Debug)]
 pub struct Session {
+    session_id: String,
+
     execute_mode: ExecuteMode,
     buy_orders: OrderList,
     sell_orders: OrderList,
-    real_account: AccountPair,
-    psudo_account: AccountPair,
+    real_account: AccountCoins,
+    psudo_account: AccountCoins,
     exchange: PyObject,
     market: PyObject,
     current_timestamp: MicroSec,
@@ -85,8 +85,8 @@ pub struct Session {
 
     clock_interval_sec: i64,
 
-    asks_edge: Decimal,
-    bids_edge: Decimal,
+    ask_edge: Decimal,
+    bid_edge: Decimal,
 
     exchange_name: String,
     trade_category: String,
@@ -97,6 +97,8 @@ pub struct Session {
     client_mode: bool,
 
     log: Logger,
+
+    db: TradeTable,
 }
 
 #[pymethods]
@@ -125,8 +127,15 @@ impl Session {
             }
         };
 
+        let production = Python::with_gil(|py| {
+            let production = exchange.getattr(py, "production").unwrap();
+            let production: bool = production.extract(py).unwrap();
+
+            production
+        });
+
         let config = Python::with_gil(|py| {
-            let config = market.getattr(py, "market_config").unwrap();
+            let config = market.getattr(py, "config").unwrap();
             let config: MarketConfig = config.extract(py).unwrap();
 
             config
@@ -140,13 +149,23 @@ impl Session {
         });
 
         let category = config.trade_category.clone();
+        let now_time = NOW() / 1_000_000;
+
+        let db_path = TradeTable::make_db_path(
+            &exchange_name, 
+            &category, 
+            &config.trade_symbol, 
+            ! production);
+
+        let db = TradeTable::open(&db_path);
 
         let mut session = Self {
+            session_id: Self::int_to_base64(now_time),
             execute_mode: execute_mode,
             buy_orders: OrderList::new(OrderSide::Buy),
             sell_orders: OrderList::new(OrderSide::Sell),
-            real_account: AccountPair::default(),
-            psudo_account: AccountPair::default(),
+            real_account: AccountCoins::default(),
+            psudo_account: AccountCoins::default(),
             exchange,
             market,
             current_timestamp: 0,
@@ -172,8 +191,8 @@ impl Session {
 
             clock_interval_sec: 0,
 
-            asks_edge: dec![0.0],
-            bids_edge: dec![0.0],
+            ask_edge: dec![0.0],
+            bid_edge: dec![0.0],
 
             exchange_name,
             trade_category: category,
@@ -184,6 +203,8 @@ impl Session {
             client_mode: client_mode,
 
             log: Logger::new(log_memory),
+
+            db: db.unwrap(),
         };
 
         session.load_order_list().unwrap();
@@ -193,12 +214,17 @@ impl Session {
 
     // -----   market information -----
     // call market with current_timestamp
-    pub fn ohlcv(&self, interval: i64, count: i64) -> Result<Py<PyAny>, PyErr> {
+    pub fn py_ohlcv(&self, interval: i64, count: i64) -> Result<Py<PyAny>, PyErr> {
         let window_sec = interval * count;
         let time_from = self.current_timestamp - SEC(window_sec);
         let time_to = self.current_timestamp;
 
+        println!("ohlcv: time_from={}, time_to={}, interval={}", time_from, time_to, interval);
+
         Python::with_gil(|py| {
+
+            println!("call start ohlcv: time_from={}, time_to={}, interval={}", time_from, time_to, interval);
+
             let result = self
                 .market
                 .call_method1(py, "ohlcv", (time_from, time_to, interval));
@@ -215,6 +241,17 @@ impl Session {
         })
     }
 
+    pub fn ohlcv(&mut self, interval: i64, count: i64) -> anyhow::Result<PyDataFrame> {
+        let window_sec = interval * count;
+        let time_from = self.current_timestamp - SEC(window_sec);
+        let time_to = self.current_timestamp;
+
+        let df = self.db.ohlcv_df(time_from, time_to, interval)?;
+
+        Ok(PyDataFrame(df))
+    }
+
+
     #[getter]
     pub fn get_timestamp(&self) -> MicroSec {
         self.current_timestamp
@@ -307,8 +344,8 @@ impl Session {
     #[getter]
     pub fn get_last_price(&self) -> (f64, f64) {
         (
-            self.bids_edge.to_f64().unwrap(),
-            self.asks_edge.to_f64().unwrap(),
+            self.bid_edge.to_f64().unwrap(),
+            self.ask_edge.to_f64().unwrap(),
         )
     }
 
@@ -318,17 +355,17 @@ impl Session {
     }
 
     #[getter]
-    pub fn get_psudo_account(&self) -> AccountPair {
+    pub fn get_psudo_account(&self) -> AccountCoins {
         self.psudo_account.clone()
     }
 
     #[getter]
-    pub fn get_real_account(&self) -> AccountPair {
+    pub fn get_real_account(&self) -> AccountCoins {
         self.real_account.clone()
     }
 
     #[getter]
-    pub fn get_account(&self) -> AccountPair {
+    pub fn get_account(&self) -> AccountCoins {
         match self.execute_mode {
             ExecuteMode::Real => self.real_account.clone(),
             ExecuteMode::BackTest => self.psudo_account.clone(),
@@ -462,9 +499,9 @@ impl Session {
         }
 
         let execute_price = if side == OrderSide::Buy {
-            self.asks_edge + self.market_config.market_order_price_slip
+            self.ask_edge + self.market_config.market_order_price_slip
         } else {
-            self.bids_edge - self.market_config.market_order_price_slip
+            self.bid_edge - self.market_config.market_order_price_slip
         };
 
         return execute_price;
@@ -660,7 +697,7 @@ impl Session {
     }
 
     pub fn update_psudo_account_by_order(&mut self, order: &Order) -> bool {
-        self.psudo_account.apply_order(order);
+        self.psudo_account.apply_order(&self.market_config, order);
 
         if order.status == OrderStatus::Filled || order.status == OrderStatus::PartiallyFilled {
             return true;
@@ -721,7 +758,7 @@ impl Session {
 
 impl Session {
     pub fn on_message(&mut self, message: &MarketMessage) -> Vec<Order> {
-        let config = self.market_config.clone();
+        let _config = self.market_config.clone();
         let mut new_orders = vec![];
 
         match message {
@@ -750,8 +787,7 @@ impl Session {
             MarketMessage::Account(coins) => {
                 log::debug!("on_message: account={:?}", coins);
 
-                let account: AccountPair = coins.extract_pair(&config);
-                self.on_account_update(&account);
+                self.on_account_update(coins);
             }
             MarketMessage::Orderbook(orderbook) => {
                 log::warn!("IGNORED MESSAGE: on_message: orderbook={:?}", orderbook);
@@ -825,14 +861,14 @@ impl Session {
         self.current_timestamp = tick.time;
 
         if tick.order_side == OrderSide::Buy {
-            self.asks_edge = tick.price;
-            if self.asks_edge <= self.bids_edge {
-                self.bids_edge = self.asks_edge - self.market_config.price_unit;
+            self.ask_edge = tick.price;
+            if self.ask_edge <= self.bid_edge {
+                self.bid_edge = self.ask_edge - self.market_config.price_unit;
             }
         } else if tick.order_side == OrderSide::Sell {
-            self.bids_edge = tick.price;
-            if self.asks_edge <= self.bids_edge {
-                self.asks_edge = self.bids_edge + self.market_config.price_unit;
+            self.bid_edge = tick.price;
+            if self.ask_edge <= self.bid_edge {
+                self.ask_edge = self.bid_edge + self.market_config.price_unit;
             }
         }
 
@@ -843,15 +879,22 @@ impl Session {
         }
     }
 
-    pub fn on_account_update(&mut self, account: &AccountPair) {
-        self.real_account = account.clone();
+    pub fn on_account_update(&mut self, account: &AccountCoins) {
+        self.real_account.update(account);
 
-        if self.log_account(account).is_err() {
+        let account_pair: AccountPair = self.real_account.extract_pair(&self.market_config);        
+
+        if self.log_account(&account_pair).is_err() {
             log::error!("log_account_status error");
         };
     }
 
     pub fn on_order_update(&mut self, order: &mut Order) {
+        if ! order.is_my_order(&self.session_name) {
+            log::debug!("on_order_update: skip my order: {:?}", order);
+            return;
+        }
+
         self.log_id += 1;
         order.log_id = self.log_id;
         order.update_balance(&self.market_config);
@@ -882,7 +925,7 @@ impl Session {
     fn new_order_id(&mut self) -> String {
         self.order_number += 1;
 
-        format!("{}-{:04}", self.session_name, self.order_number)
+        format!("{}-{}{:04}", self.session_name, self.session_id, self.order_number)
     }
 
     fn load_order_list(&mut self) -> Result<(), PyErr> {
@@ -909,6 +952,10 @@ impl Session {
                             continue;
                         }
 
+                        if ! order.is_my_order(&self.session_name) {
+                            continue;
+                        }
+
                         log::debug!("OpenOrder {:?}", order);
                         if order.order_side == OrderSide::Buy {
                             self.buy_orders.update_or_insert(&order);
@@ -1110,9 +1157,34 @@ impl Session {
         }
     }
 
-    pub fn set_real_account(&mut self, account: &AccountPair) {
-        self.real_account = account.clone();
+    // base 64 generated by co pilot!!
+    pub fn int_to_base64(num: i64) -> String {
+        let mut num = num;
+        let mut result = String::new();
+
+        loop {
+            let r = num % 64;
+            num = num / 64;
+
+            let c = match r {
+                0..=25 => (r + 65) as u8 as char,
+                26..=51 => (r + 71) as u8 as char,
+                52..=61 => (r - 4) as u8 as char,
+                62 => '-',
+                63 => '_',
+                _ => panic!("Invalid number"),
+            };
+
+            result.push(c);
+
+            if num == 0 {
+                break;
+            }
+        }
+
+        result
     }
+
 }
 
 #[cfg(test)]


Test patch:
diff --git a/rbot/test/binance_test/backtest.ipynb b/rbot/test/binance_test/backtest.ipynb
index b6d8b0c6..d36d6f4c 100644
--- a/rbot/test/binance_test/backtest.ipynb
+++ b/rbot/test/binance_test/backtest.ipynb
@@ -6,11 +6,12 @@
    "metadata": {},
    "outputs": [],
    "source": [
+    "import rbot\n",
     "\n",
     "from rbot import Session\n",
     "from rbot import Runner\n",
+    "from rbot import Binance\n",
     "from rbot import BinanceConfig\n",
-    "from rbot import BinanceMarket\n",
     "from rbot import init_debug_log\n",
     "from rbot import init_log\n",
     "from rbot import OrderSide\n",
@@ -31,30 +32,23 @@
     "    def __init__(self):\n",
     "        self.last_update = 0\n",
     "        self.order_size = 0.001\n",
+    "    \n",
+    "    def on_init(self, session):\n",
+    "        session.clock_interval_sec = 10\n",
     "\n",
     "    def on_clock(self, session, clock):\n",
-    "        if 0 < len(session.buy_orders):\n",
-    "            if session.current_time + SEC(60) < session.buy_orders[0].create_time:\n",
-    "                session.cancel_order(session.buy_orders[0].order_id)\n",
-    "                print(\"cancel buy order\")\n",
-    "        \n",
-    "        if 0 < len(session.sell_orders):\n",
-    "            if session.current_time + SEC(60) < session.sell_orders[0].create_time:\n",
-    "                session.cancel_order(session.sell_orders[0].order_id)\n",
-    "                print(\"cancel sell order\")\n",
+    "        ohlcv = session.ohlcv(60, 5)\n",
+    "        pass\n",
+    "\n",
     "\n",
     "    def on_tick(self, session, side, price, size):\n",
-    "        if session.buy_order_amount == 0 and ((- self.order_size * session.buy_edge) < session.home_change):\n",
-    "            session.limit_order(OrderSide.Buy, session.buy_edge - MARGIN, self.order_size)\n",
-    "            #print(\"try make buy order\", session.home_change, session.foreign_change)\n",
-    "            \n",
-    "        if session.sell_order_amount == 0 and ((- self.order_size) < session.foreign_change):\n",
-    "            session.limit_order(OrderSide.Sell, session.sell_edge + MARGIN, self.order_size)\n",
-    "            #print(\"try make sell order\", session.home_change, session.foreign_change)\n",
+    "        pass\n",
+    "    \n",
+    "    def on_update(self, session, updated_order):\n",
+    "        print(\"on_update\", updated_order)\n",
     "    \n",
     "#    def on_update(self, session, updated_order):\n",
     "#        print(\"on_update\", updated_order)\n",
-    "        \n",
     "    def on_account_update(self, session, account):\n",
     "        print(\"account update: \", session.current_time, account)\n",
     "\n",
@@ -69,12 +63,39 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "market = BinanceMarket(BinanceConfig.BTCUSDT)\n",
+    "binance = Binance(True)\n",
+    "\n",
+    "market = binance.open_market(BinanceConfig.BTCUSDT)\n",
+    "\n",
+    "market\n",
+    "\n",
+    "\n"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "print(binance.production)\n",
+    "print(rbot.__version__)\n",
+    "print(rbot.get_db_root())\n",
+    "print(rbot.set_db_root(\"/tmp/rbot\"))\n",
+    "print(rbot.get_db_root())"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
     "\n",
     "agent = MyAgent()\n",
     "runner = Runner()\n",
     "\n",
-    "session = runner.back_test(market, agent, 60, 0, 0)\n",
+    "session = runner.back_test(exchange=binance, market=market, agent=agent, start_time=0, end_time=0, verbose=True)\n",
     "\n",
     "session"
    ]
@@ -88,6 +109,15 @@
     "session.log_df[session.log_df[\"status\"] == \"Canceled\"]"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "market.ohlcv(0, 0, 600)"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
diff --git a/rbot/test/binance_test/download.py b/rbot/test/binance_test/download.py
index 1c5f2dd4..4d3acec9 100644
--- a/rbot/test/binance_test/download.py
+++ b/rbot/test/binance_test/download.py
@@ -1,43 +1,22 @@
-from rbot import BinanceConfig, BinanceMarket
+from rbot import BinanceConfig, Binance
 from time import sleep
 from rbot import init_log, init_debug_log
 
-
-config = BinanceConfig.BTCUSDT
-config.db_base_dir="./"
-
 #init_debug_log()
-market = BinanceMarket(config)
-
-
-#market.download(ndays=100, verbose=True)
-print("download 1")
-
-market.download(ndays=1, verbose=True)
 
-print("download 2")
-
-if market.is_db_thread_running():
-    print("1:db thread is running")
-else:
-    print("1:db thread is not running")
-
-sleep(4)
-
-market.download(ndays=1, verbose=True)
+config = BinanceConfig.BTCUSDT
+binance = Binance(True)
 
-if market.is_db_thread_running():
-    print("2:db thread is running")
-else:
-    print("2:db thread is not running")
 
+market = binance.open_market(config)
 
-market.stop_db_thread()
+#market.expire_unfix_data(True)
 
-if market.is_db_thread_running():
-    print("3:db thread is running")
-else:
-    print("3:db thread is not running")
+#market.start_market_stream()
+#market.download_archive(ndays=2, verbose=True)
 
-sleep(3)
+#init_debug_log()
+#market.download_latest(verbose=True)
+market.download_gap(force=False, verbose=True)
 
+sleep(5)
\ No newline at end of file
diff --git a/rbot/test/binance_test/market.ipynb b/rbot/test/binance_test/market.ipynb
index 68132e1b..bd91a3bb 100644
--- a/rbot/test/binance_test/market.ipynb
+++ b/rbot/test/binance_test/market.ipynb
@@ -6,84 +6,10 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "import json\n",
-    "from json2html import json2html\n",
-    "from IPython.display import HTML\n",
-    "\n",
-    "from rbot import *\n",
-    "config = BinanceConfig.SPOT(\"BTC\", \"USDT\")\n",
-    "\n",
-    "HTML(json2html.convert(config.__str__()))"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "import rbot\n",
-    "from rbot import BinanceMarket\n",
-    "\n",
-    "          #BinanceMarket\n",
-    "          \n",
-    "b = Binance(test_net=False)          \n",
-    "binance = b.open_market(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.is_db_thread_running()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "binance.start_market_stream()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "binance.stop_db_thread()"
-   ]
-  },
-  {
-   "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)"
+    "from rbot import Binance\n",
+    "from rbot import BinanceConfig\n",
+    "from rbot import init_debug_log\n",
+    "init_debug_log()"
    ]
   },
   {
@@ -92,7 +18,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "binance.download_latest(True)"
+    "binance = Binance(production=False)"
    ]
   },
   {
@@ -101,8 +27,9 @@
    "metadata": {},
    "outputs": [],
    "source": [
+    "config = BinanceConfig.BTCUSDT\n",
     "\n",
-    "binance.download(1, True, True)"
+    "market = binance.open_market(config)"
    ]
   },
   {
@@ -111,7 +38,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "binance.start_market_stream()"
+    "market.download_latest(verbose=True)"
    ]
   },
   {
@@ -120,25 +47,9 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "import polars as pl\n",
+    "from rbot import get_orderbook_list\n",
     "\n",
-    "skip_size = 0.03\n",
-    "ask = binance.asks.filter(pl.col('size') > skip_size).head()['price'][0]\n",
-    "bid = binance.bids.filter(pl.col('size') > skip_size).head()['price'][0]\n",
-    "\n",
-    "print(ask, bid, ask - bid)\n",
-    "\n",
-    "print(bid*0.0002)\n",
-    "        \n"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "binance.market_order(OrderSide.Buy, 0.05, \"BTCUSDT\")"
+    "get_orderbook_list()"
    ]
   },
   {
@@ -147,10 +58,9 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "price = 28000\n",
+    "from rbot import get_orderbook\n",
     "\n",
-    "o = binance.limit_order(OrderSide.Sell, price + 0.1, 0.001)\n",
-    "HTML(json2html.convert(json=o.__str__()))"
+    "get_orderbook(\"BINANCE/SPOT/BTCUSDT\")"
    ]
   },
   {
@@ -159,7 +69,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "binance.new_limit_order_raw(OrderSide.Buy, 24000.1, 0.001)"
+    "market.download_archive(ndays=2, force=False, verbose=True, low_priority=True)"
    ]
   },
   {
@@ -168,142 +78,14 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "HTML(json2html.convert(binance.open_orders.__str__()))\n"
+    "market.download_gap(True)"
    ]
   },
   {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "binance.cancel_all_orders()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "HTML(json2html.convert(binance.trade_list.__str__()))"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "binance.start_market_stream()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "bid, ask = binance.board"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "bid"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "import polars as pl\n",
-    "\n",
-    "bid = binance.bids.filter(pl.col('cusum')< 0.2).tail()['price'][0]\n",
-    "ask = binance.asks.filter(pl.col('cusum')< 0.2).tail()['price'][0]\n",
-    "\n",
-    "print(bid, ask, ask-bid)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "binance.asks"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "import time\n",
-    "from IPython.display import clear_output\n",
-    "count = 100\n",
-    "while True:\n",
-    "    print(binance.asks.head().iloc[::-1])\n",
-    "    print(\"   -- ↑ Asks / Bids ↓ --\")\n",
-    "    print(binance.bids.head())\n",
-    "    time.sleep(0.5)\n",
-    "    clear_output(wait=True)\n",
-    "\n",
-    "    if count < 0:\n",
-    "        break\n",
-    "    count -= 1"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "binance.stop_market_stream()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "j = \"\"\"[{\"symbol\":\"BTCUSDT\",\"orderId\":2782215,\"orderListId\":-1,\"clientOrderId\":\"LnvxKDYYXdtn7m8oweHqyh\",\"price\":\"25000.00000000\",\"origQty\":\"0.01000000\",\"executedQty\":\"0.01000000\",\"cummulativeQuoteQty\":\"250.00000000\",\"status\":\"FILLED\",\"timeInForce\":\"GTC\",\"type\":\"LIMIT\",\"side\":\"BUY\",\"stopPrice\":\"0.00000000\",\"icebergQty\":\"0.00000000\",\"time\":1694396686024,\"updateTime\":1694398622469,\"isWorking\":true,\"workingTime\":1694396686024,\"origQuoteOrderQty\":\"0.00000000\",\"selfTradePreventionMode\":\"NONE\"},\n",
-    " {\"symbol\":\"BTCUSDT\",\"orderId\":2782217,\"orderListId\":-1,\"clientOrderId\":\"tpELVj8sifYLV0RKUA8Ojx\",\"price\":\"25000.00000000\",\"origQty\":\"0.01000000\",\"executedQty\":\"0.01000000\",\"cummulativeQuoteQty\":\"250.00000000\",\"status\":\"FILLED\",\"timeInForce\":\"GTC\",\"type\":\"LIMIT\",\"side\":\"BUY\",\"stopPrice\":\"0.00000000\",\"icebergQty\":\"0.00000000\",\"time\":1694396695637,\"updateTime\":1694398622469,\"isWorking\":true,\"workingTime\":1694396695637,\"origQuoteOrderQty\":\"0.00000000\",\"selfTradePreventionMode\":\"NONE\"},\n",
-    "]\"\"\""
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": []
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "json2html.convert(json=j)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
+   "cell_type": "markdown",
    "metadata": {},
-   "outputs": [],
    "source": [
-    "init_debug_log()\n",
-    "binance.start_market_stream()"
+    "## Order Test"
    ]
   },
   {
@@ -311,15 +93,6 @@
    "metadata": {},
    "source": []
   },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "binance.stop_market_stream()"
-   ]
-  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -327,8 +100,8 @@
    "outputs": [],
    "source": [
     "init_debug_log()\n",
-    "binance.start_user_stream()\n",
-    "#binance.stop_user_stream()"
+    "binance.enable_order_with_my_own_risk = True\n",
+    "binance.limit_order(config, \"Buy\", 40000, 0.001, \"test1\")"
    ]
   },
   {
@@ -337,7 +110,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "binance.market_order(OrderSide.Sell, 0.05)"
+    "binance.market_order(config, \"Sell\", 0.001, \"test2\")"
    ]
   },
   {
@@ -346,7 +119,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "stream = binance.channel"
+    "binance.get_open_orders(config)"
    ]
   },
   {
@@ -355,17 +128,13 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "stream"
+    "binance.cancel_order(config, \"11248143\")"
    ]
   },
   {
-   "cell_type": "code",
-   "execution_count": null,
+   "cell_type": "markdown",
    "metadata": {},
-   "outputs": [],
-   "source": [
-    "session = _LiveSession()"
-   ]
+   "source": []
   },
   {
    "cell_type": "code",
@@ -373,7 +142,8 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "session.connect(stream)"
+    "from rbot import ExecuteMode\n",
+    "from rbot import Session"
    ]
   },
   {
@@ -382,7 +152,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "binance.cancel_all_orders()"
+    "session = Session(binance, market, ExecuteMode.Real, False)"
    ]
   },
   {
@@ -391,8 +161,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "ohlcv1h = binance.ohlcv(NOW()-DAYS(1), 0, 3600)\n",
-    "ohlcv1h"
+    "session.limit_order(\"Buy\", 40000, 0.001)"
    ]
   },
   {
@@ -401,7 +170,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "(ohlcv1h['high']-ohlcv1h['low']).mean()"
+    "session.market_order(\"Buy\", 0.001)"
    ]
   },
   {
diff --git a/rbot/test/binance_test/nullagent.py b/rbot/test/binance_test/nullagent.py
index 4a2d39de..51b5f104 100644
--- a/rbot/test/binance_test/nullagent.py
+++ b/rbot/test/binance_test/nullagent.py
@@ -1,8 +1,10 @@
 
 from rbot import Session
 from rbot import Runner
+
+from rbot import Binance
 from rbot import BinanceConfig
-from rbot import BinanceMarket
+
 from rbot import init_debug_log
 from rbot import init_log
 from rbot import OrderSide
@@ -14,7 +16,7 @@ def __init__(self):
         pass
 
     def on_init(self, session):
-        session.clock_interval = 60
+        session.clock_interval_sec = 60
         print("init: ", session.timestamp)
 
         pass
@@ -40,23 +42,32 @@ def on_account_update(self, session, account):
         pass
 
 init_log()
-   
-market = BinanceMarket(BinanceConfig.TEST_BTCUSDT)
-#market.download(1, force=False, verbose=True)
 
-market.cancel_all_orders()
-    
+
+binance = Binance(False)   
+
+market = binance.open_market(BinanceConfig.BTCUSDT)
 agent = MyAgent()
 runner = Runner()
 
 #init_debug_log()
-session = runner.back_test(market, agent, start_time=NOW()-DAYS(1), end_time=0, verbose=True,execute_time=60*3)
-#session = runner.dry_run(market, agent, verbose=True, execute_time=60)
-#session = runner.real_run(market, agent, verbose=True, execute_time=60)
+#session = runner.back_test(exchange=binance, market=market, agent=agent, start_time=NOW()-DAYS(1), end_time=0, verbose=True,execute_time=60*3)
+#print(session)
 
 
-print(session)
+#runner = Runner()
+#session = runner.dry_run(
+    #exchange=binance, market=market, agent=agent, verbose=True, execute_time=60*3, log_memory=True)
+#print(session)
+
+
+binance.enable_order_with_my_own_risk = True
+
+runner = Runner()
+session = runner.real_run(
+    exchange=binance, market=market, agent=agent, verbose=True, execute_time=60*3, log_memory=True)
 
+print(session.log)
 
 
 
diff --git a/rbot/test/binance_test/session.ipynb b/rbot/test/binance_test/session.ipynb
index e1493a0b..6802e6a3 100644
--- a/rbot/test/binance_test/session.ipynb
+++ b/rbot/test/binance_test/session.ipynb
@@ -17,7 +17,6 @@
     "from rbot import Session\n",
     "from rbot import Runner\n",
     "from rbot import BinanceConfig\n",
-    "from rbot import BinanceMarket\n",
     "from rbot import init_debug_log\n",
     "from rbot import init_log\n",
     "from rbot import OrderSide\n",
@@ -57,7 +56,11 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "config = BinanceConfig.TEST_BTCUSDT\n",
+    "config = BinanceConfig.BTCUSDT\n",
+    "exchagen = Binance(False)\n",
+    "\n",
+    "\n",
+    "\n",
     "market = BinanceMarket(config)"
    ]
   },
diff --git a/rbot/test/bybit_test/basilbot_backtest.ipynb b/rbot/test/bybit_test/basilbot_backtest.ipynb
index 24826035..501b8ad2 100644
--- a/rbot/test/bybit_test/basilbot_backtest.ipynb
+++ b/rbot/test/bybit_test/basilbot_backtest.ipynb
@@ -179,6 +179,32 @@
     "\n"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "MARGIN = 0.05\n",
+    "\n",
+    "class MyAgent:\n",
+    "    def __init__(self):\n",
+    "        self.last_update = 0\n",
+    "        self.order_size = 0.001\n",
+    "\n",
+    "    def on_clock(self, session, clock):\n",
+    "        pass\n",
+    "\n",
+    "\n",
+    "    def on_tick(self, session, side, price, size):\n",
+    "        pass\n",
+    "    \n",
+    "#    def on_update(self, session, updated_order):\n",
+    "#        print(\"on_update\", updated_order)\n",
+    "    def on_account_update(self, session, account):\n",
+    "        print(\"account update: \", session.current_time, account)"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {
@@ -200,7 +226,7 @@
    },
    "outputs": [],
    "source": [
-    "from rbot import Bybit, BybitMarket, BybitConfig\n",
+    "from rbot import Bybit, BybitConfig\n",
     "from rbot import Runner\n",
     "from rbot import NOW, DAYS"
    ]
@@ -217,19 +243,18 @@
    },
    "outputs": [],
    "source": [
-    "config = BybitConfig.SPOT_BTCUSDT\n",
+    "config = BybitConfig.BTCUSDT\n",
     "\n",
-    "bybit = Bybit()\n",
+    "bybit = Bybit(True)\n",
     "market = bybit.open_market(config)\n",
     "\n",
     "# バックテストの期間を設定\n",
     "BACKTEST_DAYS = 2\n",
     "\n",
     "# バックテスト用に過去データをダウンロード\n",
-    "market.download(\n",
+    "market.download_archive(\n",
     "    ndays=BACKTEST_DAYS,          # 過去１日分のデータをダウンロード\n",
     "    verbose=True,\n",
-    "    archive_only=True\n",
     ")"
    ]
   },
@@ -241,10 +266,21 @@
    },
    "outputs": [],
    "source": [
-    "agent = BasilAgent()        # 実行対象のエージェントを指定\n",
+    "#agent = BasilAgent()        # 実行対象のエージェントを指定\n",
+    "agent = MyAgent()        # 実行対象のエージェントを指定\n",
     "runner = Runner()           # 実行モジュール Runner を作成"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from rbot import init_debug_log;\n",
+    "init_debug_log()"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -258,6 +294,7 @@
    "outputs": [],
    "source": [
     "session = runner.back_test(\n",
+    "    exchange=bybit,                 # 取引所を指定\n",
     "    agent=agent,                    # エージェントを指定\n",
     "    market=market,                  # マーケットを指定\n",
     "    start_time=NOW()-DAYS(BACKTEST_DAYS), # 開始時刻を指定\n",
diff --git a/rbot/test/bybit_test/market.ipynb b/rbot/test/bybit_test/market.ipynb
index 62ccee66..3548a0f7 100644
--- a/rbot/test/bybit_test/market.ipynb
+++ b/rbot/test/bybit_test/market.ipynb
@@ -1,15 +1,5 @@
 {
  "cells": [
-  {
-   "cell_type": "markdown",
-   "metadata": {},
-   "source": []
-  },
-  {
-   "cell_type": "markdown",
-   "metadata": {},
-   "source": []
-  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -33,7 +23,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "bybit = Bybit(production=False)"
+    "bybit = Bybit(production=True)"
    ]
   },
   {
@@ -92,19 +82,6 @@
     "market.download_archive(ndays=2, force=False, verbose=True, low_priority=False)"
    ]
   },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "from rbot import time_string\n",
-    "\n",
-    "a, b  = market.find_latest_gap()\n",
-    "\n",
-    "print(time_string(a), time_string(b))\n"
-   ]
-  },
   {
    "cell_type": "code",
    "execution_count": null,
diff --git a/rbot/test/bybit_test/nullagent.py b/rbot/test/bybit_test/nullagent.py
index ba65973d..5774d0aa 100644
--- a/rbot/test/bybit_test/nullagent.py
+++ b/rbot/test/bybit_test/nullagent.py
@@ -84,6 +84,3 @@ def on_update(self, session, updated_order):
             )
 
     
-
-
-