Coverage for src/finbot/models/core.py: 100%

1import hashlib 

2import json 

3import math 

4from abc import ABC, abstractmethod 

5from dataclasses import dataclass 

6from datetime import datetime 

7from pathlib import Path 

8from typing import TypeVar 

9 

10from finbot.features import FeatureDataset, SupervisedDataset 

11 

12ModelScalar = str | int | float | bool | None 

13ModelT = TypeVar("ModelT", bound="Model") 

14 

15 

16@dataclass(frozen=True) 

17class TrainingConfiguration: 

18 dataset_id: str 

19 target_name: str 

20 sample_count: int 

21 

22 def to_dict(self) -> dict[str, object]: 

23 return { 

24 "dataset_id": self.dataset_id, 

25 "target_name": self.target_name, 

26 "sample_count": self.sample_count, 

27 } 

28 

29 

30@dataclass(frozen=True) 

31class ModelMetadata: 

32 model_type: str 

33 hyperparameters: tuple[tuple[str, ModelScalar], ...] 

34 feature_schema: tuple[str, ...] 

35 random_seed: int 

36 training_configuration: TrainingConfiguration 

37 

38 @property 

39 def model_id(self) -> str: 

40 """Return a stable identity for this trained model configuration.""" 

41 payload = json.dumps( 

42 self.to_dict(), 

43 sort_keys=True, 

44 separators=(",", ":"), 

45 allow_nan=False, 

46 ) 

47 return hashlib.sha256(payload.encode("utf-8")).hexdigest() 

48 

49 def to_dict(self) -> dict[str, object]: 

50 return { 

51 "model_type": self.model_type, 

52 "hyperparameters": dict(self.hyperparameters), 

53 "feature_schema": list(self.feature_schema), 

54 "random_seed": self.random_seed, 

55 "training_configuration": self.training_configuration.to_dict(), 

56 } 

57 

58 

59@dataclass(frozen=True) 

60class Prediction: 

61 symbol: str 

62 timestamp: datetime 

63 probability_positive: float 

64 model_id: str 

65 model_type: str 

66 metadata: tuple[tuple[str, ModelScalar], ...] = () 

67 

68 def __post_init__(self) -> None: 

69 if not isinstance(self.symbol, str) or not self.symbol.strip(): 

70 raise ValueError("Prediction symbol must be a non-empty string.") 

71 if not isinstance(self.timestamp, datetime): 

72 raise ValueError("Prediction timestamp must be a datetime.") 

73 if ( 

74 isinstance(self.probability_positive, bool) 

75 or not isinstance(self.probability_positive, (int, float)) 

76 or not math.isfinite(self.probability_positive) 

77 or not 0.0 <= self.probability_positive <= 1.0 

78 ): 

79 raise ValueError("Prediction probability must be finite and between 0 and 1.") 

80 if not isinstance(self.model_id, str) or not self.model_id.strip(): 

81 raise ValueError("Prediction model ID must be a non-empty string.") 

82 if not isinstance(self.model_type, str) or not self.model_type.strip(): 

83 raise ValueError("Prediction model type must be a non-empty string.") 

84 if any(not isinstance(key, str) or not key.strip() for key, _ in self.metadata): 

85 raise ValueError("Prediction metadata keys must be non-empty strings.") 

86 metadata_keys = [key for key, _ in self.metadata] 

87 if len(set(metadata_keys)) != len(metadata_keys): 

88 raise ValueError("Prediction metadata keys must be unique.") 

89 if any( 

90 value is not None and not isinstance(value, (str, int, float, bool)) 

91 for _, value in self.metadata 

92 ): 

93 raise ValueError("Prediction metadata values must be scalar.") 

94 

95 def to_dict(self) -> dict[str, object]: 

96 return { 

97 "symbol": self.symbol, 

98 "timestamp": self.timestamp.isoformat(), 

99 "probability_positive": float(self.probability_positive), 

100 "model_id": self.model_id, 

101 "model_type": self.model_type, 

102 "metadata": dict(self.metadata), 

103 } 

104 

105 

106class Model(ABC): 

107 """Strategy-independent interface for supervised models.""" 

108 

109 @property 

110 @abstractmethod 

111 def metadata(self) -> ModelMetadata: 

112 """Describe the trained model and the data used to fit it.""" 

113 ... 

114 

115 @abstractmethod 

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

117 """Fit the model using a leakage-safe supervised dataset.""" 

118 ... 

119 

120 def fit(self: ModelT, dataset: SupervisedDataset) -> ModelT: 

121 """Fit the model and return it for convenient composition.""" 

122 self.train(dataset) 

123 return self 

124 

125 @abstractmethod 

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

127 """Return positive-class probabilities for feature rows.""" 

128 ... 

129 

130 @abstractmethod 

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

132 """Save a trained model artifact.""" 

133 ...