From bb2e387f870495703fd663ca8f525028c3a8ced5 Mon Sep 17 00:00:00 2001 From: TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com> Date: Thu, 2 Apr 2026 09:37:24 +0900 Subject: feat(strategy): add Asian Session RSI strategy for SOL/USDT scalping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Simple time-based + RSI strategy for small capital day trading: - Trading window: KST 9:00-11:00 (UTC 0:00-2:00) - Entry: RSI(14) < 25 + volume above average - Exit: +1.5% TP, -0.7% SL, or session end time exit - Risk: max 3 trades/day, pause after 2 consecutive losses - Config: ~$75 per trade (10% of 100만원 capital) --- .../tests/test_asian_session_rsi.py | 161 +++++++++++++++++++++ 1 file changed, 161 insertions(+) create mode 100644 services/strategy-engine/tests/test_asian_session_rsi.py (limited to 'services/strategy-engine/tests') diff --git a/services/strategy-engine/tests/test_asian_session_rsi.py b/services/strategy-engine/tests/test_asian_session_rsi.py new file mode 100644 index 0000000..b311220 --- /dev/null +++ b/services/strategy-engine/tests/test_asian_session_rsi.py @@ -0,0 +1,161 @@ +"""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 -- cgit v1.2.3