| from datetime import datetime, timedelta |
|
|
| import numpy as np |
|
|
| from app.models.schemas import KlineData |
| from app.strategies.bollinger_bands import BollingerBandsStrategy |
| from app.strategies.dual_thrust import DualThrustStrategy |
| from app.strategies.ma_crossover import MACrossoverStrategy |
|
|
|
|
| def make_klines(prices: list[float], symbol: str = "IF2406") -> list[KlineData]: |
| now = datetime.utcnow() |
| return [ |
| KlineData( |
| symbol=symbol, |
| open=p * 0.999, |
| high=p * 1.002, |
| low=p * 0.998, |
| close=p, |
| volume=100, |
| timestamp=now - timedelta(minutes=len(prices) - i), |
| ) |
| for i, p in enumerate(prices) |
| ] |
|
|
|
|
| class TestMACrossover: |
| def test_buy_signal_on_golden_cross(self): |
| strategy = MACrossoverStrategy("test-ma", "IF2406", {"fast_period": 3, "slow_period": 5, "quantity": 1}) |
| prices = [100, 99, 98, 97, 96, 95, 94, 96, 99, 103, 108] |
| klines = make_klines(prices) |
| signal = strategy.calculate_signal(klines) |
| if signal: |
| assert signal.side.value == "BUY" |
|
|
| def test_sell_signal_on_death_cross(self): |
| strategy = MACrossoverStrategy("test-ma", "IF2406", {"fast_period": 3, "slow_period": 5, "quantity": 1}) |
| prices = [100, 101, 102, 103, 104, 105, 106, 104, 101, 97, 92] |
| klines = make_klines(prices) |
| signal = strategy.calculate_signal(klines) |
| if signal: |
| assert signal.side.value == "SELL" |
|
|
| def test_no_signal_with_insufficient_data(self): |
| strategy = MACrossoverStrategy("test-ma", "IF2406", {"fast_period": 5, "slow_period": 20}) |
| klines = make_klines([100, 101, 102]) |
| signal = strategy.calculate_signal(klines) |
| assert signal is None |
|
|
|
|
| class TestBollingerBands: |
| def test_no_signal_with_insufficient_data(self): |
| strategy = BollingerBandsStrategy("test-bb", "IF2406", {"period": 20}) |
| klines = make_klines([100] * 5) |
| signal = strategy.calculate_signal(klines) |
| assert signal is None |
|
|
| def test_buy_signal_at_lower_band(self): |
| strategy = BollingerBandsStrategy("test-bb", "IF2406", {"period": 10, "std_dev": 1.5, "quantity": 1}) |
| np.random.seed(42) |
| prices = [100 + np.random.normal(0, 0.5) for _ in range(12)] |
| prices.append(prices[-1] - 5) |
| klines = make_klines(prices) |
| signal = strategy.calculate_signal(klines) |
| if signal: |
| assert signal.side.value == "BUY" |
|
|
| def test_sell_signal_at_upper_band(self): |
| strategy = BollingerBandsStrategy("test-bb", "IF2406", {"period": 10, "std_dev": 1.5, "quantity": 1}) |
| np.random.seed(42) |
| prices = [100 + np.random.normal(0, 0.5) for _ in range(12)] |
| prices.append(prices[-1] + 5) |
| klines = make_klines(prices) |
| signal = strategy.calculate_signal(klines) |
| if signal: |
| assert signal.side.value == "SELL" |
|
|
|
|
| class TestDualThrust: |
| def test_no_signal_with_insufficient_data(self): |
| strategy = DualThrustStrategy("test-dt", "IF2406", {"lookback": 5}) |
| klines = make_klines([100, 101]) |
| signal = strategy.calculate_signal(klines) |
| assert signal is None |
|
|
| def test_breakout_signal(self): |
| strategy = DualThrustStrategy("test-dt", "IF2406", {"lookback": 5, "k1": 0.3, "k2": 0.3, "quantity": 1}) |
| prices = [100, 101, 99, 100.5, 100.2, 100.1, 108] |
| klines = make_klines(prices) |
| signal = strategy.calculate_signal(klines) |
| if signal: |
| assert signal.side.value == "BUY" |
|
|