summaryrefslogtreecommitdiff
path: root/services/strategy-engine/tests/test_asian_session_rsi.py
diff options
context:
space:
mode:
authorTheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com>2026-04-02 09:37:24 +0900
committerTheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com>2026-04-02 09:37:24 +0900
commitbb2e387f870495703fd663ca8f525028c3a8ced5 (patch)
tree9293ccedf65a5218fcbab17f0508577cd6c2f3f5 /services/strategy-engine/tests/test_asian_session_rsi.py
parent98039ac910ab9afcdcb1813d00f3de8de0d2803c (diff)
feat(strategy): add Asian Session RSI strategy for SOL/USDT scalping
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)
Diffstat (limited to 'services/strategy-engine/tests/test_asian_session_rsi.py')
-rw-r--r--services/strategy-engine/tests/test_asian_session_rsi.py161
1 files changed, 161 insertions, 0 deletions
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