Coverage for src/finbot/features/market.py: 100%

1import statistics 

2from collections.abc import Sequence 

3from dataclasses import dataclass 

4 

5from finbot.features.core import Feature, FeatureValue 

6from finbot.market.models import Bar 

7 

8 

9def _validate_window(name: str, value: int) -> None: 

10 if isinstance(value, bool) or not isinstance(value, int) or value < 1: 

11 raise ValueError(f"{name} must be a positive integer.") 

12 

13 

14@dataclass(frozen=True) 

15class ReturnFeature(Feature): 

16 period: int 

17 

18 def __post_init__(self) -> None: 

19 _validate_window("Return period", self.period) 

20 

21 @property 

22 def name(self) -> str: 

23 return f"return_{self.period}" 

24 

25 def calculate(self, history: Sequence[Bar]) -> FeatureValue: 

26 if len(history) <= self.period: 

27 return None 

28 return history[-1].close / history[-self.period - 1].close - 1.0 

29 

30 

31@dataclass(frozen=True) 

32class SimpleMovingAverageFeature(Feature): 

33 window: int 

34 

35 def __post_init__(self) -> None: 

36 _validate_window("SMA window", self.window) 

37 

38 @property 

39 def name(self) -> str: 

40 return f"sma_{self.window}" 

41 

42 def calculate(self, history: Sequence[Bar]) -> FeatureValue: 

43 if len(history) < self.window: 

44 return None 

45 return statistics.fmean(bar.close for bar in history[-self.window:]) 

46 

47 

48@dataclass(frozen=True) 

49class SmaRatioFeature(Feature): 

50 short_window: int = 10 

51 long_window: int = 30 

52 

53 def __post_init__(self) -> None: 

54 _validate_window("Short SMA window", self.short_window) 

55 _validate_window("Long SMA window", self.long_window) 

56 if self.short_window >= self.long_window: 

57 raise ValueError("Short SMA window must be smaller than long SMA window.") 

58 

59 @property 

60 def name(self) -> str: 

61 return f"sma_ratio_{self.short_window}_{self.long_window}" 

62 

63 def calculate(self, history: Sequence[Bar]) -> FeatureValue: 

64 if len(history) < self.long_window: 

65 return None 

66 short_sma = statistics.fmean(bar.close for bar in history[-self.short_window:]) 

67 long_sma = statistics.fmean(bar.close for bar in history[-self.long_window:]) 

68 return short_sma / long_sma - 1.0 

69 

70 

71@dataclass(frozen=True) 

72class RollingVolatilityFeature(Feature): 

73 window: int = 20 

74 

75 def __post_init__(self) -> None: 

76 _validate_window("Volatility window", self.window) 

77 

78 @property 

79 def name(self) -> str: 

80 return f"rolling_volatility_{self.window}" 

81 

82 def calculate(self, history: Sequence[Bar]) -> FeatureValue: 

83 if len(history) <= self.window: 

84 return None 

85 closes = [bar.close for bar in history[-self.window - 1:]] 

86 returns = [current / previous - 1.0 for previous, current in zip(closes, closes[1:])] 

87 return statistics.pstdev(returns) 

88 

89 

90@dataclass(frozen=True) 

91class VolumeChangeFeature(Feature): 

92 period: int = 1 

93 

94 def __post_init__(self) -> None: 

95 _validate_window("Volume-change period", self.period) 

96 

97 @property 

98 def name(self) -> str: 

99 return f"volume_change_{self.period}" 

100 

101 def calculate(self, history: Sequence[Bar]) -> FeatureValue: 

102 if len(history) <= self.period: 

103 return None 

104 previous_volume = history[-self.period - 1].volume 

105 if previous_volume == 0.0: 

106 return None 

107 return history[-1].volume / previous_volume - 1.0 

108 

109 

110@dataclass(frozen=True) 

111class IntrabarRangeFeature(Feature): 

112 @property 

113 def name(self) -> str: 

114 return "intrabar_range" 

115 

116 def calculate(self, history: Sequence[Bar]) -> FeatureValue: 

117 current = history[-1] 

118 return (current.high - current.low) / current.close 

119 

120 

121@dataclass(frozen=True) 

122class MomentumFeature(Feature): 

123 window: int = 10 

124 

125 def __post_init__(self) -> None: 

126 _validate_window("Momentum window", self.window) 

127 

128 @property 

129 def name(self) -> str: 

130 return f"momentum_{self.window}" 

131 

132 def calculate(self, history: Sequence[Bar]) -> FeatureValue: 

133 if len(history) <= self.window: 

134 return None 

135 return history[-1].close / history[-self.window - 1].close - 1.0 

136 

137 

138def initial_feature_set() -> tuple[Feature, ...]: 

139 return ( 

140 ReturnFeature(1), 

141 ReturnFeature(5), 

142 ReturnFeature(20), 

143 SimpleMovingAverageFeature(10), 

144 SimpleMovingAverageFeature(30), 

145 SmaRatioFeature(10, 30), 

146 RollingVolatilityFeature(20), 

147 VolumeChangeFeature(1), 

148 IntrabarRangeFeature(), 

149 MomentumFeature(10), 

150 )