Coverage for src/finbot/features/core.py: 98%
101 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-04 12:05 +0000
1import hashlib
2import json
3import math
4from abc import ABC, abstractmethod
5from collections.abc import Iterable, Sequence
6from dataclasses import dataclass, fields, is_dataclass
7from datetime import datetime
9from finbot.market.models import Bar
11FeatureValue = float | None
12FeatureParameter = str | int | float | bool | None
13FEATURE_GENERATOR_VERSION = 1
16class Feature(ABC):
17 """A causal transformation of one symbol's bar history."""
19 @property
20 @abstractmethod
21 def name(self) -> str:
22 """Return the stable name used in the generated dataset."""
23 ...
25 @abstractmethod
26 def calculate(self, history: Sequence[Bar]) -> FeatureValue:
27 """Calculate a value using history ending at the current bar."""
28 ...
30 def configuration(self) -> tuple[tuple[str, FeatureParameter], ...]:
31 """Return stable parameters describing this feature's configuration."""
32 if not is_dataclass(self):
33 return ()
35 configuration: list[tuple[str, FeatureParameter]] = []
36 for field_definition in fields(self):
37 value = getattr(self, field_definition.name)
38 if value is not None and not isinstance(value, (str, int, float, bool)): 38 ↛ 39line 38 didn't jump to line 39 because the condition on line 38 was never true
39 raise TypeError(
40 f"Feature parameter {field_definition.name!r} must be a scalar value.",
41 )
42 configuration.append((field_definition.name, value))
43 return tuple(configuration)
46@dataclass(frozen=True)
47class FeatureSchema:
48 name: str
49 parameters: tuple[tuple[str, FeatureParameter], ...] = ()
51 def to_dict(self) -> dict[str, object]:
52 return {"name": self.name, "parameters": dict(self.parameters)}
55@dataclass(frozen=True)
56class FeatureDatasetMetadata:
57 source_dataset_id: str
58 feature_schema: tuple[FeatureSchema, ...]
59 generator_version: int = FEATURE_GENERATOR_VERSION
61 def to_dict(self) -> dict[str, object]:
62 return {
63 "source_dataset_id": self.source_dataset_id,
64 "feature_schema": [feature.to_dict() for feature in self.feature_schema],
65 "generator_version": self.generator_version,
66 }
69@dataclass(frozen=True)
70class FeatureRow:
71 symbol: str
72 timestamp: datetime
73 values: tuple[FeatureValue, ...]
76@dataclass(frozen=True)
77class FeatureDataset:
78 feature_names: tuple[str, ...]
79 rows: tuple[FeatureRow, ...]
80 metadata: FeatureDatasetMetadata
82 @property
83 def dataset_id(self) -> str:
84 payload = json.dumps(
85 self.to_dict(),
86 sort_keys=True,
87 separators=(",", ":"),
88 allow_nan=False,
89 )
90 return hashlib.sha256(payload.encode("utf-8")).hexdigest()
92 def to_dict(self) -> dict[str, object]:
93 return {
94 "feature_names": list(self.feature_names),
95 "metadata": self.metadata.to_dict(),
96 "rows": [
97 {
98 "symbol": row.symbol,
99 "timestamp": row.timestamp.isoformat(),
100 "values": list(row.values),
101 }
102 for row in self.rows
103 ],
104 }
107class FeaturePipeline:
108 """Generate deterministic, time-ordered features from normalized bars."""
110 def __init__(self, features: Iterable[Feature]) -> None:
111 self.features = tuple(features)
112 raw_names = tuple(feature.name for feature in self.features)
114 if any(not isinstance(name, str) or not name.strip() for name in raw_names):
115 raise ValueError("Feature names must be non-empty strings.")
117 names = tuple(name.strip() for name in raw_names)
118 if len(set(names)) != len(names):
119 raise ValueError("Feature names must be unique.")
121 self.feature_names = names
123 def generate(
124 self,
125 bars: Iterable[Bar],
126 *,
127 source_dataset_id: str | None = None,
128 ) -> FeatureDataset:
129 ordered_bars = sorted(bars, key=lambda bar: (bar.timestamp, bar.symbol))
130 resolved_source_id = (
131 self._source_fingerprint(ordered_bars)
132 if source_dataset_id is None
133 else source_dataset_id
134 )
135 if not isinstance(resolved_source_id, str) or not resolved_source_id.strip():
136 raise ValueError("Source dataset ID must be a non-empty string.")
137 seen_events: set[tuple[str, datetime]] = set()
138 histories: dict[str, list[Bar]] = {}
139 rows: list[FeatureRow] = []
141 for bar in ordered_bars:
142 event_key = (bar.symbol, bar.timestamp)
143 if event_key in seen_events:
144 raise ValueError(
145 f"Duplicate bar for {bar.symbol} at {bar.timestamp.isoformat()}.",
146 )
147 seen_events.add(event_key)
149 history = histories.setdefault(bar.symbol, [])
150 history.append(bar)
151 causal_history = tuple(history)
153 values = tuple(
154 self._validate_value(feature.name, feature.calculate(causal_history))
155 for feature in self.features
156 )
157 rows.append(
158 FeatureRow(
159 symbol=bar.symbol,
160 timestamp=bar.timestamp,
161 values=values,
162 ),
163 )
165 metadata = FeatureDatasetMetadata(
166 source_dataset_id=resolved_source_id.strip(),
167 feature_schema=tuple(
168 FeatureSchema(name=name, parameters=feature.configuration())
169 for name, feature in zip(self.feature_names, self.features)
170 ),
171 )
172 return FeatureDataset(
173 feature_names=self.feature_names,
174 rows=tuple(rows),
175 metadata=metadata,
176 )
178 @staticmethod
179 def _source_fingerprint(bars: Sequence[Bar]) -> str:
180 payload = [
181 {
182 "symbol": bar.symbol,
183 "timestamp": bar.timestamp.isoformat(),
184 "open": bar.open,
185 "high": bar.high,
186 "low": bar.low,
187 "close": bar.close,
188 "volume": bar.volume,
189 }
190 for bar in bars
191 ]
192 serialized = json.dumps(payload, sort_keys=True, separators=(",", ":"))
193 return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
195 @staticmethod
196 def _validate_value(feature_name: str, value: FeatureValue) -> FeatureValue:
197 if value is None:
198 return None
200 if isinstance(value, bool) or not isinstance(value, (int, float)):
201 raise TypeError(f"Feature {feature_name!r} must return a number or None.")
203 normalized_value = float(value)
204 if not math.isfinite(normalized_value):
205 raise ValueError(f"Feature {feature_name!r} must return a finite value or None.")
207 return normalized_value