| from pathlib import Path |
| import sys |
|
|
| import polars as pl |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parent)) |
|
|
| from data_handler import _resample_to_timeframe |
|
|
|
|
| def _sample_ohlcv_frame() -> pl.DataFrame: |
| return pl.DataFrame( |
| { |
| "date": [ |
| "2026-01-02 10:45:00", |
| "2026-01-02 09:45:00", |
| "2026-01-02 10:00:00", |
| "2026-01-02 09:15:00", |
| "2026-01-02 10:15:00", |
| "2026-01-02 09:30:00", |
| "2026-01-02 10:30:00", |
| "2026-01-02 11:00:00", |
| ], |
| "open": [7.0, 3.0, 4.0, 1.0, 5.0, 2.0, 6.0, 8.0], |
| "high": [7.5, 3.5, 4.5, 1.5, 5.5, 2.5, 6.5, 8.5], |
| "low": [6.5, 2.5, 3.5, 0.5, 4.5, 1.5, 5.5, 7.5], |
| "close": [7.25, 3.25, 4.25, 1.25, 5.25, 2.25, 6.25, 8.25], |
| "volume": [70, 30, 40, 10, 50, 20, 60, 80], |
| } |
| ) |
|
|
|
|
| def test_resample_to_hour_preserves_expected_ohlcv_groups(): |
| resampled = _resample_to_timeframe(_sample_ohlcv_frame(), "1_hour") |
|
|
| assert resampled["date"].dt.strftime("%Y-%m-%d %H:%M:%S").to_list() == [ |
| "2026-01-02 09:00:00", |
| "2026-01-02 10:00:00", |
| "2026-01-02 11:00:00", |
| ] |
| assert resampled["open"].to_list() == [1.0, 4.0, 8.0] |
| assert resampled["high"].to_list() == [3.5, 7.5, 8.5] |
| assert resampled["low"].to_list() == [0.5, 3.5, 7.5] |
| assert resampled["close"].to_list() == [3.25, 7.25, 8.25] |
| assert resampled["volume"].to_list() == [60.0, 220.0, 80.0] |
|
|
|
|
| def test_resample_to_max_keeps_full_history_extremes(): |
| resampled = _resample_to_timeframe(_sample_ohlcv_frame(), "MAX") |
|
|
| assert resampled.height == 1 |
| assert resampled["date"].dt.strftime("%Y-%m-%d %H:%M:%S").item() == "2026-01-02 09:15:00" |
| assert resampled["open"].item() == 1.0 |
| assert resampled["high"].item() == 8.5 |
| assert resampled["low"].item() == 0.5 |
| assert resampled["close"].item() == 8.25 |
| assert resampled["volume"].item() == 360.0 |
|
|
