File size: 2,007 Bytes
e7811c7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | 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
|