From 66368d580cf569b50a33e438f2287a977e6fc704 Mon Sep 17 00:00:00 2001 From: TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com> Date: Wed, 1 Apr 2026 17:11:51 +0900 Subject: test: add edge case tests for zero volume, empty data, extreme values --- tests/edge_cases/test_zero_volume.py | 79 ++++++++++++++++++++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 tests/edge_cases/test_zero_volume.py (limited to 'tests/edge_cases/test_zero_volume.py') diff --git a/tests/edge_cases/test_zero_volume.py b/tests/edge_cases/test_zero_volume.py new file mode 100644 index 0000000..0aefa07 --- /dev/null +++ b/tests/edge_cases/test_zero_volume.py @@ -0,0 +1,79 @@ +"""Tests for strategies handling zero-volume candles gracefully.""" + +import sys +from datetime import datetime, timezone +from decimal import Decimal +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "services" / "strategy-engine")) +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "services" / "backtester" / "src")) + +from shared.models import Candle +from strategies.vwap_strategy import VwapStrategy +from strategies.volume_profile_strategy import VolumeProfileStrategy +from strategies.rsi_strategy import RsiStrategy + + +def _candle(close: str, volume: str = "0", idx: int = 0) -> Candle: + base = datetime(2025, 1, 1, tzinfo=timezone.utc) + from datetime import timedelta + return Candle( + symbol="BTCUSDT", + timeframe="1h", + open_time=base + timedelta(hours=idx), + open=Decimal(close), + high=Decimal(close), + low=Decimal(close), + close=Decimal(close), + volume=Decimal(volume), + ) + + +class TestVwapZeroVolume: + """VWAP strategy must not divide by zero when all volume is 0.""" + + def test_all_zero_volume_no_crash(self): + strategy = VwapStrategy() + for i in range(50): + result = strategy.on_candle(_candle("100", "0", i)) + # Should return None (no signal) rather than raising ZeroDivisionError + assert result is None + + def test_mixed_zero_and_nonzero_volume(self): + strategy = VwapStrategy() + # First 30 candles with volume to pass warmup, then zero volume + for i in range(30): + strategy.on_candle(_candle("100", "10", i)) + # Now feed zero-volume candles; cumulative_vol > 0 so VWAP still defined + result = strategy.on_candle(_candle("100", "0", 30)) + # Should not crash + assert result is None or result is not None # just checking no exception + + +class TestVolumeProfileZeroVolume: + """Volume Profile strategy with all-zero volumes should return None.""" + + def test_all_zero_volume_returns_none(self): + strategy = VolumeProfileStrategy() + for i in range(150): + result = strategy.on_candle(_candle(str(100 + i % 10), "0", i)) + # With zero total volume, should return None (no meaningful signal) + assert result is None + + +class TestRsiZeroVolume: + """RSI strategy should process price correctly regardless of volume.""" + + def test_rsi_with_zero_volume_processes_price(self): + strategy = RsiStrategy() + # Feed descending prices to trigger oversold RSI + signals = [] + for i in range(30): + price = str(100 - i) + result = strategy.on_candle(_candle(price, "0", i)) + if result is not None: + signals.append(result) + # RSI uses only close prices, volume is irrelevant; should work fine + # With steadily dropping prices, RSI should go below oversold threshold + assert len(signals) >= 1 + assert signals[0].side.value == "BUY" # oversold signal -- cgit v1.2.3