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"