Coverage for src/finbot/strategy/strategies.py: 98%

1import random 

2from collections import defaultdict, deque 

3from collections.abc import Callable, Iterable 

4from typing import Any 

5 

6from finbot.market.models import Bar 

7from finbot.strategy.core import Signal, SignalDirection, Strategy 

8 

9 

10class RandomStrategy(Strategy): 

11 def __init__(self, rng: random.Random | None = None): 

12 self._rng = rng or random.Random() 

13 

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

15 n = self._rng.randint(1, 100) 

16 

17 if n == 42: 

18 return Signal(symbol=bar.symbol, direction=SignalDirection.LONG) 

19 if n == 69: 

20 return Signal(symbol=bar.symbol, direction=SignalDirection.SHORT) 

21 

22 return None 

23 

24 

25class BuyOnceStrategy(Strategy): 

26 def __init__(self) -> None: 

27 self._signaled_symbols: set[str] = set() 

28 

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

30 if bar.symbol in self._signaled_symbols: 

31 return None 

32 

33 self._signaled_symbols.add(bar.symbol) 

34 

35 return Signal(symbol=bar.symbol, direction=SignalDirection.LONG) 

36 

37 

38class SmaCrossoverStrategy(Strategy): 

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

40 if short_window <= 0: 

41 raise ValueError("short_window must be positive") 

42 

43 if long_window <= short_window: 

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

45 

46 self.short_window = short_window 

47 self.long_window = long_window 

48 

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

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

51 

52 def prime(self, bars: Iterable[Bar]) -> None: 

53 """Seed rolling prices while keeping the evaluation portfolio flat.""" 

54 for bar in sorted(bars, key=lambda item: (item.timestamp, item.symbol)): 

55 self._prices[bar.symbol].append(bar.close) 

56 self._previous_direction = { 

57 symbol: SignalDirection.FLAT 

58 for symbol, prices in self._prices.items() 

59 if len(prices) == self.long_window 

60 } 

61 

62 @staticmethod 

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

64 if short_sma > long_sma: 

65 return SignalDirection.LONG 

66 if short_sma < long_sma: 66 ↛ 68line 66 didn't jump to line 68 because the condition on line 66 was always true

67 return SignalDirection.SHORT 

68 return SignalDirection.FLAT 

69 

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

71 prices = self._prices[bar.symbol] 

72 prices.append(bar.close) 

73 

74 if len(prices) < self.long_window: 

75 return None 

76 

77 values = list(prices) 

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

79 long_sma = sum(values) / self.long_window 

80 

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

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

83 

84 self._previous_direction[bar.symbol] = current_direction 

85 

86 if previous_direction is None: 

87 return None 

88 

89 if ( 

90 previous_direction != SignalDirection.LONG 

91 and current_direction == SignalDirection.LONG 

92 ): 

93 return Signal( 

94 symbol=bar.symbol, 

95 direction=SignalDirection.LONG, 

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

97 ) 

98 

99 if ( 

100 previous_direction != SignalDirection.SHORT 

101 and current_direction == SignalDirection.SHORT 

102 ): 

103 return Signal( 

104 symbol=bar.symbol, 

105 direction=SignalDirection.SHORT, 

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

107 ) 

108 

109 return None 

110 

111 

112StrategyFactory = Callable[..., Strategy] 

113 

114 

115def _create_buy_once(**params: Any) -> Strategy: 

116 if params: 

117 raise ValueError("buy_once does not accept strategy parameters") 

118 

119 return BuyOnceStrategy() 

120 

121 

122_STRATEGY_FACTORIES: dict[str, StrategyFactory] = { 

123 "sma_crossover": SmaCrossoverStrategy, 

124 "buy_once": _create_buy_once, 

125} 

126 

127 

128def create_strategy(name: str, params: dict[str, Any] | None = None) -> Strategy: 

129 try: 

130 factory = _STRATEGY_FACTORIES[name] 

131 except KeyError: 

132 raise ValueError(f"Unknown strategy: {name!r}") from None 

133 

134 return factory(**(params or {}))