Coverage for src/finbot/features/supervised.py: 93%

1import hashlib 

2import json 

3import math 

4from collections.abc import Iterable 

5from dataclasses import dataclass 

6from datetime import datetime 

7 

8from finbot.features.core import FeatureDataset 

9from finbot.market.models import Bar 

10 

11SUPERVISED_DATASET_GENERATOR_VERSION = 1 

12 

13 

14@dataclass(frozen=True) 

15class ForwardReturnTarget: 

16 """A future close-to-close return used as a supervised-learning target.""" 

17 

18 horizon: int = 1 

19 

20 def __post_init__(self) -> None: 

21 if isinstance(self.horizon, bool) or not isinstance(self.horizon, int) or self.horizon < 1: 

22 raise ValueError("Target horizon must be a positive integer.") 

23 

24 @property 

25 def name(self) -> str: 

26 return f"forward_return_{self.horizon}" 

27 

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

29 return {"name": self.name, "horizon": self.horizon} 

30 

31 

32@dataclass(frozen=True) 

33class SupervisedDatasetMetadata: 

34 source_feature_dataset_id: str 

35 target: ForwardReturnTarget 

36 generator_version: int = SUPERVISED_DATASET_GENERATOR_VERSION 

37 

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

39 return { 

40 "source_feature_dataset_id": self.source_feature_dataset_id, 

41 "target": self.target.to_dict(), 

42 "generator_version": self.generator_version, 

43 } 

44 

45 

46@dataclass(frozen=True) 

47class SupervisedRow: 

48 symbol: str 

49 timestamp: datetime 

50 target_timestamp: datetime 

51 features: tuple[float, ...] 

52 target: float 

53 

54 

55@dataclass(frozen=True) 

56class SupervisedDataset: 

57 feature_names: tuple[str, ...] 

58 target_name: str 

59 rows: tuple[SupervisedRow, ...] 

60 metadata: SupervisedDatasetMetadata 

61 

62 @property 

63 def dataset_id(self) -> str: 

64 serialized = json.dumps( 

65 self.to_dict(), 

66 sort_keys=True, 

67 separators=(",", ":"), 

68 allow_nan=False, 

69 ) 

70 return hashlib.sha256(serialized.encode("utf-8")).hexdigest() 

71 

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

73 return { 

74 "feature_names": list(self.feature_names), 

75 "target_name": self.target_name, 

76 "metadata": self.metadata.to_dict(), 

77 "rows": [ 

78 { 

79 "symbol": row.symbol, 

80 "timestamp": row.timestamp.isoformat(), 

81 "target_timestamp": row.target_timestamp.isoformat(), 

82 "features": list(row.features), 

83 "target": row.target, 

84 } 

85 for row in self.rows 

86 ], 

87 } 

88 

89 

90class SupervisedDatasetBuilder: 

91 """Align causal feature rows with future returns for the same symbol.""" 

92 

93 def __init__(self, target: ForwardReturnTarget | None = None) -> None: 

94 self.target = target or ForwardReturnTarget() 

95 

96 def build( 

97 self, 

98 features: FeatureDataset, 

99 bars: Iterable[Bar], 

100 ) -> SupervisedDataset: 

101 feature_rows = {(row.symbol, row.timestamp): row for row in features.rows} 

102 if len(feature_rows) != len(features.rows): 102 ↛ 103line 102 didn't jump to line 103 because the condition on line 102 was never true

103 raise ValueError("Feature dataset contains duplicate symbol/timestamp rows.") 

104 

105 bars_by_symbol: dict[str, list[Bar]] = {} 

106 seen_bars: set[tuple[str, datetime]] = set() 

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

108 key = (bar.symbol, bar.timestamp) 

109 if key in seen_bars: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true

110 raise ValueError( 

111 f"Duplicate bar for {bar.symbol} at {bar.timestamp.isoformat()}.", 

112 ) 

113 seen_bars.add(key) 

114 bars_by_symbol.setdefault(bar.symbol, []).append(bar) 

115 

116 rows: list[SupervisedRow] = [] 

117 matched_features: set[tuple[str, datetime]] = set() 

118 for symbol, symbol_bars in bars_by_symbol.items(): 

119 for index, bar in enumerate(symbol_bars): 

120 key = (symbol, bar.timestamp) 

121 feature_row = feature_rows.get(key) 

122 if feature_row is None: 122 ↛ 123line 122 didn't jump to line 123 because the condition on line 122 was never true

123 continue 

124 matched_features.add(key) 

125 

126 target_index = index + self.target.horizon 

127 if target_index >= len(symbol_bars) or any( 

128 value is None for value in feature_row.values 

129 ): 

130 continue 

131 

132 target_bar = symbol_bars[target_index] 

133 target_value = target_bar.close / bar.close - 1.0 

134 if not math.isfinite(target_value): 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true

135 raise ValueError("Supervised target must be finite.") 

136 rows.append( 

137 SupervisedRow( 

138 symbol=symbol, 

139 timestamp=bar.timestamp, 

140 target_timestamp=target_bar.timestamp, 

141 features=tuple(float(value) for value in feature_row.values if value is not None), 

142 target=float(target_value), 

143 ), 

144 ) 

145 

146 missing_bars = set(feature_rows) - matched_features 

147 if missing_bars: 

148 symbol, timestamp = min(missing_bars, key=lambda item: (item[1], item[0])) 

149 raise ValueError( 

150 f"Feature row for {symbol} at {timestamp.isoformat()} has no matching bar.", 

151 ) 

152 

153 rows.sort(key=lambda row: (row.timestamp, row.symbol)) 

154 metadata = SupervisedDatasetMetadata( 

155 source_feature_dataset_id=features.dataset_id, 

156 target=self.target, 

157 ) 

158 return SupervisedDataset( 

159 feature_names=features.feature_names, 

160 target_name=self.target.name, 

161 rows=tuple(rows), 

162 metadata=metadata, 

163 ) 

164 

165 

166def build_supervised_dataset( 

167 features: FeatureDataset, 

168 bars: Iterable[Bar], 

169 *, 

170 horizon: int = 1, 

171) -> SupervisedDataset: 

172 """Build a supervised dataset using a forward-return target.""" 

173 return SupervisedDatasetBuilder(ForwardReturnTarget(horizon)).build(features, bars)