Repository: yasstake/rbot (language: python)

Issue:
Session.ohlcvで１本だけデータが欲しい場合でも２本返ってくる場合がある
OHLCVを取得するとき、
　　　・「現在時刻　ー　（足の幅）　x 本数」から「現在時刻」までのOHLCVをつくっていた。

そのため、12:00時点において１時間足を１本つくったときは　11:00-12:00の１本が出来上がるが、11:30で取得すると 10:30-11:00, 11:00-11:30までの２本が取得されていた。

Add 8MB buffer to download web archive



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

これらの変更は、`BybitMarket`のデータベース管理と取引データの取得機能を大幅に強化します。新しいメソッドと、既存のメソッドのシグネチャが変更されることで、トレードデータの処理がより構造化され、効率的になりました。これにより、APIの利用が向上し、全体的なパフォーマンスも改善されます。

## Changes

| ファイル                          | 変更の概要                                                                                                      |
|----------------------------------|----------------------------------------------------------------------------------------------------------------|
| `exchanges/bybit/src/market.rs` | `vaccum`メソッドの追加、`get_trades`への変更、データベースチャンネルの管理改善。                                |
| `exchanges/bybit/src/rest.rs`   | `get_trade_klines`が`get_trades`に改名され、ページネーション対応の新しい戻り値を導入。                            |
| `rbot/test/bybit_test/nullagent.py` | ログ機能の更新と`download_archive`メソッドのパラメータ修正。                                                    |
| `rbot_lib/src/db/archive.rs`    | `log_download_tmp`にバッファリングを導入し、テストモジュール`archive_test`を追加。                               |
| `rbot_lib/src/db/sqlite.rs`     | `vacuum`メソッドを公開に変更し、ロギング機能を追加。                                                              |
| `rbot_lib/src/db/tradedf.rs`    | `vacuum`メソッドの追加、`start_thread`から`open_channel`への改名、リファクタリング。                             |
| `rbot_lib/src/net/rest.rs`       | `TradePage` enumを追加し、`get_trade_klines`を`get_trades`に改名。                                               |
| `rbot_market/src/market.rs`      | データベース操作の改善、`open_channel`を使用し、取引データ処理を強化。                                          |
| `rbot_session/src/runner.rs`     | 不要なインポートを削除。                                                                                         |

## Sequence Diagram(s)

```mermaid
sequenceDiagram
    participant User
    participant API
    participant Database

    User->>API: get_trades(config, start_time, end_time, page)
    API->>Database: fetch_trades(start_time, end_time, page)
    Database->>API: return (trades, tradePage)
    API->>User: return (trades, tradePage)
```

> 
> ぴょんぴょん跳ねるラビットが、
> 新しいコードを見てうれしそう。
> データが整って、読みやすくなり、
> みんなのトレードもスムーズに。
> さあ、共に祝おう、コードの進化を！🐇✨
> 

</details>

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

<!--

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

The diff introduces several significant changes to the functionality of the `BybitMarket` implementation in the `market.rs` file. Notably, a new method `vaccum` is added, which locks the database and calls the `vacuum` method on it, indicating a potential cleanup or optimization operation for the database. This addition enhances the database management capabilities of the `BybitMarket` class.

Furthermore, the logic surrounding the retrieval of trade data has been modified. The original method that fetched trade data using `get_trade_klines` has been replaced with `get_trades`, which now returns a tuple containing both trades and a `TradePage` object. This change suggests an enhancement in the way trade data is processed, allowing for more structured handling of trade information. The `TradePage::Done` parameter is now being passed, which may indicate a new state or filtering mechanism for the trade data being retrieved.

Additionally, the method for starting the database channel has been updated from `start_thread` to `open_channel`. This change likely reflects a shift in how database interactions are managed, possibly improving concurrency or resource handling.

Overall, these modifications enhance the `BybitMarket` class's functionality by introducing new methods and altering existing ones to improve database management and trade data retrieval.

## Alterations to the declarations of exported or public entities

- `fn vaccum(&self) -> anyhow::Result<()>` in `impl BybitMarket` in `exchanges/bybit/src/market.rs`
- `fn get_trades(...)` in `impl RestApi` in `exchanges/bybit/src/net.rs` (signature change inferred from usage)
- `fn open_channel(&self) -> anyhow::Result<Channel>` in `impl TradeDb` in `exchanges/bybit/src/db.rs` (signature change inferred from usage)

---

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

The diff introduces significant modifications to the `BybitRestApi` implementation within the `exchanges/bybit/src/rest.rs` file. Notably, the function `get_trade_klines` has been renamed to `get_trades`, reflecting a shift in its purpose and output. The function now accepts an additional parameter of type `TradePage`, which is used to manage pagination in the response. The return type has also changed from `anyhow::Result<Vec<Kline>>` to `anyhow::Result<(Vec<Trade>, TradePage)>`, indicating that the function now returns a tuple containing a vector of `Trade` objects and a `TradePage` instance, enhancing its functionality to provide more detailed trade information.

Within the function body, the previous logic that collected `Kline` data has been replaced with a call to `convert_klines_to_trades`, which transforms the collected `Kline` data into `Trade` objects. This change signifies a shift in focus from handling `Kline` data directly to managing `Trade` data, aligning the function's output with the new requirements.

Additionally, the test function `test_trade_kline` has been updated to reflect these changes. It now tests the `get_trades` function instead of `get_trade_klines`, and the assertions have been adjusted to accommodate the new return type. The test now verifies the number of trades returned and prints the timestamps of the first and last trades, ensuring that the new functionality is correctly implemented and validated.

Overall, these changes enhance the API's capability to handle trade data more effectively, improving both the logic and the control flow of the involved functions.

## Alterations to the declarations of exported or public entities

- `async fn get_trade_klines(&self, config: &MarketConfig, start_time: MicroSec, end_time: MicroSec) -> anyhow::Result<Vec<Kline>>` in `impl RestApi for BybitRestApi` → `async fn get_trades(&self, config: &MarketConfig, start_time: MicroSec, end_time: MicroSec, _page: TradePage) -> anyhow::Result<(Vec<Trade>, TradePage)>` in `impl RestApi for BybitRestApi`
- `async fn test_trade_kline()` in `mod bybit_rest_test` → `async fn test_trade_kline() -> anyhow::Result<()>` in `mod bybit_rest_test`

---

rbot/test/bybit_test/nullagent.py: ## AI-generated summary of changes

The changes in the `nullagent.py` file primarily involve updates to logging functionality and adjustments to the parameters used in the `download_archive` method. The import statement for the logging function has been modified from `init_debug_log` to `init_log`, suggesting a shift from a debug-specific logging mechanism to a more general logging approach. This change may impact how logging is handled throughout the application, potentially affecting the verbosity or format of the logs generated during execution.

Additionally, the `download_archive` method call has been altered from downloading an archive of size 10 to size 13, with the addition of a `verbose=True` parameter. This change indicates an enhancement in the functionality of the `download_archive` method, likely providing more detailed output during the execution of the method. The overall control flow of the `on_update` function remains intact, but these modifications suggest a refinement in logging practices and an increase in the volume of data processed during the archive download.

## Alterations to the declarations of exported or public entities

- `def on_update(self, session, updated_order)` in class `NullAgent` in `rbot/test/bybit_test/nullagent.py` → No changes
- `from rbot import init_debug_log` in `rbot/test/bybit_test/nullagent.py` → `from rbot import init_log` in `rbot/test/bybit_test/nullagent.py`

---

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

The diff introduces several significant changes to the `log_download_tmp` function within the `TradeArchive` implementation in the `archive.rs` file. The most notable modification is the introduction of a constant `BUFFER_SIZE`, which is set to 8 MB. This constant is utilized to create a `BufWriter` that wraps the file being created for the downloaded content. This change enhances the efficiency of file writing by buffering the output, which is particularly beneficial for handling larger downloads.

Additionally, the code has removed the retrieval of `total_size` from the response, which previously allowed tracking the total bytes expected to be downloaded. The variable `downloaded`, which was used to keep track of the amount of data written, has also been eliminated. Instead, the logic now directly writes the chunks received from the response stream to the buffered writer without maintaining a count of the total bytes downloaded. This simplifies the code but may impact the ability to handle scenarios where monitoring the download progress is necessary.

The flushing of the file has been updated to flush the buffered writer instead of the file directly, which is a more efficient approach when using buffering. Overall, these changes streamline the download process and improve performance by leveraging buffered I/O operations.

The diff also includes a new test module `archive_test` with a test function `test_download` that verifies the functionality of `log_download_tmp`. This addition ensures that the download functionality is tested, which is crucial for maintaining code reliability.

Overall, the changes reflect a shift towards more efficient file handling and the introduction of testing for the download functionality.

## Alterations to the declarations of exported or public entities

- `pub async fn log_download_tmp(url: &str, tmp_dir: &Path) -> anyhow::Result<PathBuf>` in `rbot_lib/src/db/archive.rs` → no changes in signature. 
- New constant declaration: `const BUFFER_SIZE: usize = 8 * 1024 * 1024;` in `rbot_lib/src/db/archive.rs`.

---

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

The diff introduces a significant change in the visibility of the `vacuum` function within the `TradeDb` implementation. Previously, the `vacuum` function was a private method, but it has been modified to be a public method by changing its declaration from `fn vacuum(&self) -> anyhow::Result<()>` to `pub fn vacuum(&self) -> anyhow::Result<()>`. This change allows external access to the `vacuum` method, enabling other parts of the codebase to invoke it. Additionally, a logging statement has been added at the beginning of the method, which logs a debug message indicating that the database vacuum operation is being performed. This enhances the observability of the function's execution without altering its core functionality.

## Alterations to the declarations of exported or public entities

- `fn vacuum(&self) -> anyhow::Result<()>` in `impl TradeDb` in `rbot_lib/src/db/sqlite.rs` → `pub fn vacuum(&self) -> anyhow::Result<()>` in `impl TradeDb` in `rbot_lib/src/db/sqlite.rs`

---

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

The diff introduces several modifications to the `TradeDataFrame` implementation in the Rust file `tradedf.rs`, enhancing its functionality and refactoring certain methods for clarity and efficiency. A new public method, `vacuum`, is added, which calls the `vacuum` function on the database, potentially improving database management by reclaiming storage. This method enhances the interface of `TradeDataFrame` by allowing users to trigger a vacuum operation directly.

The method `start_thread` has been renamed to `open_channel`, reflecting a more descriptive purpose for its functionality, which involves opening a communication channel with the database. This change improves code readability and aligns the method name with its actual behavior.

Additionally, the method `select_cachedf` has been refactored. A new private method, `archive_end_time`, is introduced to encapsulate the logic for retrieving the end time of the archive. This refactoring allows `select_cachedf` to utilize `archive_end_time`, streamlining the retrieval process and enhancing code maintainability. The logic within `select_cachedf` remains consistent, but the method now benefits from improved clarity.

Furthermore, there are adjustments in the comments and minor modifications to the logic within the `expire_cache_df` and data loading sections, where the comments have been updated for better understanding. The cache expiration logic has been slightly altered to expire by date rather than a specific timestamp.

Overall, the changes enhance the functionality, readability, and maintainability of the `TradeDataFrame` implementation, while also introducing a new method for database management.

## Alterations to the declarations of exported or public entities

- `pub fn start_thread(&mut self) -> anyhow::Result<Sender<Vec<Trade>>>` in `impl TradeDataFrame` in `rbot_lib/src/db/tradedf.rs` → `pub fn open_channel(&mut self) -> anyhow::Result<Sender<Vec<Trade>>>` in `impl TradeDataFrame` in `rbot_lib/src/db/tradedf.rs`
- `pub fn select_cachedf(&mut self, start_time: MicroSec, end_time: MicroSec) -> anyhow::Result<DataFrame>` in `impl TradeDataFrame` in `rbot_lib/src/db/tradedf.rs` → `pub fn archive_end_time(&self) -> MicroSec` in `impl TradeDataFrame` in `rbot_lib/src/db/tradedf.rs`
- `pub fn select_cachedf(&mut self, start_time: MicroSec, end_time: MicroSec) -> anyhow::Result<DataFrame>` in `impl TradeDataFrame` in `rbot_lib/src/db/tradedf.rs` (refactored to use the new `archive_end_time` method)
- `pub fn expire_cache_df(&mut self, forget_before: MicroSec)` in `impl TradeDataFrame` in `rbot_lib/src/db/tradedf.rs` (minor internal logic changes, no signature modification)
- `pub fn vacuum(&self) -> anyhow::Result<()>` in `impl TradeDataFrame` in `rbot_lib/src/db/tradedf.rs` (newly added method)

---

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

The diff introduces significant modifications to the `RestApi` trait within the `rbot_lib/src/net/rest.rs` file. A new enum, `TradePage`, is defined, which includes variants for representing different pagination states: `Done`, `Time(MicroSec)`, `Int(i64)`, and `Key(String)`. This enum enhances the ability to manage trade data retrieval with specific pagination requirements.

The method previously named `get_trade_klines` has been renamed to `get_trades`, reflecting a broader functionality. The signature of this method has been altered to accept a `TradePage` parameter, which allows the method to return a tuple containing a vector of `Trade` objects and a `TradePage` instance. This change indicates a shift from a fixed retrieval of trade data to a more flexible approach that can handle pagination and state management.

Additionally, the diff removes a substantial commented-out block of code that included functions for downloading and reading CSV archives. This indicates a potential refactoring or a decision to remove unused or deprecated functionality, streamlining the codebase.

Overall, these changes enhance the functionality of the `RestApi` trait, particularly in how it handles trade data retrieval, while also cleaning up the code by removing unnecessary comments.

## Alterations to the declarations of exported or public entities

- `async fn get_trade_klines(&self, config: &MarketConfig, start_time: MicroSec, end_time: MicroSec)` in trait `RestApi` in `rbot_lib/src/net/rest.rs` → `async fn get_trades(&self, config: &MarketConfig, page: TradePage)` in trait `RestApi` in `rbot_lib/src/net/rest.rs`
- `pub enum TradePage` in `rbot_lib/src/net/rest.rs` → New declaration of `TradePage` enum with variants `Done`, `Time(MicroSec)`, `Int(i64)`, `Key(String)` in `rbot_lib/src/net/rest.rs`

---

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

The diff introduces several modifications to the functionality of the `market.rs` file, primarily affecting the handling of trade data and the interactions with the database. Key changes include the addition of the `TradeDb` import, which indicates a shift towards utilizing this structure for managing trade-related messages. 

In the `find_latest_gap` function, the initialization of `start_time` has been moved inside the database lock scope, ensuring that it is set just before the database operations, which may improve clarity regarding the timing of database queries. This change also emphasizes the necessity of locking the database during the computation of `fix_time` and `unfix_time`.

The `start_thread` method call has been replaced with `open_channel` in the context of database interactions. This suggests a change in the underlying mechanism for managing threads or channels, potentially improving the responsiveness or efficiency of the system.

Moreover, in the `async_download_recent_trades` function, the result from `download_archive` is now captured and checked for errors using the `?` operator. Following this, the code checks if `archive_end` is non-zero, and if so, it generates an expiration control message using `TradeDb::expire_control_message`. This indicates an enhancement in the logic to handle trade data more robustly, as it now includes additional checks and operations that were not present before.

Overall, these changes reflect a refinement in the control flow and logic surrounding trade data management, with a focus on ensuring data integrity and improving the handling of asynchronous operations.

## Alterations to the declarations of exported or public entities

- `fn find_latest_gap(&self, force: bool) -> anyhow::Result<(MicroSec, MicroSec)` in `rbot_market/src/market.rs` → No change
- `fn async_download_recent_trades(&mut self, verbose: bool) -> anyhow::Result<i64>` in `rbot_market/src/market.rs` → No change
- `fn async_download_latest(&mut self, verbose: bool) -> anyhow::Result<i64>` in `rbot_market/src/market.rs` → No change
- `use rbot_lib::db::TradeDb;` in `rbot_market/src/market.rs` → New import added

---

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

The diff presents a minor modification in the import statements of the `runner.rs` file. Specifically, it removes the unused `marker` import from the `pyo3` crate. This change does not affect the functionality or logic of the code, as it only cleans up the import list by eliminating an unnecessary dependency. The remaining imports remain unchanged, and the overall structure and control flow of the code are preserved.

## Alterations to the declarations of exported or public entities

- No alterations to the declarations of exported or public entities were made in `rbot_session/src/runner.rs`.
```

-->

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

<!--

Title: Add 8MB buffer to download web archive
Number: 249
User: yasstake

-->

<!-- 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 -->

Fix patch:
diff --git a/rbot_lib/src/db/archive.rs b/rbot_lib/src/db/archive.rs
index 55fb4965..7928abb2 100644
--- a/rbot_lib/src/db/archive.rs
+++ b/rbot_lib/src/db/archive.rs
@@ -322,10 +322,6 @@ impl TradeArchive {
         let file = File::open(file)?;
         let reader = SerializedFileReader::new(file)?;
 
-        // スキーマを取得
-        let schema = reader.metadata().file_metadata().schema();
-        println!("File Schema: {:?}", schema);
-
         for row in reader {
             let row = row?;
 
diff --git a/rbot_lib/src/db/tradedf.rs b/rbot_lib/src/db/tradedf.rs
index b0113aab..ee109ede 100644
--- a/rbot_lib/src/db/tradedf.rs
+++ b/rbot_lib/src/db/tradedf.rs
@@ -435,7 +435,7 @@ impl TradeDataFrame {
         Ok(())
     }
 
-    pub fn ohlcvv_df(
+    fn _ohlcvv_df(
         &mut self,
         mut start_time: MicroSec,
         end_time: MicroSec,
@@ -460,14 +460,14 @@ impl TradeDataFrame {
     ) -> 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)?;
+        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(
+    fn _ohlcv_df(
         &mut self,
         mut start_time: MicroSec,
         end_time: MicroSec,
@@ -492,7 +492,7 @@ impl TradeDataFrame {
     ) -> anyhow::Result<PyDataFrame> {
         start_time = ohlcv_start(start_time); // 開始tickは確定足、終了は未確定足もOK.
 
-        let mut df = self.ohlcv_df(start_time, end_time, window_sec)?;
+        let mut df = self._ohlcv_df(start_time, end_time, window_sec)?;
         convert_timems_to_datetime(&mut df)?;
         let df = PyDataFrame(df);
 
diff --git a/rbot_session/src/runner.rs b/rbot_session/src/runner.rs
index f8b68400..a9976f5a 100644
--- a/rbot_session/src/runner.rs
+++ b/rbot_session/src/runner.rs
@@ -53,6 +53,8 @@ pub struct Runner {
     exchange_name: String,
     category: String,
     symbol: String,
+
+    warmup_count: i64,
 }
 
 #[pymethods]
@@ -86,6 +88,7 @@ impl Runner {
             exchange_name: "".to_string(),
             category: "".to_string(),
             symbol: "".to_string(),
+            warmup_count: 0,
         }
     }
 
@@ -727,7 +730,10 @@ impl Runner {
             if let MarketMessage::Trade(trade) = message {
                 let new_clock = FLOOR_SEC(trade.time, interval_sec);
 
-                if self.current_clock < new_clock {
+                if self.current_clock == 0 {
+                    self.current_clock = new_clock;
+                }
+                else if self.current_clock < new_clock {
                     self.current_clock = new_clock;
 
                     self.call_agent_on_clock(py, agent, py_session, new_clock)?;
diff --git a/rbot_session/src/session.rs b/rbot_session/src/session.rs
index ebc4e1a1..baed16a5 100644
--- a/rbot_session/src/session.rs
+++ b/rbot_session/src/session.rs
@@ -1,22 +1,30 @@
 // Copyright(c) 2022-2024. yasstake. All rights reserved.
 
-use std::{collections::VecDeque, sync::Arc};
 use std::sync::Mutex;
+use std::{collections::VecDeque, sync::Arc};
 
+use polars::lazy::dsl::last;
 use pyo3::{pyclass, pymethods, PyAny, PyObject, Python};
 
 use pyo3_polars::PyDataFrame;
+use rbot_lib::common::FLOOR_SEC;
+use rbot_lib::db::ohlcv_start;
 use rbot_server::get_rest_orderbook;
 use rust_decimal::{prelude::ToPrimitive, Decimal};
 use rust_decimal_macros::dec;
 
 use super::{Logger, OrderList};
 use pyo3::prelude::*;
-use rbot_lib::{common::{
-    date_string, get_orderbook, hour_string, min_string, time_string, AccountCoins, AccountPair, MarketConfig, MarketMessage, MicroSec, Order, OrderBookList, OrderSide, OrderStatus, OrderType, Trade, NOW, SEC
-}, db::TradeDataFrame};
+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::TradeDataFrame,
+};
 
-use anyhow;
+use anyhow::anyhow;
 
 #[derive(Debug, Clone, PartialEq)]
 #[pyclass]
@@ -194,48 +202,70 @@ impl Session {
         return session;
     }
 
-
     #[pyo3(signature = (interval, count, market=None))]
-    pub fn ohlcv(&mut self, interval: i64, count: i64, market: Option<&MarketConfig>) -> anyhow::Result<PyDataFrame> {
-        let window_sec = interval * count;
-        let time_from = self.current_timestamp - SEC(window_sec);
+    pub fn ohlcv(
+        &mut self,
+        interval: i64,
+        count: i64,
+        market: Option<&MarketConfig>,
+    ) -> anyhow::Result<PyDataFrame> {
+        let time_from = calc_ohlcv_start(self.current_timestamp, interval, count)?;
         let time_to = self.current_timestamp;
 
         let df = {
-            log::debug!("ohlcv: time_from={}, time_to={}, interval={}", time_from, time_to, interval);
+            log::debug!(
+                "ohlcv: time_from={}, time_to={}, interval={}",
+                time_from,
+                time_to,
+                interval
+            );
 
             let db = self.get_db(market)?;
             let lock = db.lock();
 
-            let ohlcv = lock.unwrap().ohlcv_df(time_from, time_to, interval)?;
+            let ohlcv = lock.unwrap().py_ohlcv_polars(time_from, time_to, interval)?;
 
             ohlcv
         };
 
-        Ok(PyDataFrame(df))
+        Ok(df)
     }
 
     #[pyo3(signature = (interval, count, market=None))]
-    pub fn ohlcvv(&mut self, interval: i64, count: i64, market: Option<&MarketConfig>) -> anyhow::Result<PyDataFrame> {
-        let window_sec = interval * count;
-        let time_from = self.current_timestamp - SEC(window_sec);
+    pub fn ohlcvv(
+        &mut self,
+        interval: i64,
+        count: i64,
+        market: Option<&MarketConfig>,
+    ) -> anyhow::Result<PyDataFrame> {
+        let time_from = calc_ohlcv_start(self.current_timestamp, interval, count)?;
         let time_to = self.current_timestamp;
 
         let df = {
-            log::debug!("ohlcv: time_from={}, time_to={}, interval={}", time_from, time_to, interval);
+            log::debug!(
+                "ohlcv: time_from={}, time_to={}, interval={}",
+                time_from,
+                time_to,
+                interval
+            );
 
             let db = self.get_db(market)?;
             let lock = db.lock();
 
-            let ohlcv = lock.unwrap().ohlcvv_df(time_from, time_to, interval)?;
+            let ohlcv = lock.unwrap().py_ohlcvv_polars(time_from, time_to, interval)?;
 
             ohlcv
         };
 
-        Ok(PyDataFrame(df))
+        Ok((df))
     }
 
-    pub fn vap(&mut self, start_time: MicroSec, end_time: MicroSec, price_unit: i64) -> anyhow::Result<PyDataFrame> {
+    pub fn vap(
+        &mut self,
+        start_time: MicroSec,
+        end_time: MicroSec,
+        price_unit: i64,
+    ) -> anyhow::Result<PyDataFrame> {
         let db = self.get_db(None).unwrap();
         let mut lock = db.lock().unwrap();
 
@@ -255,12 +285,15 @@ impl Session {
     }
 
     #[getter]
-    pub fn get_board(&self) -> anyhow::Result<(PyDataFrame, PyDataFrame)>{
+    pub fn get_board(&self) -> anyhow::Result<(PyDataFrame, PyDataFrame)> {
         self.select_board(None)
     }
 
     #[pyo3(signature = (market_config=None))]
-    pub fn select_board(&self, market_config: Option<&MarketConfig>) -> anyhow::Result<(PyDataFrame, PyDataFrame)> {
+    pub fn select_board(
+        &self,
+        market_config: Option<&MarketConfig>,
+    ) -> anyhow::Result<(PyDataFrame, PyDataFrame)> {
         let board_config = if let Some(config) = market_config {
             config
         } else {
@@ -750,15 +783,17 @@ impl Session {
 }
 
 impl Session {
-    pub fn get_db(&self, market_config: Option<&MarketConfig>) -> anyhow::Result<Arc<Mutex<TradeDataFrame>>> {
-
+    pub fn get_db(
+        &self,
+        market_config: Option<&MarketConfig>,
+    ) -> anyhow::Result<Arc<Mutex<TradeDataFrame>>> {
         if market_config.is_none() {
             return TradeDataFrame::get(&self.market_config, self.production);
         }
 
         let market_config = market_config.unwrap();
 
-        TradeDataFrame::get(&market_config, true)       // always use production other than primary market.
+        TradeDataFrame::get(&market_config, true) // always use production other than primary market.
     }
 
     /// Message処理
@@ -864,7 +899,7 @@ impl Session {
     pub fn on_account_update(&mut self, account: &AccountCoins) {
         self.real_account.update(account);
 
-        let account_pair: AccountPair = self.real_account.extract_pair(&self.market_config);        
+        let account_pair: AccountPair = self.real_account.extract_pair(&self.market_config);
 
         if self.log_account(&account_pair).is_err() {
             log::error!("log_account_status error");
@@ -872,7 +907,7 @@ impl Session {
     }
 
     pub fn on_order_update(&mut self, order: &mut Order) {
-        if ! order.is_my_order(&self.session_name) {
+        if !order.is_my_order(&self.session_name) {
             log::debug!("on_order_update: skip my order: {:?}", order);
             return;
         }
@@ -880,7 +915,7 @@ impl Session {
         self.log_id += 1;
         order.log_id = self.log_id;
         order.update_balance(&self.market_config);
-        self.update_psudo_position(order);        
+        self.update_psudo_position(order);
 
         if order.order_side == OrderSide::Buy {
             if order.status == OrderStatus::Filled || order.status == OrderStatus::Canceled {
@@ -906,7 +941,10 @@ impl Session {
     fn new_order_id(&mut self) -> String {
         self.order_number += 1;
 
-        format!("{}-{}{:04}", self.session_name, self.session_id, self.order_number)
+        format!(
+            "{}-{}{:04}",
+            self.session_name, self.session_id, self.order_number
+        )
     }
 
     fn load_order_list(&mut self) -> Result<(), PyErr> {
@@ -933,7 +971,7 @@ impl Session {
                             continue;
                         }
 
-                        if ! order.is_my_order(&self.session_name) {
+                        if !order.is_my_order(&self.session_name) {
                             continue;
                         }
 
@@ -1159,13 +1197,67 @@ impl Session {
 
         result
     }
+}
+
+pub fn calc_ohlcv_start(
+    ohlcv_end_time: MicroSec,
+    window_sec: i64,
+    nbar: i64,
+) -> anyhow::Result<MicroSec> {
+    //        log::debug("time={} ({})", time_string(last_time), last_time);
+    if nbar < 1 {
+        return Err(anyhow!("nbar is zero, or minus. nbar={}", nbar));
+    }
+
+    let start_time = (ohlcv_end_time - 1) - SEC(window_sec) * (nbar -1); 
+    let start_time = FLOOR_SEC(start_time, window_sec);
 
+    Ok(start_time)
 }
 
 #[cfg(test)]
 mod session_tests {
+    use super::*;
+    use rbot_lib::common::{init_debug_log, parse_time};
+
+
+    #[test]
+    fn test_calc_ohlcv_start() -> anyhow::Result<()>{
+        init_debug_log();
+
+        let t1 = parse_time("2024-07-10T00:00:00.000000+00:00");
+
+        // if ndays = 0, raize an error.
+        let result = calc_ohlcv_start(t1, 60, 0);
+        log::debug!("{:?}", result);
+        assert!(result.is_err());
+
+        let result = calc_ohlcv_start(t1, 60, 1);
+        log::debug!("{:?}", result);
+        assert!(result.is_ok());
+        let start = result.unwrap();
+        log::debug!("start: {:?} ({:?})", time_string(start), start);
+        log::debug!("end: {:?} ({:?})", time_string(t1), t1);
 
+        assert_eq!(calc_ohlcv_start(parse_time("2024-07-10T00:00:00.000000+00:00"), 60, 1)?, parse_time("2024-07-09T23:59:00.000000+00:00"));
 
+        let result = calc_ohlcv_start(t1, 60, 2);
+        log::debug!("{:?}", result);
+        assert!(result.is_ok());
+        let start = result.unwrap();
+        log::debug!("start: {:?} ({:?})", time_string(start), start);
+        log::debug!("end: {:?} ({:?})", time_string(t1), t1);
+
+
+        assert_eq!(calc_ohlcv_start(parse_time("2024-07-10T00:00:00.000000+00:00"), 60, 2)?, parse_time("2024-07-09T23:58:00.000000+00:00"));
+
+
+
+        assert_eq!(calc_ohlcv_start(parse_time("2024-07-10T00:00:00.000000+00:00"), 3600, 1)?, parse_time("2024-07-09T23:00:00.000000+00:00"));
+        assert_eq!(calc_ohlcv_start(parse_time("2024-07-10T00:00:00.000000+00:00"), 3600, 2)?, parse_time("2024-07-09T22:00:00.000000+00:00"));
+
+        Ok(()) 
+    }
 
     /*
     use rbot_lib::common::init_debug_log;


Test patch:
diff --git a/rbot/test/bybit_test/warmup.py b/rbot/test/bybit_test/warmup.py
new file mode 100644
index 00000000..b616c86b
--- /dev/null
+++ b/rbot/test/bybit_test/warmup.py
@@ -0,0 +1,97 @@
+# test warm up period
+
+
+
+from rbot import time_string        # μsの時間を文字列に変換するユーティリティ関数です。
+
+class SkeltonAgent:      # クラス名は任意です
+    def __init__(self):
+        """Botの初期化処理です。パラメータなどを設定するといいでしょう。利用しなくても構いません。
+        """
+        self.tick_count = 0         # on_tickが呼び出された回数をカウントします。
+    
+    def on_init(self, session):
+        """Botの初期化処理。Botの初期化時に一度だけ呼ばれます。
+        通常はsession.clock_interval_secを指定しon_clockの呼び出し間隔を設定します。
+        Args:
+            session: セッション情報（Botの初期化時用に渡されます）
+        """
+        session.clock_interval_sec = 60        # 10秒ごとにon_clockを呼び出します。
+
+    
+    def on_tick(self, session, side, price, size):
+        """取引所からの全ての約定イベント毎に呼び出される処理です（高頻度で呼び出されます）
+        Args:
+            session: セッション情報（市況情報の取得や注文するために利用します)
+            side: 売買区分です。"Buy"または"Sell"が設定されます。
+            price: 約定価格です。
+            size: 約定数量です。
+        """
+        #print(side, price, size)
+        #if self.tick_count < 10:
+            #print("on tick")
+
+        
+        # on_tickは高頻度によびだされるので、100回に1回だけ内容をプリントします。
+        #print("on_tick: ", side, price, size)
+        pass
+    
+    def on_clock(self, session, clock):
+        """定期的にフレームワークから呼び出される処理です。
+        session.clock_interval_secで指定した間隔で呼び出されます。
+
+        Args:
+            session: セッション情報（市況情報の取得や注文するために利用します)        
+            clock: 現在時刻です。エポック時間からのマイクロ秒で表されます。
+        """
+        # 現在の時刻をプリントします。
+#        print("on_clock: ", clock, ": ", time_string(clock))
+        self.tick_count += 1
+        if self.tick_count < 10:
+            # rbotから最新のタイムバーを1本取得
+            ohlcv = session.ohlcv(session.clock_interval_sec, 1)
+            print(f"on_clock(clock = {time_string(clock)} {clock}")
+            print(ohlcv)
+
+
+    def on_update(self, session, updated_order):
+        """自分の注文状態が変化した場合に呼び出される処理です。
+        Args:
+            session: セッション情報（市況情報の取得や注文するために利用します)                
+            session: セッション情報（市況情報の取得や注文するために利用します)        
+            updated_order: 注文状態が変化した注文情報です。
+        """
+        # 注文状態が変化した注文情報をプリントします。オーダーを発行しない限り呼び出されません。
+#        print("on_update", updated_order)    
+
+    
+from rbot import Bybit, BybitConfig
+
+from rbot import init_log
+
+
+bybit  = Bybit(production=False)
+#bybit.enable_order_with_my_own_risk = True
+
+market = bybit.open_market(BybitConfig.BTCUSDT)
+
+from rbot import Runner
+from rbot import NOW, DAYS
+
+agent = SkeltonAgent()
+runner = Runner()
+
+from rbot import init_debug_log
+#init_log()
+
+market.download_archive(13, verbose=True)
+
+
+session = runner.back_test(
+                exchange=bybit,
+                market=market,
+                agent=agent, 
+#                execute_time = 10,
+                verbose=True,
+                log_file="skelton_bot.log",
+            )