Coverage for src/finbot/strategy/sma_crossover.py: 96%

1from collections import defaultdict, deque 

2 

3from finbot.market.models import Bar 

4from finbot.strategy.base import Strategy 

5from finbot.strategy.signal import Signal, SignalDirection 

6 

7 

8class SmaCrossoverStrategy(Strategy): 

9 def __init__(self, short_window: int, long_window: int): 

10 if short_window <= 0: 

11 raise ValueError("short_window must be positive") 

12 

13 if long_window <= short_window: 

14 raise ValueError("long_window must be greater than short_window") 

15 

16 self.short_window = short_window 

17 self.long_window = long_window 

18 

19 self._prices: dict[str, deque[float]] = defaultdict(lambda: deque(maxlen=self.long_window)) 

20 self._previous_direction: dict[str, SignalDirection] = {} 

21 

22 @staticmethod 

23 def _get_direction(short_sma: float, long_sma: float) -> SignalDirection: 

24 if short_sma > long_sma: return SignalDirection.LONG 

25 if short_sma < long_sma: return SignalDirection.SHORT 25 ↛ 26line 25 didn't jump to line 26 because the condition on line 25 was always true

26 return SignalDirection.FLAT 

27 

28 def on_bar(self, bar: Bar) -> Signal | None: 

29 prices = self._prices[bar.symbol] 

30 prices.append(bar.close) 

31 

32 if len(prices) < self.long_window: 

33 return None 

34 

35 values = list(prices) 

36 short_sma = sum(values[-self.short_window:]) / self.short_window 

37 long_sma = sum(values) / self.long_window 

38 

39 current_direction = self._get_direction(short_sma=short_sma, long_sma=long_sma) 

40 previous_direction = self._previous_direction.get(bar.symbol) 

41 

42 self._previous_direction[bar.symbol] = current_direction 

43 

44 if previous_direction is None: 

45 return None 

46 

47 if ( 

48 previous_direction != SignalDirection.LONG 

49 and current_direction == SignalDirection.LONG 

50 ): 

51 return Signal( 

52 symbol=bar.symbol, 

53 direction=SignalDirection.LONG, 

54 metadata={"short_sma": short_sma, "long_sma": long_sma}, 

55 ) 

56 

57 if ( 

58 previous_direction != SignalDirection.SHORT 

59 and current_direction == SignalDirection.SHORT 

60 ): 

61 return Signal( 

62 symbol=bar.symbol, 

63 direction=SignalDirection.SHORT, 

64 metadata={"short_sma": short_sma, "long_sma": long_sma}, 

65 ) 

66 

67 return None