diff options
Diffstat (limited to 'services/strategy-engine/tests/test_asian_session_rsi.py')
| -rw-r--r-- | services/strategy-engine/tests/test_asian_session_rsi.py | 190 |
1 files changed, 0 insertions, 190 deletions
diff --git a/services/strategy-engine/tests/test_asian_session_rsi.py b/services/strategy-engine/tests/test_asian_session_rsi.py deleted file mode 100644 index db031f0..0000000 --- a/services/strategy-engine/tests/test_asian_session_rsi.py +++ /dev/null @@ -1,190 +0,0 @@ -"""Tests for Asian Session RSI strategy.""" - -import sys -from pathlib import Path - -sys.path.insert(0, str(Path(__file__).resolve().parents[1])) - -from datetime import datetime, timezone -from decimal import Decimal - -from shared.models import Candle, OrderSide -from strategies.asian_session_rsi import AsianSessionRsiStrategy - - -def _candle(price, hour=0, minute=30, volume=100.0, day=1): - return Candle( - symbol="SOLUSDT", - timeframe="5m", - open_time=datetime(2025, 1, day, hour, minute, tzinfo=timezone.utc), - open=Decimal(str(price)), - high=Decimal(str(price + 1)), - low=Decimal(str(price - 1)), - close=Decimal(str(price)), - volume=Decimal(str(volume)), - ) - - -def _make_strategy(**overrides): - s = AsianSessionRsiStrategy() - params = { - "rsi_period": 5, - "rsi_oversold": 30, - "rsi_overbought": 70, - "quantity": "0.5", - "take_profit_pct": 1.5, - "stop_loss_pct": 0.7, - "session_start_utc": 0, - "session_end_utc": 2, - "max_trades_per_day": 3, - "max_consecutive_losses": 2, - } - params.update(overrides) - s.configure(params) - return s - - -def test_no_signal_outside_session(): - s = _make_strategy() - # Hour 5 UTC = outside session (0-2 UTC) - for i in range(10): - sig = s.on_candle(_candle(100 - i * 3, hour=5)) - assert sig is None - - -def test_buy_signal_during_session_on_oversold(): - s = AsianSessionRsiStrategy() - s._rsi_period = 5 - s._rsi_oversold = 30 - s._quantity = Decimal("0.5") - s._take_profit_pct = 1.5 - s._stop_loss_pct = 0.7 - s._session_start_utc = 0 - s._session_end_utc = 2 - s._max_trades_per_day = 3 - s._max_consecutive_losses = 10 # High limit so test isn't blocked - - # Feed declining prices — collect all signals - signals = [] - for i in range(10): - sig = s.on_candle(_candle(100 - i * 3, hour=0, minute=i * 5)) - if sig is not None: - signals.append(sig) - - # Should have generated at least one BUY signal - buy_signals = [s for s in signals if s.side == OrderSide.BUY] - assert len(buy_signals) > 0 - assert buy_signals[0].strategy == "asian_session_rsi" - - -def test_take_profit_exit(): - s = _make_strategy(rsi_period=5, rsi_oversold=40) - # Force entry - for i in range(8): - s.on_candle(_candle(100 - i * 2, hour=0, minute=i * 5)) - - # Should be in position now — push price up for TP - sig = s.on_candle(_candle(100, hour=0, minute=50)) # entry ~around 84-86 - if s._in_position: - tp_price = s._entry_price * (1 + s._take_profit_pct / 100) - sig = s.on_candle(_candle(tp_price + 1, hour=1, minute=0)) - if sig is not None: - assert sig.side == OrderSide.SELL - assert "Take profit" in sig.reason - - -def test_stop_loss_exit(): - s = _make_strategy(rsi_period=5, rsi_oversold=40) - for i in range(8): - s.on_candle(_candle(100 - i * 2, hour=0, minute=i * 5)) - - if s._in_position: - sl_price = s._entry_price * (1 - s._stop_loss_pct / 100) - sig = s.on_candle(_candle(sl_price - 1, hour=1, minute=0)) - if sig is not None: - assert sig.side == OrderSide.SELL - assert "Stop loss" in sig.reason - - -def test_time_exit_when_session_ends(): - s = _make_strategy(rsi_period=5, rsi_oversold=40) - for i in range(8): - s.on_candle(_candle(100 - i * 2, hour=0, minute=i * 5)) - - if s._in_position: - # Session ends at hour 2 - sig = s.on_candle(_candle(s._entry_price, hour=3, minute=0)) - if sig is not None: - assert sig.side == OrderSide.SELL - assert "Time exit" in sig.reason - - -def test_max_trades_per_day(): - s = _make_strategy(rsi_period=3, rsi_oversold=40, max_trades_per_day=1) - # Force one trade - for i in range(6): - s.on_candle(_candle(100 - i * 5, hour=0, minute=i * 5)) - # Exit - if s._in_position: - s.on_candle(_candle(200, hour=0, minute=35)) # TP exit - # Try to enter again — should be blocked - for i in range(6): - s.on_candle(_candle(100 - i * 5, hour=1, minute=i * 5)) - # After 1 trade, no more allowed - assert not s._in_position or s._trades_today >= 1 - - -def test_consecutive_losses_stop(): - s = _make_strategy(rsi_period=3, rsi_oversold=40, max_consecutive_losses=2) - # Simulate 2 losses - s._consecutive_losses = 2 - # Even with valid conditions, should not enter - for i in range(6): - sig = s.on_candle(_candle(100 - i * 5, hour=0, minute=i * 5)) - assert sig is None - - -def test_reset_clears_all(): - s = _make_strategy() - s.on_candle(_candle(100, hour=0)) - s._in_position = True - s._trades_today = 2 - s._consecutive_losses = 1 - s.reset() - assert not s._in_position - assert s._trades_today == 0 - assert len(s._closes) == 0 - - -def test_warmup_period(): - s = _make_strategy(rsi_period=14) - assert s.warmup_period == 15 - - -def test_ema_filter_blocks_below_ema(): - """Entry blocked when price is below EMA.""" - s = AsianSessionRsiStrategy() - s._rsi_period = 5 - s._rsi_oversold = 40 - s._quantity = Decimal("0.5") - s._take_profit_pct = 1.5 - s._stop_loss_pct = 0.7 - s._session_start_utc = 0 - s._session_end_utc = 2 - s._max_trades_per_day = 3 - s._max_consecutive_losses = 10 - s._ema_period = 5 - s._require_bullish_candle = False # Test EMA only - - # Feed rising prices to set EMA high, then sharp drop - for i in range(10): - s.on_candle(_candle(200 + i * 5, hour=0, minute=i * 5)) - # Now feed low price -- below EMA, RSI should be low - signals = [] - for i in range(5): - sig = s.on_candle(_candle(100 - i * 5, hour=0, minute=(15 + i) * 5 % 60)) - if sig is not None: - signals.append(sig) - # Should have no BUY signals because price is way below EMA - buy_sigs = [s for s in signals if s.side == OrderSide.BUY] - assert len(buy_sigs) == 0 |