Coverage for src/finbot/models/logistic.py: 85%

1import json 

2import math 

3from pathlib import Path 

4from typing import cast 

5 

6from sklearn.linear_model import LogisticRegression 

7 

8from finbot.features import FeatureDataset, SupervisedDataset 

9from finbot.models.core import ( 

10 Model, 

11 ModelMetadata, 

12 ModelScalar, 

13 Prediction, 

14 TrainingConfiguration, 

15) 

16 

17MODEL_ARTIFACT_FORMAT_VERSION = 1 

18LOGISTIC_REGRESSION_MODEL_TYPE = "logistic_regression" 

19 

20 

21class LogisticRegressionModel(Model): 

22 """Deterministic binary classifier for positive forward returns.""" 

23 

24 def __init__( 

25 self, 

26 *, 

27 random_seed: int = 0, 

28 c: float = 1.0, 

29 max_iter: int = 1_000, 

30 ) -> None: 

31 if isinstance(random_seed, bool) or not isinstance(random_seed, int): 

32 raise ValueError("Random seed must be an integer.") 

33 if isinstance(c, bool) or not isinstance(c, (int, float)) or not math.isfinite(c) or c <= 0: 

34 raise ValueError("Logistic-regression C must be a finite positive number.") 

35 if isinstance(max_iter, bool) or not isinstance(max_iter, int) or max_iter < 1: 

36 raise ValueError("Maximum iterations must be a positive integer.") 

37 

38 self.random_seed = random_seed 

39 self.c = float(c) 

40 self.max_iter = max_iter 

41 self._metadata: ModelMetadata | None = None 

42 self._coefficients: tuple[float, ...] | None = None 

43 self._intercept: float | None = None 

44 

45 @property 

46 def metadata(self) -> ModelMetadata: 

47 if self._metadata is None: 

48 raise RuntimeError("Model has not been trained.") 

49 return self._metadata 

50 

51 def train(self, dataset: SupervisedDataset) -> None: 

52 feature_schema = self._validate_schema(dataset.feature_names) 

53 if not dataset.rows: 

54 raise ValueError("Training dataset must not be empty.") 

55 

56 features: list[list[float]] = [] 

57 labels: list[int] = [] 

58 for index, row in enumerate(dataset.rows): 

59 features.append( 

60 list(self._validate_values(row.features, feature_schema, context=f"Training row {index}")), 

61 ) 

62 if isinstance(row.target, bool) or not isinstance(row.target, (int, float)): 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true

63 raise ValueError(f"Training target at row {index} must be a number.") 

64 target = float(row.target) 

65 if not math.isfinite(target): 

66 raise ValueError(f"Training target at row {index} must be finite.") 

67 labels.append(int(target > 0.0)) 

68 

69 if len(set(labels)) < 2: 

70 raise ValueError("Training targets must contain both positive and non-positive classes.") 

71 

72 estimator = LogisticRegression( 

73 C=self.c, 

74 max_iter=self.max_iter, 

75 random_state=self.random_seed, 

76 solver="liblinear", 

77 ) 

78 estimator.fit(features, labels) 

79 

80 self._coefficients = tuple(float(value) for value in estimator.coef_[0]) 

81 self._intercept = float(estimator.intercept_[0]) 

82 self._metadata = ModelMetadata( 

83 model_type=LOGISTIC_REGRESSION_MODEL_TYPE, 

84 hyperparameters=( 

85 ("C", self.c), 

86 ("max_iter", self.max_iter), 

87 ("solver", "liblinear"), 

88 ), 

89 feature_schema=feature_schema, 

90 random_seed=self.random_seed, 

91 training_configuration=TrainingConfiguration( 

92 dataset_id=dataset.dataset_id, 

93 target_name=dataset.target_name, 

94 sample_count=len(dataset.rows), 

95 ), 

96 ) 

97 

98 def predict_proba(self, dataset: FeatureDataset) -> tuple[Prediction, ...]: 

99 metadata = self.metadata 

100 if tuple(dataset.feature_names) != metadata.feature_schema: 

101 raise ValueError( 

102 "Inference feature schema must exactly match the trained feature names and ordering.", 

103 ) 

104 assert self._coefficients is not None 

105 assert self._intercept is not None 

106 

107 predictions: list[Prediction] = [] 

108 for index, row in enumerate(dataset.rows): 

109 values = self._validate_values( 

110 row.values, 

111 metadata.feature_schema, 

112 context=f"Inference row {index}", 

113 ) 

114 score = self._intercept + sum( 

115 coefficient * value 

116 for coefficient, value in zip(self._coefficients, values) 

117 ) 

118 probability = self._sigmoid(score) 

119 predictions.append( 

120 Prediction( 

121 symbol=row.symbol, 

122 timestamp=row.timestamp, 

123 probability_positive=probability, 

124 model_id=metadata.model_id, 

125 model_type=metadata.model_type, 

126 metadata=( 

127 ("training_dataset_id", metadata.training_configuration.dataset_id), 

128 ("target_name", metadata.training_configuration.target_name), 

129 ), 

130 ), 

131 ) 

132 return tuple(predictions) 

133 

134 def save(self, path: Path) -> None: 

135 metadata = self.metadata 

136 assert self._coefficients is not None 

137 assert self._intercept is not None 

138 payload = { 

139 "format_version": MODEL_ARTIFACT_FORMAT_VERSION, 

140 "metadata": metadata.to_dict(), 

141 "parameters": { 

142 "coefficients": list(self._coefficients), 

143 "intercept": self._intercept, 

144 }, 

145 } 

146 serialized = json.dumps(payload, sort_keys=True, separators=(",", ":"), allow_nan=False) 

147 path.parent.mkdir(parents=True, exist_ok=True) 

148 temporary = path.with_suffix(f"{path.suffix}.tmp") 

149 temporary.write_text(serialized, encoding="utf-8") 

150 temporary.replace(path) 

151 

152 @classmethod 

153 def load(cls, path: Path) -> "LogisticRegressionModel": 

154 try: 

155 payload = json.loads(path.read_text(encoding="utf-8")) 

156 except FileNotFoundError: 

157 raise 

158 except (OSError, json.JSONDecodeError) as exc: 

159 raise ValueError("Model artifact is not valid JSON.") from exc 

160 

161 metadata, coefficients, intercept = cls._parse_artifact(payload) 

162 hyperparameters = dict(metadata.hyperparameters) 

163 model = cls( 

164 random_seed=metadata.random_seed, 

165 c=cast(float, hyperparameters["C"]), 

166 max_iter=cast(int, hyperparameters["max_iter"]), 

167 ) 

168 model._metadata = metadata 

169 model._coefficients = coefficients 

170 model._intercept = intercept 

171 return model 

172 

173 @staticmethod 

174 def _validate_schema(feature_names: tuple[str, ...]) -> tuple[str, ...]: 

175 if not feature_names: 

176 raise ValueError("Training feature schema must not be empty.") 

177 if any(not isinstance(name, str) or not name.strip() for name in feature_names): 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true

178 raise ValueError("Training feature names must be non-empty strings.") 

179 if len(set(feature_names)) != len(feature_names): 179 ↛ 180line 179 didn't jump to line 180 because the condition on line 179 was never true

180 raise ValueError("Training feature names must be unique.") 

181 return tuple(feature_names) 

182 

183 @staticmethod 

184 def _validate_values( 

185 raw_values: tuple[float | None, ...], 

186 feature_schema: tuple[str, ...], 

187 *, 

188 context: str, 

189 ) -> tuple[float, ...]: 

190 if len(raw_values) != len(feature_schema): 190 ↛ 191line 190 didn't jump to line 191 because the condition on line 190 was never true

191 raise ValueError(f"{context} values must match the feature schema.") 

192 values: list[float] = [] 

193 for value in raw_values: 

194 if value is None or isinstance(value, bool) or not isinstance(value, (int, float)): 

195 raise ValueError(f"{context} features must be finite numbers.") 

196 normalized = float(value) 

197 if not math.isfinite(normalized): 

198 raise ValueError(f"{context} features must be finite numbers.") 

199 values.append(normalized) 

200 return tuple(values) 

201 

202 @staticmethod 

203 def _sigmoid(score: float) -> float: 

204 if score >= 0: 

205 return 1.0 / (1.0 + math.exp(-score)) 

206 exponential = math.exp(score) 

207 return exponential / (1.0 + exponential) 

208 

209 @staticmethod 

210 def _parse_artifact(payload: object) -> tuple[ModelMetadata, tuple[float, ...], float]: 

211 if not isinstance(payload, dict) or payload.get("format_version") != MODEL_ARTIFACT_FORMAT_VERSION: 

212 raise ValueError("Unsupported model artifact format version.") 

213 raw_metadata = payload.get("metadata") 

214 raw_parameters = payload.get("parameters") 

215 if not isinstance(raw_metadata, dict) or not isinstance(raw_parameters, dict): 215 ↛ 216line 215 didn't jump to line 216 because the condition on line 215 was never true

216 raise ValueError("Model artifact is missing metadata or parameters.") 

217 if raw_metadata.get("model_type") != LOGISTIC_REGRESSION_MODEL_TYPE: 217 ↛ 218line 217 didn't jump to line 218 because the condition on line 217 was never true

218 raise ValueError("Model artifact has an unexpected model type.") 

219 

220 raw_hyperparameters = raw_metadata.get("hyperparameters") 

221 raw_schema = raw_metadata.get("feature_schema") 

222 raw_seed = raw_metadata.get("random_seed") 

223 raw_training = raw_metadata.get("training_configuration") 

224 if not isinstance(raw_hyperparameters, dict): 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true

225 raise ValueError("Model hyperparameters must be an object.") 

226 if set(raw_hyperparameters) != {"C", "max_iter", "solver"} or raw_hyperparameters.get("solver") != "liblinear": 226 ↛ 227line 226 didn't jump to line 227 because the condition on line 226 was never true

227 raise ValueError("Model artifact contains unsupported hyperparameters.") 

228 if not isinstance(raw_schema, list) or not all(isinstance(name, str) for name in raw_schema): 228 ↛ 229line 228 didn't jump to line 229 because the condition on line 228 was never true

229 raise ValueError("Model feature schema must be a list of names.") 

230 feature_schema = LogisticRegressionModel._validate_schema(tuple(raw_schema)) 

231 if isinstance(raw_seed, bool) or not isinstance(raw_seed, int): 231 ↛ 232line 231 didn't jump to line 232 because the condition on line 231 was never true

232 raise ValueError("Model random seed must be an integer.") 

233 if not isinstance(raw_training, dict): 233 ↛ 234line 233 didn't jump to line 234 because the condition on line 233 was never true

234 raise ValueError("Model training configuration must be an object.") 

235 dataset_id = raw_training.get("dataset_id") 

236 target_name = raw_training.get("target_name") 

237 sample_count = raw_training.get("sample_count") 

238 if not isinstance(dataset_id, str) or not dataset_id or not isinstance(target_name, str) or not target_name: 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true

239 raise ValueError("Model training configuration is invalid.") 

240 if isinstance(sample_count, bool) or not isinstance(sample_count, int) or sample_count < 1: 240 ↛ 241line 240 didn't jump to line 241 because the condition on line 240 was never true

241 raise ValueError("Model training sample count must be positive.") 

242 

243 raw_coefficients = raw_parameters.get("coefficients") 

244 raw_intercept = raw_parameters.get("intercept") 

245 if not isinstance(raw_coefficients, list) or len(raw_coefficients) != len(feature_schema): 

246 raise ValueError("Model coefficients must match the feature schema.") 

247 coefficients = LogisticRegressionModel._parse_finite_numbers(raw_coefficients, "coefficients") 

248 intercept_values = LogisticRegressionModel._parse_finite_numbers([raw_intercept], "intercept") 

249 

250 c = raw_hyperparameters.get("C") 

251 max_iter = raw_hyperparameters.get("max_iter") 

252 model = LogisticRegressionModel(random_seed=raw_seed, c=cast(float, c), max_iter=cast(int, max_iter)) 

253 metadata = ModelMetadata( 

254 model_type=LOGISTIC_REGRESSION_MODEL_TYPE, 

255 hyperparameters=model._hyperparameters(), 

256 feature_schema=feature_schema, 

257 random_seed=raw_seed, 

258 training_configuration=TrainingConfiguration( 

259 dataset_id=dataset_id, 

260 target_name=target_name, 

261 sample_count=sample_count, 

262 ), 

263 ) 

264 return metadata, coefficients, intercept_values[0] 

265 

266 def _hyperparameters(self) -> tuple[tuple[str, ModelScalar], ...]: 

267 return (("C", self.c), ("max_iter", self.max_iter), ("solver", "liblinear")) 

268 

269 @staticmethod 

270 def _parse_finite_numbers(values: list[object], name: str) -> tuple[float, ...]: 

271 if any( 

272 isinstance(value, bool) 

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

274 or not math.isfinite(value) 

275 for value in values 

276 ): 

277 raise ValueError(f"Model {name} must contain finite numbers.") 

278 return tuple(float(value) for value in values) 

279 

280 

281def load_model(path: Path) -> Model: 

282 """Load a supported model artifact from disk.""" 

283 try: 

284 payload = json.loads(path.read_text(encoding="utf-8")) 

285 except FileNotFoundError: 

286 raise 

287 except (OSError, json.JSONDecodeError) as exc: 

288 raise ValueError("Model artifact is not valid JSON.") from exc 

289 if not isinstance(payload, dict) or not isinstance(payload.get("metadata"), dict): 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true

290 raise ValueError("Model artifact is missing metadata.") 

291 if payload["metadata"].get("model_type") == LOGISTIC_REGRESSION_MODEL_TYPE: 

292 return LogisticRegressionModel.load(path) 

293 raise ValueError("Model artifact uses an unsupported model type.")