172 lines
6.0 KiB
Python
172 lines
6.0 KiB
Python
"""Tests for timeframe utilities."""
|
|
|
|
import pytest
|
|
|
|
from datetime import timedelta
|
|
|
|
from exchange_data_manager.utils.timeframes import (
|
|
timeframe_to_seconds,
|
|
timeframe_to_ms,
|
|
seconds_to_timeframe,
|
|
timeframe_to_timedelta,
|
|
estimate_record_count,
|
|
align_to_timeframe,
|
|
TIMEFRAME_SECONDS,
|
|
)
|
|
|
|
|
|
class TestTimeframeToSeconds:
|
|
"""Tests for timeframe_to_seconds function."""
|
|
|
|
def test_standard_timeframes(self):
|
|
"""Test all standard timeframe mappings."""
|
|
assert timeframe_to_seconds("1s") == 1
|
|
assert timeframe_to_seconds("1m") == 60
|
|
assert timeframe_to_seconds("5m") == 300
|
|
assert timeframe_to_seconds("15m") == 900
|
|
assert timeframe_to_seconds("30m") == 1800
|
|
assert timeframe_to_seconds("1h") == 3600
|
|
assert timeframe_to_seconds("4h") == 14400
|
|
assert timeframe_to_seconds("1d") == 86400
|
|
assert timeframe_to_seconds("1w") == 604800
|
|
|
|
def test_case_insensitive(self):
|
|
"""Test that uppercase timeframes work (converted to lowercase)."""
|
|
assert timeframe_to_seconds("1H") == 3600
|
|
assert timeframe_to_seconds("5M") == 300
|
|
assert timeframe_to_seconds("1D") == 86400
|
|
|
|
def test_month_timeframe(self):
|
|
"""Test that uppercase 1M maps to one month (approximate)."""
|
|
assert timeframe_to_seconds("1M") == 2592000
|
|
|
|
def test_custom_timeframes(self):
|
|
"""Test parsing of non-standard timeframes."""
|
|
assert timeframe_to_seconds("2m") == 120
|
|
assert timeframe_to_seconds("10m") == 600
|
|
assert timeframe_to_seconds("2h") == 7200
|
|
assert timeframe_to_seconds("6h") == 21600
|
|
assert timeframe_to_seconds("2d") == 172800
|
|
assert timeframe_to_seconds("2w") == 1209600
|
|
|
|
def test_invalid_timeframe_raises(self):
|
|
"""Test that invalid timeframes raise ValueError."""
|
|
with pytest.raises(ValueError):
|
|
timeframe_to_seconds("invalid")
|
|
|
|
with pytest.raises(ValueError):
|
|
timeframe_to_seconds("x")
|
|
|
|
with pytest.raises(ValueError):
|
|
timeframe_to_seconds("5x")
|
|
|
|
with pytest.raises(ValueError):
|
|
timeframe_to_seconds("")
|
|
|
|
|
|
class TestTimeframeToMs:
|
|
"""Tests for timeframe_to_ms function."""
|
|
|
|
def test_converts_to_milliseconds(self):
|
|
"""Test conversion to milliseconds."""
|
|
assert timeframe_to_ms("1s") == 1000
|
|
assert timeframe_to_ms("1m") == 60000
|
|
assert timeframe_to_ms("5m") == 300000
|
|
assert timeframe_to_ms("1h") == 3600000
|
|
|
|
|
|
class TestSecondsToTimeframe:
|
|
"""Tests for seconds_to_timeframe function."""
|
|
|
|
def test_exact_matches(self):
|
|
"""Test exact matches from TIMEFRAME_SECONDS."""
|
|
for tf, secs in TIMEFRAME_SECONDS.items():
|
|
assert seconds_to_timeframe(secs) == tf
|
|
|
|
def test_calculated_timeframes(self):
|
|
"""Test calculated timeframe strings."""
|
|
assert seconds_to_timeframe(120) == "2m"
|
|
assert seconds_to_timeframe(7200) == "2h"
|
|
assert seconds_to_timeframe(172800) == "2d"
|
|
assert seconds_to_timeframe(1209600) == "2w"
|
|
|
|
def test_seconds_fallback(self):
|
|
"""Test fallback to seconds for odd values."""
|
|
assert seconds_to_timeframe(45) == "45s"
|
|
assert seconds_to_timeframe(90) == "90s"
|
|
|
|
|
|
class TestTimeframeToTimedelta:
|
|
"""Tests for timeframe_to_timedelta function."""
|
|
|
|
def test_standard_timeframes(self):
|
|
"""Test conversion to timedelta."""
|
|
assert timeframe_to_timedelta("1m") == timedelta(minutes=1)
|
|
assert timeframe_to_timedelta("5m") == timedelta(minutes=5)
|
|
assert timeframe_to_timedelta("1h") == timedelta(hours=1)
|
|
assert timeframe_to_timedelta("4h") == timedelta(hours=4)
|
|
assert timeframe_to_timedelta("1d") == timedelta(days=1)
|
|
|
|
def test_can_use_for_arithmetic(self):
|
|
"""Test that timedelta can be used for datetime arithmetic."""
|
|
from datetime import datetime, timezone
|
|
|
|
dt = datetime(2024, 3, 2, 0, 0, 0, tzinfo=timezone.utc)
|
|
td = timeframe_to_timedelta("5m")
|
|
|
|
result = dt + td
|
|
assert result.minute == 5
|
|
|
|
|
|
class TestEstimateRecordCount:
|
|
"""Tests for estimate_record_count function."""
|
|
|
|
def test_one_hour_1m(self):
|
|
"""Test estimate for 1 hour of 1m candles."""
|
|
start_ms = 1709337600000 # 2024-03-02 00:00:00
|
|
end_ms = 1709341200000 # 2024-03-02 01:00:00
|
|
|
|
count = estimate_record_count(start_ms, end_ms, "1m")
|
|
assert count == 61 # 60 intervals + 1 (inclusive)
|
|
|
|
def test_one_day_1h(self):
|
|
"""Test estimate for 1 day of 1h candles."""
|
|
start_ms = 1709337600000 # 2024-03-02 00:00:00
|
|
end_ms = 1709337600000 + (24 * 3600 * 1000) # +24 hours
|
|
|
|
count = estimate_record_count(start_ms, end_ms, "1h")
|
|
assert count == 25 # 24 intervals + 1 (inclusive)
|
|
|
|
def test_invalid_range_returns_zero(self):
|
|
"""Test that invalid range returns 0."""
|
|
assert estimate_record_count(1000, 1000, "1m") == 0 # Same start/end
|
|
assert estimate_record_count(2000, 1000, "1m") == 0 # End before start
|
|
|
|
|
|
class TestAlignToTimeframe:
|
|
"""Tests for align_to_timeframe function."""
|
|
|
|
def test_already_aligned(self):
|
|
"""Test timestamp already on boundary."""
|
|
# 2024-03-02 00:00:00 is on 5m boundary
|
|
ts = 1709337600000
|
|
assert align_to_timeframe(ts, "5m") == ts
|
|
|
|
def test_mid_period(self):
|
|
"""Test timestamp in middle of period."""
|
|
# 2024-03-02 00:02:30 -> should align to 00:00:00
|
|
ts = 1709337600000 + (2 * 60 + 30) * 1000 # +2m30s
|
|
assert align_to_timeframe(ts, "5m") == 1709337600000
|
|
|
|
def test_end_of_period(self):
|
|
"""Test timestamp at end of period."""
|
|
# 2024-03-02 00:04:59 -> should align to 00:00:00
|
|
ts = 1709337600000 + (4 * 60 + 59) * 1000 # +4m59s
|
|
assert align_to_timeframe(ts, "5m") == 1709337600000
|
|
|
|
def test_1h_alignment(self):
|
|
"""Test hourly alignment."""
|
|
# 2024-03-02 00:30:00 -> should align to 00:00:00
|
|
ts = 1709337600000 + 30 * 60 * 1000 # +30m
|
|
assert align_to_timeframe(ts, "1h") == 1709337600000
|