64 lines
3.0 KiB
Python
64 lines
3.0 KiB
Python
|
import pytest
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from data.common.aggregation.bucket import TimeframeBucket
|
||
|
|
from data.common.aggregation.batch import BatchCandleProcessor
|
||
|
|
|
||
|
|
class TestTimeframeBucket:
|
||
|
|
@pytest.mark.parametrize("timeframe, expected", [
|
||
|
|
("1s", timedelta(seconds=1)),
|
||
|
|
("5s", timedelta(seconds=5)),
|
||
|
|
("10s", timedelta(seconds=10)),
|
||
|
|
("15s", timedelta(seconds=15)),
|
||
|
|
("30s", timedelta(seconds=30)),
|
||
|
|
("1m", timedelta(minutes=1)),
|
||
|
|
("5m", timedelta(minutes=5)),
|
||
|
|
("15m", timedelta(minutes=15)),
|
||
|
|
("30m", timedelta(minutes=30)),
|
||
|
|
("1h", timedelta(hours=1)),
|
||
|
|
("4h", timedelta(hours=4)),
|
||
|
|
("1d", timedelta(days=1)),
|
||
|
|
])
|
||
|
|
def test_parse_timeframe_to_timedelta_valid(self, timeframe, expected):
|
||
|
|
assert TimeframeBucket._parse_timeframe_to_timedelta(timeframe) == expected
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("timeframe", [
|
||
|
|
"1w", "abc", "60x", "", "m5", "-1m", "0h"
|
||
|
|
])
|
||
|
|
def test_parse_timeframe_to_timedelta_invalid(self, timeframe):
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
TimeframeBucket._parse_timeframe_to_timedelta(timeframe)
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("timeframe", [
|
||
|
|
"-1m", "-5h", "-10s", "-2d"
|
||
|
|
])
|
||
|
|
def test_parse_timeframe_to_timedelta_negative(self, timeframe):
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
TimeframeBucket._parse_timeframe_to_timedelta(timeframe)
|
||
|
|
|
||
|
|
def test_calculate_end_time(self):
|
||
|
|
start = datetime(2024, 1, 1, 0, 0, 0)
|
||
|
|
bucket = TimeframeBucket(symbol="BTC-USDT", timeframe="5m", start_time=start)
|
||
|
|
expected_end = start + timedelta(minutes=5)
|
||
|
|
assert bucket.end_time == expected_end
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("timeframe, timestamp, expected_start", [
|
||
|
|
("1s", datetime(2024, 1, 1, 12, 0, 0, 123456), datetime(2024, 1, 1, 12, 0, 0)),
|
||
|
|
("5s", datetime(2024, 1, 1, 12, 0, 7), datetime(2024, 1, 1, 12, 0, 5)),
|
||
|
|
("10s", datetime(2024, 1, 1, 12, 0, 17), datetime(2024, 1, 1, 12, 0, 10)),
|
||
|
|
("1m", datetime(2024, 1, 1, 12, 7, 45), datetime(2024, 1, 1, 12, 7, 0)),
|
||
|
|
("5m", datetime(2024, 1, 1, 9, 7, 30), datetime(2024, 1, 1, 9, 5, 0)),
|
||
|
|
("15m", datetime(2024, 1, 1, 9, 23, 0), datetime(2024, 1, 1, 9, 15, 0)),
|
||
|
|
("1h", datetime(2024, 1, 1, 13, 45, 0), datetime(2024, 1, 1, 13, 0, 0)),
|
||
|
|
("4h", datetime(2024, 1, 1, 11, 15, 0), datetime(2024, 1, 1, 8, 0, 0)),
|
||
|
|
("1d", datetime(2024, 1, 1, 18, 30, 0), datetime(2024, 1, 1, 0, 0, 0)),
|
||
|
|
])
|
||
|
|
def test_get_bucket_start_time_valid(timeframe, timestamp, expected_start):
|
||
|
|
processor = BatchCandleProcessor(symbol="BTC-USDT", exchange="okx", timeframes=[timeframe])
|
||
|
|
start = processor._get_bucket_start_time(timestamp, timeframe)
|
||
|
|
assert start == expected_start
|
||
|
|
|
||
|
|
@pytest.mark.parametrize("timeframe", ["0h", "-5m", "abc", "1w", "", "m5"])
|
||
|
|
def test_get_bucket_start_time_invalid(timeframe):
|
||
|
|
processor = BatchCandleProcessor(symbol="BTC-USDT", exchange="okx", timeframes=[timeframe])
|
||
|
|
with pytest.raises(ValueError):
|
||
|
|
processor._get_bucket_start_time(datetime(2024, 1, 1, 12, 0, 0), timeframe)
|