Repository: yasstake/rbot (language: python)

Issue:
Create archive as file
Implement archive as csv.gz.

But its slow as much as sqlite, so this is the baseline for alter to use paquest file format.
migrate profit to order struct
* migrate profit to order struct

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

## Summary by CodeRabbit

- **新機能**
	- 注文ステータスに新しいフィールド（開いたポジション、閉じたポジション、ポジション、利益、手数料、総利益）を追加しました。
- **改善点**
	- バックテストの表示を最後の5件から最初の100件に変更しました。
- **リファクタ**
	- 注文ログとアカウント情報のロギングに焦点を当てるために、`Logger`から利益関連の構造とメソッドを削除しました。
	- セッション管理の改善として、不要な注文のスキップと注文詳細の更新処理を強化しました。

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


Discussion / hints:
<!-- This is an auto-generated comment: summarize by coderabbit.ai -->
<!-- This is an auto-generated comment: failure by coderabbit.ai -->

> [!CAUTION]
> ## Review failed
> 
> The pull request is closed.

<!-- end of auto-generated comment: failure by coderabbit.ai --><!-- walkthrough_start -->

<details>
<summary>Walkthrough</summary>

## Walkthrough

この変更では、BinanceとBybitの取引プラットフォームに関連する構成やAPIの改善が行われました。新しいメソッドの追加や、既存のメソッドの引数の名称変更により、コードの可読性と機能性が向上しました。また、データのCSVおよびParquetフォーマットへの処理機能が強化され、取引データの管理がより効率的になりました。

## Changes

| ファイルパス                                           | 変更概要                                                        |
|-----------------------------------------------------|---------------------------------------------------------------|
| `exchanges/binance/src/config.rs`                  | `BinanceServerConfig`に`is_production`メソッドを追加。`size_scale`が4から3に変更。 |
| `exchanges/bybit/src/config.rs`                     | `BybitServerConfig`に`is_production`メソッドを追加。                   |
| `exchanges/binance/src/market.rs`                   | `async_new`メソッド内の`test_mode`を`production`に変更。DB接続方法の更新。    |
| `exchanges/bybit/src/market.rs`                     | `test_net`を`production`に変更。構造体のフィールド名変更。           |
| `exchanges/binance/src/rest.rs`                     | `line_to_trade`と`convert_archive_line`の新メソッドを追加。              |
| `exchanges/bybit/src/rest.rs`                       | `line_to_trade`と`convert_archive_line`の新メソッドを追加。              |
| `rbot_lib/src/common/config.rs`                     | `ServerConfig`トレイトに`is_production`メソッドを追加。               |
| `rbot_lib/src/common/order.rs`                      | `Trade`構造体にCSV処理機能の強化。新たにCSV形式への変換メソッドを追加。     |
| `rbot_lib/src/common/time.rs`                       | `parse_date`メソッドを追加し、日付文字列をマイクロ秒に変換。            |
| `rbot_lib/src/db/archive.rs`                        | `TradeTableArchive`構造体を追加し、アーカイブデータ処理の機能を実装。     |
| `rbot_lib/src/db/avro.rs`                           | `TradeAvro`構造体を追加し、Avro形式のデータ処理を実装。                 |
| `rbot_lib/src/db/compress.rs`                       | Gzip圧縮ログの読み書き処理を追加。非同期関数`log_convert`と`log_foreach`を実装。 |
| `rbot_lib/src/net/rest.rs`                          | CSVおよびParquet関連の新メソッドを追加。                           |
| `rbot_market/src/market.rs`                         | `for_each_record`メソッドを追加し、指定範囲内の取引レコードを処理。       |
| `rbot_session/src/session.rs`                       | `production`に基づくパラメータの変更。                               |
| `src/exchange/binance/mod.rs`                       | `BinanceMarket`構造体を追加し、Binance取引データとのインタフェースを実装。   |

## Sequence Diagram(s)

```mermaid
sequenceDiagram
    participant User
    participant BinanceMarket
    participant Database

    User->>BinanceMarket: download(ndays, force)
    BinanceMarket->>Database: check existing data
    Database-->>BinanceMarket: return existing trades
    BinanceMarket->>BinanceMarket: determine time chunks
    BinanceMarket->>Database: download data
    Database-->>BinanceMarket: save new trades
    BinanceMarket-->>User: return download status
```

> うさぎの詩 🐇✨  
>  
> > 取引のデータ、流れよ、  
> > 新しいメソッド、まるで春のよう。  
> > 笑顔で跳ねる、うさぎの心、  
> > 便利になった、喜びの連鎖。  
> > データも踊る、楽しい夜、  
> > 未来へ続く、この道を行こう。  
>

</details>

<!-- walkthrough_end --><!-- This is an auto-generated comment: raw summary by coderabbit.ai -->

<!--

```markdown
exchanges/binance/src/config.rs: ## AI-generated summary of changes

The diff introduces a new method `is_production` to the `BinanceServerConfig` struct, which returns the value of the `production` field. This addition enhances the functionality of the `ServerConfig` implementation by allowing the retrieval of the production status of the server configuration. Additionally, the `size_scale` attribute in the `BinanceConfig` struct is modified from `4` to `3`, indicating a change in the scaling factor for size, which may affect how sizes are interpreted or calculated in the system. These changes collectively improve the configurability and clarity of the Binance server settings.

## Alterations to the declarations of exported or public entities

- `fn is_production(&self) -> bool` in struct `BinanceServerConfig` in `exchanges/binance/src/config.rs` → new method added
- `size_scale: 4` in struct `BinanceConfig` in `exchanges/binance/src/config.rs` → `size_scale: 3` in struct `BinanceConfig` in `exchanges/binance/src/config.rs`

---

exchanges/binance/src/market.rs: ## AI-generated summary of changes

The diff introduces modifications to the `BinanceMarket` implementation, specifically in the `new` and `async_new` methods. The primary change involves renaming the `test_mode` parameter to `production` in the `async_new` method, which alters the semantic meaning of the boolean argument passed to this function. The logic now reflects that the parameter indicates whether the market is operating in production mode rather than test mode. Consequently, the argument passed in the `new` method has also been updated to align with this new naming convention, flipping the boolean value to directly reflect the production status of the server configuration.

Additionally, the method for opening the database connection has been changed from `TradeTable::open` to `TradeTable::get`, suggesting a possible shift in how database connections are managed, though the precise implications of this change would depend on the definitions of these methods. The `db` field of the `BinanceMarket` struct is now assigned the result of `TradeTable::get`, which may indicate a more streamlined or updated approach to accessing database resources.

Overall, these changes enhance clarity regarding the operational mode of the market while potentially improving the handling of database connections.

## Alterations to the declarations of exported or public entities

- `pub fn new(server_config: &BinanceServerConfig, config: &MarketConfig) -> Self` in `impl BinanceMarket` in `exchanges/binance/src/market.rs` → `pub fn new(server_config: &BinanceServerConfig, config: &MarketConfig) -> Self` in `impl BinanceMarket` in `exchanges/binance/src/market.rs` (no signature change, but internal logic modified)
- `async fn async_new(server_config: &BinanceServerConfig, config: &MarketConfig, test_mode: bool) -> anyhow::Result<Self>` in `impl BinanceMarket` in `exchanges/binance/src/market.rs` → `async fn async_new(server_config: &BinanceServerConfig, config: &MarketConfig, production: bool) -> anyhow::Result<Self>` in `impl BinanceMarket` in `exchanges/binance/src/market.rs`

---

exchanges/binance/src/rest.rs: ## AI-generated summary of changes

The diff introduces two new methods to the `RestApi<BinanceServerConfig>` implementation for the `BinanceRestApi` struct. The first method, `line_to_trade`, takes a string slice as an argument and returns a `Trade` instance, currently implemented to print a placeholder message indicating that it is not yet implemented and returns a default `Trade`. The second method, `convert_archive_line`, also takes a string slice and returns a `String`, similarly printing a placeholder message and returning an empty string. These additions suggest that the functionality for processing trade data and converting archive lines is being set up but not yet fully realized.

## Alterations to the declarations of exported or public entities

- `fn line_to_trade(_line: &str) -> Trade` in `src/rest.rs` → `fn line_to_trade(_line: &str) -> Trade` in `src/rest.rs`
- `fn convert_archive_line(_line: &str) -> String` in `src/rest.rs` → `fn convert_archive_line(_line: &str) -> String` in `src/rest.rs`

---

exchanges/bybit/src/archive_test.rs: ## AI-generated summary of changes

The new file `archive_test.rs` introduces functionality for handling data archiving and retrieval using the Avro and Parquet formats, specifically tailored for a trading application. It provides four primary functions for saving and loading data frames: `save_df_to_avro`, `load_df_from_avro`, `save_df_to_parquet`, and `load_df_from_parquet`. Each function takes a mutable reference to a `DataFrame` and a mutable `PathBuf` for file paths, allowing for the creation and reading of files in the specified formats. The data frames are manipulated using the `polars` library, which is designed for efficient data handling.

The control flow begins with the `save_df_to_avro` and `save_df_to_parquet` functions, which set the appropriate file extensions, create files, and write the data frames to these files using their respective writers. The loading functions, `load_df_from_avro` and `load_df_from_parquet`, open the specified files and read the data back into `DataFrame` objects. Each function returns a result wrapped in `anyhow::Result`, facilitating error handling.

Additionally, the file contains a series of tests encapsulated within a module named `archive_test`. These tests cover various functionalities, including creating archives, downloading data, converting formats, and validating the integrity of the loaded data. The tests utilize asynchronous programming with `tokio`, allowing for non-blocking operations, which is essential for handling I/O-bound tasks in a trading environment.

Overall, this file enhances the application's ability to manage historical trading data efficiently, supporting both storage and retrieval processes in a structured manner.

## Alterations to the declarations of exported or public entities

- `pub fn save_df_to_avro(df: &mut DataFrame, path: &mut PathBuf) -> anyhow::Result<PathBuf>` in `archive_test.rs`
- `pub fn load_df_from_avro(path: &PathBuf) -> anyhow::Result<DataFrame>` in `archive_test.rs`
- `pub fn save_df_to_parquet(df: &mut DataFrame, path: &mut PathBuf) -> anyhow::Result<PathBuf>` in `archive_test.rs`
- `pub fn load_df_from_parquet(path: &PathBuf) -> anyhow::Result<DataFrame>` in `archive_test.rs`

---

exchanges/bybit/src/config.rs: ## AI-generated summary of changes

The diff introduces a new method `is_production` to the `BybitServerConfig` struct, which implements the `ServerConfig` trait. This method returns a boolean indicating whether the server configuration is set for production by accessing the `production` field of the struct. The addition of this method enhances the functionality of the `BybitServerConfig` by providing a direct way to check the production status, which could be crucial for decision-making processes in the application that depend on the environment configuration.

The overall control flow of the `BybitServerConfig` remains unchanged, but the inclusion of the `is_production` method allows for more explicit checks regarding the server's operational context. This change could facilitate better handling of different environments (e.g., production vs. development) within the application logic.

## Alterations to the declarations of exported or public entities

- `fn is_production(&self) -> bool` in struct `BybitServerConfig` in `exchanges/bybit/src/config.rs`

---

exchanges/bybit/src/lib.rs: ## AI-generated summary of changes

The diff introduces a new module named `archive_test` into the `exchanges/bybit/src/lib.rs` file. This addition does not alter any existing functionality or logic within the file but expands the module's structure by including this new component. The inclusion of `archive_test` suggests a potential focus on testing related to the archive functionality, although no specific details about its implementation or purpose are provided in the diff. The existing `pub use` statements remain unchanged, indicating that the public interface of the module is not affected by this addition.

## Alterations to the declarations of exported or public entities

- `mod archive_test;` in `exchanges/bybit/src/lib.rs` (new module added)

---

exchanges/bybit/src/market.rs: ## AI-generated summary of changes

The diff introduces significant changes to the `Bybit` struct and its associated methods within the `market.rs` file. The primary modification is the renaming of the `test_net` field to `production`, which alters the semantic meaning of this boolean flag. This change is reflected in the struct definition, where the `production` field is now used to indicate whether the instance operates in a production environment rather than a test network.

In the `Bybit` implementation, the constructor now initializes the `production` field directly, replacing the previous logic that set `test_net` to the negation of `production`. Consequently, the `open_market` method has been updated to pass the `production` flag to the `BybitMarket` constructor instead of the `test_net` flag, thus ensuring that the market operates under the correct configuration based on the new field semantics.

Furthermore, the `BybitMarket` struct's constructor has been adjusted to accept the `production` parameter instead of `test_net`, and this change propagates to the `async_new` method, which also replaces `test_mode` with `production`. This consistent renaming across methods ensures that all references to the operational mode of the market are aligned with the new terminology.

Additionally, a block of commented-out code related to the `async_download_lastest` function has been removed, indicating a potential cleanup or refactoring effort. This function was previously intended for downloading the latest trades but is no longer present in the implementation, possibly suggesting a shift in focus or a decision to simplify the codebase.

Overall, these changes enhance clarity regarding the operational mode of the `Bybit` struct and its market interactions, while also streamlining the code by removing unused or unnecessary components.

## Alterations to the declarations of exported or public entities

- `pub struct Bybit` in `exchanges/bybit/src/market.rs` → `pub struct Bybit { production: bool, ... }` (renamed from `test_net` to `production`)
- `pub fn open_market(&self, config: &MarketConfig) -> BybitMarket` in `impl Bybit` in `exchanges/bybit/src/market.rs` → `pub fn open_market(&self, config: &MarketConfig) -> BybitMarket` (changed parameter from `test_net` to `production` in `BybitMarket::new`)
- `pub fn new(server_config: &BybitServerConfig, config: &MarketConfig, test_net: bool) -> Self` in `impl BybitMarket` in `exchanges/bybit/src/market.rs` → `pub fn new(server_config: &BybitServerConfig, config: &MarketConfig, production: bool) -> Self` (changed parameter name from `test_net` to `production`)
- `pub async fn async_new(server_config: &BybitServerConfig, config: &MarketConfig, test_mode: bool) -> anyhow::Result<Self>` in `impl BybitMarket` in `exchanges/bybit/src/market.rs` → `pub async fn async_new(server_config: &BybitServerConfig, config: &MarketConfig, production: bool) -> anyhow::Result<Self>` (changed parameter name from `test_mode` to `production`)

---

exchanges/bybit/src/rest.rs: ## AI-generated summary of changes

The diff introduces several modifications to the `BybitRestApi` implementation in `exchanges/bybit/src/rest.rs`, primarily enhancing the clarity and structure of asynchronous function calls and adding new utility functions. Key changes include:

1. **Refactoring of Asynchronous Calls**: The asynchronous calls to `Self::get` have been streamlined. The previous multiline formatting for error context has been condensed into a single line, improving readability without altering the underlying logic. This change is evident in the methods `get_board_snapshot` and `get_recent_trades`, where the error handling context is now more succinctly presented.

2. **Addition of New Functions**: Two new functions, `line_to_trade` and `convert_archive_line`, have been introduced. The `line_to_trade` function processes a CSV line into a `Trade` object, parsing necessary fields such as timestamp, order side, size, price, and ID. This function enhances the ability to handle trade data from CSV formats effectively. The `convert_archive_line` function similarly processes a CSV line but formats it back into a string representation, which may be useful for logging or storage purposes.

3. **Improved Logging**: The logging statements have been reformatted for better clarity. For example, the debug logging in the `get_trade_klines` function now uses a more structured format, making it easier to read the output. This is a minor improvement but contributes to better maintainability and debugging.

4. **Minor Adjustments in Function Signatures**: The `get_account` function's signature was slightly adjusted for consistency, although the core functionality remains unchanged. The spacing in the function parameters has been modified to align with coding standards, enhancing code readability.

5. **Test Enhancements**: A new test case, `test_line_2_trade`, has been added to verify the functionality of the `line_to_trade` method. This addition indicates a focus on ensuring the correctness of new functionality through unit testing.

Overall, these changes enhance the code's readability, maintainability, and functionality without altering the fundamental logic of the existing methods.

## Alterations to the declarations of exported or public entities

- `async fn get_account(server: &BybitServerConfig) -> anyhow::Result<AccountCoins>` in `impl RestApi<BybitServerConfig> for BybitRestApi` remains unchanged.
- `fn line_to_trade(line: &str) -> Trade` in `impl RestApi<BybitServerConfig> for BybitRestApi` added.
- `fn convert_archive_line(line: &str) -> String` in `impl RestApi<BybitServerConfig> for BybitRestApi` added.

---

rbot_lib/src/common/config.rs: ## AI-generated summary of changes

The diff introduces a new method to the `ServerConfig` trait in the `rbot_lib/src/common/config.rs` file. Specifically, it adds the `is_production(&self) -> bool` function, which is likely intended to determine whether the server is running in a production environment. This addition enhances the functionality of the `ServerConfig` trait by providing a mechanism to check the operational context of the server, which may influence configuration management and behavior in different environments. No other changes to existing methods or logic are present in the diff.

## Alterations to the declarations of exported or public entities

- `fn is_production(&self) -> bool` in trait `ServerConfig` in `rbot_lib/src/common/config.rs`

---

rbot_lib/src/common/order.rs: ## AI-generated summary of changes

The diff introduces several enhancements to the `Trade` struct in the `order.rs` file, primarily focusing on CSV serialization and deserialization functionalities. The existing `to_csv` method has been modified to streamline its output format, reducing the number of fields serialized from six to five, omitting the `status` field. This change reflects a shift in the data representation for CSV output, likely aimed at simplifying the exported data.

New methods have been added to facilitate the conversion from and to CSV format. The `from_csv` method parses a CSV string into a `Trade` instance, extracting values for `time`, `order_side`, `price`, `size`, and `id`. This method utilizes Rust's error handling to unwrap parsed values, ensuring that the input format adheres to expectations. Additionally, a `from_record` method has been introduced, which constructs a `Trade` instance from a `StringRecord`, providing an alternative way to create `Trade` objects from CSV-like data structures.

A new `csv_header` method has also been added, which generates a CSV header string based on predefined keys, enhancing the usability of the CSV output by providing a clear structure for the data. This method ensures that the header aligns with the data fields outputted by the `to_csv` method.

Overall, these changes significantly enhance the CSV handling capabilities of the `Trade` struct, introducing new methods that allow for both serialization and deserialization, while also refining the existing functionality.

## Alterations to the declarations of exported or public entities

- `pub fn to_csv(&self) -> String` in `impl Trade` in `rbot_lib/src/common/order.rs` → `pub fn to_csv(&self) -> String` (modified implementation)
- `pub fn from_csv(csv: String) -> Self` in `impl Trade` in `rbot_lib/src/common/order.rs` → `pub fn from_csv(csv: String) -> Self`
- `pub fn from_record(rec: &StringRecord) -> Trade` in `impl Trade` in `rbot_lib/src/common/order.rs` → `pub fn from_record(rec: &StringRecord) -> Trade`
- `pub fn csv_header() -> String` in `impl Trade` in `rbot_lib/src/common/order.rs` → `pub fn csv_header() -> String`

---

rbot_lib/src/common/time.rs: ## AI-generated summary of changes

The diff introduces significant enhancements to the time handling functionality within the `rbot_lib/src/common/time.rs` file. A new public function, `parse_date`, has been added, which converts a date string in the YYYYMMDD format into a `MicroSec` timestamp. This function includes validation for the input format, returning an error if the string does not conform to the expected length. It parses the year, month, and day components from the string and constructs a `DateTime` object using the `Utc` timezone, subsequently returning the corresponding timestamp in microseconds.

Additionally, the `date_string` function has been documented with a comment that clarifies its purpose of converting time to the YYYYMMDD format. The test suite has been expanded with a new test case, `test_yymmdd`, which verifies the correctness of the `parse_date` function by asserting expected outcomes for specific date inputs.

The changes also include imports for `FromStr` from the standard library and `anyhow` for error handling, which are essential for the new parsing functionality. Overall, these modifications enhance the library's capabilities in handling date strings and improve its robustness through error handling.

## Alterations to the declarations of exported or public entities

- `pub fn date_string(t: MicroSec) -> String` in `rbot_lib/src/common/time.rs` → `pub fn date_string(t: MicroSec) -> String` (no signature change, but added documentation)
- `pub fn parse_date(date: &str) -> anyhow::Result<MicroSec>` in `rbot_lib/src/common/time.rs` (new function added)

---

rbot_lib/src/db/archive.rs: ## AI-generated summary of changes

The newly introduced file `archive.rs` implements the `TradeTableArchive` struct, which provides functionality for managing and processing archived trade data. This struct is generic over two types, `T` and `U`, where `T` must implement the `RestApi` trait and `U` must implement the `ServerConfig` trait along with the `Clone` trait. The struct maintains several fields to track server configurations, timestamps, and archive data.

Key functionalities include methods for creating a new instance, checking for the latest archive date, downloading historical data, and managing the archive files. The `new` method initializes the struct and performs an analysis of existing archives. The `latest_archive_date` method retrieves the most recent archive date, with a retry mechanism in case of failure. The `download` method facilitates downloading historical trade data, with options to force downloads and control verbosity.

The struct also provides methods for loading data into a DataFrame, selecting records based on time ranges, and iterating over records with a user-defined function. It includes utility methods for creating empty DataFrames, checking for existing archive files, and managing file paths. The `analyze` method inspects the archive directory to determine the start and end times of available data, while the `list_dates` method retrieves a list of available archive dates.

Additionally, the file supports converting CSV archives to Parquet format and provides detailed logging throughout the processes to aid in debugging and monitoring. Overall, this implementation enhances the capability to manage trade archives efficiently, ensuring robust data handling and retrieval mechanisms.

## Alterations to the declarations of exported or public entities

- `pub struct TradeTableArchive<T, U>` in `rbot_lib/src/db/archive.rs` → `pub struct TradeTableArchive<T, U>` in `rbot_lib/src/db/archive.rs`
- `pub fn new(server: &U, config: &MarketConfig) -> Self` in `rbot_lib/src/db/archive.rs` → `pub fn new(server: &U, config: &MarketConfig) -> Self` in `rbot_lib/src/db/archive.rs`
- `pub async fn latest_archive_date(&mut self) -> anyhow::Result<MicroSec>` in `rbot_lib/src/db/archive.rs` → `pub async fn latest_archive_date(&mut self) -> anyhow::Result<MicroSec>` in `rbot_lib/src/db/archive.rs`
- `pub fn has_archive_file(&self, date: MicroSec) -> bool` in `rbot_lib/src/db/archive.rs` → `pub fn has_archive_file(&self, date: MicroSec) -> bool` in `rbot_lib/src/db/archive.rs`
- `pub async fn download(&mut self, ndays: i64, force: bool, verbose: bool) -> anyhow::Result<i64>` in `rbot_lib/src/db/archive.rs` → `pub async fn download(&mut self, ndays: i64, force: bool, verbose: bool) -> anyhow::Result<i64>` in `rbot_lib/src/db/archive.rs`
- `pub fn load_df(&self, date: MicroSec) -> anyhow::Result<DataFrame>` in `rbot_lib/src/db/archive.rs` → `pub fn load_df(&self, date: MicroSec) -> anyhow::Result<DataFrame>` in `rbot_lib/src/db/archive.rs`
- `pub fn select_df(&self, start_time: MicroSec, end_time: MicroSec) -> anyhow::Result<DataFrame>` in `rbot_lib/src/db/archive.rs` → `pub fn select_df(&self, start_time: MicroSec, end_time: MicroSec) -> anyhow::Result<DataFrame>` in `rbot_lib/src/db/archive.rs`
- `pub fn for_each_record<F>(&self, start_time: MicroSec, end_time: MicroSec, f: &mut F) -> anyhow::Result<i64>` in `rbot_lib/src/db/archive.rs` → `pub fn for_each_record<F>(&self, start_time: MicroSec, end_time: MicroSec, f: &mut F) -> anyhow::Result<i64>` in `rbot_lib/src/db/archive.rs`
- `pub fn select_dates(&self, start_time: MicroSec, end_time: MicroSec) -> anyhow::Result<Vec<MicroSec>>` in `rbot_lib/src/db/archive.rs` → `pub fn select_dates(&self, start_time: MicroSec, end_time: MicroSec) -> anyhow::Result<Vec<MicroSec>>` in `rbot_lib/src/db/archive.rs`
- `pub fn file_date(&self, path: &PathBuf) -> anyhow::Result<MicroSec>` in `rbot_lib/src/db/archive.rs` → `pub fn file_date(&self, path: &PathBuf) -> anyhow::Result<MicroSec>` in `rbot_lib/src/db/archive.rs`
- `pub fn analyze(&mut self) -> anyhow::Result<()>` in `rbot_lib/src/db/archive.rs` → `pub fn analyze(&mut self) -> anyhow::Result<()>` in `rbot_lib/src/db/archive.rs`
- `pub fn list_dates(&self) -> anyhow::Result<Vec<MicroSec>>` in `rbot_lib/src/db/archive.rs` → `pub fn list_dates(&self) -> anyhow::Result<Vec<MicroSec>>` in `rbot_lib/src/db/archive.rs`
- `pub fn archive_directory(&self) -> PathBuf` in `rbot_lib/src/db/archive.rs` → `pub fn archive_directory(&self) -> PathBuf` in `rbot_lib/src/db/archive.rs`
- `pub fn file_path(&self, date: MicroSec) -> PathBuf` in `rbot_lib/src/db/archive.rs` → `pub fn file_path(&self, date: MicroSec) -> PathBuf` in `rbot_lib/src/db/archive.rs`
- `pub fn delete(&self, date: MicroSec) -> anyhow::Result<()>` in `rbot_lib/src/db/archive.rs` → `pub fn delete(&self, date: MicroSec) -> anyhow::Result<()>` in `rbot_lib/src/db/archive.rs`
- `pub async fn archive_to_parquet(&mut self, date: MicroSec, force: bool, verbose: bool) -> anyhow::Result<i64>` in `rbot_lib/src/db/archive.rs` → `pub async fn archive_to_parquet(&mut self, date: MicroSec, force: bool, verbose: bool) -> anyhow::Result<i64>` in `rbot_lib/src/db/archive.rs`
- `pub async fn archive_to_csv(&mut self, date: MicroSec, force: bool, verbose: bool) -> anyhow::Result<i64>` in `rbot_lib/src/db/archive.rs` → `pub async fn archive_to_csv(&mut self, date: MicroSec, force: bool, verbose: bool) -> anyhow::Result<i64>` in `rbot_lib/src/db/archive.rs`

---

rbot_lib/src/db/avro.rs: ## AI-generated summary of changes

The newly introduced file `avro.rs` implements functionality for handling trade data in the Avro serialization format. It defines a `TradeAvro` struct that encapsulates key attributes of a trade, including execution time, order side, price, size, and a unique identifier. This struct is annotated with Serde's `Serialize` and `Deserialize` traits, allowing for easy conversion to and from Avro format.

The file also includes a test function `save_load`, which demonstrates how to define an Avro schema for the `TradeAvro` struct. This schema specifies the data types for each field and is parsed from a string representation. Within the test, a `Writer` is created to serialize `TradeAvro` instances into an Avro file named `measurements.avro`, using the Deflate codec for compression. The function initializes a sample `TradeAvro` object but does not currently implement the serialization logic to write the object to the file.

Overall, this file lays the groundwork for serializing trade data in Avro format, facilitating efficient storage and transport of trade records.

## Alterations to the declarations of exported or public entities

- `pub struct TradeAvro` in `rbot_lib/src/db/avro.rs` → `pub struct TradeAvro` in `rbot_lib/src/db/avro.rs` (new declaration)
- `fn save_load() -> anyhow::Result<()>` in `rbot_lib/src/db/avro.rs` → `fn save_load() -> anyhow::Result<()>` in `rbot_lib/src/db/avro.rs` (new declaration)

---

rbot_lib/src/db/compress.rs: ## AI-generated summary of changes

The newly introduced file implements two asynchronous functions, `log_convert` and `log_foreach`, which handle reading and processing Gzip-compressed log files. Both functions utilize the `async_compression` crate for Gzip decoding and encoding, leveraging Tokio's asynchronous I/O capabilities for efficient file operations.

The `log_convert` function takes a source file path, a target file path, a boolean indicating the presence of a header, and a transformation function `f`. It opens the source file and creates a target file, then reads the compressed log lines. If a header is present, it reads and logs it. For each subsequent line, it applies the transformation function `f`, writes the modified line to the target file, and ensures that all data is flushed and the encoder is properly shut down at the end.

The `log_foreach` function similarly opens a source file and reads its contents. It also checks for a header and logs it if present. Instead of transforming the lines, it applies a mutable function `f` to each line as it is read, allowing for flexible processing of the log data without writing to a new file.

Both functions are designed to handle asynchronous operations effectively, providing a robust mechanism for processing Gzip-compressed logs in a non-blocking manner.

## Alterations to the declarations of exported or public entities

- `pub async fn log_convert<F>(source_file: PathBuf, target_file: PathBuf, has_header: bool, f: F) -> anyhow::Result<()>` in `rbot_lib/src/db/compress.rs` 
- `pub async fn log_foreach<F>(source_file: PathBuf, has_header: bool, mut f: F) -> anyhow::Result<()>` in `rbot_lib/src/db/compress.rs`

---

rbot_lib/src/db/df.rs: ## AI-generated summary of changes

The provided diff modifies the `ohlcv_df` and `ohlcv_from_ohlcvv_df` functions in the `rbot_lib/src/db/df.rs` file by removing commented-out sorting logic. Specifically, it eliminates multiple blocks of commented-out code that previously defined sorting behavior based on the `time_stamp` key. This change simplifies the code by removing unnecessary comments that do not contribute to the current functionality. The remaining code continues to utilize the `group_by_dynamic` and `agg` functions to process data based on the `time_stamp` column, maintaining the existing logic for aggregating open prices. The removal of these comments does not alter the control flow or the logic of the functions, as the sorting operations were not active in the original implementation.

## Alterations to the declarations of exported or public entities

- No changes to the declarations of exported or public entities were made.

---

rbot_lib/src/db/fs.rs: ## AI-generated summary of changes

The diff introduces significant changes to the functionality of the database path management in the Rust module. The function `db_full_path` has been modified to accommodate a new parameter, `production`, replacing the previous `test_net` boolean. This change alters the logic for determining the directory structure for database files. 

The new function `db_path_root` is introduced, which constructs the root path for the database based on whether the environment is production or test. The logic now differentiates between production and test directories by appending either "PRODUCTION" or "TEST" to the path, enhancing clarity and organization in the file structure. The creation of directories has been adjusted to reflect this new hierarchy, ensuring that the relevant directories are created based on the specified environment.

In the `db_full_path` function, the call to `db_path_root` has been updated to utilize the new parameter, and the path construction for the database file now incorporates the newly defined root path. This results in a more structured and logical approach to managing database file paths, improving the maintainability of the code.

Overall, these changes improve the clarity of the code and enhance its functionality by clearly separating production and test environments within the database file structure.

## Alterations to the declarations of exported or public entities

- `pub fn db_full_path(exchange_name: &str, category: &str, symbol: &str, test_net: bool) -> PathBuf` in `rbot_lib/src/db/fs.rs` → `pub fn db_full_path(exchange_name: &str, category: &str, symbol: &str, production: bool) -> PathBuf` in `rbot_lib/src/db/fs.rs`
- `pub fn db_full_path(exchange_name: &str, category: &str, symbol: &str, test_net: bool) -> PathBuf` in `rbot_lib/src/db/fs.rs` → `pub fn db_path_root(exchange_name: &str, category: &str, symbol: &str, production: bool) -> PathBuf` in `rbot_lib/src/db/fs.rs`

---

rbot_lib/src/db/mod.rs: ## AI-generated summary of changes

The diff introduces several new modules to the `rbot_lib/src/db/mod.rs` file, enhancing the functionality of the database library. Specifically, it adds the `archive`, `compress`, and `avro` modules, which are expected to provide additional capabilities related to data handling and storage. Alongside the declarations for these new modules, the `pub use` statements have been updated to re-export the newly added modules, ensuring that they are accessible for use in other parts of the library. This change improves the modularity and extensibility of the codebase by allowing for more diverse data management strategies.

## Alterations to the declarations of exported or public entities

- `pub mod archive;` in `rbot_lib/src/db/mod.rs` → `pub mod archive;` in `rbot_lib/src/db/mod.rs`
- `pub mod compress;` in `rbot_lib/src/db/mod.rs` → `pub mod compress;` in `rbot_lib/src/db/mod.rs`
- `pub mod avro;` in `rbot_lib/src/db/mod.rs` → `pub mod avro;` in `rbot_lib/src/db/mod.rs`
- `pub use archive::*;` in `rbot_lib/src/db/mod.rs` → `pub use archive::*;` in `rbot_lib/src/db/mod.rs`
- `pub use compress::*;` in `rbot_lib/src/db/mod.rs` → `pub use compress::*;` in `rbot_lib/src/db/mod.rs`
- `pub use avro::*;` in `rbot_lib/src/db/mod.rs` → `pub use avro::*;` in `rbot_lib/src/db/mod.rs`

---

rbot_lib/src/db/sqlite.rs: ## AI-generated summary of changes

The diff introduces several modifications primarily focused on improving the code structure and enhancing functionality within the `TradeTable` implementation in `rbot_lib/src/db/sqlite.rs`. Notable changes include the removal of unnecessary blank lines, which streamlines the code and improves readability. The `test_net` parameter in the `db_full_path` function has been renamed to `production`, indicating a clearer intent regarding its usage in production environments.

Additionally, the logic surrounding database operations has been refined. The `open` method of `TradeTable` has been changed from public to private, suggesting that this method is intended for internal use only, which may enhance encapsulation. The handling of the database vacuuming process has been elaborated with additional logging, providing better insight into the operation's success or failure. 

The `set_cache_ohlcvv` method has been reformatted for clarity, with chained method calls on the DataFrame now presented in a more readable manner. The return statements in the `ohlcv_floor_fix_time` method have been simplified, maintaining functionality while improving code style.

Overall, these changes reflect an effort to enhance code clarity, maintainability, and functionality, particularly in the context of database operations and method accessibility.

## Alterations to the declarations of exported or public entities

- `pub fn open(name: &str) -> anyhow::Result<Self>` in `impl TradeTable` → `fn open(name: &str) -> anyhow::Result<Self>` in `impl TradeTable`

---

rbot_lib/src/net/rest.rs: ## AI-generated summary of changes

The diff introduces several significant changes to the functionality of the `RestApi` trait and its associated implementations in the `rbot_lib/src/net/rest.rs` file. Key modifications include the addition of new methods for handling CSV and Parquet data, which enhance data processing capabilities.

1. **New Methods Added**: 
   - The `line_to_trade(rec: &str) -> Trade` method has been added to convert a line of text into a `Trade` object.
   - The `convert_archive_line(line: &str) -> String` method has been introduced to facilitate the conversion of archive lines into a specific string format.
   - The `has_archive(server: &T, config: &MarketConfig, date: MicroSec) -> bool` method has been added to check for the existence of an archive based on the provided parameters.

2. **DataFrame Handling**: 
   - The `logfile_to_df(logfile: &str) -> anyhow::Result<DataFrame>` method has been introduced to read a `.csv.gz` file and convert it into a Polars `DataFrame`. This method incorporates error handling for file reading and parsing.
   - The `logdf_to_archivedf(df: &DataFrame) -> anyhow::Result<DataFrame>` method has been added to transform a DataFrame into a specific archived format, renaming columns and ensuring the correct data types.

3. **File Writing**: 
   - The `archivedf_to_file(mut df: DataFrame, archive_file: &str) -> anyhow::Result<()>` method has been implemented to write a DataFrame to a Parquet file, including error handling for file creation and writing.

4. **Refactoring of Existing Methods**: 
   - The `read_csv_archive<F>(file_path: &PathBuf, has_header: bool, mut f: F) -> anyhow::Result<()>` method has been modified to accept a `PathBuf` instead of a `&str`, enhancing type safety and flexibility. The internal logic has also been updated to streamline the reading process for different file types (gzipped, zipped, or plain CSV).
   - The `read_line_archive<F>(file_path: &str, has_header: bool, mut f: F) -> anyhow::Result<()>` method has been added to read lines from an archive file, improving the handling of CSV files with or without headers.

5. **Removal of Legacy Code**: 
   - Several commented-out sections of code related to the old `log_download` and `extract` functions have been removed, indicating a shift towards a more streamlined and efficient approach to downloading and processing logs.

Overall, these changes significantly enhance the data processing capabilities of the library, particularly in terms of handling CSV and Parquet formats, while also improving the clarity and maintainability of the code.

## Alterations to the declarations of exported or public entities

- `fn line_to_trade(rec: &str) -> Trade` in trait `RestApi<T>` in `rbot_lib/src/net/rest.rs` → `fn line_to_trade(rec: &str) -> Trade` in trait `RestApi<T>` in `rbot_lib/src/net/rest.rs`
- `fn convert_archive_line(line: &str) -> String` in trait `RestApi<T>` in `rbot_lib/src/net/rest.rs` → `fn convert_archive_line(line: &str) -> String` in trait `RestApi<T>` in `rbot_lib/src/net/rest.rs`
- `fn logfile_to_df(logfile: &str) -> anyhow::Result<DataFrame>` in trait `RestApi<T>` in `rbot_lib/src/net/rest.rs` → `fn logfile_to_df(logfile: &str) -> anyhow::Result<DataFrame>` in trait `RestApi<T>` in `rbot_lib/src/net/rest.rs`
- `fn logdf_to_archivedf(df: &DataFrame) -> anyhow::Result<DataFrame>` in trait `RestApi<T>` in `rbot_lib/src/net/rest.rs` → `fn logdf_to_archivedf(df: &DataFrame) -> anyhow::Result<DataFrame>` in trait `RestApi<T>` in `rbot_lib/src/net/rest.rs`
- `fn archivedf_to_file(mut df: DataFrame, archive_file: &str) -> anyhow::Result<()>` in trait `RestApi<T>` in `rbot_lib/src/net/rest.rs` → `fn archivedf_to_file(mut df: DataFrame, archive_file: &str) -> anyhow::Result<()>` in trait `RestApi<T>` in `rbot_lib/src/net/rest.rs`
- `pub fn read_csv_archive<F>(file_path: &PathBuf, has_header: bool, mut f: F) -> anyhow::Result<()>` in `rbot_lib/src/net/rest.rs` → `pub fn read_csv_archive<F>(file_path: &PathBuf, has_header: bool, mut f: F) -> anyhow::Result<()>` in `rbot_lib/src/net/rest.rs`
- `pub fn read_line_archive<F>(file_path: &str, has_header: bool, mut f: F) -> anyhow::Result<()>` in `rbot_lib/src/net/rest.rs` → `pub fn read_line_archive<F>(file_path: &str, has_header: bool, mut f: F) -> anyhow::Result<()>` in `rbot_lib/src/net/rest.rs`

---

rbot_lib/src/net/ws.rs: ## AI-generated summary of changes

The diff introduces a new method, `is_production`, to the `test_exchange_ws` module, which returns a boolean value of `false`. This addition alters the functionality by providing a way to determine the production status within the context of the module, potentially influencing how instances of this module are utilized in production versus testing environments. Additionally, the diff removes an unused import statement for `url::Url`, which helps streamline the code by eliminating unnecessary dependencies. Overall, these changes enhance the clarity and functionality of the code while maintaining its existing structure.

## Alterations to the declarations of exported or public entities

- `fn is_production(&self) -> bool` in module `test_exchange_ws` in `rbot_lib/src/net/ws.rs` → `fn is_production(&self) -> bool` in module `test_exchange_ws` in `rbot_lib/src/net/ws.rs` (new method added)

---

rbot_market/src/market.rs: ## AI-generated summary of changes

The diff introduces a new method called `for_each_record` to the implementation of a struct in the `rbot_market/src/market.rs` file. This method takes two parameters, `time_from` and `time_to`, both of type `MicroSec`, along with a mutable reference to a closure `f` that processes each `Trade` record. The method's purpose is to retrieve records from a database within the specified time range and apply the provided closure to each record, returning a result wrapped in `anyhow::Result<()>`. The logic includes locking the database for safe access, selecting the records based on the time range, and executing the closure on each selected `Trade`. The method concludes by returning an `Ok(())` result, indicating successful completion or propagating any errors encountered during the process.

This addition enhances the functionality of the struct by allowing users to easily iterate over trade records within a specified time frame, thereby improving data handling capabilities.

## Alterations to the declarations of exported or public entities

- `fn for_each_record<F>(&self, time_from: MicroSec, time_to: MicroSec, f: &mut F) -> anyhow::Result<()>` in struct implementation in `rbot_market/src/market.rs` → New method added.

---

rbot_session/src/session.rs: ## AI-generated summary of changes

The changes in the provided diff primarily involve modifications to the parameters passed to function calls within the `Session` implementation. Specifically, the boolean argument indicating whether to use a production environment has been altered from negation (`!production`) to a direct reference to the `production` variable in one instance, and from `false` to `true` in another. This indicates a shift in logic regarding the usage of the production database. The first change suggests that the function will now directly utilize the value of the `production` variable instead of its negation, potentially altering the intended behavior based on the context in which `production` is set. The second change explicitly sets the database reference to always use the production database, reinforcing a more stringent requirement for production data access. These adjustments reflect a clear intent to ensure that the session operates under production conditions, which may impact how data is processed and retrieved.

## Alterations to the declarations of exported or public entities

- No alterations to the declarations of exported or public entities were made in this diff.

---

src/exchange/binance/mod.rs: ## AI-generated summary of changes

The newly introduced file `src/exchange/binance/mod.rs` implements the `BinanceMarket` struct, which serves as an interface for interacting with the Binance trading platform's historical trade data. The struct encapsulates essential properties such as the market name, a dummy flag, and a database connection represented by `TradeTable`. 

The primary functionality revolves around downloading historical trade data over specified time periods. The `download` method orchestrates this process by determining the time chunks that need to be downloaded based on existing data in the database and the specified number of days. It utilizes multi-threading to handle downloads concurrently, sending data through a channel to be processed and inserted into the database. The method also includes error handling and logging for monitoring download progress and issues.

Additionally, the `BinanceMarket` struct provides methods for selecting trades (`select_trades`), generating OHLCV data (`ohlcv` and `ohlcvv`), and retrieving market information (`info`). The helper methods `make_historical_data_url_timestamp` and `make_historical_data_url` construct URLs for accessing the trade data based on timestamps. The `rec_to_trade` method converts CSV records into `Trade` objects, facilitating the parsing of downloaded data.

The file also includes unit tests that validate the URL generation logic and the download functionality, ensuring that the methods behave as expected. Overall, this implementation enhances the ability to interact with Binance's historical trade data programmatically, making it accessible for further analysis or integration into larger systems.

## Alterations to the declarations of exported or public entities

- `pub struct BinanceMarket` in `src/exchange/binance/mod.rs` → `pub struct BinanceMarket { name: String, pub dummy: bool, pub db: TradeTable }` in `src/exchange/binance/mod.rs`
- `pub fn new(market_name: &str, dummy: bool) -> Self` in `src/exchange/binance/mod.rs`
- `pub fn download(&mut self, ndays: i64, force: bool) -> i64` in `src/exchange/binance/mod.rs`
- `pub fn select_trades(&mut self, from_time: MicroSec, to_time: MicroSec) -> PyResult<Py<PyArray2<f64>>>` in `src/exchange/binance/mod.rs`
- `pub fn ohlcvv(&mut self, from_time: MicroSec, to_time: MicroSec, window_sec: i64) -> PyResult<Py<PyArray2<f64>>>` in `src/exchange/binance/mod.rs`
- `pub fn ohlcv(&mut self, from_time: MicroSec, to_time: MicroSec, window_sec: i64) -> PyResult<Py<PyArray2<f64>>>` in `src/exchange/binance/mod.rs`
- `pub fn info(&mut self) -> String` in `src/exchange/binance/mod.rs`
- `pub fn _repr_html_(&self) -> String` in `src/exchange/binance/mod.rs`
```

-->

<!-- end of auto-generated comment: raw summary by coderabbit.ai --><!-- This is an auto-generated comment: pr objectives by coderabbit.ai -->

<!--

## PR Summary

The pull request titled "Create archive as file" was submitted by the user yasstake. It introduces functionality to implement an archive in the CSV.GZ format. The primary goal of this implementation is to provide a method for archiving data efficiently.

However, the author notes that the current performance of this implementation is comparable to that of SQLite, indicating that it may not meet desired speed requirements. As a result, the author suggests that this implementation serves as a baseline for future modifications, specifically mentioning the potential to transition to using the Parquet file format, which is known for better performance in handling large datasets.

Overall, this pull request aims to enhance the data archiving capabilities of the project while acknowledging the need for further optimization in terms of speed.

-->

<!-- end of auto-generated comment: pr objectives by coderabbit.ai --><!-- tips_start -->

---



Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media?

<details>
<summary>Share</summary>

- [X](https://twitter.com/intent/tweet?text=I%20just%20used%20%40coderabbitai%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20the%20proprietary%20code.%20Check%20it%20out%3A&url=https%3A//coderabbit.ai)
- [Mastodon](https://mastodon.social/share?text=I%20just%20used%20%40coderabbitai%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20the%20proprietary%20code.%20Check%20it%20out%3A%20https%3A%2F%2Fcoderabbit.ai)
- [Reddit](https://www.reddit.com/submit?title=Great%20tool%20for%20code%20review%20-%20CodeRabbit&text=I%20just%20used%20CodeRabbit%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20proprietary%20code.%20Check%20it%20out%3A%20https%3A//coderabbit.ai)
- [LinkedIn](https://www.linkedin.com/sharing/share-offsite/?url=https%3A%2F%2Fcoderabbit.ai&mini=true&title=Great%20tool%20for%20code%20review%20-%20CodeRabbit&summary=I%20just%20used%20CodeRabbit%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20proprietary%20code)

</details>

<details>
<summary>Tips</summary>

### Chat

There are 3 ways to chat with [CodeRabbit](https://coderabbit.ai):

- Review comments: Directly reply to a review comment made by CodeRabbit. Example:
	- `I pushed a fix in commit <commit_id>.`
	- `Generate unit testing code for this file.`
	- `Open a follow-up GitHub issue for this discussion.`
- Files and specific lines of code (under the "Files changed" tab): Tag `@coderabbitai` in a new review comment at the desired location with your query. Examples:
	- `@coderabbitai generate unit testing code for this file.`
	-	`@coderabbitai modularize this function.`
- PR comments: Tag `@coderabbitai` in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
	- `@coderabbitai generate interesting stats about this repository and render them as a table.`
	- `@coderabbitai show all the console.log statements in this repository.`
	- `@coderabbitai read src/utils.ts and generate unit testing code.`
	- `@coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.`
	- `@coderabbitai help me debug CodeRabbit configuration file.`

Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments.

### CodeRabbit Commands (invoked as PR comments)

- `@coderabbitai pause` to pause the reviews on a PR.
- `@coderabbitai resume` to resume the paused reviews.
- `@coderabbitai review` to trigger an incremental review. This is useful when automatic reviews are disabled for the repository.
- `@coderabbitai full review` to do a full review from scratch and review all the files again.
- `@coderabbitai summary` to regenerate the summary of the PR.
- `@coderabbitai resolve` resolve all the CodeRabbit review comments.
- `@coderabbitai configuration` to show the current CodeRabbit configuration for the repository.
- `@coderabbitai help` to get help.


Additionally, you can add `@coderabbitai ignore` anywhere in the PR description to prevent this PR from being reviewed.

### CodeRabbit Configuration File (`.coderabbit.yaml`)

- You can programmatically configure CodeRabbit by adding a `.coderabbit.yaml` file to the root of your repository.
- Please see the [configuration documentation](https://docs.coderabbit.ai/guides/configure-coderabbit) for more information.
- If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: `# yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json`

### Documentation and Community

- Visit our [Documentation](https://coderabbit.ai/docs) for detailed information on how to use CodeRabbit.
- Join our [Discord Community](https://discord.com/invite/GsXnASn26c) to get help, request features, and share feedback.
- Follow us on [X/Twitter](https://twitter.com/coderabbitai) for updates and announcements.

</details>

<!-- tips_end -->
<!-- This is an auto-generated comment: summarize by coderabbit.ai -->
<!-- walkthrough_start -->

<details>
<summary>Walkthrough</summary>

## Walkthrough
プロジェクト全体で、`Order`構造体に新しいフィールドが追加され、利益関連のロギングが削除されました。これにより、注文の管理が強化され、コードベースがシンプルになりました。

## Changes
| ファイル | 変更内容 |
| --- | --- |
| `exchanges/bybit/src/...` | `BybitOrderStatus`と`Order`に新しいフィールドを追加 |
| `rbot/test/tutorial/...` | 表示する注文の数を最後の5つから最初の100つに変更 |
| `rbot_lib/src/common/order.rs` | `Order`構造体に新しいフィールドを追加、新しいインポートを導入 |
| `rbot_session/src/logger.rs` | 利益関連のデータ構造とメソッドを削除 |
| `rbot_session/src/session.rs` | `on_message`メソッドの更新、`log_profit`メソッドの削除 |

> 🐰
> ぴょんぴょん、コードが変わる  
> 新しいフィールド、古いログはさようなら  
> ぴょこぴょこ、バグを追いかけ  
> コードの森を、一緒に駆け巡ろう  
> 🌟🌿📈  
> ぴょん、変更の祝いに、一句  
> 「コード更新、春の息吹」

</details>

<!-- walkthrough_end --><!-- This is an auto-generated comment: raw summary by coderabbit.ai -->
<!--


exchanges/bybit/src/message.rs, exchanges/bybit/src/rest.rs: 
- Added new fields `open_position`, `close_position`, `position`, `profit`, `fee`, and `total_profit` to the `BybitOrderStatus` struct implementation for `Into<Order>`.
- In the `BybitRestApi` implementation, new fields `open_position`, `close_position`, `position`, `profit`, `fee`, and `total_profit` with initial values of `dec![0.0]` have been added to the struct `Order`.

rbot/test/tutorial/basilbot_backtest.ipynb: 
- Changed the code to display the first 100 orders instead of the last 5.

rbot_lib/src/common/order.rs: 
- In the `order.rs` file, the changes involve adding new fields to the `Order` struct such as `open_position`, `close_position`, `position`, `profit`, `fee`, and `total_profit`. Additionally, new imports for `async_std::stream::Cloned` and `reqwest::Proxy` were introduced. The `Order` struct's constructor and the function `ordervec_to_dataframe` were updated to accommodate these new fields.

rbot_session/src/logger.rs: 
- The changes in `logger.rs` involve removing the `profit` data structure and related methods from the `Logger` struct and associated functions. The `log_profit` method and related getters for `profit` have been eliminated, streamlining the codebase by focusing on logging orders and account information. Additionally, the `__repr__` method no longer includes `profit` information in its output.

rbot_session/src/session.rs: 
- In the `Session` implementation in `session.rs`, the `on_message` method now handles a `MarketMessage::Order` by skipping orders not belonging to the session and updating order details. The `log_profit` method has been removed, and `update_psudo_position` now modifies the `Order` struct directly.

-->
<!-- end of auto-generated comment: raw summary by coderabbit.ai --><!-- This is an auto-generated comment: short summary by coderabbit.ai -->
<!--


### PR Objectives

The primary objective of this pull request, titled "migrate profit to order struct," is to integrate profit-related data directly into the `Order` struct across various components of the codebase. This migration involves adding new fields related to profit calculation and order positioning, such as `open_position`, `close_position`, `position`, `profit`, `fee`, and `total_profit`, into the `Order` struct. These changes are applied within different parts of the project, including the Bybit exchange integration and the core order management system. Additionally, the pull request aims to streamline the logging and session management by removing outdated profit logging mechanisms and updating methods to accommodate the newly integrated profit information within the `Order` struct. This migration is intended to centralize profit tracking within the order data structure, thereby simplifying the handling of order-related profit information throughout the system.

### AI-generated Summary of Generated Summaries

This pull request introduces a significant restructuring of how profit information is handled within the system by migrating profit-related data to the `Order` struct. Key changes include:

1. **Structural Updates to `Order` and Bybit Integration**: New fields, namely `open_position`, `close_position`, `position`, `profit`, `fee`, and `total_profit`, have been added to the `Order` struct across various files, including `exchanges/bybit/src/message.rs`, `exchanges/bybit/src/rest.rs`, and `rbot_lib/src/common/order.rs`. These additions are aimed at incorporating direct profit tracking and order positioning within the `Order` struct itself. The Bybit integration specifically sees these fields initialized with default values, ensuring compatibility with the Bybit API.

2. **Adjustments in Logging and Session Management**: The pull request eliminates the previously separate profit logging mechanism within `rbot_session/src/logger.rs` by removing the `profit` data structure and associated methods from the `Logger` struct. This change is complemented by adjustments in session management within `rbot_session/src/session.rs`, where the `log_profit` method is removed, and the `update_psudo_position` method is updated to modify the `Order` struct directly, reflecting the new approach to handling profit information.

3. **Enhancements in Data Presentation and Utility Functions**: The tutorial notebook `rbot/test/tutorial/basilbot_backtest.ipynb` has been updated to display the first 100 orders instead of the last 5, improving data visibility during backtesting. Additionally, the `ordervec_to_dataframe` function within `rbot_lib/src/common/order.rs` is updated to accommodate the new fields in the `Order` struct, ensuring that profit and position data are correctly represented in dataframes.

Overall, this pull request centralizes profit tracking within the order management system by embedding relevant data directly into the `Order` struct, streamlines logging and session management by removing outdated profit logging mechanisms, and enhances data presentation and utility functions to accommodate these structural changes.

-->
<!-- end of auto-generated comment: short summary by coderabbit.ai --><!-- commit_ids_reviewed_start -->

<!-- 59983bee045358d3e4e1652e28242fb57aa86919 -->

<!-- commit_ids_reviewed_end --><!-- tweet_start -->

Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media?

<details>
<summary>Share</summary>

- [X](https://twitter.com/intent/tweet?text=I%20just%20used%20%40coderabbitai%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20the%20proprietary%20code.%20Check%20it%20out%3A&url=https%3A//coderabbit.ai)
- [Mastodon](https://mastodon.social/share?text=I%20just%20used%20%40coderabbitai%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20the%20proprietary%20code.%20Check%20it%20out%3A%20https%3A%2F%2Fcoderabbit.ai)
- [Reddit](https://www.reddit.com/submit?title=Great%20tool%20for%20code%20review%20-%20CodeRabbit&text=I%20just%20used%20CodeRabbit%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20proprietary%20code.%20Check%20it%20out%3A%20https%3A//coderabbit.ai)
- [LinkedIn](https://www.linkedin.com/sharing/share-offsite/?url=https%3A%2F%2Fcoderabbit.ai&mini=true&title=Great%20tool%20for%20code%20review%20-%20CodeRabbit&summary=I%20just%20used%20CodeRabbit%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20proprietary%20code)

</details>

<!-- tweet_end --><!-- tips_start -->

---

<details>
<summary>Tips</summary>

### Chat

There are 3 ways to chat with CodeRabbit:

- Review comments: Directly reply to a review comment made by CodeRabbit. Example:
	- `I pushed a fix in commit <commit_id>.`
	- `Generate unit-tests for this file.`
	- `Open a follow-up GitHub issue for this discussion.`
- Files and specific lines of code (under the "Files changed" tab): Tag `@coderabbitai` in a new review comment at the desired location with your query. Examples:
	- `@coderabbitai generate unit tests for this file.`
	-	`@coderabbitai modularize this function.`
- PR comments: Tag `@coderabbitai` in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
	- `@coderabbitai generate interesting stats about this repository and render them as a table.`
	- `@coderabbitai show all the console.log statements in this repository.`
	- `@coderabbitai read src/utils.ts and generate unit tests.`
	- `@coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.`

Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments.

### CodeRabbit Commands (invoked as PR comments)

- `@coderabbitai pause` to pause the reviews on a PR.
- `@coderabbitai resume` to resume the paused reviews.
- `@coderabbitai review` to trigger a review. This is useful when automatic reviews are disabled for the repository.
- `@coderabbitai resolve` resolve all the CodeRabbit review comments.
- `@coderabbitai help` to get help.


Additionally, you can add `@coderabbitai ignore` anywhere in the PR description to prevent this PR from being reviewed.

### CodeRabbit Configration File (`.coderabbit.yaml`)

- You can programmatically configure CodeRabbit by adding a `.coderabbit.yaml` file to the root of your repository.
- The JSON schema for the configuration file is available [here](https://coderabbit.ai/integrations/coderabbit-overrides.v2.json).
- If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: `# yaml-language-server: $schema=https://coderabbit.ai/integrations/coderabbit-overrides.v2.json`

### CodeRabbit Discord Community

Join our [Discord Community](https://discord.com/invite/GsXnASn26c) to get help, request features, and share feedback.

</details>

<!-- tips_end -->

Fix patch:
diff --git a/Cargo.toml b/Cargo.toml
index a434e40f..b920471f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -18,7 +18,6 @@ members = [
     "exchanges/bitbank",
     "exchanges/bitflyer",
     "exchanges/bybit", 
-    "exchanges/binance", 
 ]
 
 resolver = "2"
@@ -28,7 +27,6 @@ name = "rbot"
 version = "0.3.2"
 edition = "2021"
 
-
 [workspace.dependencies]
 rbot = { path = "./rbot" }
 rbot_lib = { path = "./rbot_lib" }
@@ -67,12 +65,12 @@ strum_macros = { version = "0.26.0" }
 strum = { version = "0.26.0", features = ["derive"] }
 
 async-stream = { version = "0.3.5" }
-async-trait = { version = "0.1.77" }
 
 tokio = { version = "1", features = ["full"] }
 tokio-stream = { version = "0.1.14" }
+async-trait = {version="0.1.81"}
 
-rusqlite = { version = "0.31.0", features = ["bundled"] }
+rusqlite = { version = "0.32.0", features = ["bundled"] }
     
 polars = { version = "0.41.3", features = [
         "lazy", 
diff --git a/exchanges/binance/src/market.rs b/exchanges/binance/src/market.rs
index 989dd1bb..cbf4d2f0 100644
--- a/exchanges/binance/src/market.rs
+++ b/exchanges/binance/src/market.rs
@@ -20,7 +20,7 @@ use rbot_lib::common::OrderBook;
 use rbot_lib::common::MARKET_HUB;
 use rbot_lib::common::{flush_log, LogStatus};
 use rbot_lib::common::{time_string, NOW};
-use rbot_lib::db::TradeTable;
+use rbot_lib::db::TradeDataFrame;
 use rbot_lib::net::{BroadcastMessage, RestApi};
 use rust_decimal::Decimal;
 // Copyright(c) 2022-2024. yasstake. All rights reserved.
@@ -38,6 +38,8 @@ use crate::BinanceServerConfig;
 
 use pyo3::prelude::*;
 
+use anyhow::anyhow;
+
 pub const BINANCE:&str = "BINANCE";
 
 #[pyclass]
@@ -193,7 +195,7 @@ impl OrderInterfaceImpl<BinanceRestApi, BinanceServerConfig> for Binance {
 pub struct BinanceMarket {
     server_config: BinanceServerConfig,
     config: MarketConfig,
-    pub db: Arc<Mutex<TradeTable>>,
+    pub db: Arc<Mutex<TradeDataFrame>>,
     pub board: Arc<RwLock<OrderBook>>,
     pub public_handler: Option<tokio::task::JoinHandle<()>>,
 }
@@ -319,15 +321,19 @@ impl BinanceMarket {
         MarketImpl::_repr_html_(self)
     }
 
-    #[pyo3(signature = (ndays, force=false, verbose=false, low_priority=true))]
+    #[pyo3(signature = (ndays, force=false, verbose=false))]
     fn download_archive(
         &mut self,
         ndays: i64,
         force: bool,
         verbose: bool,
-        low_priority: bool,
     ) -> anyhow::Result<i64> {
-        MarketImpl::download_archives(self, ndays, force, verbose, low_priority)
+        BLOCK_ON(async{
+            
+            
+            //MarketImpl::async_download_archives(self, ndays, force, verbose).await
+            Err(anyhow!("not implemented"))
+        })
     }
 
     #[pyo3(signature = (verbose=false))]
@@ -395,7 +401,7 @@ impl MarketImpl<BinanceRestApi, BinanceServerConfig> for BinanceMarket {
         BLOCK_ON(async { self.async_download_latest(verbose).await })
     }
 
-    fn get_db(&self) -> Arc<Mutex<TradeTable>> {
+    fn get_db(&self) -> Arc<Mutex<TradeDataFrame>> {
         self.db.clone()
     }
 
@@ -421,19 +427,6 @@ impl MarketImpl<BinanceRestApi, BinanceServerConfig> for BinanceMarket {
             self.async_download_gap(force, verbose).await
         })
     }
-
-    fn download_archives(
-        &mut self,
-        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 {
@@ -442,14 +435,14 @@ impl BinanceMarket {
         config: &MarketConfig,
         production: bool,
     ) -> anyhow::Result<Self> {
-        let db_path = TradeTable::make_db_path(
+        let db_path = TradeDataFrame::make_db_path(
             &config.exchange_name,
             &config.trade_category,
             &config.trade_symbol,
             production
         );
 
-        let db = TradeTable::get(&db_path)
+        let db = TradeDataFrame::get(&db_path)
             .with_context(|| format!("Error in TradeTable::open: {:?}", db_path))?;
 
         // let public_ws = BinancePublicWsClient::new(&server_config, &config).await;
diff --git a/exchanges/bitflyer/src/market.rs b/exchanges/bitflyer/src/market.rs
index 868ace4b..1f2a7c82 100644
--- a/exchanges/bitflyer/src/market.rs
+++ b/exchanges/bitflyer/src/market.rs
@@ -985,50 +985,31 @@ impl BitflyerMarket {
 
     #[getter]
     pub fn get_order_status(&self) -> PyResult<Vec<Order>> {
-        let _status = order_status(&self.server_config.rest_server, &self.config);
+        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",
-        ));
+        convert_pyresult(status)
     }
 
     #[getter]
     pub fn get_open_orders(&self) -> PyResult<Vec<Order>> {
-        /*
         let status = open_orders(&self.server_config, &self.config);
 
         log::debug!("OpenOrder: {:?}", status);
 
-        // convert_pyresult_vec(status)
-        // TODO: implement
-        */
-        return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
-            "Not implemented",
-        ));
+        convert_pyresult_vec(status)
     }
 
     #[getter]
     pub fn get_trade_list(&self) -> PyResult<Vec<Order>> {
-        /*
         let status = trade_list(&self.server_config.rest_server, &self.config);
-        */
-        //         convert_pyresult(status)
-        return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
-            "Not implemented",
-        ));
+        convert_pyresult(status)
     }
 
     #[getter]
     pub fn get_account(&self) -> PyResult<AccountStatus> {
         let status = get_balance(&self.server_config.rest_server, &self.config);
 
-        status.unwrap();
-
-        //convert_pyresult(status)
-        return Err(PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(
-            "Not implemented",
-        ));
+        convert_pyresult(status)
     }
 
     #[getter]
diff --git a/exchanges/bybit/src/config.rs b/exchanges/bybit/src/config.rs
index 0d1a74e0..11062ae0 100644
--- a/exchanges/bybit/src/config.rs
+++ b/exchanges/bybit/src/config.rs
@@ -2,6 +2,7 @@
 use std::env;
 
 use pyo3::prelude::*;
+use rust_decimal_macros::dec;
 use serde_derive::{Deserialize, Serialize};
 
 use rbot_lib::common::{FeeType, MarketConfig, PriceType, SecretString, ServerConfig};
@@ -45,8 +46,23 @@ impl BybitServerConfig {
         }
         .to_string();
 
-        let api_key = env::var("BYBIT_API_KEY").unwrap_or_default();
-        let api_secret = env::var("BYBIT_API_SECRET").unwrap_or_default();
+        let api_key = if let Ok(key) = env::var("BYBIT_API_KEY") {
+            key
+        }
+        else {
+            println!("API KEY environment variable [BYBIT_API_KEY] is not set");
+            log::warn!("API KEY environment variable [BYBIT_API_KEY] is not set");
+            "".to_string()
+        };
+
+        let api_secret = if let Ok(secret) = env::var("BYBIT_API_SECRET") {
+            secret
+        }
+        else {
+            println!("API SECRET environment variable [BYBIT_API_SECRET] is not set");
+            log::warn!("API SECRET environment variable [BYBIT_API_SECRET] is not set");
+            "".to_string()
+        };
 
         return BybitServerConfig {
             production,
@@ -109,42 +125,17 @@ impl BybitConfig {
         return BybitConfig {};
     }
 
-    /*
-    #[classattr]
-    pub fn SPOT_BTCUSDT() -> MarketConfig {
-        MarketConfig {
-            price_unit: dec![0.05],
-            price_scale: 3,
-            size_unit: dec![0.001],
-            size_scale: 4,
-            maker_fee: dec![0.00_01],
-            taker_fee: dec![0.00_01],
-            price_type: PriceType::Home,
-            fee_type: FeeType::Home,
-            home_currency: "USDT".to_string(),
-            foreign_currency: "BTC".to_string(),
-            market_order_price_slip: dec![0.01],
-            board_depth: 200,
-            trade_category: "spot".to_string(),
-            trade_symbol: "BTCUSDT".to_string(),
-            public_subscribe_channel: vec![
-                "publicTrade.BTCUSDT".to_string(),
-                "orderbook.200.BTCUSDT".to_string(),
-            ],
-        }
-    }
-    */
 
     #[classattr]
     pub fn BTCUSDT() -> MarketConfig {
         MarketConfig::new(
             BYBIT,
             "linear",
-            "USDT",
             "BTC",
-            1,
+            "USDT",
+            dec![0.1],
             PriceType::Home,
-            3,
+            dec![0.001],
             200,
             0.1,
             0.00_01,
@@ -162,11 +153,11 @@ impl BybitConfig {
         let mut config = MarketConfig::new(
             BYBIT,
             "linear",
-            "USDC",
             "BTC",
-            1,
+            "USDC",
+            dec![0.1],
             PriceType::Home,
-            3,
+            dec![0.001],
             200,
             0.1,
             0.00_01,
@@ -188,11 +179,11 @@ impl BybitConfig {
         let config = MarketConfig::new(
             BYBIT,
             "linear",
-            "USDT",            
             "ETH",
-            2,
+            "USDT",            
+            dec![0.01],
             PriceType::Home,
-            2,
+            dec![0.01],
             200,
             0.1,
             0.00_01,
@@ -212,11 +203,11 @@ impl BybitConfig {
         let mut config = MarketConfig::new(
             BYBIT,
             "linear",
-            "USDC",            
             "ETH",
-            2,
+            "USDC",            
+            dec![0.01],
             PriceType::Home,
-            2,
+            dec![0.01],
             200,
             0.1,
             0.00_01,
@@ -237,11 +228,11 @@ impl BybitConfig {
         MarketConfig::new(
             BYBIT,
             "linear",
-            "USDT",            
             "MNT",
-            1,
+            "USDT",            
+            dec![0.0001],
             PriceType::Home,
-            0,
+            dec![1],
             200,
             0.1,
             0.00_01,
@@ -254,17 +245,39 @@ impl BybitConfig {
         )
     }
 
+    #[classattr]
+    pub fn MNTUSDC() -> MarketConfig {
+        MarketConfig::new(
+            BYBIT,
+            "linear",
+            "MNT",
+            "USDT",            
+            dec![0.0001],
+            PriceType::Home,
+            dec![0.1],
+            200,
+            0.1,
+            0.00_01,
+            0.00_01,
+            FeeType::Home,
+            vec![
+                "publicTrade.MNT-PERP".to_string(),
+                "orderbook.200.MNT-PERP".to_string(),
+            ],
+        )
+    }
+
 
     #[classattr]
     pub fn SOLUSDT() -> MarketConfig {
         MarketConfig::new(
             BYBIT,
             "linear",
-            "USDT",            
             "SOL",
-            2,
+            "USDT",            
+            dec![0.0001],
             PriceType::Home,
-            1,
+            dec![0.1],
             200,
             0.1,
             0.00_01,
@@ -277,17 +290,39 @@ impl BybitConfig {
         )
     }
 
+    #[classattr]
+    pub fn SOLUSDC() -> MarketConfig {
+        MarketConfig::new(
+            BYBIT,
+            "linear",
+            "SOL",
+            "USDC",            
+            dec![0.01],
+            PriceType::Home,
+            dec![0.1],
+            200,
+            0.1,
+            0.00_01,
+            0.00_01,
+            FeeType::Home,
+            vec![
+                "publicTrade.SOLPERP".to_string(),
+                "orderbook.200.SOLPERP".to_string(),
+            ],
+        )
+    }
+
 
     #[classattr]
     pub fn USDCUSDT() -> MarketConfig {
         MarketConfig::new(
             BYBIT,
             "linear",
-            "USDT",            
             "USDC",
-            4,
+            "USDT",            
+            dec![0.0001],
             PriceType::Home,
-            1,
+            dec![0.1],
             200,
             0.1,
             0.00_01,
@@ -330,11 +365,11 @@ mod test_bybit_config {
         let new_config = MarketConfig::new(
             BYBIT,
             "linear",
-            "USDT",
             "BTC",
-            1,
+            "USDT",
+            dec![0.1],
             PriceType::Home,
-            3,
+            dec![0.001],
             200,
             0.1,
             0.00_01,
diff --git a/exchanges/bybit/src/market.rs b/exchanges/bybit/src/market.rs
index 4bf9047c..7cc29b7a 100644
--- a/exchanges/bybit/src/market.rs
+++ b/exchanges/bybit/src/market.rs
@@ -24,7 +24,7 @@ use rbot_lib::common::{
     ServerConfig, Trade, DAYS, FLOOR_DAY, HHMM, MARKET_HUB, NOW, SEC,
 };
 
-use rbot_lib::db::{db_full_path, TradeTable, TradeTableDb, KEY};
+use rbot_lib::db::{db_full_path, TradeArchive, TradeDataFrame, TradeDb, KEY};
 use rbot_lib::net::{latest_archive_date, BroadcastMessage, RestApi, UdpSender};
 
 use rbot_market::MarketImpl;
@@ -51,16 +51,17 @@ use anyhow::Context;
 use rbot_blockon::BLOCK_ON;
 
 use tokio::task::JoinHandle;
+use anyhow::anyhow;
 
 pub const BYBIT: &str = "BYBIT";
 
 #[pyclass]
-#[derive(Debug)]
 pub struct Bybit {
     production: bool,
     enable_order: bool,
     server_config: BybitServerConfig,
     user_handler: Option<JoinHandle<()>>,
+    api: BybitRestApi
 }
 
 #[pymethods]
@@ -69,12 +70,14 @@ impl Bybit {
     #[pyo3(signature = (production=false))]
     pub fn new(production: bool) -> Self {
         let server_config = BybitServerConfig::new(production);
+        let api = BybitRestApi::new(&server_config);
 
         return Bybit {
             production: production,
             enable_order: false,
             server_config: server_config,
             user_handler: None,
+            api: api
         };
     }
 
@@ -90,11 +93,11 @@ impl Bybit {
     //--- OrderInterfaceImpl ----
     #[setter]
     pub fn set_enable_order_with_my_own_risk(&mut self, enable_order: bool) {
-        OrderInterfaceImpl::set_enable_order_feature(self, enable_order)
+        self.enable_order = enable_order;
     }
     #[getter]
     pub fn get_enable_order_with_my_own_risk(&self) -> bool {
-        OrderInterfaceImpl::get_enable_order_feature(self)
+        self.enable_order
     }
 
     pub fn limit_order(
@@ -145,7 +148,11 @@ impl Bybit {
     }
 }
 
-impl OrderInterfaceImpl<BybitRestApi, BybitServerConfig> for Bybit {
+impl OrderInterfaceImpl<BybitRestApi> for Bybit {
+    fn get_restapi(&self) -> &BybitRestApi {
+        &self.api
+    }    
+
     fn set_enable_order_feature(&mut self, enable_order: bool) {
         self.enable_order = enable_order;
     }
@@ -154,10 +161,6 @@ impl OrderInterfaceImpl<BybitRestApi, BybitServerConfig> for Bybit {
         self.enable_order
     }
 
-    fn get_server_config(&self) -> &BybitServerConfig {
-        &self.server_config
-    }
-
     async fn async_start_user_stream(&mut self) -> anyhow::Result<()> {
         let exchange_name = BYBIT.to_string();
         let server_config = self.server_config.clone();
@@ -205,14 +208,16 @@ impl OrderInterfaceImpl<BybitRestApi, BybitServerConfig> for Bybit {
 
         Ok(())
     }
+    
+    
 }
 
-#[derive(Debug)]
 #[pyclass]
 pub struct BybitMarket {
     pub server_config: BybitServerConfig,
+    pub api: BybitRestApi,
     pub config: MarketConfig,
-    pub db: Arc<Mutex<TradeTable>>,
+    pub db: Arc<Mutex<TradeDataFrame>>,
     pub board: Arc<RwLock<OrderBook>>,
     pub public_handler: Option<tokio::task::JoinHandle<()>>,
     // pub user_ws: WebSocketClient<BybitServerConfig, BybitWsOpMessage>,
@@ -235,23 +240,6 @@ impl BybitMarket {
     fn get_config(&self) -> MarketConfig {
         MarketImpl::get_config(self)
     }
-    #[getter]
-    fn get_exchange_name(&self) -> String {
-        MarketImpl::get_exchange_name(self)
-    }
-    #[getter]
-    fn get_trade_category(&self) -> String {
-        MarketImpl::get_trade_category(self)
-    }
-
-    #[getter]
-    fn get_trade_symbol(&self) -> String {
-        MarketImpl::get_trade_symbol(self)
-    }
-
-    fn drop_table(&mut self) -> anyhow::Result<()> {
-        MarketImpl::drop_table(self)
-    }
 
     #[getter]
     fn get_cache_duration(&self) -> MicroSec {
@@ -262,11 +250,6 @@ 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)
@@ -330,33 +313,20 @@ impl BybitMarket {
         MarketImpl::get_edge_price(self)
     }
 
-    #[getter]
-    fn get_file_name(&self) -> String {
-        MarketImpl::get_file_name(self)
-    }
-
-    #[getter]
-    fn get_running(&self) -> bool {
-        MarketImpl::get_running(self)
-    }
-
-    fn vacuum(&self) {
-        MarketImpl::vacuum(self)
-    }
-
     fn _repr_html_(&self) -> String {
         MarketImpl::_repr_html_(self)
     }
 
-    #[pyo3(signature = (ndays, force=false, verbose=false, low_priority=true))]    
+    #[pyo3(signature = (ndays, force=false, verbose=false))]    
     fn download_archive(
         &mut self,
         ndays: i64,
         force: bool,
         verbose: bool,
-        low_priority: bool,
     ) -> anyhow::Result<i64> {
-            MarketImpl::download_archives(self, ndays, force, verbose, low_priority)
+        BLOCK_ON(async{
+            MarketImpl::async_download_archive(self, ndays, force, verbose).await
+        })
     }
 
     #[pyo3(signature = (verbose=false))]    
@@ -399,20 +369,15 @@ impl BybitMarket {
         config: &MarketConfig,
         production: bool,
     ) -> anyhow::Result<Self> {
-        let db_path = TradeTable::make_db_path(
-            &config.exchange_name,
-            &config.trade_category,
-            &config.trade_symbol,
-            production
-        );
 
-        let db = TradeTable::get(&db_path)
-            .with_context(|| format!("Error in TradeTable::open: {:?}", db_path))?;
+        let db = TradeDataFrame::get(config, server_config.production)
+            .with_context(|| format!("Error in TradeTable::open: {:?}", config))?;
 
         let public_ws = BybitPublicWsClient::new(&server_config, &config).await;
 
         let mut market = BybitMarket {
             server_config: server_config.clone(),
+            api: BybitRestApi::new(server_config),
             config: config.clone(),
             db: db,
             board: Arc::new(RwLock::new(OrderBook::new(&config))),
@@ -423,29 +388,17 @@ impl BybitMarket {
     }
 }
 
-impl MarketImpl<BybitRestApi, BybitServerConfig> for BybitMarket {
-    fn get_config(&self) -> MarketConfig {
-        self.config.clone()
-    }
-
-    fn get_server_config(&self) -> BybitServerConfig {
-        self.server_config.clone()
+impl MarketImpl<BybitRestApi> for BybitMarket {
+    fn get_restapi(&self) -> &BybitRestApi {
+        &self.api
     }
 
-    fn get_exchange_name(&self) -> String {
-        self.config.exchange_name.clone()
-    }
-
-    fn get_trade_category(&self) -> String {
-        self.config.trade_category.clone()
-    }
-
-    fn get_trade_symbol(&self) -> String {
-        self.config.trade_symbol.clone()
+    fn get_config(&self) -> MarketConfig {
+        self.config.clone()
     }
 
     fn download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
-        BLOCK_ON(async { self.async_download_latest(verbose).await })
+       BLOCK_ON(async { self.async_download_latest(verbose).await })
     }
 
     fn download_gap(&mut self, force: bool, verbose: bool) -> anyhow::Result<i64> {
@@ -453,7 +406,7 @@ impl MarketImpl<BybitRestApi, BybitServerConfig> for BybitMarket {
     }
 
 
-    fn get_db(&self) -> Arc<Mutex<TradeTable>> {
+    fn get_db(&self) -> Arc<Mutex<TradeDataFrame>> {
         self.db.clone()
     }
 
@@ -476,19 +429,6 @@ impl MarketImpl<BybitRestApi, BybitServerConfig> for BybitMarket {
             self.async_start_market_stream().await
         })
     }
-    
-    fn download_archives(
-        &mut self,
-        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 BybitMarket {
@@ -500,7 +440,7 @@ impl BybitMarket {
 
         let db_channel = {
             let mut lock = self.db.lock().unwrap();
-            lock.start_thread().await
+            lock.start_thread()?
         };
 
         let orderbook = self.board.clone();
@@ -609,8 +549,8 @@ impl BybitMarket {
             return Ok(0);
         }
 
-        let klines = BybitRestApi::get_trade_klines(
-            &self.server_config,
+        let api = self.get_restapi();
+        let klines = api.get_trade_klines(
             &self.get_config(),
             unfix_start,
             unfix_end,
@@ -619,11 +559,11 @@ impl BybitMarket {
         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 expire_message = TradeDb::expire_control_message(unfix_start, unfix_end, false);
 
         let tx = {
             let mut lock = self.db.lock().unwrap();
-            lock.start_thread().await
+            lock.start_thread()?
         };
 
         tx.send(expire_message)?;
@@ -638,7 +578,8 @@ impl BybitMarket {
     }
 
     async fn async_reflesh_order_book(&mut self) -> anyhow::Result<()> {
-        let board = BybitRestApi::get_board_snapshot(&self.server_config, &self.config).await?;
+        let api = self.get_restapi();
+        let board = api.get_board_snapshot(&self.config).await?;
         let mut b = self.board.write().unwrap();
         b.update(&board);
 
@@ -649,6 +590,7 @@ impl BybitMarket {
 
 #[cfg(test)]
 mod bybit_test {
+    use rbot_lib::common::init_debug_log;
     use rust_decimal_macros::dec;
 
     use crate::{Bybit, BybitConfig};
@@ -668,6 +610,7 @@ mod bybit_test {
 
     #[test]
     fn test_limit_order() {
+        init_debug_log();
         let mut bybit = Bybit::new(false);
         let config = BybitConfig::BTCUSDT();
 
@@ -686,6 +629,8 @@ mod bybit_test {
         let mut bybit = Bybit::new(false);
         let config = BybitConfig::BTCUSDT();
 
+        init_debug_log();
+
         let rec = bybit.market_order(&config, "Buy", dec![0.001], None);
         println!("{:?}", rec);
         assert!(rec.is_err()); // first enable flag.
@@ -752,15 +697,16 @@ mod market_test {
     }
 
     #[ignore]
-    #[tokio::test]
-    async fn test_download_archive() {
+    #[test]
+    fn test_download_archive() {
         use super::*;
         let server_config = BybitServerConfig::new(false);
         let market_config = BybitConfig::BTCUSDT();
 
+        init_debug_log();
         let mut market = BybitMarket::new(&server_config, &market_config, true);
 
-        let rec = market.async_download_archives(1, true, true, false).await;
+        let rec = market.download_archive(3, false, true);
         assert!(rec.is_ok());
     }
 
diff --git a/exchanges/bybit/src/rest.rs b/exchanges/bybit/src/rest.rs
index f97ff403..ea92a87e 100644
--- a/exchanges/bybit/src/rest.rs
+++ b/exchanges/bybit/src/rest.rs
@@ -4,16 +4,26 @@
 use std::convert;
 use std::fmt::format;
 
+use chrono::Datelike as _;
 use csv::StringRecord;
+use polars::chunked_array::ops::ChunkCast as _;
+use polars::datatypes::DataType;
 use polars::export::num::FromPrimitive;
 use polars::frame::DataFrame;
+use polars::series::Series;
+use rbot_lib::common::split_yyyymmdd;
 use rbot_lib::common::time_string;
+use rbot_lib::common::to_naive_datetime;
 use rbot_lib::common::AccountCoins;
 use rbot_lib::common::AccountPair;
 use rbot_lib::common::BoardTransfer;
 use rbot_lib::common::Kline;
 use rbot_lib::common::LogStatus;
-use rbot_lib::db::TradeTable;
+use rbot_lib::db::ohlcv_end;
+use rbot_lib::db::ohlcv_start;
+use rbot_lib::db::TradeDataFrame;
+use rbot_lib::db::KEY;
+use rbot_lib::net::check_exist;
 use rust_decimal_macros::dec;
 use serde_derive::Deserialize;
 use serde_derive::Serialize;
@@ -83,13 +93,24 @@ struct CancelOrderMessage {
     order_id: String,
 }
 
-pub struct BybitRestApi {}
 
-impl RestApi<BybitServerConfig> for BybitRestApi {
-    async fn get_board_snapshot(
-        server: &BybitServerConfig,
-        config: &MarketConfig,
-    ) -> anyhow::Result<BoardTransfer> {
+
+pub struct BybitRestApi {
+    server_config: BybitServerConfig
+}
+
+impl BybitRestApi {
+    pub fn new(server_config: &BybitServerConfig) -> Self {
+        Self {
+            server_config: server_config.clone()
+        }
+    }
+}
+
+impl RestApi for BybitRestApi {
+    async fn get_board_snapshot(&self, config: &MarketConfig) -> anyhow::Result<BoardTransfer> {
+        let server = &self.server_config;
+
         let path = "/v5/market/orderbook";
 
         let params = format!(
@@ -99,7 +120,7 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
             config.board_depth
         );
 
-        let r = Self::get(&server, path, &params).await.with_context(|| {
+        let r = Self::get(server, path, &params).await.with_context(|| {
             format!(
                 "get_board_snapshot: server={:?} / path={:?} / params={:?}",
                 server, path, params
@@ -114,10 +135,9 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
         Ok(result.into())
     }
 
-    async fn get_recent_trades(
-        server: &BybitServerConfig,
-        config: &MarketConfig,
-    ) -> anyhow::Result<Vec<Trade>> {
+    async fn get_recent_trades(&self, config: &MarketConfig) -> anyhow::Result<Vec<Trade>> {
+        let server = &self.server_config;
+
         let path = "/v5/market/recent-trade";
 
         let params = format!(
@@ -138,24 +158,24 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
             .with_context(|| format!("parse error in get_recent_trades"))?;
 
         Ok(result.into())
-
-        //        Err(anyhow!("not implemented"))
     }
 
     async fn get_trade_klines(
-        server: &BybitServerConfig,
+        &self,
         config: &MarketConfig,
         start_time: MicroSec,
         end_time: MicroSec,
     ) -> anyhow::Result<Vec<Kline>> {
+        let server = &self.server_config;
+
         let mut klines_buf: Vec<Kline> = vec![];
 
-        let start_time = TradeTable::ohlcv_start(start_time);
+        let start_time = ohlcv_start(start_time);
 
         let mut end_time = if end_time == 0 {
-            TradeTable::ohlcv_end(NOW()) - 1
+            ohlcv_end(NOW()) - 1
         } else {
-            TradeTable::ohlcv_end(end_time) - 1
+            ohlcv_end(end_time) - 1
         };
 
         loop {
@@ -207,7 +227,7 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
     }
 
     async fn new_order(
-        server: &BybitServerConfig,
+        &self,
         config: &MarketConfig,
         side: OrderSide,
         price: Decimal, // when order_type is Market, this value is ignored.
@@ -215,6 +235,8 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
         order_type: OrderType,
         client_order_id: Option<&str>,
     ) -> anyhow::Result<Vec<Order>> {
+        let server = &self.server_config;
+
         let category = config.trade_category.clone();
         let symbol = config.trade_symbol.clone();
 
@@ -296,16 +318,18 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
             total_profit: dec![0.0],
         };
 
-        order.update_balance(config);
+        order.update_balance(&config);
 
         return Ok(vec![order]);
     }
 
     async fn cancel_order(
-        server: &BybitServerConfig,
+        &self,
         config: &MarketConfig,
         order_id: &str,
     ) -> anyhow::Result<Order> {
+        let server = &self.server_config;
+
         let category = config.trade_category.clone();
         let message = CancelOrderMessage {
             category: category.clone(),
@@ -369,11 +393,12 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
         return Ok(order);
     }
 
-    // TODO: implement paging.
     async fn open_orders(
-        server: &BybitServerConfig,
+        &self,
         config: &MarketConfig,
     ) -> anyhow::Result<Vec<Order>> {
+        let server = &self.server_config;
+
         let query_string = format!(
             "category={}&symbol={}&limit=50",
             config.trade_category, config.trade_symbol
@@ -407,7 +432,11 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
         Ok(orders)
     }
 
-    async fn get_account(server: &BybitServerConfig) -> anyhow::Result<AccountCoins> {
+    async fn get_account(
+        &self,
+    ) -> anyhow::Result<AccountCoins> {
+        let server = &self.server_config;
+
         let path = "/v5/account/wallet-balance";
         // TODO: implement otherthn accountType=UNIFIED(e.g.inverse)
         let query_string = format!("accountType=UNIFIED");
@@ -439,124 +468,59 @@ impl RestApi<BybitServerConfig> for BybitRestApi {
         Ok(coins)
     }
 
-    fn format_historical_data_url(
-        history_web_base: &str,
-        category: &str,
-        symbol: &str,
-        yyyy: i64,
-        mm: i64,
-        dd: i64,
-    ) -> String {
+    fn history_web_url(&self, config: &MarketConfig, date: MicroSec) -> String {
+        let web_base = self.server_config.get_historical_web_base();
+
+        let (yyyy, mm, dd) = split_yyyymmdd(date);
+
         format!(
             "{}/trading/{}/{}{:04}-{:02}-{:02}.csv.gz",
-            history_web_base, symbol, symbol, yyyy, mm, dd
+            web_base, config.trade_symbol, config.trade_symbol, yyyy, mm, dd
         )
     }
 
-    /// 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;
+    fn archive_has_header(&self) -> bool {
+        true
     }
 
-    /// 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    
+    async fn has_archive(&self, config: &MarketConfig, date: MicroSec) -> anyhow::Result<bool> {
+        let url = self.history_web_url(config, date);
 
-    fn line_to_trade(line: &str) -> Trade {
-        let col: Vec<&str> = line.split(",").collect();
+        let result = check_exist(&url).await?;
 
-        let timestamp: f64 = col[0].parse().unwrap();
-        let timestamp = timestamp * 1_000_000.0;
-        let timestamp = timestamp as MicroSec;
-
-        let order_side = col[2];
-        let order_side = OrderSide::from(order_side);
-
-        let size: f64 = col[3].parse().unwrap();
-        let size = Decimal::from_f64(size).unwrap();
+        Ok(result)
+    }
 
-        let price: f64 = col[4].parse().unwrap();
-        let price = Decimal::from_f64(price).unwrap();
+    /// create DataFrame with columns;
+    ///  KEY:time_stamp(Int64), KEY:order_side(Bool), KEY:price(f64), KEY:size(f64)
+    fn logdf_to_archivedf(&self, df: &DataFrame) -> anyhow::Result<DataFrame> {
+        let df = df.clone();
 
-        let id = col[6];
+        let timestamp = df.column("timestamp")?.f64()? * 1_000_000.0;
+        let timestamp = timestamp.cast(&DataType::Int64)?;
 
-        let trade = Trade::new(
-            timestamp,
-            order_side,
-            price,
-            size,
-            LogStatus::FixArchiveBlock,
-            id,
-        );
+        let timestamp = timestamp.clone();
+        let mut timestamp = Series::from(timestamp.clone());
+        timestamp.rename(KEY::time_stamp);
 
-        return trade;
-    }
+        let mut id = df.column("trdMatchID")?.clone();
+        id.rename(KEY::id);
 
-    fn convert_archive_line(line: &str) -> String {
-        let col: Vec<&str> = line.split(",").collect();
+        let mut side = df.column("side")?.clone();
+        side.rename(KEY::order_side);
 
-        let timestamp: f64 = col[0].parse().unwrap();
-        let timestamp = timestamp * 1_000_000.0;
-        let timestamp = timestamp as MicroSec;
+        let mut price = df.column("price")?.clone();
+        price.rename(KEY::price);
 
-        let order_side = col[2];
-        let size = col[3];
-        let price = col[4];
-        let id = col[6];
+        let mut size = df.column("size")?.clone();
+        size.rename(KEY::size);
 
-        format!("{},{},{},{},{}", timestamp, order_side, size, price, id)
-    }
+        let df = DataFrame::new(vec![timestamp, side, price, size, id])?;
 
-    fn archive_has_header() -> bool {
-        true
+        Ok(df)
     }
 }
 
-use tracing::instrument;
-
 impl BybitRestApi {
     async fn get(
         server: &BybitServerConfig,
@@ -726,7 +690,9 @@ mod bybit_rest_test {
         let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
 
-        let r = BybitRestApi::get_board_snapshot(&server_config, &config).await?;
+        let api = BybitRestApi::new(&server_config);
+
+        let r = api.get_board_snapshot(&config).await?;
         println!("{:?}", r);
 
         Ok(())
@@ -736,8 +702,9 @@ mod bybit_rest_test {
     async fn test_recent_trades() {
         let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
+        let api = BybitRestApi::new(&server_config);
 
-        let r = BybitRestApi::get_recent_trades(&server_config, &config).await;
+        let r = api.get_recent_trades(&config).await;
         assert!(r.is_ok());
         println!("{:?}", r);
     }
@@ -747,11 +714,12 @@ mod bybit_rest_test {
         init_debug_log();
         let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
+        let api = BybitRestApi::new(&server_config);
 
         let now = NOW();
         let start_time = now - HHMM(24, 0);
 
-        let r = BybitRestApi::get_trade_klines(&server_config, &config, start_time, now).await;
+        let r = api.get_trade_klines(&config, start_time, now).await;
         assert!(r.is_ok());
 
         let r = r.unwrap();
@@ -768,17 +736,18 @@ mod bybit_rest_test {
     async fn test_new_limit_order() {
         let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
-
-        let r = BybitRestApi::new_order(
-            &server_config,
-            &config,
-            OrderSide::Buy,
-            dec![40000.0],
-            dec![0.001],
-            OrderType::Limit,
-            None,
-        )
-        .await;
+        let api = BybitRestApi::new(&server_config);
+
+        let r = api
+            .new_order(
+                &config,
+                OrderSide::Buy,
+                dec![40000.0],
+                dec![0.001],
+                OrderType::Limit,
+                None,
+            )
+            .await;
 
         assert!(r.is_ok());
         println!("{:?}", r);
@@ -788,17 +757,18 @@ mod bybit_rest_test {
     async fn test_new_market_order() {
         let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
-
-        let r = BybitRestApi::new_order(
-            &server_config,
-            &config,
-            OrderSide::Sell,
-            dec![0.0],
-            dec![0.001],
-            OrderType::Market,
-            None,
-        )
-        .await;
+        let api = BybitRestApi::new(&server_config);
+
+        let r = api
+            .new_order(
+                &config,
+                OrderSide::Sell,
+                dec![0.0],
+                dec![0.001],
+                OrderType::Market,
+                None,
+            )
+            .await;
 
         assert!(r.is_ok());
         println!("{:?}", r);
@@ -808,22 +778,23 @@ mod bybit_rest_test {
     async fn test_cancel_order() {
         let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
-
-        let r = BybitRestApi::new_order(
-            &server_config,
-            &config,
-            OrderSide::Buy,
-            dec![40000.0],
-            dec![0.001],
-            OrderType::Limit,
-            None,
-        )
-        .await;
+        let api = BybitRestApi::new(&server_config);
+
+        let r = api
+            .new_order(
+                &config,
+                OrderSide::Buy,
+                dec![40000.0],
+                dec![0.001],
+                OrderType::Limit,
+                None,
+            )
+            .await;
 
         assert!(r.is_ok());
         println!("{:?}", r);
 
-        let r = BybitRestApi::cancel_order(&server_config, &config, &r.unwrap()[0].order_id).await;
+        let r = api.cancel_order(&config, &r.unwrap()[0].order_id).await;
 
         assert!(r.is_ok());
         println!("{:?}", r);
@@ -835,8 +806,9 @@ mod bybit_rest_test {
 
         let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
+        let api = BybitRestApi::new(&server_config);
 
-        let r = BybitRestApi::open_orders(&server_config, &config).await;
+        let r = api.open_orders(&config).await;
 
         assert!(r.is_ok());
         println!("{:?}", r);
@@ -847,19 +819,10 @@ mod bybit_rest_test {
         init_debug_log();
         let server_config = BybitServerConfig::new(false);
         let config = BybitConfig::BTCUSDT();
+        let api = BybitRestApi::new(&server_config);
 
-        let r = BybitRestApi::get_account(&server_config).await;
+        let r = api.get_account().await;
         println!("{:?}", r);
         assert!(r.is_ok());
     }
-
-    #[test]
-    fn test_line_2_trade() {
-        let line = "1620086396.8268,BTCUSDT,Buy,0.02,57199.5,ZeroMinusTick,224061a0-e105-508c-9696-b53ab4b5bb03,114399000000.0,0.02,1143.99";
-        let r = BybitRestApi::line_to_trade(&line.to_string());
-
-        init_debug_log();
-
-        log::debug!("{:?}", r);
-    }
 }
diff --git a/rbot/Cargo.toml b/rbot/Cargo.toml
index a6e6b244..00890c5f 100644
--- a/rbot/Cargo.toml
+++ b/rbot/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "rbot"
-version = "0.3.7"
+version = "0.4.0"
 edition = "2021"
 exclude = ["test/**"]
 
@@ -10,7 +10,7 @@ crate-type = ["cdylib", "rlib"]
 
 [dependencies]
 bybit = {path = "../exchanges/bybit"}
-binance = {path = "../exchanges/binance"}
+# binance = {path = "../exchanges/binance"}
 
 rbot_lib = {path="../rbot_lib"}
 rbot_session = {path="../rbot_session"}
diff --git a/rbot/src/lib.rs b/rbot/src/lib.rs
index 104fecc7..d9d850c4 100644
--- a/rbot/src/lib.rs
+++ b/rbot/src/lib.rs
@@ -11,7 +11,8 @@ use rbot_lib::{common::{
 
 use rbot_session::{Logger, Session, Runner, ExecuteMode};
 use bybit::{Bybit, BybitConfig};
-use binance::{Binance, BinanceConfig};
+
+// use binance::{Binance, BinanceConfig};
 
 
 // use console_subscriber;
@@ -65,8 +66,8 @@ fn rbot(m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<PriceType>()?;
 
     // Binance
-    m.add_class::<Binance>()?;
-    m.add_class::<BinanceConfig>()?;
+//    m.add_class::<Binance>()?;
+//    m.add_class::<BinanceConfig>()?;
     
     // ByBit
     m.add_class::<Bybit>()?;
diff --git a/rbot_lib/src/common/ch.rs b/rbot_lib/src/common/ch.rs
index af03cb48..a16faf33 100644
--- a/rbot_lib/src/common/ch.rs
+++ b/rbot_lib/src/common/ch.rs
@@ -118,211 +118,42 @@ impl MarketStream {
         let (sender, receiver) = bounded(CHANNEL_SIZE);
         (sender, Self { reciver: receiver })
     }
-}
-
-
-/*
-
-
-
-#[derive(Debug)]
-struct Channel<T> {
-    sender: Sender<T>,
-    valid: bool,
-}
-
-#[derive(Debug)]
-pub struct BroadcastChannel<T>
-where
-    T: Clone,
-{
-    channels: Vec<Channel<T>>,
-}
-
-impl<T> BroadcastChannel<T>
-where
-    T: Clone,
-{
-    pub fn new() -> Self {
-        Self {
-            channels: Vec::new(),
-        }
-    }
-
-    pub fn close(&mut self) {
-        loop {
-            match self.channels.pop() {
-                Some(channel) => drop(channel),
-                None => break,
-            }
-        }
-    }
-
-    pub fn add_channel(&mut self, channel: Sender<T>) {
-        self.channels.push(Channel {
-            sender: channel,
-            valid: true,
-        });
-    }
-
-    pub fn open_channel(&mut self, buffer_size: usize) -> Receiver<T> {
-        let (sender, receiver) = if buffer_size == 0 {
-            unbounded()
-        } else {
-            bounded(buffer_size)
-        };
-        self.add_channel(sender);
-
-        receiver
-    }
 
-    pub fn send(&mut self, message: T) -> Result<()> {
-        let mut has_error: bool = false;
+    pub fn recv(&self) -> anyhow::Result<MarketMessage> {
+        let r = self.reciver.recv()?;
 
-        for channel in self.channels.iter_mut() {
-            let result = channel.sender.send(message.clone());
-
-            if result.is_err() {
-                log::warn!("Send ERROR: {:?}. remove channel", result);
-                channel.valid = false;
-                has_error = true;
-            }
-        }
-
-        // remove invalid channels
-        if has_error {
-            log::warn!("Send ERROR: removing invalid channels");
-            self.channels.retain(|x| x.valid);
-        }
-
-        Ok(())
+        Ok(r)        
     }
 }
-*/
-/*
-#[cfg(test)]
-mod channel_test {
-    use rust_decimal_macros::dec;
-
-    use super::*;
-    use crate::common::{init_debug_log, init_log, LogStatus, OrderSide, OrderStatus};
 
-    #[test]
-    /// チャネルを開いて、メッセージを送信する。
-    /// 送信したメッセージと同じものが帰ってきているか確認する。
-    /// かつ、複数チャネルへ同じメッセージが来ていることを確認する。
-    fn test_channel() {
-        let mut channel = BroadcastChannel::new();
-        let receiver1 = channel.open_channel(0);
-        let receiver2 = channel.open_channel(0);
-
-        let trade = Trade::new(
-            1111,
-            OrderSide::Buy,
-            dec![1.0],
-            dec![2.0],
-            LogStatus::UnFix,
-            "ORDERID",
-        );
-
-        let message = MarketMessage::Trade(trade.clone());
-        channel.send(message.clone()).unwrap();
-
-        let result = receiver1.recv();
-        assert!(result.is_ok());
-
-        let result = result.unwrap();
-        match result {
-            MarketMessage::Trade(t) => {
-                assert_eq!(t, trade);
-            }
-            _ => {
-                assert!(false);
-            }
-        }
+#[cfg(test)]
+mod test_market_stream {
+    use crate::common::Trade;
 
-        let result = receiver2.recv();
-        match result.unwrap() {
-            MarketMessage::Trade(t) => {
-                assert_eq!(t, trade);
-            }
-            _ => {
-                assert!(false);
-            }
-        }
-    }
+    use super::{MarketMessage, MarketStream};
 
     #[test]
-    /// 4096個のメッセージを送信して、最後のメッセージが送信できるか確認する。
-    fn test_channel_full() {
-        // init_log();
+    fn test_market_stream() -> anyhow::Result<()>{
+        let (sender, ms) = MarketStream::open();
 
-        let mut channel = BroadcastChannel::new();
-        let _receiver = channel.open_channel(0);
+        let mut trade = Trade::default();
+        trade.id = "test".to_string();
+        let message1 = MarketMessage::from_trade(trade);
+        sender.send(message1.clone())?;
 
-        for i in 0..4096 {
-            let message = MarketMessage::Message(format!("test message {:?}", i));
-            channel.send(message.clone()).unwrap();
-        }
+        let mut trade = Trade::default();
+        trade.id = "test".to_string();
+        let message2 = MarketMessage::from_trade(trade);
+        sender.send(message2.clone())?;
 
-        let message = MarketMessage::Message("LAST test message 4096".to_string());
+        let r = ms.reciver.recv()?;
+        assert_eq!(message1, r);
 
-        let result = channel.send(message.clone());
-        assert_eq!(result.is_ok(), true);
-    }
+        let r = ms.reciver.recv()?;
+        assert_eq!(message2, r);
 
-    #[test]
-    fn test_channel_disconnect() {
-        //init_log();
-        let mut channel = BroadcastChannel::new();
-        let receiver = channel.open_channel(0);
-
-        let message = MarketMessage::Message("test message1".to_string());
-        channel.send(message.clone()).unwrap();
-        drop(receiver);
-
-        let message = MarketMessage::Message("test message2".to_string());
-        let result = channel.send(message.clone());
-        assert_eq!(result.is_ok(), true);
-
-        let receiver = channel.open_channel(0);
-        // send again, should be ok
-        let result = channel.send(message.clone());
-        let _m = receiver.recv().unwrap();
-        assert_eq!(result.is_ok(), true);
+        Ok(())
     }
-*/
-
-
-
-#[tokio::test]
-async fn test_handling_lag() {
-    use tokio::sync::broadcast;
-
-    let (tx, mut rx) = broadcast::channel(2);
-
-    // send mote than buffer size
-    tx.send(10).unwrap();
-
-    let mut rx2 = tx.subscribe();
 
-    tx.send(20).unwrap();
-    tx.send(30).unwrap();
 
-    // The receiver lagged behind
-    assert!(rx.recv().await.is_err());
-
-    // At this point, we can abort or continue with lost messages
-
-    assert_eq!(20, rx.recv().await.unwrap());
-    assert_eq!(30, rx.recv().await.unwrap());
-
-    assert_eq!(20, rx2.recv().await.unwrap());
-    assert_eq!(30, rx2.recv().await.unwrap());
-    println!("rx2: {:?}", rx2.recv().await);
-    println!("rx2: {:?}", rx2.recv().await);
-
-    for _ in 0..1000 {
-        tx.send(40).unwrap();
-    }
-}
+}
\ No newline at end of file
diff --git a/rbot_lib/src/common/config.rs b/rbot_lib/src/common/config.rs
index 60b74cb8..a55b9e57 100644
--- a/rbot_lib/src/common/config.rs
+++ b/rbot_lib/src/common/config.rs
@@ -6,12 +6,21 @@ use rusqlite::ffi::SQLITE_LIMIT_FUNCTION_ARG;
 use rust_decimal::{prelude::FromPrimitive, Decimal};
 use rust_decimal_macros::dec;
 use serde_derive::{Serialize, Deserialize};
-
-use crate::db::KEY::price;
-
+use zip::read::Config;
 use super::SecretString;
+use anyhow::anyhow;
 
 
+pub trait ServerConfig : Send + Sync {
+    fn get_historical_web_base(&self) -> String;
+    fn get_public_ws_server(&self) -> String;
+    fn get_user_ws_server(&self) -> String;
+    fn get_rest_server(&self) -> String;
+    fn get_api_key(&self) -> SecretString;
+    fn get_api_secret(&self) -> SecretString;
+    fn is_production(&self) -> bool;
+}
+
 #[pyclass]
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub enum FeeType {
@@ -39,13 +48,9 @@ pub struct MarketConfig {
 
     #[pyo3(set)]
     pub price_unit: Decimal,
-    #[pyo3(set)]    
-    pub price_scale: u32,
     
     #[pyo3(set)]
     pub size_unit: Decimal,
-    #[pyo3(set)]    
-    pub size_scale: u32,
 
     #[pyo3(set)]
     pub maker_fee: Decimal,
@@ -72,6 +77,20 @@ pub struct MarketConfig {
     pub public_subscribe_channel: Vec<String>,
 }
 
+fn round(unit: Decimal, value: Decimal) -> anyhow::Result<Decimal> {
+    let scale = unit.scale();
+    
+    let v = (value / unit).floor() * unit; // price_unitで切り捨て
+    let v = v.round_dp(scale);
+
+    if v == dec![0.0] {
+        log::warn!("Price or size becomes zero value= {} / unit= {} => {}", value, unit, v);
+        return Err(anyhow!("Price or size becomes zero value= {} / unit= {} => {}", value, unit, v))
+    }
+    Ok(v)
+}
+
+
 #[pymethods]
 impl MarketConfig {
     pub fn __repr__(&self) -> String {
@@ -82,15 +101,24 @@ impl MarketConfig {
         self.__repr__()
     }
 
+    pub fn round_price(&self, price: Decimal) -> anyhow::Result<Decimal> {
+        round(self.price_unit, price)
+    }
+
+    pub fn round_size(&self, size: Decimal) -> anyhow::Result<Decimal> {
+        round(self.size_unit, size)
+    }
+
+
     #[new]
     pub fn new(
         exchange_name: &str,
         trade_category: &str,
-        home_currency: &str,
         foreign_currency: &str,
-        price_scale: u32,
+        home_currency: &str,
+        price_unit: Decimal,
         price_type: PriceType,
-        size_scale: u32,
+        size_unit: Decimal,
         board_depth: u32,
         market_order_price_slip: f64,
         maker_fee: f64,
@@ -103,10 +131,8 @@ impl MarketConfig {
 
         Self {
             exchange_name: exchange_name.to_string(),
-            price_unit: Decimal::new(1, price_scale),
-            price_scale,
-            size_unit: Decimal::new(1, size_scale),
-            size_scale,
+            price_unit: price_unit,
+            size_unit: size_unit,
             maker_fee,
             taker_fee,
             price_type,
@@ -120,77 +146,80 @@ impl MarketConfig {
             public_subscribe_channel: public_subscribe_channel
         }
     }
-}
 
-impl MarketConfig {
-    pub fn new_bitflyer(
-        trade_category: &str,
-        home_currency: &str,
-        foreign_currency: &str,
-        price_scale: u32,
-        size_scale: u32,
-    ) -> Self {
-        let symbol = if trade_category == "FX" {
-            format!("FX_{}_{}", foreign_currency, home_currency)
+    pub fn key_string(&self, production: bool) -> String {
+        if production {
+            format!("{}/{}/{}", self.exchange_name, self.trade_category, self.trade_symbol)
         }
         else {
-            format!("{}_{}", foreign_currency, home_currency)
-        };
-
-        Self {
-            exchange_name: "DUMMY".to_string(),
-            price_unit: Decimal::new(1, price_scale),
-            price_scale,
-            size_unit: Decimal::new(1, size_scale),
-            size_scale,
-            maker_fee: dec![0.0], // dec![0.00_015],  // 0.015%
-            taker_fee: dec![0.0], // dec![0.00_015],  // 0.015%
-            price_type: PriceType::Foreign,
-            fee_type: FeeType::Home,
-            home_currency: home_currency.to_string(),
-            foreign_currency: foreign_currency.to_string(),
-            market_order_price_slip: dec![0.0],
-            board_depth: 1000,
-            trade_category: trade_category.to_string(),
-            trade_symbol: symbol,
-            public_subscribe_channel: vec![],
+            format!("{}/{}/{}/test", self.exchange_name, self.trade_category, self.trade_symbol)
         }
     }
 }
 
 impl Default for MarketConfig {
     fn default() -> Self {
-        Self {
-            exchange_name: "DUMMY".to_string(),
-            trade_category: "DUMMY".to_string(),
-            trade_symbol: "DUMMY".to_string(),
-            price_unit: dec![0.01],
-            price_scale: 2,
-            size_unit: dec![1000.0],
-            size_scale: 0,
-            maker_fee: dec![0.0], // dec![0.00_015],  // 0.015%
-            taker_fee: dec![0.0], // dec![0.00_015],  // 0.015%
-            price_type: PriceType::Foreign,
-            fee_type: FeeType::Home,
-            home_currency: "DUMMY".to_string(),
-            foreign_currency: "DUMMY".to_string(),
-            market_order_price_slip: dec![0.0],
-            board_depth: 1000,
-            public_subscribe_channel: vec![],
-        }
+        MarketConfig::new(
+            "default_exchange",
+            "default_trade_category",
+            "default_foreign_currency",
+            "default_home_currency",
+            Decimal::from_f64(0.01).unwrap(),
+            PriceType::Home,
+            Decimal::from_f64(0.01).unwrap(),
+            10,
+            0.0,
+            0.0,
+            0.0,
+            FeeType::Home,
+            vec![],
+        )
     }
-
 }
 
 
 
-pub trait ServerConfig : Send + Sync {
-    fn get_historical_web_base(&self) -> String;
-    fn get_public_ws_server(&self) -> String;
-    fn get_user_ws_server(&self) -> String;
-    fn get_rest_server(&self) -> String;
-    fn get_api_key(&self) -> SecretString;
-    fn get_api_secret(&self) -> SecretString;
-    fn is_production(&self) -> bool;
+
+#[cfg(test)]
+mod test_market_config {
+    use rust_decimal_macros::dec;
+
+    use crate::common::init_debug_log;
+
+    use super::MarketConfig;
+
+    #[test]
+    fn round_price() -> anyhow::Result<()>{
+        let mut config = MarketConfig::default();
+        config.price_unit = dec![0.5];
+
+        let price = dec![0.51];
+        let round = config.round_price(price)?;
+        assert_eq!(round, dec![0.5]);
+
+        let price = dec![0.6];
+        let round = config.round_price(price)?;
+        assert_eq!(round, dec![0.5]);
+
+        Ok(())
+    }
+
+    #[test]
+    fn round_size() -> anyhow::Result<()> {
+        init_debug_log();
+        let mut config = MarketConfig::default();
+        config.price_unit = dec![0.001];
+
+        // when trunc into zero, receive err.
+        let price = dec![0.0001];
+        let round = config.round_price(price);
+        assert!(round.is_err());
+
+        let price = dec![0.11];
+        let round = config.round_price(price)?;
+        assert_eq!(round, dec![0.11]);
+
+        Ok(())
+    }
 }
 
diff --git a/rbot_lib/src/common/env.rs b/rbot_lib/src/common/env.rs
index bf2abc13..cc9bcb2d 100644
--- a/rbot_lib/src/common/env.rs
+++ b/rbot_lib/src/common/env.rs
@@ -43,3 +43,4 @@ pub fn env_rbot_multicast_port() -> i64 {
 
     port.unwrap()
 }
+
diff --git a/rbot_lib/src/common/hub.rs b/rbot_lib/src/common/hub.rs
index a21ddf06..74ba039b 100644
--- a/rbot_lib/src/common/hub.rs
+++ b/rbot_lib/src/common/hub.rs
@@ -210,7 +210,7 @@ mod test_market_hub {
                 exchange: "a".to_string(),
                 category: "b".to_string(),
                 symbol: "c".to_string(),
-                msg: MarketMessage::make_message(&format!("message {}", i)),
+                msg: MarketMessage::make_message(&format!("{}", i)),
             };
             let r = tx.send(msg.clone());
             if r.is_err() {
@@ -220,9 +220,15 @@ mod test_market_hub {
 
         log::debug!("send message done");
 
-        for _i in 0..10 {
+        for i in 0..10 {
             if let Some(r) = rx.next().await {
-                println!("ch: m.essage: {:?}", r);
+                let m = r.unwrap();
+                if let MarketMessage::Message(message) = m {
+                    assert_eq!(message, i.to_string());
+                }
+                else {
+                    assert!(false);
+                }
             }
         }
     }
@@ -230,10 +236,10 @@ mod test_market_hub {
     #[test]
     fn test_market_hub() {
         init_debug_log();
+
         let tx = MARKET_HUB.open_channel();
-        let rx = MARKET_HUB.subscribe("a", "b", "c", "").unwrap();
-        let rx2 = MARKET_HUB.subscribe("a", "b", "c", "").unwrap();
 
+        // first make channel full
         for i in 0..CHANNEL_SIZE * 2 {
             let msg = BroadcastMessage {
                 exchange: "a".to_string(),
@@ -243,48 +249,64 @@ mod test_market_hub {
             };
             let r = tx.send(msg.clone());
             if r.is_err() {
-                println!("error: {:?}", r);
+                log::debug!("send error: {:?}", r);
+                assert!(false, "send error{:?}", r);
             }
         }
 
+        // then subscribe channel
+        let rx = MARKET_HUB.subscribe("a", "b", "c", "").unwrap();
+        let rx2 = MARKET_HUB.subscribe("a", "b", "c", "").unwrap();
 
+        // send 10 message
         for i in 10..20 {
             let msg = BroadcastMessage {
                 exchange: "a".to_string(),
                 category: "b".to_string(),
                 symbol: "c".to_string(),
-                msg: MarketMessage::make_message(&format!("message {}", i)),
+                msg: MarketMessage::make_message(&format!("{}", i)),
             };
 
             let r = tx.send(msg.clone());
             if r.is_err() {
-                println!("error: {:?}", r);
+                log::debug!("receive1 error: {:?}", r);
             }
         }
 
         log::debug!("-------rx1--------");
-
-        for _i in 10..20 {
+        for i in 10..20 {
             let r = rx.recv();
             if r.is_err() {
-                println!("error: {:?}", r);
+                log::debug!("error: {:?}", r);
             } else {
                 let r = r.unwrap();
-                println!("message: {:?}", r);
+                if let MarketMessage::Message(m) = r {
+                    assert_eq!(m, i.to_string());
+                    log::debug!("message: {:?}", m);
+                }
+                else {
+                    assert!(false, "unexpected message type");
+                }
             }
         }
 
         log::debug!("-------rx2--------");
-
-        for _i in 10..20 {
+        for i in 10..20 {
             let r = rx2.recv();
             if r.is_err() {
-                println!("error: {:?}", r);
+                log::debug!("error: {:?}", r);
             } else {
                 let r = r.unwrap();
-                println!("message: {:?}", r);
+                if let MarketMessage::Message(m) = r {
+                    assert_eq!(m, i.to_string());
+                    log::debug!("message: {:?}", m);
+                }
+                else {
+                    assert!(false, "unexpected message type");
+                }
             }
         }
+
     }
 
     #[tokio::test]
@@ -300,7 +322,7 @@ mod test_market_hub {
                 exchange: "a".to_string(),
                 category: "b".to_string(),
                 symbol: "c".to_string(),
-                msg: MarketMessage::make_message(&format!("message {}", i)),
+                msg: MarketMessage::make_message(&format!("{}", i)),
             };
             let r = tx.send(msg.clone());
             if r.is_err() {
diff --git a/rbot_lib/src/common/order.rs b/rbot_lib/src/common/order.rs
index 0c92a919..761214e7 100644
--- a/rbot_lib/src/common/order.rs
+++ b/rbot_lib/src/common/order.rs
@@ -1401,9 +1401,9 @@ mod order_tests {
             "SPOT",
             "BTC",
             "USDT",
-            2,
+            dec![0.01],
             PriceType::Home,
-            4,
+            dec![0.001],
             1000,
             0.0001,
             0.001,
diff --git a/rbot_lib/src/common/time.rs b/rbot_lib/src/common/time.rs
index 0f9394f6..1f017cdc 100644
--- a/rbot_lib/src/common/time.rs
+++ b/rbot_lib/src/common/time.rs
@@ -5,7 +5,7 @@
 
 use std::str::FromStr;
 
-use chrono::{DateTime, NaiveDate, NaiveDateTime, TimeZone, Utc};
+use chrono::{DateTime, Datelike as _, NaiveDate, NaiveDateTime, TimeZone, Utc};
 use pyo3::prelude::*;
 
 use anyhow::anyhow;
@@ -142,6 +142,17 @@ pub fn SEC(sec: i64) -> MicroSec {
     return sec * MICRO_SECOND as MicroSec;
 }
 
+pub fn split_yyyymmdd(t: MicroSec) -> (i64, i64, i64)
+{
+    let timestamp = to_naive_datetime(t);
+
+    let yyyy = timestamp.year() as i64;
+    let mm = timestamp.month() as i64;
+    let dd = timestamp.day() as i64;
+
+    (yyyy, mm, dd)
+}
+
 ///
 /// 現在時刻を返す(Microsecond)
 /// ```
@@ -236,4 +247,12 @@ mod time_test {
     
         Ok(())
     }
+
+    #[test]
+    fn test_split_yyyymmdd() {
+        let (yyyy, mm, dd)  = split_yyyymmdd(0);
+        assert_eq!(yyyy, 1970);
+        assert_eq!(mm, 1);
+        assert_eq!(dd, 1);
+    }
 }
diff --git a/rbot_lib/src/db/archive.rs b/rbot_lib/src/db/archive.rs
index 962e8894..51c4a122 100644
--- a/rbot_lib/src/db/archive.rs
+++ b/rbot_lib/src/db/archive.rs
@@ -1,37 +1,37 @@
 use crate::{
     common::{
-        date_string, parse_date, time_string, MarketConfig, MicroSec, OrderSide, ServerConfig, Trade, DAYS, FLOOR_DAY, MIN, NOW, TODAY
+        date_string, parse_date, time_string, MarketConfig, MicroSec, OrderSide, Trade, DAYS,
+        FLOOR_DAY, MIN, NOW, TODAY,
     },
-    db::{csv_to_df, df_to_parquet, merge_df, parquet_to_df, KEY},
-    net::{check_exist, log_download_tmp, read_csv_archive, RestApi},
+    db::{append_df, csv_to_df, df_to_parquet, parquet_to_df, KEY},
+    net::{check_exist, RestApi},
 };
 use anyhow::{anyhow, Context};
+use futures::StreamExt;
 use parquet::{
     file::reader::{FileReader, SerializedFileReader},
     record::RowAccessor,
 };
+use reqwest::Client;
 use rust_decimal::{prelude::FromPrimitive, Decimal};
+use tokio::io::AsyncWriteExt as _;
 // Import the `anyhow` crate and the `Result` type.
-use super::db_path_root;
+use super::{db_path_root, select_df};
 use polars::lazy::{
     dsl::{col, lit},
     frame::IntoLazy,
 };
 use polars::prelude::{DataFrame, NamedFrom};
 use polars::series::Series;
+
 use std::{
     fs::{self, File},
-    io::BufWriter,
-    path::PathBuf,
+    path::{Path, PathBuf},
     str::FromStr,
     vec,
 };
 use tempfile::tempdir;
 
-use parquet::record::Row;
-use std::sync::Arc;
-
-const PARQUET: bool = true;
 const EXTENSION: &str = "parquet";
 
 ///
@@ -50,35 +50,44 @@ const EXTENSION: &str = "parquet";
 /// log_df    ->   raw archvie file it may be different from exchanges.
 /// archive_df -> archvie file that is stored in the local directory
 /// chache_df -> df to use TradeTable's cache.
-#[derive(Debug)]
-pub struct TradeTableArchive<T, U>
-where
-    T: RestApi<U>,
-    U: ServerConfig + Clone,
-{
-    server: U,
+pub struct TradeArchive {
     config: MarketConfig,
+    production: bool,
     last_archive_check_time: MicroSec,
     latest_archive_date: MicroSec,
     start_time: MicroSec,
     end_time: MicroSec,
-    _dummy: Option<T>,
 }
 
-impl<T, U> TradeTableArchive<T, U>
-where
-    T: RestApi<U>,
-    U: ServerConfig + Clone,
-{
-    pub fn new(server: &U, config: &MarketConfig) -> Self {
+impl Clone for TradeArchive {
+    fn clone(&self) -> Self {
+        let mut archive = Self {
+            config: self.config.clone(),
+            production: self.production.clone(),
+            last_archive_check_time: self.last_archive_check_time.clone(),
+            latest_archive_date: self.latest_archive_date.clone(),
+            start_time: self.start_time.clone(),
+            end_time: self.end_time.clone(),
+        };
+
+        let r = archive.analyze();
+        if r.is_err() {
+            log::warn!("trade archive analyze error");
+        }
+
+        archive
+    }
+}
+
+impl TradeArchive {
+    pub fn new(config: &MarketConfig, production: bool) -> Self {
         let mut my = Self {
-            server: server.clone(),
             config: config.clone(),
+            production: production,
             last_archive_check_time: 0,
             latest_archive_date: 0,
             start_time: 0,
             end_time: 0,
-            _dummy: None,
         };
 
         let r = my.analyze();
@@ -86,62 +95,34 @@ where
             log::debug!("Archive analyze error {:?}", r);
         }
 
-        log::debug!("S: {} ({})", my.start_time(), time_string(my.start_time()));
-        log::debug!("E: {} ({})", my.end_time(), time_string(my.end_time()));
+        log::debug!(
+            "S: {:?} ({:?})",
+            my.start_time(),
+            time_string(my.start_time().unwrap_or_default())
+        );
+        log::debug!(
+            "E: {:?} ({:?})",
+            my.end_time(),
+            time_string(my.end_time().unwrap_or_default())
+        );
 
         return my;
     }
 
-    pub fn start_time(&self) -> MicroSec {
-        self.start_time
-    }
-
-    pub fn end_time(&self) -> MicroSec {
-        self.end_time
-    }
-
-    /// check the lates date in archive web site
-    /// check the latest check time, within 60 min call this function, reuse cache value.
-    pub async fn latest_archive_date(&mut self) -> anyhow::Result<MicroSec> {
-        let now = NOW();
-        if now - self.last_archive_check_time < MIN(60) {
-            return Ok(self.latest_archive_date);
+    pub fn start_time(&self) -> anyhow::Result<MicroSec> {
+        if self.start_time == 0 {
+            return Err(anyhow!("no data in archive"));
         }
-        self.last_archive_check_time = now;
 
-        let mut latest = TODAY();
-        let mut i = 0;
-
-        loop {
-            log::debug!("check log exist = {}({})", time_string(latest), latest);
-
-            if self.has_web_archive(latest).await? {
-                self.latest_archive_date = latest;
-                return Ok(latest);
-            }
-            latest -= DAYS(1);
-            i += 1;
-            if 5 < i {
-                return Err(anyhow!(
-                    "Find archive retry over {}/{}/{}",
-                    i,
-                    latest,
-                    time_string(latest)
-                ));
-            }
-        }
+        Ok(self.start_time)
     }
 
-    /// check if achive date is avairable at specified date
-    async fn has_web_archive(&self, date: MicroSec) -> anyhow::Result<bool> {
-        let url = T::history_web_url(&self.server, &self.config, date);
-        let result = check_exist(url.as_str()).await;
-
-        if result.is_err() {
-            return Ok(false);
+    pub fn end_time(&self) -> anyhow::Result<MicroSec> {
+        if self.end_time == 0 {
+            return Err(anyhow!("no data in archive"));
         }
 
-        Ok(result.unwrap())
+        Ok(self.end_time)
     }
 
     // check if the log data is already stored in local.
@@ -154,12 +135,16 @@ where
     }
 
     /// download historical data from the web and store csv in the Archive directory
-    pub async fn download(
+    pub async fn download<T>(
         &mut self,
+        api: &T,
         ndays: i64,
         force: bool,
         verbose: bool,
-    ) -> anyhow::Result<i64> {
+    ) -> anyhow::Result<i64>
+    where
+        T: RestApi,
+    {
         let mut date = FLOOR_DAY(NOW());
 
         log::debug!("download log from {:?}({:?})", date_string(date), date);
@@ -168,7 +153,9 @@ where
         let mut i = 0;
 
         while i <= ndays {
-            count += self.web_archive_to_parquet(date, force, verbose).await?;
+            count += self
+                .web_archive_to_parquet(api, date, force, verbose)
+                .await?;
             date -= DAYS(1);
 
             i += 1;
@@ -179,6 +166,41 @@ where
         Ok(count)
     }
 
+    /// check the lates date in archive web site
+    /// check the latest check time, within 60 min call this function, reuse cache value.
+    pub async fn latest_archive_date<T>(&mut self, api: &T) -> anyhow::Result<MicroSec>
+    where
+        T: RestApi,
+    {
+        let now = NOW();
+        if now - self.last_archive_check_time < MIN(60) {
+            return Ok(self.latest_archive_date);
+        }
+        self.last_archive_check_time = now;
+
+        let mut latest = TODAY();
+        let mut i = 0;
+
+        loop {
+            log::debug!("check log exist = {}({})", time_string(latest), latest);
+
+            if has_web_archive(api, &self.config, latest).await? {
+                self.latest_archive_date = latest;
+                return Ok(latest);
+            }
+            latest -= DAYS(1);
+            i += 1;
+            if 5 < i {
+                return Err(anyhow!(
+                    "Find archive retry over {}/{}/{}",
+                    i,
+                    latest,
+                    time_string(latest)
+                ));
+            }
+        }
+    }
+
     /// generate 0 row empty cache(stored in memory) df
     pub fn make_empty_cachedf() -> DataFrame {
         let time = Series::new(KEY::time_stamp, Vec::<MicroSec>::new());
@@ -204,24 +226,47 @@ where
         Ok(df)
     }
 
+    /// load from archived paquet file retrive specifed time frame.
+    pub fn select_cachedf(
+        &self,
+        start_time: MicroSec,
+        end_time: MicroSec,
+    ) -> anyhow::Result<DataFrame> {
+        let dates = self.select_dates(start_time, end_time)?;
+
+        let mut df = Self::make_empty_cachedf();
+
+        for date in dates {
+            log::debug!("{:?}", date_string(date));
+
+            let new_df = self.load_cache_df(date)?;
+
+            df = append_df(&df, &new_df)?;
+        }
+
+        df = select_df(&df, start_time, end_time);
+
+        Ok(df)
+    }
+
     /// load from parquet file and retrive as cachedf.
     pub fn load_cache_df(&self, date: MicroSec) -> anyhow::Result<DataFrame> {
         let date = FLOOR_DAY(date);
 
-        if date < self.start_time() {
+        if date < self.start_time()? {
             log::warn!(
                 "Not found in archive[too early] query={:?} < start_time{:?}",
                 date_string(date),
-                date_string(self.start_time())
+                date_string(self.start_time()?)
             );
             return Ok(Self::make_empty_cachedf());
         }
 
-        if self.end_time() <= date {
+        if self.end_time()? <= date {
             log::warn!(
                 "Not found in archive[too new] query={:?} >= end_time{:?}",
                 date_string(date),
-                date_string(self.end_time())
+                date_string(self.end_time()?)
             );
             return Ok(Self::make_empty_cachedf());
         }
@@ -246,31 +291,15 @@ where
         Ok(df)
     }
 
-    /// load from archived paquet file retrive specifed time frame.
-    pub fn select_cache_df(
+    /// execute f for each rec in archive within specifed time frame.
+    pub fn foreach<F>(
         &self,
         start_time: MicroSec,
         end_time: MicroSec,
-    ) -> anyhow::Result<DataFrame> {
-        let dates = self.select_dates(start_time, end_time)?;
-
-        let mut df = Self::make_empty_cachedf();
-
-        for date in dates {
-            log::debug!("{:?}", date_string(date));
-
-            let new_df = self.load_cache_df(date)?;
-
-            df = merge_df(&df, &new_df);
-        }
-
-        Ok(df)
-    }
-
-    /// execute f for each rec in archive within specifed time frame.
-    pub fn foreach<F>(&self, start_time: MicroSec, end_time: MicroSec, f: &mut F) -> anyhow::Result<i64>
+        f: &mut F,
+    ) -> anyhow::Result<i64>
     where
-    F: FnMut(&Trade) -> anyhow::Result<()>
+        F: FnMut(&Trade) -> anyhow::Result<()>,
     {
         let dates = self.select_dates(start_time, end_time)?;
 
@@ -281,11 +310,11 @@ where
 
         Ok(count)
     }
-    
+
     /// execite f for each rec(trade) specifed a date.
-    pub fn foreach_paquet<F>(&self, date: MicroSec, f: &mut F) -> anyhow::Result<i64> 
+    pub fn foreach_paquet<F>(&self, date: MicroSec, f: &mut F) -> anyhow::Result<i64>
     where
-    F: FnMut(&Trade) -> anyhow::Result<()>
+        F: FnMut(&Trade) -> anyhow::Result<()>,
     {
         let mut count = 0;
 
@@ -306,12 +335,16 @@ where
             let size = row.get_double(3)?;
             let id = row.get_string(4)?;
 
-            let trade = Trade::new(timestamp, 
-                OrderSide::from(order_side.as_str()), 
-                Decimal::from_f64(price).unwrap(), 
-                Decimal::from_f64(size).unwrap(), crate::common::LogStatus::FixArchiveBlock, id);
+            let trade = Trade::new(
+                timestamp,
+                OrderSide::from(order_side.as_str()),
+                Decimal::from_f64(price).unwrap(),
+                Decimal::from_f64(size).unwrap(),
+                crate::common::LogStatus::FixArchiveBlock,
+                id,
+            );
 
-                f(&trade)?;
+            f(&trade)?;
             // log::debug!("{:?}" , trade);
 
             count += 1;
@@ -320,20 +353,20 @@ where
         Ok(count)
     }
 
-    /// 
+    ///
     pub fn select_dates(
         &self,
         start_time: MicroSec,
         end_time: MicroSec,
     ) -> anyhow::Result<Vec<MicroSec>> {
-        let start_time = if start_time == 0 || start_time < self.start_time() {
-            self.start_time()
+        let start_time = if start_time == 0 || start_time < self.start_time()? {
+            self.start_time()?
         } else {
             FLOOR_DAY(start_time)
         };
 
-        let end_time = if end_time == 0 || self.end_time() <= end_time {
-            FLOOR_DAY(self.end_time() - 1)
+        let end_time = if end_time == 0 || self.end_time()? <= end_time {
+            FLOOR_DAY(self.end_time()? - 1)
         } else {
             FLOOR_DAY(end_time)
         };
@@ -366,6 +399,7 @@ where
 
     /// analyze archive directory, find start_time, end_time
     /// if there is fragmented date, delete it.
+    /// Note: endday means next day's begining(archive date 24:00 = next day 00:00)
     pub fn analyze(&mut self) -> anyhow::Result<()> {
         let mut dates = self.list_dates()?;
         dates.reverse();
@@ -444,7 +478,7 @@ where
             &self.config.exchange_name,
             &self.config.trade_category,
             &self.config.trade_symbol,
-            self.server.is_production(),
+            self.production,
         );
 
         let archive_dir = db_path_root.join("ARCHIVE");
@@ -483,13 +517,16 @@ where
         Ok(())
     }
 
-    pub async fn web_archive_to_parquet(
+    pub async fn web_archive_to_parquet<T>(
         &mut self,
+        api: &T,
         date: MicroSec,
         force: bool,
         verbose: bool,
-    ) -> anyhow::Result<i64> {
-        let server = &self.server.clone();
+    ) -> anyhow::Result<i64>
+    where
+        T: RestApi,
+    {
         let config = &self.config.clone();
 
         let date = FLOOR_DAY(date);
@@ -498,12 +535,13 @@ where
 
         if has_csv_file && !force {
             if verbose {
+                log::debug!("archive csv file exist {}", time_string(date));
                 println!("archive csv file exist {}", time_string(date));
                 return Ok(0);
             }
         }
 
-        let latest = { self.latest_archive_date().await? };
+        let latest = { self.latest_archive_date(api).await? };
 
         if latest < date {
             log::warn!(
@@ -515,10 +553,11 @@ where
         }
 
         if verbose && force {
-            println!("force download")
+            log::debug!("force download");
+            println!("force download");
         }
 
-        let url = T::history_web_url(server, config, date);
+        let url = api.history_web_url(config, date);
 
         log::debug!("Downloading ...[{}]", url);
 
@@ -532,11 +571,11 @@ where
 
         // load to paquet
         log::debug!("read log csv to df");
-        let has_header = T::archive_has_header();
+        let has_header = api.archive_has_header();
         let df = csv_to_df(&file_path, has_header)?;
         log::debug!("load to df {:?}", df.shape());
 
-        let mut archive_df = T::logdf_to_archivedf(&df)?;
+        let mut archive_df = api.logdf_to_archivedf(&df)?;
         log::debug!("archive df shape={:?}", archive_df.shape());
 
         log::debug!("store paquet");
@@ -548,3 +587,67 @@ where
     }
 }
 
+pub async fn log_download_tmp(url: &str, tmp_dir: &Path) -> anyhow::Result<PathBuf> {
+    let client = Client::new();
+
+    let response = client
+        .get(url)
+        .header("User-Agent", "Mozilla/5.0")
+        .header("Accept", "text/html")
+        .send()
+        .await
+        .with_context(|| format!("URL get error {}", url))?;
+
+    log::debug!(
+        "Response code = {} / download size {}",
+        response.status().as_str(),
+        response.content_length().unwrap_or_default() // if error, return 0
+    );
+
+    if !response.status().is_success() {
+        return Err(anyhow!("Download error response={:?}", response));
+    }
+
+    let total_size = response.content_length().unwrap_or(0);
+
+    let fname = response
+        .url()
+        .path_segments()
+        .and_then(|segments| segments.last())
+        .and_then(|name| if name.is_empty() { None } else { Some(name) })
+        .unwrap_or("tmp.bin");
+
+    let path = tmp_dir.join(fname);
+
+    let mut file = tokio::fs::File::create(path.clone()).await?;
+    let mut downloaded: u64 = 0;
+    let mut stream = response.bytes_stream();
+
+    log::debug!("start reading from web");
+
+    while let Some(item) = stream.next().await {
+        let chunk = item?;
+        file.write_all(&chunk).await?;
+        let new = std::cmp::min(downloaded + (chunk.len() as u64), total_size);
+        downloaded = new;
+    }
+
+    file.flush().await?;
+
+    Ok(path)
+}
+
+/// check if achive date is avairable at specified date
+async fn has_web_archive<T>(api: &T, config: &MarketConfig, date: MicroSec) -> anyhow::Result<bool>
+where
+    T: RestApi,
+{
+    let url = api.history_web_url(config, date);
+    let result = check_exist(url.as_str()).await;
+
+    if result.is_err() {
+        return Ok(false);
+    }
+
+    Ok(result.unwrap())
+}
diff --git a/rbot_lib/src/db/avro.rs b/rbot_lib/src/db/avro.rs
deleted file mode 100644
index b67cafa6..00000000
--- a/rbot_lib/src/db/avro.rs
+++ /dev/null
@@ -1,53 +0,0 @@
-use std::fs::File;
-
-use apache_avro::{Codec, Schema, Writer, Reader, from_avro_datum};
-use apache_avro::types::Record;
-
-use crate::common::MicroSec;
-use serde::{Serialize, Deserialize};
-
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
-/// Represents a trade made on an exchange.
-pub struct TradeAvro {
-    /// The time the trade was executed, in microseconds since the epoch.
-    pub time: i64,
-    /// The side of the order that was filled.
-    pub buy_side: bool,
-    /// The price at which the trade was executed.
-    pub price: f64,
-    /// The size of the trade.
-    pub size: f64,
-    /// The unique identifier for the trade.
-    pub id: String,
-}
-
-#[test]
-fn save_load() -> anyhow::Result<()> {
-    let trade_schema = Schema::parse_str(r#"
-        "type": "record",
-        "name": "TradeAvro",
-        "fields": [
-            {"name": "time", "type": "int"},
-            {"name": "buy_side", "type": "bool"},
-            {"name": "price", "type": "double"},
-            {"name": "size", "type": "double"},
-            {"name": "id", "type": "string"},
-        ]
-    "#)?;
-
-    let mut writer = Writer::with_codec(&trade_schema, File::create("measurements.avro")?, Codec::Deflate);
-
-    let rec1 = TradeAvro{
-        time: 1,
-        buy_side: true,
-        price: 10.1,
-        size: 20.1,
-        id: "asdfasdfasdfasfd".to_string(),
-    };
-
-            
-
-    Ok(())
-}
-
-
diff --git a/rbot_lib/src/db/compress.rs b/rbot_lib/src/db/compress.rs
deleted file mode 100644
index e4a1ce67..00000000
--- a/rbot_lib/src/db/compress.rs
+++ /dev/null
@@ -1,69 +0,0 @@
-use async_compression::tokio::bufread::GzipDecoder;
-use async_compression::tokio::write::GzipEncoder;
-use std::path::PathBuf;
-use tokio::io::{AsyncBufReadExt, AsyncWriteExt};
-
-
-pub async fn log_convert<F>(source_file: PathBuf, target_file: PathBuf, has_header: bool, f: F) -> anyhow::Result<()>
-where
-    F: Fn(String) -> String
-{
-    let source_file = tokio::fs::File::open(source_file).await?;
-    let target_file = tokio::fs::File::create(target_file).await?;
-
-    let reader = tokio::io::BufReader::new(source_file);
-    let writer = tokio::io::BufWriter::new(target_file);
-
-    let decoder = GzipDecoder::new(reader);
-    let decoder = tokio::io::BufReader::new(decoder);
-
-    let encoder = GzipEncoder::new(writer);
-    let mut encoder = tokio::io::BufWriter::new(encoder);
-
-    let mut lines = decoder.lines();
-
-    if has_header {
-        let header = lines.next_line().await?;
-        log::debug!("header: {:?}", header);
-    }
-
-    while let Some(mut line) = lines.next_line().await? {
-        line = f(line);
-        encoder.write_all(line.as_bytes()).await?;
-        // println!("length = {}", line.len());
-    }
-
-    let r = encoder.flush().await;
-    if r.is_err() {
-        log::warn!("encoder close err {:?}", r);
-    }
-
-    encoder.shutdown().await?;
-
-    Ok(())
-}
-
-pub async fn log_foreach<F>(source_file: PathBuf, has_header: bool, mut f: F) -> anyhow::Result<()>
-where
-    F: FnMut(String)
-{
-    let source_file = tokio::fs::File::open(source_file).await?;
-
-    let reader = tokio::io::BufReader::new(source_file);
-    let decoder = GzipDecoder::new(reader);
-    let decoder = tokio::io::BufReader::new(decoder);
-
-    let mut lines = decoder.lines();
-
-    if has_header {
-        let header = lines.next_line().await?;
-        log::debug!("header: {:?}", header);
-    }
-
-    while let Some(line) = lines.next_line().await? {
-        f(line);
-    }
-
-    Ok(())
-}
-
diff --git a/rbot_lib/src/db/df.rs b/rbot_lib/src/db/df.rs
index 1c1e03ab..51ee8462 100644
--- a/rbot_lib/src/db/df.rs
+++ b/rbot_lib/src/db/df.rs
@@ -20,6 +20,8 @@ use polars::lazy::prelude::IntoLazy;
 use polars::lazy::prelude::{col, LazyFrame};
 use polars::time::ClosedWindow;
 
+use anyhow::anyhow;
+
 #[allow(non_upper_case_globals)]
 #[allow(non_snake_case)]
 pub mod KEY {
@@ -53,7 +55,7 @@ pub mod KEY {
 }
 
 /// Convert DataFrame to Parquet format and save it to the specified path.
-pub fn df_to_parquet(df: &mut DataFrame, target_path: &PathBuf) -> anyhow::Result<i64>{
+pub fn df_to_parquet(df: &mut DataFrame, target_path: &PathBuf) -> anyhow::Result<i64> {
     let mut target_path = target_path.clone();
     target_path.set_extension("parquet");
     let tmp = target_path.clone();
@@ -61,8 +63,7 @@ pub fn df_to_parquet(df: &mut DataFrame, target_path: &PathBuf) -> anyhow::Resul
 
     let mut file = File::create(&tmp).expect("could not create file");
 
-    ParquetWriter::new(&mut file)
-        .finish(df)?;
+    ParquetWriter::new(&mut file).finish(df)?;
 
     std::fs::rename(&tmp, &target_path)?;
 
@@ -70,22 +71,24 @@ pub fn df_to_parquet(df: &mut DataFrame, target_path: &PathBuf) -> anyhow::Resul
 }
 
 /// This function reads a Parquet file and converts it into a DataFrame.
-pub fn parquet_to_df(path: &PathBuf) -> anyhow::Result<DataFrame>{
+pub fn parquet_to_df(path: &PathBuf) -> anyhow::Result<DataFrame> {
     let file = File::open(path).expect("file not found");
 
-    let df = ParquetReader::new(file)
-            .finish()?;    
+    let df = ParquetReader::new(file).finish()?;
 
-    Ok(df)    
+    Ok(df)
 }
 
+/// import csv format into dataframe
 pub fn csv_to_df(source_path: &PathBuf, has_header: bool) -> anyhow::Result<DataFrame> {
-        let df = CsvReadOptions::default()
+    log::debug!("reading csv file = {:?}", source_path);
+
+    let df = CsvReadOptions::default()
         .with_has_header(has_header)
         .try_into_reader_with_file_path(Some(source_path.clone()))?
         .finish()?;
 
-        Ok(df)
+    Ok(df)
 }
 
 /// Cutoff start_time to end_time(not include)
@@ -124,6 +127,7 @@ pub fn select_df(df: &DataFrame, start_time: MicroSec, end_time: MicroSec) -> Da
     return df;
 }
 
+/// select df with lazy dataframe
 pub fn select_df_lazy(df: &DataFrame, start_time: MicroSec, end_time: MicroSec) -> LazyFrame {
     log::debug!(
         "Select from {} -> {}",
@@ -161,31 +165,39 @@ pub fn end_time_df(df: &DataFrame) -> Option<MicroSec> {
     df.column(KEY::time_stamp).unwrap().max().unwrap()
 }
 
-pub fn merge_df(df1: &DataFrame, df2: &DataFrame) -> DataFrame {
-    let df2_start_time = start_time_df(df2);
+/// append df2 after df1
+pub fn append_df(df1: &DataFrame, df2: &DataFrame) -> anyhow::Result<DataFrame> {
+    let df = df1.vstack(df2)?;
+    
+    Ok(df)
+}
 
-    if let Some(df2_start_time) = df2_start_time {
-        let df = select_df(df1, 0, df2_start_time);
+/// merge 2 dataframe, if overlap df2 on df1, df1 will be trimmed(overritten by df2)
+pub fn merge_df(df1: &DataFrame, df2: &DataFrame) -> anyhow::Result<DataFrame> {
+    log::debug!("merge df1={:?}  df2={:?}", df1.shape(), df2.shape());
 
-        log::debug!("merge len {}", df.shape().0);
+    if df1.shape().0 == 0 {
+        return Ok(df2.clone());
+    }
 
-        if df.shape().0 == 0 {
-            df2.clone()
-        } else {
-            log::debug!("merge df1={:?}  df2={:?}", df1.shape(), df2.shape());
+    if df2.shape().0 == 0 {
+        return Ok(df1.clone());
+    }
 
-            let df = df.vstack(df2);
+    let df2_start_time = start_time_df(df2);
+    if df2_start_time.is_none() {
+        return Err(anyhow!("cannot find time stamp {:?}", df2));
+    }
+    let df2_start_time = df2_start_time.unwrap();
 
-            if df.is_err() {
-                log::error!("merge_df error {:?} and {:?}", df, df2);
-                df2.clone()
-            } else {
-                df.unwrap()
-            }
-        }
-    } else {
-        df1.clone()
+    let df = select_df(df1, 0, df2_start_time);
+    if df.shape().0 == 0 {
+        return Ok(df2.clone());
     }
+
+    let df = df.vstack(df2)?;
+
+    Ok(df)
 }
 
 pub fn ohlcv_df(
@@ -221,15 +233,6 @@ pub fn ohlcv_df(
     let df = select_df_lazy(df, start_time, end_time);
 
     let result = df
-        .sort(
-            vec![(KEY::time_stamp).to_string()],
-            SortMultipleOptions {
-                descending: vec![false],
-                nulls_last: vec![false],
-                maintain_order: true,
-                multithreaded: true,
-            },
-        )
         .group_by_dynamic(col(KEY::time_stamp), [], option)
         .agg([
             col(KEY::price).first().alias(KEY::open),
@@ -239,6 +242,15 @@ pub fn ohlcv_df(
             col(KEY::size).sum().alias(KEY::volume),
             col(KEY::price).count().alias(KEY::count),
         ])
+        .sort(
+            vec![(KEY::time_stamp).to_string()],
+            SortMultipleOptions {
+                descending: vec![false],
+                nulls_last: vec![false],
+                maintain_order: true,
+                multithreaded: true,
+            },
+        )
         .collect();
 
     match result {
@@ -282,15 +294,6 @@ pub fn ohlcvv_df(
     let df = select_df_lazy(df, start_time, end_time);
 
     let result = df
-        .sort(
-            vec![KEY::time_stamp.to_string()],
-            SortMultipleOptions {
-                descending: vec![false],
-                nulls_last: vec![false],
-                maintain_order: true,
-                multithreaded: true,
-            },
-        )
         .group_by_dynamic(col(KEY::time_stamp), [col(KEY::order_side)], option)
         .agg([
             col(KEY::price).first().alias(KEY::open),
@@ -302,6 +305,15 @@ pub fn ohlcvv_df(
             col(KEY::time_stamp).min().alias(KEY::start_time),
             col(KEY::time_stamp).max().alias(KEY::end_time),
         ])
+        .sort(
+            vec![KEY::time_stamp.to_string()],
+            SortMultipleOptions {
+                descending: vec![false],
+                nulls_last: vec![false],
+                maintain_order: true,
+                multithreaded: true,
+            },
+        )
         .collect();
 
     match result {
@@ -345,26 +357,17 @@ pub fn ohlcv_from_ohlcvv_df(
     let df = select_df_lazy(df, start_time, end_time);
 
     let result = df
-        .sort(
-            vec![(KEY::time_stamp).to_string()],
-            SortMultipleOptions {
-                descending: vec![false],
-                nulls_last: vec![false],
-                maintain_order: true,
-                multithreaded: true,
-            },
-        )
         .group_by_dynamic(col(KEY::time_stamp), [], option)
         .agg([
             col(KEY::open)
                 .sort_by(
                     vec![col(KEY::start_time)],
-                    SortMultipleOptions{
+                    SortMultipleOptions {
                         descending: vec![false],
                         nulls_last: vec![false],
                         multithreaded: true,
-                        maintain_order: true
-                    }
+                        maintain_order: true,
+                    },
                 )
                 .first()
                 .alias(KEY::open),
@@ -373,18 +376,27 @@ pub fn ohlcv_from_ohlcvv_df(
             col(KEY::close)
                 .sort_by(
                     vec![col(KEY::end_time)],
-                    SortMultipleOptions{
+                    SortMultipleOptions {
                         descending: vec![false],
                         nulls_last: vec![false],
                         multithreaded: true,
-                        maintain_order: true
-                    }
-                    )
+                        maintain_order: true,
+                    },
+                )
                 .last()
                 .alias(KEY::close),
             col(KEY::volume).sum().alias(KEY::volume),
             col(KEY::count).sum().alias(KEY::count),
         ])
+        .sort(
+            vec![(KEY::time_stamp).to_string()],
+            SortMultipleOptions {
+                descending: vec![false],
+                nulls_last: vec![false],
+                maintain_order: true,
+                multithreaded: true,
+            },
+        )
         .collect();
 
     match result {
@@ -423,26 +435,6 @@ pub fn ohlcvv_from_ohlcvv_df(
     let df = select_df_lazy(df, start_time, end_time);
 
     let result = df
-        .sort(
-            vec![(KEY::time_stamp).to_string()],
-            SortMultipleOptions {
-                descending: vec![false],
-                nulls_last: vec![false],
-                maintain_order: true,
-                multithreaded: true,
-            },
-        )
-        /*
-        .sort(
-                KEY::time_stamp,
-                SortOptions {
-                    descending: false,
-                    nulls_last: false,
-                    maintain_order: true,
-                    multithreaded: true,
-                },
-            )
-            */
         .group_by_dynamic(col(KEY::time_stamp), [col(KEY::order_side)], option)
         .agg([
             col(KEY::open).first().alias(KEY::open),
@@ -454,6 +446,15 @@ pub fn ohlcvv_from_ohlcvv_df(
             col(KEY::start_time).min().alias(KEY::start_time),
             col(KEY::end_time).max().alias(KEY::end_time),
         ])
+        .sort(
+            vec![(KEY::time_stamp).to_string()],
+            SortMultipleOptions {
+                descending: vec![false],
+                nulls_last: vec![false],
+                maintain_order: true,
+                multithreaded: true,
+            },
+        )
         .collect();
 
     match result {
@@ -466,44 +467,6 @@ pub fn ohlcvv_from_ohlcvv_df(
     }
 }
 
-/// Calc Value At Price
-/// group by unit price and order_side
-pub fn vap_df_bak(df: &DataFrame, start_time: MicroSec, end_time: MicroSec) -> DataFrame {
-    let df = select_df_lazy(df, start_time, end_time);
-
-    let vap = df
-        .group_by([KEY::price, KEY::order_side])
-        .agg([col(KEY::size).sum().alias(KEY::size)])
-        .collect()
-        .unwrap();
-
-    vap
-}
-
-/// Calc Value At Price
-/// group by unit price and order_side
-pub fn vap_df_bak2(df: &DataFrame, start_time: MicroSec, end_time: MicroSec) -> DataFrame {
-    let df = select_df_lazy(df, start_time, end_time);
-
-    //    let floor_price = col(KEY::price)
-    //    .map(|v: f64| (v / 10.0).floor() * 10.0, Arc::new(DataType::Float64) as Arc<dyn FunctionOutputField>);
-
-    //.div(10.0).alias("floor_price").cast(&DataType::Float64);
-
-    //    let floor_price =
-    //col(KEY::price).div(10.0).alias("floor_price").cast(&DataType::Float64);
-
-    let floor_price = col(KEY::price).floor();
-
-    let vap_gb = df.group_by([floor_price, col(KEY::order_side)]);
-
-    let vap = vap_gb
-        .agg([col(KEY::size).sum().alias(KEY::size)])
-        .collect()
-        .unwrap();
-
-    vap
-}
 
 /// Calc Value At Price
 /// group by unit price and order_side
@@ -531,14 +494,17 @@ pub fn vap_df(df: &DataFrame, start_time: MicroSec, end_time: MicroSec, size: i6
     )
     .unwrap();
 
-    let vap = vap.sort(
-        vec![KEY::price.to_string()],
-         SortMultipleOptions { 
-            descending: vec![false], 
-            nulls_last: vec![false], 
-            multithreaded: true, 
-            maintain_order: true}
-    ).unwrap();
+    let vap = vap
+        .sort(
+            vec![KEY::price.to_string()],
+            SortMultipleOptions {
+                descending: vec![false],
+                nulls_last: vec![false],
+                multithreaded: true,
+                maintain_order: true,
+            },
+        )
+        .unwrap();
 
     let price = vap.column(KEY::price).unwrap().clone();
     let buy_vol = vap
@@ -556,7 +522,7 @@ pub fn vap_df(df: &DataFrame, start_time: MicroSec, end_time: MicroSec, size: i6
         .rename(KEY::sell_volume)
         .clone();
 
-        let total_vol = (&buy_vol + &sell_vol).unwrap().rename(KEY::volume).clone();
+    let total_vol = (&buy_vol + &sell_vol).unwrap().rename(KEY::volume).clone();
 
     let df = DataFrame::new(vec![price, buy_vol, sell_vol, total_vol]).unwrap();
 
@@ -565,9 +531,9 @@ pub fn vap_df(df: &DataFrame, start_time: MicroSec, end_time: MicroSec, size: i6
 
 pub struct TradeBuffer {
     pub time_stamp: Vec<MicroSec>,
+    pub order_side: Vec<bool>,
     pub price: Vec<f64>,
     pub size: Vec<f64>,
-    pub order_side: Vec<bool>,
 }
 
 impl TradeBuffer {
@@ -604,11 +570,11 @@ impl TradeBuffer {
 
     pub fn to_dataframe(&self) -> DataFrame {
         let time_stamp = Series::new(KEY::time_stamp, self.time_stamp.to_vec());
+        let order_side = Series::new(KEY::order_side, self.order_side.to_vec());
         let price = Series::new(KEY::price, self.price.to_vec());
         let size = Series::new(KEY::size, self.size.to_vec());
-        let order_side = Series::new(KEY::order_side, self.order_side.to_vec());
 
-        let df = DataFrame::new(vec![time_stamp, price, size, order_side]).unwrap();
+        let df = DataFrame::new(vec![time_stamp, order_side, price, size]).unwrap();
 
         return df;
     }
@@ -616,13 +582,13 @@ impl TradeBuffer {
 
 pub fn make_empty_ohlcvv() -> DataFrame {
     let time = Series::new(KEY::time_stamp, Vec::<MicroSec>::new());
-    let order_side = Series::new(KEY::order_side, Vec::<f64>::new());
+    let order_side = Series::new(KEY::order_side, Vec::<bool>::new());
     let open = Series::new(KEY::open, Vec::<f64>::new());
     let high = Series::new(KEY::high, Vec::<f64>::new());
     let low = Series::new(KEY::low, Vec::<f64>::new());
     let close = Series::new(KEY::close, Vec::<f64>::new());
     let vol = Series::new(KEY::volume, Vec::<f64>::new());
-    let count = Series::new(KEY::count, Vec::<f64>::new());
+    let count = Series::new(KEY::count, Vec::<i64>::new());
     let start_time = Series::new(KEY::start_time, Vec::<MicroSec>::new());
     let end_time = Series::new(KEY::end_time, Vec::<MicroSec>::new());
 
@@ -641,7 +607,7 @@ pub fn make_empty_ohlcv() -> DataFrame {
     let low = Series::new(KEY::low, Vec::<f64>::new());
     let close = Series::new(KEY::close, Vec::<f64>::new());
     let vol = Series::new(KEY::volume, Vec::<f64>::new());
-    let count = Series::new(KEY::count, Vec::<f64>::new());
+    let count = Series::new(KEY::count, Vec::<i64>::new());
 
     let df = DataFrame::new(vec![time, open, high, low, close, vol, count]).unwrap();
 
@@ -658,12 +624,12 @@ impl AsDynamicGroupOptions for DynamicGroupOptions {
     }
 }
 
-pub fn convert_timems_to_datetime(df: &mut DataFrame) -> &DataFrame {
-    let time = df.column(KEY::time_stamp).unwrap().i64().unwrap().clone();
+pub fn convert_timems_to_datetime(df: &mut DataFrame) -> anyhow::Result<()> {
+    let time = df.column(KEY::time_stamp)?.i64()?.clone();
     let date_time = time.into_datetime(TimeUnit::Microseconds, None);
-    let df = df.with_column(date_time).unwrap();
+    df.with_column(date_time)?;
 
-    df
+    Ok(())
 }
 
 use polars::prelude::*;
@@ -672,7 +638,79 @@ use rust_decimal::prelude::ToPrimitive;
 #[cfg(test)]
 mod test_df {
     use super::*;
-    use crate::common::DAYS;
+    use crate::common::{init_debug_log, DAYS};
+
+    #[test]
+    fn test_merge_and_append_df() -> anyhow::Result<()> {
+        init_debug_log();
+
+        let df1 = df![
+            KEY::time_stamp => [1, 2, 3, 4, 5],
+            "value" => [11, 12, 13, 14, 15]
+        ]?;
+
+        let df2 = df![
+            KEY::time_stamp => [5, 6],
+            "value" => [25, 26]
+        ]?;
+
+        let merged_df = df![
+            KEY::time_stamp => [1, 2, 3, 4, 5, 6],
+            "value" => [11, 12, 13, 14, 25, 26]
+        ]?;
+
+        let df = merge_df(&df1, &df2)?;
+        log::debug!("{:?}", df);
+        assert_eq!(df, merged_df);
+
+        let empty_df = select_df(&df1, 100, 101);
+        log::debug!("{:?}", empty_df);
+
+        let df = merge_df(&empty_df, &df2)?;
+        assert_eq!(df, df2);
+
+        let df = merge_df(&df1, &empty_df)?;
+        assert_eq!(df, df1);
+
+        assert_eq!(
+            merge_df(
+                &df![
+                    KEY::time_stamp => [1, 2],
+                    "value" => [11, 12]
+                ]?,
+                &df![
+                    KEY::time_stamp => [5, 6],
+                    "value" => [25, 26]
+                ]?,
+            )?,
+            df![
+                KEY::time_stamp => [1,2,5,6],
+                "value" => [11, 12, 25,26]
+            ]?
+        );
+
+        assert_eq!(
+            append_df(
+                &df![
+                    KEY::time_stamp => [1, 2],
+                    "value" => [11, 12]
+                ]?,
+                &df![
+                    KEY::time_stamp => [2, 3],
+                    "value" => [25, 26]
+                ]?,
+            )?,
+            df![
+                KEY::time_stamp => [1,2,2,3],
+                "value" => [11, 12, 25,26]
+            ]?
+        );
+
+
+
+
+        Ok(())
+    }
 
     #[test]
     fn test_simple_dynamic_group() {
@@ -702,15 +740,6 @@ mod test_df {
 
         let groupby = df
             .lazy()
-            .sort(
-                vec!["date".to_string()],
-                SortMultipleOptions {
-                    descending: vec![false],
-                    nulls_last: vec![false],
-                    multithreaded: true,
-                    maintain_order: false,
-                },
-            )
             .group_by_dynamic(
                 col("date"),
                 //[col("category")],
@@ -721,6 +750,15 @@ mod test_df {
                 col("value").mean().alias("mean"),
                 col("value").sum().alias("sum"),
             ])
+            .sort(
+                vec!["date".to_string()],
+                SortMultipleOptions {
+                    descending: vec![false],
+                    nulls_last: vec![false],
+                    multithreaded: true,
+                    maintain_order: false,
+                },
+            )
             .collect()
             .unwrap();
 
@@ -732,25 +770,30 @@ mod test_df {
 
     fn make_ohlcv_df() -> DataFrame {
         let df = df!(
-            KEY::time_stamp => &[DAYS(1), DAYS(2), DAYS(3)],
-            KEY::order_side => &[0, 1, 0],
-            KEY::open => &[1.0, 2.0, 3.0],
-            KEY::high => &[1.0, 2.0, 3.0],
-            KEY::low => &[1.0, 2.0, 3.0],
-            KEY::close => &[1.0, 2.0, 3.0],
-            KEY::volume => &[1.0, 2.0, 3.0],
-            KEY::count => &[1.0, 2.0, 3.0],
-            KEY::start_time => &[DAYS(1), DAYS(2), DAYS(3)],
-            KEY::end_time => &[DAYS(1), DAYS(2), DAYS(3)]
+            KEY::order_side => &[false, true, false, true],
+            KEY::time_stamp => &[DAYS(1), DAYS(2), DAYS(3), DAYS(3)],
+            KEY::open => &[1.0, 2.0, 3.0, 4.0],
+            KEY::high => &[1.0, 2.0, 3.0, 4.0],
+            KEY::low => &[1.0, 2.0, 3.0, 4.0],
+            KEY::close => &[1.0, 2.0, 3.0, 4.0],
+            KEY::volume => &[1.0, 2.0, 3.0, 4.0],
+            KEY::count => &[1, 2, 3, 4],
+            KEY::start_time => &[DAYS(1), DAYS(2), DAYS(3), DAYS(3)],
+            KEY::end_time => &[DAYS(2), DAYS(3), DAYS(4), DAYS(4)]
         );
 
-        return df.unwrap().sort([KEY::time_stamp], 
-            SortMultipleOptions{
-                descending: vec![false],
-                nulls_last: vec![false],
-                multithreaded: true,
-                maintain_order: true
-            }).unwrap();
+        return df
+            .unwrap()
+            .sort(
+                [KEY::time_stamp],
+                SortMultipleOptions {
+                    descending: vec![false],
+                    nulls_last: vec![false],
+                    multithreaded: true,
+                    maintain_order: true,
+                },
+            )
+            .unwrap();
     }
 
     #[test]
@@ -758,22 +801,33 @@ mod test_df {
         let ohlc = make_ohlcv_df();
 
         println!("{:?}", ohlc);
+        assert_eq!(ohlc.shape().0, 4);
+        assert_eq!(ohlc.shape().1, 10);
     }
 
     #[test]
-    fn test_make_ohlcv_from_ohclv() {
+    fn test_make_ohlcvv_from_ohclv() -> anyhow::Result<()>{
         let ohlcv = make_ohlcv_df();
 
-        let ohlcv2 = ohlcvv_from_ohlcvv_df(&ohlcv, 0, 0, 10);
+        let ohlcv2 = ohlcvv_from_ohlcvv_df(&ohlcv, 0, 0, 10)?;
         println!("{:?}", ohlcv2);
+
+        assert_eq!(ohlcv, ohlcv2);
+
+        Ok(())
     }
 
     #[test]
-    fn test_make_ohlc_from_ohclv() {
+    fn test_make_ohlcv_from_ohclv() -> anyhow::Result<()>{
         let ohlcv = make_ohlcv_df();
 
-        let ohlcv2 = ohlcv_from_ohlcvv_df(&ohlcv, 0, 0, 10);
+        let ohlcv2 = ohlcv_from_ohlcvv_df(&ohlcv, 0, 0, 10)?;
         println!("{:?}", ohlcv2);
+
+        assert_eq!(ohlcv2.shape().0, 3);
+        assert_eq!(ohlcv2.shape().1, 7);
+
+        Ok(())
     }
 
     #[test]
@@ -847,15 +901,15 @@ mod test_df {
     }
 
     #[test]
-    fn test_convert_ohlc_datetime() {
+    fn test_convert_ohlc_datetime() -> anyhow::Result<()>{
         let mut df = make_empty_ohlcv();
 
         println!("{:?}", df);
 
-        let time = df.column("time_stamp").unwrap().i64().unwrap().clone();
-        let date_time = time.into_datetime(TimeUnit::Microseconds, None);
-        let df = df.with_column(date_time).unwrap();
+        convert_timems_to_datetime(&mut df)?;
 
         println!("{:?}", df);
+
+        Ok(())
     }
 }
diff --git a/rbot_lib/src/db/mod.rs b/rbot_lib/src/db/mod.rs
index 45bf6158..845ecc33 100644
--- a/rbot_lib/src/db/mod.rs
+++ b/rbot_lib/src/db/mod.rs
@@ -4,14 +4,12 @@ pub mod sqlite;
 pub mod df;
 pub mod fs;
 pub mod archive;
-pub mod compress;
-pub mod avro;
+pub mod tradedf;
 
 pub use sqlite::*;
 pub use df::*;
 pub use fs::*;
 pub use archive::*;
-pub use compress::*;
-pub use avro::*;
+pub use tradedf::*;
 
 
diff --git a/rbot_lib/src/db/sqlite.rs b/rbot_lib/src/db/sqlite.rs
index 6ced9d51..f460821a 100644
--- a/rbot_lib/src/db/sqlite.rs
+++ b/rbot_lib/src/db/sqlite.rs
@@ -5,74 +5,58 @@ use anyhow::ensure;
 use anyhow::Context;
 //use anyhow::Result;
 
-use numpy::IntoPyArray;
-use numpy::PyArray2;
-use once_cell::sync::Lazy;
 use polars::prelude::DataFrame;
-use polars::prelude::Float64Type;
-use polars::prelude::IndexOrder;
-use pyo3::{Py, Python};
-use pyo3_polars::PyDataFrame;
 use rusqlite::params_from_iter;
 use rusqlite::{params, Connection, Transaction};
 use rust_decimal::prelude::FromPrimitive;
 use rust_decimal::prelude::ToPrimitive;
 use rust_decimal::Decimal;
 use rust_decimal_macros::dec;
-use std::collections::HashMap;
-use std::sync::Arc;
-use std::sync::Mutex;
-use std::time::Duration;
+use std::path::PathBuf;
 use tokio::task::spawn;
 use tokio::task::JoinHandle;
 
-use crate::common::MarketStream;
+use crate::common::MarketConfig;
+use crate::common::TimeChunk;
+use crate::common::FLOOR_DAY;
+use crate::db::TradeDataFrame;
 
 use crossbeam_channel::unbounded;
 use crossbeam_channel::Sender;
 
 use crate::common::flush_log;
 use crate::common::LogStatus;
-use crate::common::MarketMessage;
 use crate::common::OrderSide;
-use crate::common::HHMM;
 use crate::common::SEC;
-use crate::common::{time_string, MicroSec, CEIL, DAYS, FLOOR_DAY, FLOOR_SEC, NOW};
-use crate::common::{TimeChunk, Trade};
-use crate::db::df::merge_df;
-use crate::db::df::ohlcvv_df;
-use crate::db::df::ohlcvv_from_ohlcvv_df;
-use crate::db::df::select_df;
-use crate::db::df::start_time_df;
+use crate::common::{time_string, MicroSec, CEIL, DAYS, FLOOR_SEC, NOW};
+use crate::common::Trade;
 use crate::db::df::TradeBuffer;
-use crate::db::df::{end_time_df, make_empty_ohlcvv, ohlcv_df, ohlcv_from_ohlcvv_df};
-
-use crate::db::df::KEY;
 
 use super::db_full_path;
-use super::df::convert_timems_to_datetime;
-use super::df::vap_df;
+use super::OHLCV_WINDOW_SEC;
 
-static EXCHANGE_DB_CACHE: Lazy<Mutex<HashMap<String, Arc<Mutex<TradeTable>>>>> =
-    Lazy::new(|| Mutex::new(HashMap::new()));
+pub fn ohlcv_floor_fix_time(t: MicroSec, unit_sec: i64) -> MicroSec {
+    return FLOOR_SEC(t, unit_sec);
+}
 
-#[derive(Debug)]
-pub struct TradeTableDb {
-    pub file_name: String,
-    connection: Connection,
+pub fn ohlcv_start(t: MicroSec) -> MicroSec {
+    return FLOOR_SEC(t, OHLCV_WINDOW_SEC);
 }
 
-impl TradeTableDb {
-    pub fn clone_connection(&self) -> TradeTableDb {
-        let conn = Connection::open(&self.file_name).unwrap();
-        let db = TradeTableDb {
-            file_name: self.file_name.clone(),
-            connection: conn,
-        };
+pub fn ohlcv_end(t: MicroSec) -> MicroSec {
+    return CEIL(t, OHLCV_WINDOW_SEC);
+}
 
-        return db;
-    }
+pub struct TradeDb {
+    config: MarketConfig,
+    production: bool,
+    connection: Connection,
 
+    tx: Option<Sender<Vec<Trade>>>,
+    handle: Option<JoinHandle<()>>,
+}
+
+impl TradeDb {
     /// delete unstable data, include both edge.
     /// start_time <= (time_stamp) <= end_time
     pub fn delete_unstable_data(
@@ -172,7 +156,7 @@ impl TradeTableDb {
 
         let trade = trade.unwrap();
 
-        Ok(TradeTable::ohlcv_end(trade.time) + 1) // plus 1 microsecond(1us) to avoid duplicate data
+        Ok(ohlcv_end(trade.time) + 1) // plus 1 microsecond(1us) to avoid duplicate data
     }
 
     pub fn first_unfix_trade(&mut self, start_time: MicroSec) -> anyhow::Result<Option<Trade>> {
@@ -194,7 +178,7 @@ impl TradeTableDb {
 
         let trade = trade.unwrap();
 
-        Ok(TradeTable::ohlcv_end(trade.time) - 1) // minus 1 microsecond(1us) to avoid duplicate data
+        Ok(ohlcv_end(trade.time) - 1) // minus 1 microsecond(1us) to avoid duplicate data
     }
 
     /// 2日以内のUnstableデータを削除するメッセージを作成する。
@@ -339,88 +323,88 @@ impl TradeTableDb {
         return Ok(true);
     }
 
-    pub fn validate_by_date(&mut self, date: MicroSec) -> anyhow::Result<bool> {
-        let start_time = FLOOR_DAY(date);
-        let end_time = start_time + DAYS(1);
+    pub fn open(config: &MarketConfig, production: bool) -> anyhow::Result<Self> 
+    {
+        let db_path = db_full_path(
+            &config.exchange_name,
+            &config.trade_category,
+            &config.trade_symbol,
+            production,
+        );
 
-        // startからendまでのレコードにS,Eが1つづつあるかどうかを確認する。
-        let sql = r#"select time_stamp, action, price, size, status, id from trades where $1 <= time_stamp and time_stamp < $2 and (status = "S" or status = "E") order by time_stamp"#;
-        let trades = self.select_query(sql, vec![start_time, end_time])?;
+        let create_new = Self::is_db_file_exsist(&db_path);
 
-        if trades.len() != 2 {
-            log::debug!("S,E is not 2 {}", trades.len());
-            return Ok(false);
+        let conn = Connection::open(db_path)?;
+        
+        if create_new {
+            conn.pragma_update(None, "journal_mode", "wal")?;
         }
 
-        let first = trades[0].clone();
-        let last = trades[1].clone();
-
-        // Sから始まりEでおわることを確認
-        if first.status != LogStatus::FixBlockStart && last.status != LogStatus::FixBlockEnd {
-            log::debug!("S,E is not S,E");
-            return Ok(false);
-        }
+        let mut db = TradeDb {
+            config: config.clone(),
+            production,
+            connection: conn,
+            tx: None,
+            handle: None,
+        };
 
-        // S, Eのレコードの間が十分にあること（トラフィックにもよるが２２時間を想定）
-        if last.time - first.time < HHMM(20, 0) {
-            log::debug!("batch is too short");
-            return Ok(false);
+        if create_new {
+            db.create_table_if_not_exists()?;
         }
 
-        Ok(true)
+        Ok(db)
     }
-}
-
-impl TradeTableDb {
-    /// create new database
-    fn create(name: &str) -> anyhow::Result<Self> {
-        let conn = Connection::open(name)?;
 
-        conn.busy_timeout(Duration::from_secs(10))?;
-
-        conn.pragma_update(None, "journal_mode", "wal")?;
-
-        let db = TradeTableDb {
-            file_name: name.to_string(),
-            connection: conn,
-        };
-
-        return Ok(db);
-    }
+    pub fn open_channel(&mut self) -> anyhow::Result<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.tx.is_some() {
+            log::info!("DB Thread is already started, reuse tx");
+            return Ok(self.tx.clone().unwrap().clone());
+        }
 
-    fn open(name: &str) -> anyhow::Result<Self> {
-        log::debug!("open database {}", name);
+        let (tx, rx) = unbounded();
 
-        if Self::is_db_file_exsist(name) == false {
-            log::debug!("database file is not exsit. create new database");
-            let mut db = TradeTableDb::create(name)?;
-            db.create_table_if_not_exists()?;
-            return Ok(db);
-        }
+        let config = self.config.clone();
+        let production = self.production;
+        self.tx = Some(tx);
 
-        // create new one
-        let conn = Connection::open(name)?;
-        log::debug!("Database open path = {}", name);
+        let handle = spawn(async move {
+            let mut db = TradeDb::open(&config, production).unwrap();
+            let rx = rx; // Move rx into the closure's environment
+            loop {
+                match rx.recv() {
+                    Ok(trades) => {
+                        let result = db.insert_records(&trades);
 
-        // TODO: set timeout parameter
-        conn.busy_timeout(Duration::from_secs(10)).unwrap();
+                        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;
+                    }
+                }
+            }
+        });
 
-        let db = TradeTableDb {
-            file_name: name.to_string(),
-            connection: conn,
-        };
+        self.handle = Some(handle);
 
-        Ok(db)
-    }
+        return Ok(self.tx.clone().unwrap());
+      
+    } 
 
     /// check if database file is exsit
-    fn is_db_file_exsist(name: &str) -> bool {
-        let path = std::path::Path::new(name);
+    fn is_db_file_exsist(path: &PathBuf) -> bool {
         return path.exists();
     }
 
+
     /// check if table is exsit by sql.
-    #[allow(dead_code)]
     fn is_table_exsit(&mut self) -> bool {
         log::debug!("is_table_exsit");
 
@@ -476,7 +460,6 @@ impl TradeTableDb {
     fn drop_table(&self) -> anyhow::Result<()> {
         println!("database is dropped. To use database, restart the program.");
         println!("To delete completely, delete the database file from OS.");
-        println!("DB file = {}", self.file_name);
         flush_log();
 
         self.connection
@@ -504,10 +487,27 @@ impl TradeTableDb {
         Ok(())
     }
 
+    /// select  cachedf from database
+    pub fn select_cachedf(
+        &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());
+    }
+
     // 時間選択は左側は含み、右側は含まない。
     // 0をいれたときは全件検索
     pub fn select<F>(
-        &mut self,
+        &self,
         start_time: MicroSec,
         end_time: MicroSec,
         mut f: F,
@@ -605,638 +605,6 @@ impl TradeTableDb {
         return Ok(trades);
     }
 
-    pub fn select_stream(&mut self, time_from: MicroSec, time_to: MicroSec) -> MarketStream {
-        let (sender, stream) = MarketStream::open();
-
-        let mut table_db = self.clone_connection();
-
-        tokio::task::spawn(async move {
-            // 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)?;
-
-                Ok(())
-            });
-
-            if r.is_err() {
-                log::error!("select_stream error {:?}", r);
-            }
-        });
-
-        stream
-    }
-}
-
-#[derive(Debug)]
-pub struct TradeTable {
-    file_name: String,
-    pub connection: TradeTableDb,
-    cache_df: DataFrame,
-    cache_ohlcvv: DataFrame,
-    cache_duration: MicroSec,
-    tx: Option<Sender<Vec<Trade>>>,
-    handle: Option<JoinHandle<()>>,
-}
-
-impl TradeTable {
-    const OHLCV_WINDOW_SEC: i64 = 60; // min
-
-    pub fn make_db_path(
-        exchange_name: &str,
-        trade_category: &str,
-        trade_symbol: &str,
-        production: bool,
-    ) -> String {
-        let db_path = db_full_path(&exchange_name, &trade_category, trade_symbol, production);
-
-        return db_path.to_str().unwrap().to_string();
-    }
-
-    pub fn get(db_path: &str) -> anyhow::Result<Arc<Mutex<Self>>> {
-        let mut db_cache = EXCHANGE_DB_CACHE.lock().unwrap();
-        let db_in_cache = db_cache.get(db_path);
-
-        if let Some(db) = db_in_cache {
-            return Ok(db.clone());
-        }
-
-        let db = TradeTable::open(db_path)?;
-
-        log::debug!("vaccum start {:?}", db_path);
-        let r = db.vacuum();
-        if r.is_err() {
-            log::warn!("db vaccum error {:?}", r);
-        }
-        log::debug!("vaccum end");
-
-
-        let db = Arc::new(Mutex::new(db));
-        db_cache.insert(db_path.to_string(), db.clone());
-
-        Ok(db)
-    }
-
-    /// check if db thread is running.
-    pub fn is_running(&self) -> bool {
-        if self.handle.is_none() {
-            return false;
-        }
-
-        if self.handle.as_ref().unwrap().is_finished() {
-            return false;
-        }
-
-        return true;
-    }
-
-    /// 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,
-        force: bool,
-    ) -> anyhow::Result<Vec<Trade>> {
-        self.connection.make_expire_control_message(now, force)
-    }
-
-    pub fn set_cache_ohlcvv(&mut self, df: DataFrame) {
-        let start_time: MicroSec = df
-            .column(KEY::time_stamp)
-            .unwrap()
-            .min()
-            .unwrap()
-            .unwrap_or(0);
-        let end_time: MicroSec = df
-            .column(KEY::time_stamp)
-            .unwrap()
-            .max()
-            .unwrap()
-            .unwrap_or(0);
-
-        let head = select_df(&self.cache_ohlcvv, 0, start_time);
-        let tail = select_df(&self.cache_ohlcvv, end_time, 0);
-
-        log::debug!(
-            "set_cache_ohlcvv head {} /tail {}/ df {}",
-            head.shape().0,
-            tail.shape().0,
-            df.shape().0
-        );
-
-        log::debug!("df {:?}", df.head(Some(2)));
-        log::debug!("head {:?}", head.head(Some(2)));
-        log::debug!("tail {:?}", tail.head(Some(2)));
-
-        let df = merge_df(&head, &df);
-        let df = merge_df(&df, &tail);
-
-        self.cache_ohlcvv = df;
-    }
-
-    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");
-        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(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 validate_by_date(&mut self, date: MicroSec) -> anyhow::Result<bool> {
-        self.connection.validate_by_date(date)
-    }
-
-    pub fn get_cache_duration(&self) -> MicroSec {
-        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);
-    }
-
-    pub fn ohlcv_end(t: MicroSec) -> MicroSec {
-        return CEIL(t, TradeTable::OHLCV_WINDOW_SEC);
-    }
-
-    pub fn reset_cache_duration(&mut self) {
-        self.cache_duration = 0;
-    }
-
-    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());
-    }
-
-    // アーカイブにある場合は、アーカイブから取得する
-    pub fn load_df(&mut self, start_time: MicroSec, end_time: MicroSec) -> anyhow::Result<()> {
-        self.cache_df = self.select_df_from_db(start_time, end_time)?;
-
-        Ok(())
-    }
-
-    pub fn update_cache_all(&mut self) -> anyhow::Result<()> {
-        self.update_cache_df(0, 0)
-    }
-
-    // TODO: 日付単位に変更する。
-    pub fn expire_cache_df(&mut self, forget_before: MicroSec) {
-        log::debug!("Expire cache {}", time_string(forget_before));
-        let cache_timing = TradeTable::ohlcv_start(forget_before);
-        self.cache_df = select_df(&self.cache_df, cache_timing, 0);
-        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<()> {
-        log::debug!("update_cache_df {} -> {}", start_time, end_time);
-
-        let df_start_time: i64;
-
-        if end_time == 0 {
-            end_time = NOW();
-        }
-
-        let cache_time = end_time - start_time;
-        if self.cache_duration < cache_time {
-            log::debug!("update cache duration {}", self.cache_duration);
-            self.cache_duration = cache_time;
-        }
-
-        match start_time_df(&self.cache_df) {
-            Some(t) => {
-                df_start_time = t;
-            }
-            _ => {
-                log::debug!(
-                    "cache update all {} -> {}",
-                    time_string(start_time),
-                    time_string(end_time)
-                );
-                // no cache / update all
-                self.load_df(start_time, end_time)?;
-
-                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(
-                    &self.cache_df,
-                    TradeTable::ohlcv_start(df_start_time),
-                    df_end_time,
-                    TradeTable::OHLCV_WINDOW_SEC,
-                )?;
-                return Ok(()); // update cache all.
-            }
-        }
-
-        let df_end_time = end_time_df(&self.cache_df).unwrap();
-
-        // load data and merge cache
-        if start_time < df_start_time {
-            // TODO: ロードされた範囲内でのUpdateに変更する。
-            let df1 = &self.select_df_from_db(start_time, df_start_time)?;
-
-            let len = df1.shape().0;
-            // データがあった場合のみ更新
-            if 0 < len {
-                let new_df_start_time = start_time_df(&df1).unwrap();
-                let new_df_end_time = end_time_df(&df1).unwrap();
-
-                self.cache_df = merge_df(&df1, &self.cache_df);
-
-                // update ohlcv
-                let ohlcv1_start = TradeTable::ohlcv_start(new_df_start_time);
-                let ohlcv1_end = TradeTable::ohlcv_start(new_df_end_time);
-
-                log::debug!(
-                    "ohlcVV cache update diff before {} -> {}",
-                    time_string(ohlcv1_start),
-                    time_string(ohlcv1_end)
-                );
-                let ohlcv1 = ohlcvv_df(
-                    &self.cache_df,
-                    ohlcv1_start,
-                    ohlcv1_end,
-                    TradeTable::OHLCV_WINDOW_SEC,
-                )?;
-
-                if ohlcv1.shape().0 != 0 {
-                    let ohlcv2 = select_df(&self.cache_ohlcvv, ohlcv1_end, 0);
-                    self.cache_ohlcvv = merge_df(&ohlcv1, &ohlcv2);
-                }
-            }
-        } else {
-            // expire cache ducarion * 2
-            if df_start_time < start_time - self.cache_duration * 2 {
-                self.expire_cache_df(start_time - self.cache_duration);
-            }
-        }
-
-        if df_end_time < end_time {
-            // 2日先までキャッシュを先読み
-            let df2 = &self.select_df_from_db(df_end_time, end_time + DAYS(2))?;
-
-            log::debug!(
-                "load data after cache(2days) df1={:?} df2={:?}",
-                self.cache_df.shape(),
-                df2.shape()
-            );
-
-            if df2.shape().0 != 0 {
-                let new_df_start_time = start_time_df(&df2).unwrap();
-                let new_df_end_time = end_time_df(&df2).unwrap();
-                // update ohlcv
-                let ohlcv2_start = TradeTable::ohlcv_start(new_df_start_time);
-                let ohlcv2_end = TradeTable::ohlcv_start(new_df_end_time);
-
-                log::debug!(
-                    "load data AFTER cache df1={:?} df2={:?}",
-                    self.cache_df.shape(),
-                    df2.shape()
-                );
-                self.cache_df = merge_df(&self.cache_df, &df2);
-
-                log::debug!(
-                    "ohlcVV cache update diff after {} ",
-                    time_string(ohlcv2_start),
-                );
-                let ohlcv1 = select_df(&self.cache_ohlcvv, 0, ohlcv2_start);
-                let ohlcv2 = ohlcvv_df(
-                    &self.cache_df,
-                    ohlcv2_start,
-                    ohlcv2_end,
-                    TradeTable::OHLCV_WINDOW_SEC,
-                )?;
-
-                self.cache_ohlcvv = merge_df(&ohlcv1, &ohlcv2);
-            }
-        }
-
-        Ok(())
-    }
-
-    pub fn ohlcvv_df(
-        &mut self,
-        mut start_time: MicroSec,
-        end_time: MicroSec,
-        time_window_sec: i64,
-    ) -> 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 {
-            ohlcvv_from_ohlcvv_df(&self.cache_ohlcvv, start_time, end_time, time_window_sec)
-        } else {
-            ohlcvv_df(&self.cache_df, start_time, end_time, time_window_sec)
-        }
-    }
-
-    pub fn ohlcvv_array(
-        &mut self,
-        mut start_time: MicroSec,
-        end_time: MicroSec,
-        time_window_sec: i64,
-    ) -> anyhow::Result<ndarray::Array2<f64>> {
-        start_time = TradeTable::ohlcv_start(start_time); // 開始tickは確定足、終了は未確定足もOK.
-
-        let df = self.ohlcvv_df(start_time, end_time, time_window_sec)?;
-
-        let array: ndarray::Array2<f64> = df
-            .select(&[
-                KEY::time_stamp,
-                KEY::order_side,
-                KEY::open,
-                KEY::high,
-                KEY::low,
-                KEY::close,
-                KEY::volume,
-                // KEY::count,
-                KEY::start_time,
-                KEY::end_time,
-            ])
-            .unwrap()
-            .to_ndarray::<Float64Type>(IndexOrder::C)
-            .unwrap();
-
-        Ok(array)
-    }
-
-    pub fn py_ohlcvv(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-        window_sec: i64,
-    ) -> anyhow::Result<Py<PyArray2<f64>>> {
-        let array = self.ohlcvv_array(start_time, end_time, window_sec)?;
-
-        let r = Python::with_gil(|py| {
-            let py_array2: &PyArray2<f64> = array.into_pyarray(py);
-
-            return py_array2.to_owned();
-        });
-
-        return Ok(r);
-    }
-
-    pub fn py_ohlcvv_polars(
-        &mut self,
-        mut start_time: MicroSec,
-        end_time: MicroSec,
-        window_sec: i64,
-    ) -> anyhow::Result<PyDataFrame> {
-        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)?;
-
-        let df = convert_timems_to_datetime(&mut df).clone();
-
-        return Ok(PyDataFrame(df));
-    }
-
-    pub fn ohlcv_df(
-        &mut self,
-        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 {
-            ohlcv_from_ohlcvv_df(&self.cache_ohlcvv, start_time, end_time, time_window_sec)
-        } else {
-            ohlcv_df(&self.cache_df, start_time, end_time, time_window_sec)
-        }
-    }
-
-    pub fn ohlcv_array(
-        &mut self,
-        mut start_time: MicroSec,
-        end_time: MicroSec,
-        time_window_sec: i64,
-    ) -> anyhow::Result<ndarray::Array2<f64>> {
-        start_time = TradeTable::ohlcv_start(start_time); // 開始tickは確定足、終了は未確定足もOK.
-
-        let df = self.ohlcv_df(start_time, end_time, time_window_sec)?;
-
-        let array: ndarray::Array2<f64> = df
-            .select(&[
-                KEY::time_stamp,
-                KEY::open,
-                KEY::high,
-                KEY::low,
-                KEY::close,
-                KEY::volume,
-                // KEY::count,
-            ])
-            .unwrap()
-            .to_ndarray::<Float64Type>(IndexOrder::C)
-            .unwrap();
-
-        Ok(array)
-    }
-
-    pub fn py_ohlcv(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-        window_sec: i64,
-    ) -> anyhow::Result<Py<PyArray2<f64>>> {
-        let array = self.ohlcv_array(start_time, end_time, window_sec)?;
-
-        let r = Python::with_gil(|py| {
-            let py_array2: &PyArray2<f64> = array.into_pyarray(py);
-
-            return py_array2.to_owned();
-        });
-
-        return Ok(r);
-    }
-
-    pub fn py_ohlcv_polars(
-        &mut self,
-        mut start_time: MicroSec,
-        end_time: MicroSec,
-        window_sec: i64,
-    ) -> anyhow::Result<PyDataFrame> {
-        start_time = TradeTable::ohlcv_start(start_time); // 開始tickは確定足、終了は未確定足もOK.
-
-        let mut df = self.ohlcv_df(start_time, end_time, window_sec)?;
-        let df = convert_timems_to_datetime(&mut df).clone();
-        let df = PyDataFrame(df);
-
-        return Ok(df);
-    }
-
-    pub fn py_vap(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-        price_unit: i64,
-    ) -> anyhow::Result<PyDataFrame> {
-        let df = self.vap(start_time, end_time, price_unit)?;
-
-        let py_df = PyDataFrame(df);
-
-        Ok(py_df)
-    }
-
-    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);
-
-        Ok(df)
-    }
-
-    pub fn py_select_trades_polars(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-    ) -> anyhow::Result<PyDataFrame> {
-        let mut df = self.select_df_from_db(start_time, end_time)?;
-        let df = convert_timems_to_datetime(&mut df).clone();
-        let df = PyDataFrame(df);
-
-        return Ok(df);
-    }
-
-    pub fn py_select_trades(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-    ) -> anyhow::Result<Py<PyArray2<f64>>> {
-        let array = self.select_array(start_time, end_time)?;
-
-        let r = Python::with_gil(|py| {
-            let py_array2: &PyArray2<f64> = array.into_pyarray(py);
-
-            return py_array2.to_owned();
-        });
-
-        return Ok(r);
-    }
-
-    pub fn select_array(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-    ) -> anyhow::Result<ndarray::Array2<f64>> {
-        self.update_cache_df(start_time, end_time)?;
-
-        let trades = self.select_df_from_db(start_time, end_time)?;
-
-        let array: ndarray::Array2<f64> = trades
-            .select(&[KEY::time_stamp, KEY::price, KEY::size, KEY::order_side])
-            .unwrap()
-            .to_ndarray::<Float64Type>(IndexOrder::C)?;
-
-        Ok(array)
-    }
-
-    pub fn info(&self) -> String {
-        let min = self.start_time().unwrap_or_default();
-        let max = self.end_time(0).unwrap_or_default();
-
-        return format!(
-            "{{\"start\": {}, \"end\": {}}}",
-            time_string(min),
-            time_string(max)
-        );
-    }
-
-    pub fn get_file_name(&self) -> String {
-        return self.file_name.clone();
-    }
-
-    pub fn _repr_html_(&self) -> String {
-        let min = self.start_time().unwrap_or_default();
-        let max = self.end_time(0).unwrap_or_default();
-
-        return format!(
-            r#"
-            <table>
-            <caption>Trade Database info table</caption>
-            <tr><th>start</th><th>end</th></tr>
-            <tr><td>{:?}</td><td>{:?}</td></tr>
-            <tr><td>{:?}</td><td>{:?}</td></tr>
-            <tr><td><b>days=</b></td><td>{}</td></tr>                
-            </table>                
-            "#,
-            min,
-            max,
-            time_string(min),
-            time_string(max),
-            (max - min) / DAYS(1),
-        );
-    }
 
     /// 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.
@@ -1244,7 +612,6 @@ impl TradeTable {
         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)?;
@@ -1262,7 +629,6 @@ impl TradeTable {
             "select time_stamp from trades where $1 < time_stamp order by time_stamp desc limit 1";
 
         let r = self
-            .connection
             .connection
             .query_row(sql, [search_from], |row| {
                 let max: i64 = row.get(0)?;
@@ -1378,7 +744,7 @@ impl TradeTable {
             where $2 < sub_time order by time_stamp
         "#;
 
-        let mut statement = self.connection.connection.prepare(sql).unwrap();
+        let mut statement = self.connection.prepare(sql).unwrap();
         let param = vec![start_time, allow_size];
 
         let chunk_iter = statement
@@ -1482,133 +848,19 @@ impl TradeTable {
             time_chunk
         };
 
-        let days_gap = TradeTable::time_chunks_to_days(&time_gap);
+        let days_gap = Self::time_chunks_to_days(&time_gap);
         log::debug!("GAP TIME: {:?}", time_gap);
         log::debug!("GAP DAYS: {:?}", days_gap);
 
         return days_gap;
     }
 
-    pub fn insert_records(&mut self, trades: &Vec<Trade>) -> anyhow::Result<i64> {
-        return self.connection.insert_records(trades);
-    }
-
-    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, force)?;
-        let unfix_time = self.first_unfix_time(fix_time)?;
-
-        Ok((fix_time, unfix_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> {
-        self.connection.first_unfix_time(start_time)
-    }
-}
-
-impl TradeTable {
-    fn open(name: &str) -> anyhow::Result<Self> {
-        let conn = TradeTableDb::open(name)?;
-
-        log::debug!("db open success{}", name);
-        let df = TradeBuffer::new().to_dataframe();
-        // let ohlcv = ohlcv_df(&df, 0, 0, TradeTable::OHLCV_WINDOW_SEC);
-        let ohlcv = make_empty_ohlcvv();
-
-        Ok(TradeTable {
-            file_name: name.to_string(),
-            connection: conn,
-            cache_df: df,
-            cache_ohlcvv: ohlcv,
-            cache_duration: 0,
-            tx: None,
-            handle: None,
-        })
-    }
-
-    pub fn create_table_if_not_exists(&mut self) -> anyhow::Result<()> {
-        self.connection.create_table_if_not_exists()
-    }
 
-    pub fn drop_table(&self) -> anyhow::Result<()> {
-        if self.is_running() {
-            println!("WARNING: thread is running, may cause dead lock");
-            println!("Execute drop_table() before download() or start_market_stream().");
-        }
-
-        self.connection.drop_table()
-    }
 
-    pub fn vacuum(&self) -> anyhow::Result<()> {
-        self.connection.vacuum()
-    }
-
-    pub fn recreate_table(&mut self) -> anyhow::Result<()> {
-        self.connection.recreate_table()
-    }
-
-    /// Selects trades from the database within the specified time range and applies the given closure to each trade.
-    ///
-    /// # Arguments
-    ///
-    /// * `from_time` - The start time of the time range to select trades from.
-    /// * `to_time` - The end time of the time range to select trades from.
-    /// * `f` - A closure that will be applied to each selected trade.
-    ///
-    /// # Example
-    ///
-    /// ```
-    /// use rbot::db::sqlite::SqliteDb;
-    /// use rbot::models::Trade;
-    ///
-    /// let mut db = SqliteDb::new(":memory:").unwrap();
-    /// let from_time = 0;
-    /// let to_time = 100;
-    /// db.insert_trade(&Trade::new(from_time, 1.0, 1.0)).unwrap();
-    /// db.insert_trade(&Trade::new(to_time + 1, 2.0, 2.0)).unwrap();
-    /// let mut trades = Vec::new();
-    /// db.select(from_time, to_time, |trade| trades.push(trade.clone()));
-    /// assert_eq!(trades.len(), 1);
-    /// assert_eq!(trades[0].timestamp, from_time);
-    /// ```
-    pub fn select<F>(
-        &mut self,
-        start_time: MicroSec,
-        end_time: MicroSec,
-        f: F,
-    ) -> anyhow::Result<()>
-    where
-        F: FnMut(&Trade) -> anyhow::Result<()>,
-    {
-        self.connection.select(start_time, end_time, f)
-    }
 }
 
-////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
-///   Test Suite
-////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
 
+/*
 #[cfg(test)]
 mod test_transaction_table {
     use rust_decimal_macros::dec;
@@ -1625,14 +877,14 @@ mod test_transaction_table {
 
     #[test]
     fn test_open() {
-        let result = TradeTable::open("test.db");
+        let result = TradeDataFrame::open("test.db");
 
         assert!(result.is_ok());
     }
 
     #[test]
     fn test_create_table_and_drop() {
-        let mut tr = TradeTable::open("test.db").unwrap();
+        let mut tr = TradeDataFrame::open("test.db").unwrap();
 
         let r = tr.create_table_if_not_exists();
         assert!(r.is_ok());
@@ -1644,7 +896,7 @@ mod test_transaction_table {
     #[test]
     fn test_insert_table() {
         init_debug_log();
-        let mut tr = TradeTable::open("test.db").unwrap();
+        let mut tr = TradeDataFrame::open("test.db").unwrap();
         let r = tr.recreate_table();
         assert!(r.is_ok());
 
@@ -1677,30 +929,17 @@ mod test_transaction_table {
         assert!(r.is_ok());
     }
 
-    #[test]
-    fn test_select_array() {
-        init_debug_log();
-
-        let db = TradeTable::open("test.db");
-        assert!(db.is_ok());
-        let mut db = db.unwrap();
-
-        let array = db.select_array(0, 0);
-
-        println!("{:?}", array);
-    }
-
     #[test]
     fn test_info() {
         init_debug_log();
-        let db = TradeTable::open("test.db").unwrap();
+        let db = TradeDataFrame::open("test.db").unwrap();
         println!("{}", db.info());
     }
 
     #[test]
     fn test_start_time() {
         init_debug_log();
-        let db = TradeTable::open("test.db").unwrap();
+        let db = TradeDataFrame::open("test.db").unwrap();
 
         let start_time = db.start_time();
 
@@ -1712,7 +951,7 @@ mod test_transaction_table {
     #[test]
     fn test_end_time() {
         init_debug_log();
-        let db = TradeTable::open("test.db").unwrap();
+        let db = TradeDataFrame::open("test.db").unwrap();
 
         let end_time = db.end_time(0);
 
@@ -1724,7 +963,7 @@ mod test_transaction_table {
     #[test]
     fn test_select_gap_chunks() -> anyhow::Result<()> {
         let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", false);
-        let db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        let db = TradeDataFrame::open(db_name.to_str().unwrap()).unwrap();
 
         let chunks = db.select_gap_chunks(NOW() - DAYS(1), NOW(), 1_000_000 * 13)?;
 
@@ -1745,7 +984,7 @@ mod test_transaction_table {
     #[test]
     fn test_select_time_chunk_from() {
         let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", false);
-        let db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        let db = TradeDataFrame::open(db_name.to_str().unwrap()).unwrap();
 
         let chunks = db.find_time_chunk_from(NOW() - DAYS(1), NOW(), 1_000_000 * 10);
 
@@ -1759,7 +998,7 @@ mod test_transaction_table {
     #[test]
     fn test_select_time_chunk_to() {
         let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", false);
-        let db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        let db = TradeDataFrame::open(db_name.to_str().unwrap()).unwrap();
 
         let chunks = db.find_time_chunk_to(NOW() - DAYS(1), NOW(), 1_000_000 * 120);
 
@@ -1773,7 +1012,7 @@ mod test_transaction_table {
     #[test]
     fn test_select_time_chunks() -> anyhow::Result<()> {
         let db_name = db_full_path("FTX", "SPOT", "BTC-PERP", false);
-        let db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        let db = TradeDataFrame::open(db_name.to_str().unwrap()).unwrap();
 
         let chunks = db.select_time_chunks_in_db(NOW() - DAYS(1), NOW(), 1_000_000 * 10)?;
 
@@ -1792,7 +1031,7 @@ mod test_transaction_table {
             end: DAYS(1) + 100,
         }];
 
-        assert_eq!(TradeTable::time_chunks_to_days(&chunks), vec![DAYS(1)]);
+        assert_eq!(TradeDataFrame::time_chunks_to_days(&chunks), vec![DAYS(1)]);
 
         let chunks = vec![TimeChunk {
             start: DAYS(1),
@@ -1800,7 +1039,7 @@ mod test_transaction_table {
         }];
 
         assert_eq!(
-            TradeTable::time_chunks_to_days(&chunks),
+            TradeDataFrame::time_chunks_to_days(&chunks),
             vec![DAYS(1), DAYS(2), DAYS(3)]
         );
     }
@@ -1810,7 +1049,7 @@ mod test_transaction_table {
         init_log();
         let db_name = db_full_path("BN", "SPOT", "BTCBUSD", false);
 
-        let mut db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        let mut db = TradeDataFrame::open(db_name.to_str().unwrap()).unwrap();
 
         let start_timer = NOW();
         let now = NOW();
@@ -1851,7 +1090,7 @@ mod test_transaction_table {
         init_log();
 
         let db_name = db_full_path("BN", "SPOT", "BTCBUSD", false);
-        let mut db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        let mut db = TradeDataFrame::open(db_name.to_str().unwrap()).unwrap();
 
         let start = NOW();
         let ohlcv = db.select(NOW() - DAYS(2), NOW(), |_trade| Ok(()));
@@ -1859,21 +1098,11 @@ mod test_transaction_table {
         println!("{:?} / {} microsec", ohlcv, NOW() - start);
     }
 
-    #[test]
-    fn test_select_df() {
-        let db_name = db_full_path("BN", "SPOT", "BTCBUSD", false);
-        let mut db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
-
-        let df = db.select_df_from_db(NOW() - DAYS(2), NOW());
-
-        println!("{:?}", df);
-    }
-
     #[test]
     fn test_update_cache() -> anyhow::Result<()> {
         init_log();
         let db_name = db_full_path("BN", "SPOT", "BTCBUSD", false);
-        let mut db = TradeTable::open(db_name.to_str().unwrap()).unwrap();
+        let mut db = TradeDataFrame::open(db_name.to_str().unwrap()).unwrap();
 
         db.update_cache_df(NOW() - DAYS(2), NOW())?;
 
@@ -1882,7 +1111,7 @@ mod test_transaction_table {
 
     #[tokio::test]
     async fn test_start_thread() {
-        let mut table = TradeTable::open(
+        let mut table = TradeDataFrame::open(
             db_full_path("BN", "SPOT", "BTCBUSD", false)
                 .to_str()
                 .unwrap(),
@@ -1929,7 +1158,7 @@ mod test_transaction_table {
     fn test_wal_mode() -> anyhow::Result<()> {
         //let table = TradeTable::open(db_full_path("BN", "SPOT", "BTCBUSD").to_str().unwrap()).unwrap();
 
-        TradeTableDb::set_wal_mode(
+        TradeDb::set_wal_mode(
             db_full_path("BN", "SPOT", "BTCBUSD", false)
                 .to_str()
                 .unwrap(),
@@ -1942,7 +1171,7 @@ mod test_transaction_table {
     fn test_table_is_exsit() {
         //let db_name = db_full_path("BN", "SPOT", "BTCUSDT");
 
-        let mut db = TradeTableDb::open("/tmp/rbottest.db").unwrap();
+        let mut db = TradeDb::open("/tmp/rbottest.db").unwrap();
 
         if db.is_table_exsit() {
             print!("table is exist");
@@ -1954,8 +1183,9 @@ mod test_transaction_table {
     #[test]
     fn test_get_db() {
         init_debug_log();
-        let mut db = TradeTable::get("/tmp/rbottest.db").unwrap();
+        let mut db = TradeDataFrame::get("/tmp/rbottest.db").unwrap();
 
         println!("{:?}", db);
     }
 }
+*/
diff --git a/rbot_lib/src/db/tradedf.rs b/rbot_lib/src/db/tradedf.rs
new file mode 100644
index 00000000..9aacc95d
--- /dev/null
+++ b/rbot_lib/src/db/tradedf.rs
@@ -0,0 +1,620 @@
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+};
+
+use crossbeam_channel::Sender;
+use once_cell::sync::Lazy;
+use polars::frame::DataFrame;
+use pyo3_polars::PyDataFrame;
+
+use crate::{
+    common::{time_string, MarketConfig, MicroSec, Trade, DAYS, FLOOR_DAY, HHMM, NOW},
+    db::{append_df, end_time_df, make_empty_ohlcvv, merge_df, ohlcv_start, ohlcvv_df, select_df, start_time_df, TradeBuffer, KEY},
+    net::RestApi,
+};
+
+use super::{
+    convert_timems_to_datetime, ohlcv_df, ohlcv_floor_fix_time, ohlcv_from_ohlcvv_df,
+    ohlcvv_from_ohlcvv_df, vap_df, TradeArchive, TradeDb,
+};
+use anyhow::anyhow;
+
+//static EXCHANGE_DB_CACHE: Lazy<Mutex<HashMap<String, Arc<Mutex<TradeDataFrame>>>>> =
+static TRADE_DATAFRAME_CACHE: Lazy<Mutex<HashMap<String, Arc<Mutex<TradeDataFrame>>>>> =
+    Lazy::new(|| Mutex::new(HashMap::new()));
+
+fn insert_trade_dataframe_cache(
+    config: &MarketConfig,
+    production: bool,
+    trade_dataframe: TradeDataFrame,
+) -> anyhow::Result<Arc<Mutex<TradeDataFrame>>> {
+    let key = config.key_string(production);
+
+    let mut lock = TRADE_DATAFRAME_CACHE.lock().unwrap();
+    let dataframe = Arc::new(Mutex::new(trade_dataframe));
+    lock.insert(key, dataframe.clone());
+
+    Ok(dataframe)
+}
+
+fn get_trade_dataframe_cache(
+    config: &MarketConfig,
+    production: bool,
+) -> anyhow::Result<Arc<Mutex<TradeDataFrame>>> {
+    let key = config.key_string(production);
+
+    let lock = TRADE_DATAFRAME_CACHE.lock().unwrap();
+    let trade_dataframe = lock.get(&key);
+
+    if trade_dataframe.is_some() {
+        return Ok(trade_dataframe.unwrap().clone());
+    }
+
+    Err(anyhow!("no TradeDataFrame [key={}] found", key))
+}
+
+pub const OHLCV_WINDOW_SEC: i64 = 60; // min
+
+pub struct TradeDataFrame {
+    db: TradeDb,
+    archive: TradeArchive,
+
+    cache_df: DataFrame,
+    cache_ohlcvv: DataFrame,
+    cache_duration: MicroSec,
+}
+
+impl TradeDataFrame {
+    pub fn get(config: &MarketConfig, production: bool) -> anyhow::Result<Arc<Mutex<Self>>> {
+        let trade_dataframe = get_trade_dataframe_cache(config, production);
+
+        if trade_dataframe.is_ok() {
+            return Ok(trade_dataframe.unwrap());
+        }
+
+        let trade_dataframe = TradeDataFrame::open(config, production)?;
+
+        /* TODO: Consider if multiprocess access the same database
+        log::debug!("vaccum start {:?}", config.key_string(production));
+        let r = trade_dataframe.vacuum();
+        if r.is_err() {
+            log::warn!("db vaccum error {:?}", r);
+        }
+        log::debug!("vaccum end");
+        */
+
+        let trade_data_frame = insert_trade_dataframe_cache(config, production, trade_dataframe)?;
+
+        Ok(trade_data_frame)
+    }
+
+    pub fn get_archive(&self) -> TradeArchive {
+        self.archive.clone()
+    }
+
+    pub fn start_time(&self) -> anyhow::Result<MicroSec> {
+        let archive_start = self.archive.start_time();
+
+        if let Ok(start) = archive_start {
+            return Ok(start);
+        }
+
+        let db_start = self.db.start_time();
+        
+        if let Ok(start)  = db_start {
+            return Ok(start);
+        }
+
+        Err(anyhow!("no data in archive or db"))
+    }
+
+    pub fn end_time(&self) -> anyhow::Result<MicroSec> {
+        let archive_end = self.archive.end_time();
+
+        let archive_end = archive_end.unwrap_or_default();
+
+        let db_end = self.db.end_time(NOW() + HHMM(1, 0));
+
+        if let Ok(end_time) = db_end {
+            if archive_end < end_time {
+                return Ok(end_time);
+            }
+        }
+
+        if archive_end != 0 {
+            return Ok(archive_end);
+        }
+
+        Err(anyhow!("no data in archive or db"))
+    }
+
+    /// 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,
+        force: bool,
+    ) -> anyhow::Result<Vec<Trade>> {
+        self.db.make_expire_control_message(now, force)
+    }
+
+    // TODO: review
+    pub fn set_cache_ohlcvv(&mut self, df: DataFrame) -> anyhow::Result<()> {
+        let start_time: MicroSec = df
+            .column(KEY::time_stamp)
+            .unwrap()
+            .min()
+            .unwrap()
+            .unwrap_or(0);
+        let end_time: MicroSec = df
+            .column(KEY::time_stamp)
+            .unwrap()
+            .max()
+            .unwrap()
+            .unwrap_or(0);
+
+        let head = select_df(&self.cache_ohlcvv, 0, start_time);
+        let tail = select_df(&self.cache_ohlcvv, end_time, 0);
+
+        log::debug!(
+            "set_cache_ohlcvv head {} /tail {}/ df {}",
+            head.shape().0,
+            tail.shape().0,
+            df.shape().0
+        );
+
+        log::debug!("df {:?}", df.head(Some(2)));
+        log::debug!("head {:?}", head.head(Some(2)));
+        log::debug!("tail {:?}", tail.head(Some(2)));
+
+        let df = append_df(&head, &df)?;
+        let df = append_df(&df, &tail)?;
+
+        self.cache_ohlcvv = df;
+
+        Ok(())    
+    }
+
+    // TODO: rename to open_channel
+    pub fn start_thread(&mut self) -> anyhow::Result<Sender<Vec<Trade>>> {
+        self.db.open_channel()
+    }
+
+    pub async fn download_archive<T>(
+        &mut self,
+        api: &T,
+        ndays: i64,
+        force: bool,
+        verbose: bool,
+    ) -> anyhow::Result<i64>
+    where
+        T: RestApi,
+    {
+        self.archive.download(api, ndays, force, verbose).await
+    }
+
+
+    /*
+    pub async fn download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
+        if verbose {
+            println!("async_download_lastest");
+            flush_log();
+        }
+
+        let config = self.config;
+
+        let trades = self.api..T::get_recent_trades(&server_config, &config).await?;
+        let rec = trades.len() as i64;
+
+        log::debug!("rec: {}", rec);
+
+        if rec == 0 {
+            return Ok(0);
+        }
+
+        if verbose {
+            println!("from rec: {:?}", trades[0].__str__());
+            println!("to   rec: {:?}", trades[(rec as usize) - 1].__str__());
+            println!("rec: {}", rec);
+            flush_log();
+        }
+        let tx = self.start_db_thread().await;
+
+        let start_time = trades.iter().map(|trade| trade.time).min();
+
+        if let Some(start_time) = start_time {
+            let expire_control = self
+                .get_db()
+                .lock()
+                .unwrap()
+                .make_expire_control_message(start_time, false);
+
+            if expire_control.is_err() {
+                println!("make_expire_control_message {:?}", expire_control.err());
+            } else {
+                let expire_control = expire_control.unwrap();
+                if verbose {
+                    println!("expire control from {:?}", expire_control[0].__str__());
+                    println!("expire control to   {:?}", expire_control[1].__str__());
+                }
+
+                tx.send(expire_control)?;
+            }
+        }
+
+        tx.send(trades)?;
+
+        Ok(rec)
+    }
+    */
+
+    pub fn get_cache_duration(&self) -> MicroSec {
+        return self.cache_duration;
+    }
+
+    pub fn reset_cache_duration(&mut self) {
+        self.cache_duration = 0;
+    }
+
+    pub fn select_cachedf(&mut self, start_time: MicroSec, end_time: MicroSec) -> anyhow::Result<DataFrame> {
+        let archive_end = self.archive.end_time().unwrap_or_default();
+
+        let df = if start_time < archive_end {
+            let df1 = self.archive.select_cachedf(start_time, end_time)?;
+            let df2 = self.db.select_cachedf(archive_end, end_time)?;
+            append_df(&df1, &df2)
+        }
+        else {
+            self.db.select_cachedf(start_time, end_time)
+        };
+
+        df
+    }
+
+    pub fn load_cachedf(
+        &mut self,
+        start_time: MicroSec,
+        end_time: MicroSec,
+    ) -> anyhow::Result<i64> {
+        self.cache_df = self.db.select_cachedf(start_time, end_time)?;
+
+        Ok(self.cache_df.shape().0 as i64)
+    }
+
+    pub fn update_cache_all(&mut self) -> anyhow::Result<()> {
+        self.update_cache_df(0, 0)
+    }
+
+    // TODO: 日付単位に変更する。
+    pub fn expire_cache_df(&mut self, forget_before: MicroSec) {
+        log::debug!("Expire cache {}", time_string(forget_before));
+        let cache_timing = ohlcv_start(forget_before);
+        self.cache_df = select_df(&self.cache_df, cache_timing, 0);
+        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<()> {
+        log::debug!("update_cache_df {} -> {}", start_time, end_time);
+
+        let df_start_time: i64;
+
+        if end_time == 0 {
+            end_time = NOW();
+        }
+
+        let cache_time = end_time - start_time;
+        if self.cache_duration < cache_time {
+            log::debug!("update cache duration {}", self.cache_duration);
+            self.cache_duration = cache_time;
+        }
+
+        match start_time_df(&self.cache_df) {
+            Some(t) => {
+                df_start_time = t;
+            }
+            _ => {
+                log::debug!(
+                    "cache update all {} -> {}",
+                    time_string(start_time),
+                    time_string(end_time)
+                );
+                // no cache / update all
+                self.load_cachedf(start_time, end_time)?;
+
+                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(
+                    &self.cache_df,
+                    ohlcv_start(df_start_time),
+                    df_end_time,
+                    OHLCV_WINDOW_SEC,
+                )?;
+                return Ok(()); // update cache all.
+            }
+        }
+
+        let df_end_time = end_time_df(&self.cache_df).unwrap();
+
+        // load data and merge cache
+        if start_time < df_start_time {
+            let df1 = &self.db.select_cachedf(start_time, df_start_time)?;
+
+            let len = df1.shape().0;
+            // データがあった場合のみ更新
+            if 0 < len {
+                let new_df_start_time = start_time_df(&df1).unwrap();
+                let new_df_end_time = end_time_df(&df1).unwrap();
+
+                self.cache_df = merge_df(&df1, &self.cache_df)?;
+
+                // update ohlcv
+                let ohlcv1_start = ohlcv_start(new_df_start_time);
+                let ohlcv1_end = ohlcv_start(new_df_end_time); // only use fixed data
+
+                log::debug!(
+                    "ohlcVV cache update diff before {} -> {}",
+                    time_string(ohlcv1_start),
+                    time_string(ohlcv1_end)
+                );
+                let ohlcv1 = ohlcvv_df(&self.cache_df, ohlcv1_start, ohlcv1_end, OHLCV_WINDOW_SEC)?;
+
+                if ohlcv1.shape().0 != 0 {
+                    let ohlcv2 = select_df(&self.cache_ohlcvv, ohlcv1_end, 0);
+                    self.cache_ohlcvv = merge_df(&ohlcv1, &ohlcv2)?;
+                }
+            }
+        } else {
+            // expire cache ducarion * 2
+            if df_start_time < start_time - self.cache_duration * 2 {
+                self.expire_cache_df(start_time - self.cache_duration);
+            }
+        }
+
+        if df_end_time < end_time {
+            // 2日先までキャッシュを先読み
+            let df2 = &self
+                .db
+                .select_cachedf(df_end_time, FLOOR_DAY(end_time + DAYS(3)))?;
+
+            log::debug!(
+                "load data after cache(2days) df1={:?} df2={:?}",
+                self.cache_df.shape(),
+                df2.shape()
+            );
+
+            if df2.shape().0 != 0 {
+                let new_df_start_time = start_time_df(&df2).unwrap();
+                let new_df_end_time = end_time_df(&df2).unwrap();
+                // update ohlcv
+                let ohlcv2_start = ohlcv_start(new_df_start_time);
+                let ohlcv2_end = ohlcv_start(new_df_end_time); // use only fix data
+
+                log::debug!(
+                    "load data AFTER cache df1={:?} df2={:?}",
+                    self.cache_df.shape(),
+                    df2.shape()
+                );
+                self.cache_df = merge_df(&self.cache_df, &df2)?;
+
+                log::debug!(
+                    "ohlcVV cache update diff after {} ",
+                    time_string(ohlcv2_start),
+                );
+                let ohlcv1 = select_df(&self.cache_ohlcvv, 0, ohlcv2_start);
+                let ohlcv2 = ohlcvv_df(&self.cache_df, ohlcv2_start, ohlcv2_end, OHLCV_WINDOW_SEC)?;
+
+                self.cache_ohlcvv = merge_df(&ohlcv1, &ohlcv2)?;
+            }
+        }
+
+        Ok(())
+    }
+
+    pub fn ohlcvv_df(
+        &mut self,
+        mut start_time: MicroSec,
+        end_time: MicroSec,
+        time_window_sec: i64,
+    ) -> anyhow::Result<DataFrame> {
+        start_time = ohlcv_floor_fix_time(start_time, time_window_sec); // 開始tickは確定足、終了は未確定足もOK.
+
+        self.update_cache_df(start_time, end_time)?;
+
+        if time_window_sec % OHLCV_WINDOW_SEC == 0 {
+            ohlcvv_from_ohlcvv_df(&self.cache_ohlcvv, start_time, end_time, time_window_sec)
+        } else {
+            ohlcvv_df(&self.cache_df, start_time, end_time, time_window_sec)
+        }
+    }
+
+    pub fn py_ohlcvv_polars(
+        &mut self,
+        mut start_time: MicroSec,
+        end_time: MicroSec,
+        window_sec: i64,
+    ) -> anyhow::Result<PyDataFrame> {
+        start_time = ohlcv_floor_fix_time(start_time, window_sec); // 開始tickは確定足、終了は未確定足もOK.
+
+        let mut df = self.ohlcvv_df(start_time, end_time, window_sec)?;
+
+        convert_timems_to_datetime(&mut df)?;
+
+        return Ok(PyDataFrame(df));
+    }
+
+    pub fn ohlcv_df(
+        &mut self,
+        mut start_time: MicroSec,
+        end_time: MicroSec,
+        time_window_sec: i64,
+    ) -> anyhow::Result<DataFrame> {
+        start_time = ohlcv_start(start_time); // 開始tickは確定足、終了は未確定足もOK.
+
+        self.update_cache_df(start_time, end_time)?;
+
+        if time_window_sec % OHLCV_WINDOW_SEC == 0 {
+            ohlcv_from_ohlcvv_df(&self.cache_ohlcvv, start_time, end_time, time_window_sec)
+        } else {
+            ohlcv_df(&self.cache_df, start_time, end_time, time_window_sec)
+        }
+    }
+
+    pub fn py_ohlcv_polars(
+        &mut self,
+        mut start_time: MicroSec,
+        end_time: MicroSec,
+        window_sec: i64,
+    ) -> anyhow::Result<PyDataFrame> {
+        start_time = ohlcv_start(start_time); // 開始tickは確定足、終了は未確定足もOK.
+
+        let mut df = self.ohlcv_df(start_time, end_time, window_sec)?;
+        convert_timems_to_datetime(&mut df)?;
+        let df = PyDataFrame(df);
+
+        return Ok(df);
+    }
+
+    pub fn py_vap(
+        &mut self,
+        start_time: MicroSec,
+        end_time: MicroSec,
+        price_unit: i64,
+    ) -> anyhow::Result<PyDataFrame> {
+        let df = self.vap(start_time, end_time, price_unit)?;
+
+        let py_df = PyDataFrame(df);
+
+        Ok(py_df)
+    }
+
+    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);
+
+        Ok(df)
+    }
+
+    pub fn py_select_trades_polars(
+        &mut self,
+        start_time: MicroSec,
+        end_time: MicroSec,
+    ) -> anyhow::Result<PyDataFrame> {
+        let mut df = self.db.select_cachedf(start_time, end_time)?;
+        convert_timems_to_datetime(&mut df)?;
+        let df = PyDataFrame(df);
+
+        return Ok(df);
+    }
+
+    pub fn info(&self) -> String {
+        let min = self.start_time().unwrap_or_default();
+        let max = self.end_time().unwrap_or_default();
+
+        return format!(
+            "{{\"start\": {}, \"end\": {}}}",
+            time_string(min),
+            time_string(max)
+        );
+    }
+
+    pub fn _repr_html_(&self) -> String {
+        let min = self.start_time().unwrap_or_default();
+        let max = self.end_time().unwrap_or_default();
+
+        return format!(
+            r#"
+            <table>
+            <caption>Trade Database info table</caption>
+            <tr><th>start</th><th>end</th></tr>
+            <tr><td>{:?}</td><td>{:?}</td></tr>
+            <tr><td>{:?}</td><td>{:?}</td></tr>
+            <tr><td><b>days=</b></td><td>{}</td></tr>                
+            </table>                
+            "#,
+            min,
+            max,
+            time_string(min),
+            time_string(max),
+            (max - min) / DAYS(1),
+        );
+    }
+
+    pub fn insert_records(&mut self, trades: &Vec<Trade>) -> anyhow::Result<i64> {
+        return self.db.insert_records(trades);
+    }
+
+    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, force)?;
+        let unfix_time = self.first_unfix_time(fix_time)?;
+
+        Ok((fix_time, unfix_time))
+    }
+
+    pub fn latest_fix_trade(
+        &mut self,
+        start_time: MicroSec,
+        force: bool,
+    ) -> anyhow::Result<Option<Trade>> {
+        self.db.latest_fix_trade(start_time, force)
+    }
+
+    pub fn latest_fix_time(
+        &mut self,
+        start_time: MicroSec,
+        force: bool,
+    ) -> anyhow::Result<MicroSec> {
+        self.db.latest_fix_time(start_time, force)
+    }
+
+    pub fn first_unfix_trade(&mut self, start_time: MicroSec) -> anyhow::Result<Option<Trade>> {
+        self.db.first_unfix_trade(start_time)
+    }
+
+    pub fn first_unfix_time(&mut self, start_time: MicroSec) -> anyhow::Result<MicroSec> {
+        self.db.first_unfix_time(start_time)
+    }
+}
+
+impl TradeDataFrame {
+    fn open(config: &MarketConfig, production: bool) -> anyhow::Result<Self> {
+        let conn = TradeDb::open(&config, production)?;
+        log::debug!("db open success");
+        let archive = TradeArchive::new(config, production);
+
+        // setup cache
+        let df = TradeBuffer::new().to_dataframe();
+        let ohlcv = make_empty_ohlcvv();
+
+        Ok(TradeDataFrame {
+            db: conn,
+            archive: archive,
+
+            cache_df: df,
+            cache_ohlcvv: ohlcv,
+            cache_duration: 0,
+        })
+    }
+}
diff --git a/rbot_lib/src/net/rest.rs b/rbot_lib/src/net/rest.rs
index ddf8e791..a3d18282 100644
--- a/rbot_lib/src/net/rest.rs
+++ b/rbot_lib/src/net/rest.rs
@@ -2,259 +2,58 @@
 // Abloultely no warranty.
 
 use anyhow::anyhow;
-use anyhow::ensure;
 use anyhow::Context;
-use chrono::Datelike;
+use reqwest::StatusCode;
 
 // use crossbeam_channel::Receiver;
-use crossbeam_channel::Sender;
-use csv::StringRecord;
-use flate2::read::GzDecoder;
-use polars::chunked_array::ops::ChunkCast;
-use polars::datatypes::DataType;
-use polars::frame::DataFrame;
-use polars::io::csv::read::CsvReadOptions;
-use polars::io::parquet::write::ParquetWriter;
-use polars::io::SerReader;
-use polars::prelude::Series;
-use reqwest::Method;
-use rust_decimal::Decimal;
-use std::io::BufRead;
-use std::io::BufWriter;
-use std::path::PathBuf;
-use std::str::FromStr;
-use std::{
-    fs::File,
-    io::{copy, BufReader, Cursor, Write},
-    path::Path,
-    thread::sleep,
-    time::Duration,
-};
-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,
+    BoardTransfer, Kline, MarketConfig, MicroSec, Order, OrderSide, OrderType, Trade, DAYS, TODAY,
 };
-use crate::db::KEY;
+use polars::frame::DataFrame;
+use reqwest::Method;
+use rust_decimal::Decimal;
 //use crate::db::KEY::low;
+use async_trait::async_trait;
 
-pub trait RestApi<T>
-where
-    T: ServerConfig,
-{
-    async fn get_board_snapshot(server: &T, config: &MarketConfig)
-        -> anyhow::Result<BoardTransfer>;
+pub trait RestApi {
+    async fn get_board_snapshot(&self, config: &MarketConfig) -> anyhow::Result<BoardTransfer>;
 
-    async fn get_recent_trades(server: &T, config: &MarketConfig) -> anyhow::Result<Vec<Trade>>;
+    async fn get_recent_trades(&self, config: &MarketConfig) -> anyhow::Result<Vec<Trade>>;
 
-    fn get_trade_klines(
-        server: &T,
+    async fn get_trade_klines(
+        &self,
         config: &MarketConfig,
         start_time: MicroSec,
         end_time: MicroSec,
-    ) -> impl std::future::Future<Output = anyhow::Result<Vec<Kline>>> + Send;
-    fn new_order(
-        server: &T,
+    ) -> anyhow::Result<Vec<Kline>>;
+
+    async fn new_order(
+        &self,
         config: &MarketConfig,
         side: OrderSide,
         price: Decimal, // when order_type is Market, this value is ignored.
         size: Decimal,
         order_type: OrderType,
         client_order_id: Option<&str>,
-    ) -> impl std::future::Future<Output = anyhow::Result<Vec<Order>>> + Send;
-    fn cancel_order(
-        server: &T,
-        config: &MarketConfig,
-        order_id: &str,
-    ) -> impl std::future::Future<Output = anyhow::Result<Order>> + Send;
-    fn open_orders(
-        server: &T,
-        config: &MarketConfig,
-    ) -> impl std::future::Future<Output = anyhow::Result<Vec<Order>>> + Send;
-
-    fn get_account(
-        server: &T,
-    ) -> impl std::future::Future<Output = anyhow::Result<AccountCoins>> + Send;
-
-    fn history_web_url(server: &T, config: &MarketConfig, date: MicroSec) -> String {
-        let history_web_base = server.get_historical_web_base();
-        let category = &config.trade_category;
-        let symbol = &config.trade_symbol;
-
-        let timestamp = to_naive_datetime(date);
-
-        let yyyy = timestamp.year() as i64;
-        let mm = timestamp.month() as i64;
-        let dd = timestamp.day() as i64;
-
-        Self::format_historical_data_url(&history_web_base, &category, &symbol, yyyy, mm, dd)
-    }
-
-    fn format_historical_data_url(
-        history_web_base: &str,
-        category: &str,
-        symbol: &str,
-        yyyy: i64,
-        mm: i64,
-        dd: i64,
-    ) -> String;
-
-    /// Convert archived CSV trade log record to Trade struct
-    /// 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;
-    fn line_to_trade(rec: &str) -> Trade;
-    fn convert_archive_line(line: &str) -> String;
-    fn archive_has_header() -> bool;
-
-    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 {
-            log::debug!("check log exist = {}({})", time_string(latest), latest);
-
-            if has_archive(latest, &f).await {
-                return Ok(latest);
-            }
-
-            latest -= DAYS(1);
-            i += 1;
-
-            if 5 < i {
-                return Err(anyhow!(
-                    "Find archive retry over {}/{}/{}",
-                    i,
-                    latest,
-                    f(latest)
-                ));
-            }
-        }
-    }
-
-    fn download_archive(
-        server: &T,
-        config: &MarketConfig,
-        tx: &Sender<Vec<Trade>>,
-        date: MicroSec,
-        low_priority: bool,
-        verbose: bool,
-    ) -> impl std::future::Future<Output = anyhow::Result<i64>> {
-        async move {
-            let date = FLOOR_DAY(date);
-            let url = Self::history_web_url(server, config, date);
-            let has_header = Self::archive_has_header();
-
-            download_archive_log(
-                &url,
-                tx,
-                has_header,
-                low_priority,
-                verbose,
-                &Self::rec_to_trade,
-            )
-            .await
-        }
-    }
-
-    async fn has_archive(server: &T, config: &MarketConfig, date: MicroSec) -> bool {
-        has_archive(date, &|d| Self::history_web_url(server, config, d)).await
-    }
-
+    ) -> anyhow::Result<Vec<Order>>;
+    async fn cancel_order(&self, config: &MarketConfig, order_id: &str) -> anyhow::Result<Order>;
+    async fn open_orders(&self, config: &MarketConfig) -> anyhow::Result<Vec<Order>>;
 
-    /// create DataFrame with columns;
-    ///  KEY:time_stamp(Int64), KEY:order_side(Bool), KEY:price(f64), KEY:size(f64)
-    fn logdf_to_archivedf(df: &DataFrame) -> anyhow::Result<DataFrame> {
-        let df = df.clone();
+    async fn get_account(&self) -> anyhow::Result<AccountCoins>;
 
-        let timestamp = df.column("timestamp")?.f64()? * 1_000_000.0;
-        let timestamp = timestamp.cast(&DataType::Int64)?;
+    async fn has_archive(&self, config: &MarketConfig, date: MicroSec) -> anyhow::Result<bool>;
 
-        let timestamp = timestamp.clone();
-        let mut timestamp = Series::from(timestamp.clone());
-        timestamp.rename(KEY::time_stamp);
-
-        let mut id = df.column("trdMatchID")?.clone();
-        id.rename(KEY::id);
-
-        let mut side = df.column("side")?.clone();
-        side.rename(KEY::order_side);
-
-        let mut price = df.column("price")?.clone();
-        price.rename(KEY::price);
-
-        let mut size = df.column("size")?.clone();
-        size.rename(KEY::size);
-
-        let df = DataFrame::new(vec![
-            timestamp,
-            side, 
-            price,
-            size,
-            id,
-        ])?;
-
-        Ok(df)
-    }
+    fn history_web_url(&self, config: &MarketConfig, date: MicroSec) -> String;
+    fn archive_has_header(&self) -> bool;
+    fn logdf_to_archivedf(&self, df: &DataFrame) -> anyhow::Result<DataFrame>;
 }
 
-pub async fn log_download_tmp(url: &str, tmp_dir: &Path) -> anyhow::Result<String> {
-    let client = reqwest::Client::new();
-
-    let response = client
-        .get(url)
-        .header("User-Agent", "Mozilla/5.0")
-        .header("Accept", "text/html")
-        .send()
-        .await
-        .with_context(|| format!("URL get error {}", url))?;
-
-    log::debug!(
-        "Response code = {} / download size {}",
-        response.status().as_str(),
-        response.content_length().unwrap_or_default() // if error, return 0
-    );
-
-    if !response.status().is_success() {
-        return Err(anyhow!("Err: response code {}", response.status().as_str()));
-    }
-
-    let fname = response
-        .url()
-        .path_segments()
-        .and_then(|segments| segments.last())
-        .and_then(|name| if name.is_empty() { None } else { Some(name) })
-        .unwrap_or("tmp.bin");
-
-    let fname = tmp_dir.join(fname);
-    let file_name = 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 mut cursor = Cursor::new(content);
-
-    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();
-
-    log::debug!("download size {}", target.metadata().unwrap().len());
-
-    Ok(file_name.to_string())
-}
+#[async_trait]
+pub trait RestTrait: RestApi + Send + Sync + Clone {}
 
+/*
 async fn download_archive_log<F>(
     url: &str,
     tx: &Sender<Vec<Trade>>,
@@ -452,9 +251,7 @@ where
     }
     urls
 }
-
-const MAX_BUFFER_SIZE: usize = 4096;
-const LOW_QUEUE_SIZE: usize = 5;
+*/
 
 pub async fn do_rest_request(
     method: Method,
@@ -484,22 +281,47 @@ pub async fn do_rest_request(
         .await
         .with_context(|| format!("URL get error {url:}"))?;
 
-    if response.status().as_str() != "200" {
-        return Err(anyhow!(
-            "Response code = {} / download size {:?} / method({:?}) /  response body = {}",
-            response.status().as_str(),
-            response.content_length(),
-            method,
-            &body,
-        ));
+    if response.status().as_str() == "200" {
+        let body = response
+            .text()
+            .await
+            .with_context(|| format!("response text error"))?;
+
+        return Ok(body);
     }
 
-    let body = response
-        .text()
-        .await
-        .with_context(|| format!("response text error"))?;
+    // -----------other errors---------------
+    let status = response.status();
+    match status {
+        StatusCode::NOT_FOUND => {
+            log::error!("NOT FOUND url={}, {}", url, body);
+            println!("NOT FOUND url={}, {}", url, body);
+        },
+        StatusCode::FORBIDDEN |
+        StatusCode::UNAUTHORIZED => {
+            log::error!("AUTH ERROR url={}, {}", url, body);
+            println!("AUTH ERROR url={}, {}", url, body);
+            println!("Please check access key and token");
+        }
+        _ => {
+            let code = status.as_u16();
 
-    Ok(body)
+            if code == 10001 {
+                print!("status code 10001. please check access key and token");
+                log::error!("status code 10001. please check access key and token");
+            } 
+
+            log::error!("request error code={} / body={}", status, body)
+        }
+    }
+
+    Err(anyhow!(
+        "Response code = {} / download size {:?} / method({:?}) /  response body = {}",
+        response.status().as_str(),
+        response.content_length(),
+        method,
+        &body,
+    ))
 }
 
 pub async fn rest_get(
@@ -625,61 +447,7 @@ where
 
 #[cfg(test)]
 mod test_exchange {
-    // const MAX_QUEUE_SIZE: usize = 100;
-
-    use super::*;
-    use crate::common::init_debug_log;
-    // use crossbeam_channel::bounded;
-
-    #[tokio::test]
-    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();
-        let _r = log_download_tmp(url, tmp_dir.path()).await?;
-
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn test_log_download() -> anyhow::Result<()> {
-        init_debug_log();
-
-        let url = "https://data.binance.vision/data/spot/daily/trades/BTCBUSD/BTCBUSD-trades-2022-11-19.zip";
-        // let (tx, rx) = bounded::<StringRecord>(MAX_QUEUE_SIZE);
-
-        let tmp_dir = tempdir()?;
-        let path = log_download_tmp(url, tmp_dir.path()).await?;
-        let path = PathBuf::from_str(&path)?;
-        log::debug!("log_download_temp: {:?}", path);
-
-        let mut rec_no = 0;
-
-        read_csv_archive(&path, true, |_rec| {
-            rec_no += 1;
-            Ok(())
-        });
-
-        log::debug!("rec_no = {}", rec_no);
-
-        Ok(())
-    }
-
-    #[tokio::test]
-    async fn test_rest_get() -> anyhow::Result<()> {
-        let r = rest_get(
-            "https://api.binance.com",
-            "/api/v3/trades?symbol=BTCBUSD&limit=5",
-            vec![],
-            None,
-            None,
-        )
-        .await?;
-
-        println!("{}", r);
-
-        Ok(())
-    }
+    use crate::net::rest_get;
 
     #[tokio::test]
     async fn test_rest_get_err() -> anyhow::Result<()> {
@@ -696,5 +464,4 @@ mod test_exchange {
 
         Ok(())
     }
-
 }
diff --git a/rbot_lib/src/net/ws.rs b/rbot_lib/src/net/ws.rs
index 27adeaba..19741590 100644
--- a/rbot_lib/src/net/ws.rs
+++ b/rbot_lib/src/net/ws.rs
@@ -688,9 +688,7 @@ mod test_exchange_ws {
         MarketConfig {
             exchange_name: "BYBIT".to_string(),            
             price_unit: dec![0.1],
-            price_scale: 3,
             size_unit: dec![0.001],
-            size_scale: 4,
             maker_fee: dec![0.00_01],
             taker_fee: dec![0.00_01],
             price_type: PriceType::Home,
diff --git a/rbot_market/src/market.rs b/rbot_market/src/market.rs
index 53cc7d0d..d920407a 100644
--- a/rbot_market/src/market.rs
+++ b/rbot_market/src/market.rs
@@ -1,9 +1,11 @@
 // Copyright(c) 2024. yasstake. All rights reserved.
 
 use crossbeam_channel::Sender;
+use rbot_lib::common::flush_log;
 use rbot_lib::common::AccountCoins;
 use rbot_lib::common::MarketMessage;
 
+use rbot_lib::db::TradeDataFrame;
 use rust_decimal_macros::dec;
 use std::sync::{Arc, Mutex, RwLock};
 
@@ -17,19 +19,20 @@ use anyhow::anyhow;
 #[allow(unused_imports)]
 use anyhow::Context;
 
+
 use rbot_lib::{
     common::{
-        flush_log, time_string, AccountPair, MarketConfig, MarketStream, MicroSec, Order,
-        OrderSide, OrderType, ServerConfig, Trade, DAYS, NOW,
+        AccountPair, MarketConfig, MarketStream, MicroSec, Order,
+        OrderSide, OrderType, Trade, DAYS, NOW,
     },
-    db::{df::KEY, sqlite::TradeTable},
+    db::df::KEY
 };
 
 macro_rules! check_if_enable_order {
     ($s: expr) => {
         if !$s.get_enable_order_feature() {
             log::error!("Order feature is disabled.");
-            anyhow::bail!("Order feature is disabled, you can enable exchange property 'enable_order_with_my_own_risk' to True");
+            return Err(anyhow!("Order feature is disabled, you can enable exchange property 'enable_order_with_my_own_risk' to True"));
         }
     };
 }
@@ -70,31 +73,15 @@ pub trait OrderInterface {
     fn get_account(&self, market_config: &MarketConfig) -> anyhow::Result<AccountPair>;
 }
 
-pub trait OrderInterfaceImpl<T, U>
+pub trait OrderInterfaceImpl<T>
 where
-    T: RestApi<U>,
-    U: ServerConfig,
+    T: RestApi
 {
+    fn get_restapi(&self) -> &T;
+
     fn set_enable_order_feature(&mut self, enable_order: bool);
     fn get_enable_order_feature(&self) -> bool;
 
-    fn get_server_config(&self) -> &U;
-
-    fn price_dp(&self, market_config: &MarketConfig, price: Decimal) -> Decimal {
-        let price_scale = market_config.price_scale;
-        let price_dp = price.round_dp(price_scale);
-        price_dp
-    }
-
-    fn size_dp(&self, market_config: &MarketConfig, size: Decimal) -> Decimal {
-        let size_scale = market_config.size_scale;
-        let size_dp = size.round_dp(size_scale);
-        size_dp
-    }
-
-    fn order_side(side: &str) -> OrderSide {
-        OrderSide::from(side)
-    }
 
     async fn make_order(
         &self,
@@ -105,12 +92,10 @@ where
         order_type: OrderType,
         client_order_id: Option<&str>,
     ) -> anyhow::Result<Vec<Order>> {
-        let price = self.price_dp(&market_config, price);
-        let size = self.size_dp(&market_config, size);
-        let order_side = Self::order_side(side);
+        let order_side = OrderSide::from(side);
 
-        T::new_order(
-            &self.get_server_config(),
+        let api = self.get_restapi();
+        api.new_order(
             &market_config,
             order_side,
             price,
@@ -119,12 +104,6 @@ where
             client_order_id,
         )
         .await
-        .with_context(|| {
-            format!(
-                "Error in make_order: {:?} {:?} {:?} {:?} {:?} {:?}",
-                &market_config, &side, &price, &size, &order_type, &client_order_id
-            )
-        })
     }
 
     //------ REST API ----
@@ -137,6 +116,9 @@ where
         client_order_id: Option<&str>,
     ) -> anyhow::Result<Vec<Order>> {
         check_if_enable_order!(self);
+        let price = market_config.round_price(price)?;
+        let size = market_config.round_size(size)?;
+
         self.make_order(
             &market_config,
             side,
@@ -156,6 +138,8 @@ where
         client_order_id: Option<&str>,
     ) -> anyhow::Result<Vec<Order>> {
         check_if_enable_order!(self);
+        let size = market_config.round_size(size)?;
+
         self.make_order(
             market_config,
             side,
@@ -174,7 +158,9 @@ where
     ) -> anyhow::Result<Order> {
         check_if_enable_order!(self);
 
-        T::cancel_order(&self.get_server_config(), &market_config, order_id)
+        let api = self.get_restapi();
+
+        api.cancel_order(market_config, order_id)
             .await
             .with_context(|| {
                 format!(
@@ -185,15 +171,17 @@ where
     }
 
     async fn get_open_orders(&self, market_config: &MarketConfig) -> anyhow::Result<Vec<Order>> {
-        T::open_orders(&self.get_server_config(), market_config)
+        let api = self.get_restapi();
+
+        api.open_orders(market_config)
             .await
-            .with_context(|| format!("Error in get_open_orders: {:?}", &market_config))
     }
 
     async fn get_account(&self) -> anyhow::Result<AccountCoins> {
-        T::get_account(&self.get_server_config())
+        let api = self.get_restapi();
+        
+        api.get_account()
             .await
-            .with_context(|| format!("Error in get_account"))
     }
 
     async fn async_start_user_stream(&mut self) -> anyhow::Result<()>;
@@ -202,11 +190,7 @@ 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;
-    fn get_trade_symbol(&self) -> String;
 
     fn get_market_config(&self) -> MarketConfig;
 
@@ -276,45 +260,25 @@ pub trait MarketInterface {
     ) -> anyhow::Result<MarketStream>;
 }
 
-pub trait MarketImpl<T, U>
+pub trait MarketImpl<T>
 where
-    T: RestApi<U>,
-    U: ServerConfig,
+    T: RestApi
 {
-    fn get_server_config(&self) -> U;
     // --- GET CONFIG INFO ----
+    fn get_restapi(&self) -> &T;
     fn get_config(&self) -> MarketConfig;
-    fn get_exchange_name(&self) -> String;
-    fn get_trade_category(&self) -> String;
-    fn get_trade_symbol(&self) -> String;
 
     /// 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, force: bool, verbose: bool) -> anyhow::Result<i64>;
 
-    fn get_db(&self) -> Arc<Mutex<TradeTable>>;
+    fn get_db(&self) -> Arc<Mutex<TradeDataFrame>>;
 
     fn get_history_web_base_url(&self) -> String;
 
-    /// Check if database is valid at the date
-    fn validate_db_by_date(&mut self, date: MicroSec) -> anyhow::Result<bool> {
-        let db = self.get_db();
-        let mut lock = db.lock().unwrap();
-
-        lock.validate_by_date(date)
-    }
-
     fn find_latest_gap(&self, force: bool) -> anyhow::Result<(MicroSec, MicroSec)> {
         log::debug!("[start] find_latest_gap");
         let start_time = NOW() - DAYS(2);
@@ -367,15 +331,6 @@ where
         Ok((latest_trade, first_unfix_trade))
     }
 
-    // --- DB ----
-    fn drop_table(&mut self) -> anyhow::Result<()> {
-        let db = self.get_db();
-        let lock = db.lock().unwrap();
-
-        lock.drop_table()?;
-
-        Ok(())
-    }
 
     fn get_cache_duration(&self) -> MicroSec {
         let db = self.get_db();
@@ -444,26 +399,6 @@ where
         lock.info()
     }
 
-    fn get_file_name(&self) -> String {
-        let db = self.get_db();
-        let lock = db.lock().unwrap();
-        lock.get_file_name()
-    }
-
-    fn get_running(&self) -> bool {
-        let db = self.get_db();
-        let lock = db.lock().unwrap();
-        lock.is_running()
-    }
-
-    fn vacuum(&self) {
-        let db = self.get_db();
-        let lock = db.lock().unwrap();
-        if lock.vacuum().is_err() {
-            log::error!("Error in vacuum");
-        }
-    }
-
     fn _repr_html_(&self) -> String {
         let db = self.get_db();
         let lock = db.lock().unwrap();
@@ -555,7 +490,7 @@ where
         };
 
         if edge_price.is_err() {
-            self.reflesh_order_book();
+            self.reflesh_order_book()?;
             let lock = orderbook.read().unwrap();
             edge_price = lock.get_edge_price();
         }
@@ -563,18 +498,18 @@ where
         Ok(edge_price.unwrap())
     }
 
-    async fn start_db_thread(&mut self) -> Sender<Vec<Trade>> {
+    async fn start_db_thread(&mut self) -> anyhow::Result<Sender<Vec<Trade>>> {
         let db = self.get_db();
         let mut lock = db.lock().unwrap();
 
-        lock.start_thread().await
+        lock.start_thread()
     }
 
     async fn async_download_recent_trades(
         &self,
         market_config: &MarketConfig,
     ) -> anyhow::Result<Vec<Trade>> {
-        T::get_recent_trades(&self.get_server_config(), market_config).await
+        self.get_restapi().get_recent_trades(market_config).await
     }
 
     async fn async_expire_unfix_data(&mut self, force: bool) -> anyhow::Result<()> {
@@ -592,7 +527,7 @@ where
 
         let expire_message = expire_message.unwrap();
 
-        let tx = self.start_db_thread().await;
+        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());
@@ -612,12 +547,14 @@ where
     ) -> anyhow::Result<MarketStream> {
         let (sender, market_stream) = MarketStream::open();
 
-        let db = self.get_db();
+        let archive = {
+            let db = self.get_db();
+            let trade_dataframe = db.lock().unwrap();
+            trade_dataframe.get_archive()
+        };
 
         std::thread::spawn(move || {
-            let mut table_db = db.lock().unwrap();
-
-            let result = table_db.select(time_from, time_to, |trade| {
+            let result = archive.foreach(time_from, time_to, &mut |trade| {
                 let message: MarketMessage = trade.into();
                 sender.send(message)?;
 
@@ -632,108 +569,19 @@ where
         return Ok(market_stream);
     }
 
-    fn for_each_record<F>(&self, time_from: MicroSec, time_to: MicroSec, f: &mut F) -> anyhow::Result<()> 
-    where
-        F: FnMut(&Trade) -> anyhow::Result<()>
-    {
-        let db = self.get_db();
-        let mut table_db = db.lock().unwrap();
-
-        table_db.select(time_from, time_to, |trade| {
-            f(trade)
-        })?;
-
-        Ok(())
-    }
-
-    /*------------   async ----------------*/
-    async fn async_get_latest_archive_date(&self) -> anyhow::Result<MicroSec> {
-        let server_config = self.get_server_config();
-        let config = self.get_config().clone();
-
-        let r = T::latest_archive_date(&server_config, &config).await;
-        r
-    }
 
-    async fn async_download_archive(
+    async fn async_download_archive   
+    (
         &self,
-        tx: &Sender<Vec<Trade>>,
-        date: MicroSec,
-        low_priority: bool,
-        verbose: bool,
-    ) -> anyhow::Result<i64> {
-        T::download_archive(
-            &self.get_server_config(),
-            &self.get_config(),
-            tx,
-            date,
-            low_priority,
-            verbose,
-        )
-        .await
-        .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();
-        }
-
-        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;
-            }
-
-            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);
-                    }
-                }
-            }
-        }
-
-        Ok(download_rec)
+        verbose: bool
+    ) -> anyhow::Result<i64> 
+    {
+        let db = self.get_db();
+        let api = self.get_restapi();
+        let mut lock =  db.lock().unwrap();
+        lock.download_archive(api, ndays, force, verbose).await
     }
 
     async fn async_download_latest(&mut self, verbose: bool) -> anyhow::Result<i64> {
@@ -742,10 +590,10 @@ where
             flush_log();
         }
 
-        let server_config = self.get_server_config();
+        let api = self.get_restapi();
         let config = self.get_config().clone();
 
-        let trades = T::get_recent_trades(&server_config, &config).await?;
+        let trades = api.get_recent_trades(&config).await?;
         let rec = trades.len() as i64;
 
         log::debug!("rec: {}", rec);
@@ -760,7 +608,7 @@ where
             println!("rec: {}", rec);
             flush_log();
         }
-        let tx = self.start_db_thread().await;
+        let tx = self.start_db_thread().await?;
 
         let start_time = trades.iter().map(|trade| trade.time).min();
 
diff --git a/rbot_session/src/runner.rs b/rbot_session/src/runner.rs
index 398e592e..69aef0e6 100644
--- a/rbot_session/src/runner.rs
+++ b/rbot_session/src/runner.rs
@@ -1,12 +1,12 @@
 // Copyright(c) 2022-2023. yasstake. All rights reserved.
 
 use crossbeam_channel::Receiver;
-use pyo3::{pyclass, pymethods, types::IntoPyDict, Py, PyAny, PyErr, PyObject, Python};
+use pyo3::{marker, pyclass, pymethods, types::IntoPyDict, Py, PyAny, PyErr, PyObject, Python};
 use rust_decimal::prelude::ToPrimitive;
 
 use super::{has_method, ExecuteMode, Session};
 use async_stream::stream;
-use futures::{Stream, StreamExt};
+use futures::Stream;
 
 use rbot_lib::{
     common::{
@@ -293,13 +293,12 @@ impl Runner {
 
     pub fn update_market_info(&mut self, market: &PyObject) -> Result<(), PyErr> {
         let r = Python::with_gil(|py| {
-            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();
+            let market_config = market.getattr(py, "config")?;
+            let market_config = market_config.extract::<MarketConfig>(py)?;
 
-            self.exchange_name = exchange_name.extract::<String>(py).unwrap();
-            self.category = category.extract::<String>(py).unwrap();
-            self.symbol = symbol.extract::<String>(py).unwrap();
+            self.exchange_name = market_config.exchange_name;
+            self.category = market_config.trade_category;
+            self.symbol = market_config.trade_symbol;
 
             Ok(())
         });
diff --git a/rbot_session/src/session.rs b/rbot_session/src/session.rs
index 10e73c3e..ebc4e1a1 100644
--- a/rbot_session/src/session.rs
+++ b/rbot_session/src/session.rs
@@ -14,7 +14,7 @@ use super::{Logger, OrderList};
 use pyo3::prelude::*;
 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};
+}, db::TradeDataFrame};
 
 use anyhow;
 
@@ -53,6 +53,8 @@ impl ExecuteMode {
 #[pyclass(name = "Session")]
 #[derive(Debug)]
 pub struct Session {
+    production: bool,
+
     session_id: String,
 
     execute_mode: ExecuteMode,
@@ -95,7 +97,6 @@ pub struct Session {
     client_mode: bool,
 
     log: Logger,
-    db_path: String,
 }
 
 #[pymethods]
@@ -143,13 +144,8 @@ impl Session {
         let category = config.trade_category.clone();
         let now_time = NOW() / 1_000_000;
 
-        let db_path = TradeTable::make_db_path(
-            &config.exchange_name, 
-            &category, 
-            &config.trade_symbol, 
-            production);
-
         let mut session = Self {
+            production: production,
             session_id: Self::int_to_base64(now_time),
             execute_mode: execute_mode,
             buy_orders: OrderList::new(OrderSide::Buy),
@@ -191,7 +187,6 @@ impl Session {
             client_mode: client_mode,
 
             log: Logger::new(log_memory),
-            db_path: db_path,
         };
 
         session.load_order_list().unwrap();
@@ -481,9 +476,7 @@ impl Session {
 
     pub fn real_market_order(&mut self, side: String, size: Decimal) -> Result<Py<PyAny>, PyErr> {
         log::debug!("market_order: side={:}, size={}", &side, size);
-
-        let size_scale = self.market_config.size_scale;
-        let size = size.round_dp(size_scale);
+        let size = self.market_config.round_size(size)?;
 
         let local_id = self.new_order_id();
 
@@ -516,8 +509,7 @@ impl Session {
     }
 
     pub fn dry_market_order(&mut self, side: String, size: Decimal) -> Result<Py<PyAny>, PyErr> {
-        let size_scale = self.market_config.size_scale;
-        let size = size.round_dp(size_scale);
+        let size = self.market_config.round_size(size)?;
 
         let local_id = self.new_order_id();
         let order_side = OrderSide::from(&side);
@@ -547,8 +539,7 @@ impl Session {
     }
 
     pub fn dummy_market_order(&mut self, side: String, size: Decimal) -> Result<Py<PyAny>, PyErr> {
-        let size_scale = self.market_config.size_scale;
-        let size = size.round_dp(size_scale);
+        let size = self.market_config.round_size(size)?;
 
         let local_id = self.new_order_id();
         let order_side = OrderSide::from(&side);
@@ -604,11 +595,8 @@ impl Session {
         price: Decimal,
         size: Decimal,
     ) -> Result<Vec<Order>, PyErr> {
-        let price_scale = self.market_config.price_scale;
-        let pricedp = price.round_dp(price_scale);
-
-        let size_scale = self.market_config.size_scale;
-        let sizedp = size.round_dp(size_scale);
+        let price = self.market_config.round_price(price)?;
+        let size = self.market_config.round_size(size)?;
 
         // first push order to order list
         let local_id = self.new_order_id();
@@ -616,8 +604,8 @@ impl Session {
         log::debug!(
             "limit_order: side={:?}, size={}, price={}",
             side,
-            sizedp,
-            pricedp
+            size,
+            price
         );
 
         // then call market.limit_order
@@ -625,7 +613,7 @@ impl Session {
             let result = self.exchange.call_method1(
                 py,
                 "limit_order",
-                (self.market_config.clone(), side, pricedp, sizedp, local_id),
+                (self.market_config.clone(), side, price, size, local_id),
             );
 
             match result {
@@ -666,11 +654,8 @@ impl Session {
         price: Decimal,
         size: Decimal,
     ) -> Result<Vec<Order>, PyErr> {
-        let price_scale = self.market_config.price_scale;
-        let pricedp = price.round_dp(price_scale);
-
-        let size_scale = self.market_config.size_scale;
-        let sizedp = size.round_dp(size_scale);
+        let price = self.market_config.round_price(price)?;
+        let size = self.market_config.round_size(size)?;
 
         // first push order to order list
         let local_id = self.new_order_id();
@@ -680,8 +665,8 @@ impl Session {
         log::debug!(
             "dummuy_limit_order: side={:?}, size={}, price={}",
             side,
-            sizedp,
-            pricedp
+            size,
+            price
         );
 
         let mut order = Order::new(
@@ -693,8 +678,8 @@ impl Session {
             order_side,
             OrderType::Limit,
             OrderStatus::New,
-            pricedp,
-            sizedp,
+            price,
+            size,
         );
 
         order.is_maker = true;
@@ -765,21 +750,15 @@ impl Session {
 }
 
 impl Session {
-    pub fn get_db(&self, market_config: Option<&MarketConfig>) -> anyhow::Result<Arc<Mutex<TradeTable>>> {
+    pub fn get_db(&self, market_config: Option<&MarketConfig>) -> anyhow::Result<Arc<Mutex<TradeDataFrame>>> {
+
         if market_config.is_none() {
-            return TradeTable::get(&self.db_path);
+            return TradeDataFrame::get(&self.market_config, self.production);
         }
 
-        let market = market_config.unwrap();
-
-        let db_path = TradeTable::make_db_path(
-            &market.exchange_name,
-            &market.trade_category,
-            &market.trade_symbol,
-            true,      // use only production db for reference
-        );
+        let market_config = market_config.unwrap();
 
-        TradeTable::get(&db_path)
+        TradeDataFrame::get(&market_config, true)       // always use production other than primary market.
     }
 
     /// Message処理


Test patch:
diff --git a/exchanges/bybit/src/archive_test.rs b/exchanges/bybit/src/archive_test.rs
index b448879f..99302db9 100644
--- a/exchanges/bybit/src/archive_test.rs
+++ b/exchanges/bybit/src/archive_test.rs
@@ -1,74 +1,48 @@
-use anyhow::anyhow;
-use polars_io::avro::AvroReader;
-use polars_io::avro::AvroWriter;
-use polars_io::parquet::read::ParquetReader;
-use polars_io::parquet::write::ParquetWriter;
-use polars_io::SerReader;
-use polars_io::SerWriter;
-use std::fs::File;
-use std::path::PathBuf;
-
-use polars::prelude::DataFrame;
 
 #[cfg(test)]
 
 mod archive_test {
-    use std::path::Path;
     use std::path::PathBuf;
+    use std::str::FromStr;
 
     use rbot_lib::common::date_string;
     use rbot_lib::common::init_debug_log;
     use rbot_lib::common::DAYS;
-    use rbot_lib::common::FLOOR_DAY;
     use rbot_lib::common::NOW;
-    use rbot_lib::db::log_foreach;
-    use rbot_lib::db::TradeTableArchive;
-    use rbot_lib::net::log_download_tmp;
-    use rbot_market::MarketImpl;
-
+    use rbot_lib::db::log_download_tmp;
+    use rbot_lib::db::TradeArchive;
+    use rbot_lib::net::RestApi;
+    
     use crate::config::BybitConfig;
     use crate::config::BybitServerConfig;
     use crate::rest::BybitRestApi;
-    use crate::Bybit;
-
-    use super::*;
+    
 
-    fn create_archive() -> TradeTableArchive<BybitRestApi, BybitServerConfig> {
+    fn create_archive() -> TradeArchive {
         let server_config = BybitServerConfig::new(true);
         let config = BybitConfig::BTCUSDT();
 
         let archive =
-            TradeTableArchive::<BybitRestApi, BybitServerConfig>::new(&server_config, &config);
+            TradeArchive::new(&config, server_config.production);
 
         archive
     }
 
-    #[test]
-    fn test_archive_create() {
-        init_debug_log();
-        let server_config = BybitServerConfig::new(true);
-        let config = BybitConfig::BTCUSDT();
-        let archive =
-            TradeTableArchive::<BybitRestApi, BybitServerConfig>::new(&server_config, &config);
-
-        log::debug!("start->{}({}) end->{}({})",
-            archive.start_time(), date_string(archive.start_time()),
-            archive.end_time(), date_string(archive.end_time())
-        );
-    }
 
     #[tokio::test]
     async fn test_archive_latest() -> anyhow::Result<()> {
         let mut archive = create_archive();
 
         init_debug_log();
+        let server_config = BybitServerConfig::new(true);
+        let api = BybitRestApi::new(&server_config);
 
         log::debug!("first try, access web archive");
-        let latest = archive.latest_archive_date().await?;
+        let latest = archive.latest_archive_date(&api).await?;
         log::debug!("date = {}({})", date_string(latest), latest);
 
         log::debug!("second try, use cached data");
-        let latest = archive.latest_archive_date().await?;
+        let latest = archive.latest_archive_date(&api).await?;
         log::debug!("date = {}({})", date_string(latest), latest);
 
 
@@ -93,13 +67,36 @@ mod archive_test {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn test_test_download_tmp() -> anyhow::Result<()> {
+        init_debug_log();
+        let server = BybitServerConfig::new(true);
+        let config = BybitConfig::BTCUSDT();
+
+        let api = BybitRestApi::new(&server);
+
+        let url = api.history_web_url(&config, NOW() - DAYS(2));
+        log::debug!("url={:?}", url);
+
+        let tmp_dir = PathBuf::from_str("/tmp/")?;
+
+        let file = log_download_tmp(&url, &tmp_dir).await?;    
+
+        log::debug!("download complete {:?}", file);
+
+        Ok(())
+    }
+
     #[tokio::test]
     async fn test_web_archive_to_parquet() {
         let mut archive = create_archive();
 
         init_debug_log();
+        let server_config = BybitServerConfig::new(true);
+        let api = BybitRestApi::new(&server_config);
+
 
-        archive.web_archive_to_parquet(NOW() - DAYS(2), false, true).await.unwrap();
+        archive.web_archive_to_parquet(&api, NOW() - DAYS(2), false, true).await.unwrap();
     }
 
     #[test]
@@ -120,7 +117,7 @@ mod archive_test {
         let archive = create_archive();
         init_debug_log();
 
-        let df = archive.select_cache_df(0, 0)?;
+        let df = archive.select_cachedf(0, 0)?;
 
         log::debug!("{:?}", df);
 
@@ -134,21 +131,25 @@ mod archive_test {
         init_debug_log();
         let mut archive = create_archive();
 
-        archive.web_archive_to_parquet(NOW() - DAYS(2), false, true).await?;
-        archive.web_archive_to_parquet(NOW() - DAYS(3), false, true).await?;
+        let server_config = BybitServerConfig::new(true);
+        let api = BybitRestApi::new(&server_config);
+
+
+        archive.web_archive_to_parquet(&api, NOW() - DAYS(2), false, true).await?;
+        archive.web_archive_to_parquet(&api, NOW() - DAYS(3), false, true).await?;
         archive
-            .web_archive_to_parquet(NOW() - DAYS(10), false, true)
+            .web_archive_to_parquet(&api, NOW() - DAYS(10), false, true)
             .await?;
 
         log::debug!(
             "start={:?}({:?})",
             archive.start_time(),
-            date_string(archive.start_time())
+            date_string(archive.start_time()?)
         );
         log::debug!(
             "end={:?}({:?})",
             archive.end_time(),
-            date_string(archive.end_time())
+            date_string(archive.end_time()?)
         );
 
         let dates = archive.list_dates()?;
@@ -165,40 +166,43 @@ mod archive_test {
         log::debug!(
             "start={:?}({:?})",
             archive.start_time(),
-            date_string(archive.start_time())
+            date_string(archive.start_time()?)
         );
         log::debug!(
             "end={:?}({:?})",
             archive.end_time(),
-            date_string(archive.end_time())
+            date_string(archive.end_time()?)
         );
 
         log::debug!("download first");
+        let server_config = BybitServerConfig::new(true);
+        let api = BybitRestApi::new(&server_config);
 
-        archive.download(4, false, true).await?;
+
+        archive.download(&api, 4, false, true).await?;
         log::debug!(
             "start={:?}({:?})",
             archive.start_time(),
-            date_string(archive.start_time())
+            date_string(archive.start_time()?)
         );
         log::debug!(
             "end={:?}({:?})",
             archive.end_time(),
-            date_string(archive.end_time())
+            date_string(archive.end_time()?)
         );
 
         log::debug!("download with cache");
 
-        archive.download(7, false, true).await?;
+        archive.download(&api, 7, false, true).await?;
         log::debug!(
             "start={:?}({:?})",
             archive.start_time(),
-            date_string(archive.start_time())
+            date_string(archive.start_time()?)
         );
         log::debug!(
             "end={:?}({:?})",
             archive.end_time(),
-            date_string(archive.end_time())
+            date_string(archive.end_time()?)
         );
 
         Ok(())
@@ -209,18 +213,20 @@ mod archive_test {
     async fn test_load_df() -> anyhow::Result<()> {
         init_debug_log();
         let mut archive = create_archive();
+        let server_config = BybitServerConfig::new(true);
+        let api = BybitRestApi::new(&server_config);
 
-        archive.download(2, false, true).await?;
+        archive.download(&api, 2, false, true).await?;
 
         log::debug!(
             "start={:?}({:?})",
             archive.start_time(),
-            date_string(archive.start_time())
+            date_string(archive.start_time()?)
         );
         log::debug!(
             "end={:?}({:?})",
             archive.end_time(),
-            date_string(archive.end_time())
+            date_string(archive.end_time()?)
         );
 
         let df = archive.load_cache_df(NOW() - DAYS(2))?;
@@ -257,7 +263,7 @@ mod archive_test {
 
         let now = NOW();
 
-        let df = archive.select_cache_df(0, 0)?;
+        let df = archive.select_cachedf(0, 0)?;
 
         log::debug!("{:?}", df);
         log::debug!("{:?}", df.shape());
@@ -272,28 +278,4 @@ mod archive_test {
         Ok(())
     }
 
-    #[test]
-    fn test_select_db_perf() -> anyhow::Result<()> {
-        let mut market = Bybit::new(false).open_market(&BybitConfig::BTCUSDT());
-        init_debug_log();
-
-        market.download_archives(7, false, true, false);
-
-        let now = NOW();
-
-        let mut count = 0;
-        market.for_each_record(0, 0, &mut |trade| {
-            count += 1;
-            Ok(())
-        });
-
-        log::debug!(
-            "Sqliteによる読み込み {}[rec]  {}[msec]",
-            count,
-            (NOW() - now) / 1_000 as i64
-        );
-
-        Ok(())
-    }
-
 }
diff --git a/rbot/test/bybit_test/nullagent.py b/rbot/test/bybit_test/nullagent.py
index 5774d0aa..62c9b910 100644
--- a/rbot/test/bybit_test/nullagent.py
+++ b/rbot/test/bybit_test/nullagent.py
@@ -39,7 +39,7 @@ def on_clock(self, session, clock):
             clock: 現在時刻です。エポック時間からのマイクロ秒で表されます。
         """
         # 現在の時刻をプリントします。
-        print("on_clock: ", clock, ": ", time_string(clock))
+#        print("on_clock: ", clock, ": ", time_string(clock))
         session.market_order("Buy", 0.001)
     
     def on_update(self, session, updated_order):
@@ -50,7 +50,7 @@ def on_update(self, session, updated_order):
             updated_order: 注文状態が変化した注文情報です。
         """
         # 注文状態が変化した注文情報をプリントします。オーダーを発行しない限り呼び出されません。
-        print("on_update", updated_order)    
+#        print("on_update", updated_order)    
 
     
 from rbot import Bybit, BybitConfig
@@ -58,9 +58,8 @@ def on_update(self, session, updated_order):
 from rbot import init_debug_log
 
 
-
 bybit  = Bybit(production=False)
-bybit.enable_order_with_my_own_risk = True
+#bybit.enable_order_with_my_own_risk = True
 
 market = bybit.open_market(BybitConfig.BTCUSDT)
 
@@ -71,16 +70,29 @@ def on_update(self, session, updated_order):
 runner = Runner()
 
 from rbot import init_debug_log
-init_debug_log()
+#init_debug_log()
 
-session = runner.real_run(
+market.download_archive(10)
+
+
+session = runner.back_test(
                 exchange=bybit,
                 market=market,
                 agent=agent, 
 #                execute_time = 10,
                 verbose=True,
                 log_file="skelton_bot.log",
-                client= True
             )
 
+
+"""
+session = runner.real_run(
+                exchange=bybit,
+                market=market,
+                agent=agent, 
+#                execute_time = 10,
+                verbose=True,
+                log_file="skelton_bot.log",
+            )
+"""