Coverage for src/finbot/strategy/ml.py: 99%

1import math 

2from dataclasses import dataclass 

3from datetime import datetime, timedelta 

4from pathlib import Path 

5 

6from finbot.features import FeatureDataset, FeaturePipeline 

7from finbot.market.models import Bar 

8from finbot.models import Model, Prediction, load_model 

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

10 

11 

12@dataclass(frozen=True) 

13class ProbabilityThresholdPolicy: 

14 """Convert positive-class probabilities into ordinary trading signals. 

15 

16 Thresholds are inclusive: probabilities at or above ``long_threshold`` are 

17 LONG, and probabilities at or below ``short_threshold`` are SHORT. Values 

18 between the thresholds are FLAT. Setting ``short_threshold`` to ``None`` 

19 disables SHORT signals. 

20 """ 

21 

22 long_threshold: float = 0.55 

23 short_threshold: float | None = 0.45 

24 max_prediction_age: timedelta = timedelta(0) 

25 

26 def __post_init__(self) -> None: 

27 self._validate_probability("Long threshold", self.long_threshold) 

28 if self.short_threshold is not None: 

29 self._validate_probability("Short threshold", self.short_threshold) 

30 if self.short_threshold >= self.long_threshold: 

31 raise ValueError("Short threshold must be smaller than long threshold.") 

32 if not isinstance(self.max_prediction_age, timedelta) or self.max_prediction_age < timedelta(0): 

33 raise ValueError("Maximum prediction age must be a non-negative timedelta.") 

34 

35 def to_signal( 

36 self, 

37 prediction: Prediction | None, 

38 *, 

39 symbol: str, 

40 timestamp: datetime, 

41 ) -> Signal | None: 

42 """Return no signal for missing, mismatched, future, or stale predictions.""" 

43 if prediction is None or prediction.symbol != symbol: 

44 return None 

45 

46 try: 

47 age = timestamp - prediction.timestamp 

48 except TypeError: 

49 return None 

50 if age < timedelta(0) or age > self.max_prediction_age: 

51 return None 

52 

53 probability = prediction.probability_positive 

54 if probability >= self.long_threshold: 

55 direction = SignalDirection.LONG 

56 strength = probability 

57 elif self.short_threshold is not None and probability <= self.short_threshold: 

58 direction = SignalDirection.SHORT 

59 strength = 1.0 - probability 

60 else: 

61 direction = SignalDirection.FLAT 

62 strength = 0.0 

63 

64 return Signal( 

65 symbol=prediction.symbol, 

66 direction=direction, 

67 strength=strength, 

68 metadata={"prediction": prediction.to_dict()}, 

69 ) 

70 

71 @staticmethod 

72 def _validate_probability(name: str, value: float) -> None: 

73 if ( 

74 isinstance(value, bool) 

75 or not isinstance(value, (int, float)) 

76 or not math.isfinite(value) 

77 or not 0.0 <= value <= 1.0 

78 ): 

79 raise ValueError(f"{name} must be finite and between 0 and 1.") 

80 

81 

82class ModelStrategy(Strategy): 

83 """Run offline model inference and emit signals without touching execution.""" 

84 

85 def __init__( 

86 self, 

87 model: Model, 

88 feature_pipeline: FeaturePipeline, 

89 policy: ProbabilityThresholdPolicy, 

90 ) -> None: 

91 if feature_pipeline.feature_names != model.metadata.feature_schema: 

92 raise ValueError( 

93 "Feature pipeline schema must exactly match the model artifact schema.", 

94 ) 

95 self.model = model 

96 self.feature_pipeline = feature_pipeline 

97 self.policy = policy 

98 self._history_by_symbol: dict[str, list[Bar]] = {} 

99 self._predictions: list[Prediction] = [] 

100 self._previous_action_by_symbol: dict[str, SignalDirection] = {} 

101 

102 @classmethod 

103 def from_artifact( 

104 cls, 

105 path: Path, 

106 *, 

107 feature_pipeline: FeaturePipeline, 

108 policy: ProbabilityThresholdPolicy, 

109 ) -> "ModelStrategy": 

110 return cls( 

111 model=load_model(path), 

112 feature_pipeline=feature_pipeline, 

113 policy=policy, 

114 ) 

115 

116 @property 

117 def predictions(self) -> tuple[Prediction, ...]: 

118 return tuple(self._predictions) 

119 

120 def prime(self, bars: list[Bar]) -> None: 

121 """Seed causal feature history without producing predictions or signals.""" 

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

123 history = self._history_by_symbol.setdefault(bar.symbol, []) 

124 if history and bar.timestamp <= history[-1].timestamp: 

125 raise ValueError("Priming bars must be unique and ordered by timestamp.") 

126 history.append(bar) 

127 

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

129 history = self._history_by_symbol.setdefault(bar.symbol, []) 

130 if history and bar.timestamp <= history[-1].timestamp: 

131 return None 

132 history.append(bar) 

133 

134 features = self.feature_pipeline.generate(history) 

135 current_row = features.rows[-1] 

136 if any(value is None for value in current_row.values): 

137 return None 

138 

139 inference_dataset = FeatureDataset( 

140 feature_names=features.feature_names, 

141 rows=(current_row,), 

142 metadata=features.metadata, 

143 ) 

144 predictions = self.model.predict_proba(inference_dataset) 

145 prediction = predictions[0] if len(predictions) == 1 else None 

146 signal = self.policy.to_signal( 

147 prediction, 

148 symbol=bar.symbol, 

149 timestamp=bar.timestamp, 

150 ) 

151 if prediction is not None: 151 ↛ 153line 151 didn't jump to line 153 because the condition on line 151 was always true

152 self._predictions.append(prediction) 

153 if signal is not None and signal.direction != SignalDirection.FLAT: 

154 previous = self._previous_action_by_symbol.get(signal.symbol) 

155 if previous == signal.direction: 

156 return None 

157 self._previous_action_by_symbol[signal.symbol] = signal.direction 

158 return signal 

159 

160 

161class PredictionStrategy(Strategy): 

162 """Replay timestamped model predictions through an ordinary signal policy.""" 

163 

164 def __init__( 

165 self, 

166 predictions: tuple[Prediction, ...], 

167 policy: ProbabilityThresholdPolicy, 

168 ) -> None: 

169 keys = [(prediction.symbol, prediction.timestamp) for prediction in predictions] 

170 if len(keys) != len(set(keys)): 

171 raise ValueError("Predictions must be unique by symbol and timestamp.") 

172 self._predictions = dict(zip(keys, predictions, strict=True)) 

173 self.policy = policy 

174 self._previous_action_by_symbol: dict[str, SignalDirection] = {} 

175 

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

177 prediction = self._predictions.get((bar.symbol, bar.timestamp)) 

178 signal = self.policy.to_signal( 

179 prediction, 

180 symbol=bar.symbol, 

181 timestamp=bar.timestamp, 

182 ) 

183 if signal is not None and signal.direction != SignalDirection.FLAT: 

184 previous = self._previous_action_by_symbol.get(signal.symbol) 

185 if previous == signal.direction: 

186 return None 

187 self._previous_action_by_symbol[signal.symbol] = signal.direction 

188 return signal