Coverage for src/finbot/market/dataset.py: 98%

1import hashlib 

2import json 

3from dataclasses import dataclass 

4from datetime import datetime 

5 

6DATASET_VERSION = 1 

7 

8 

9@dataclass(frozen=True) 

10class DatasetMetadata: 

11 source: str 

12 symbols: tuple[str, ...] 

13 timeframe: str 

14 start: datetime 

15 end: datetime 

16 adjustment: str 

17 source_options: tuple[tuple[str, str], ...] = () 

18 version: int = DATASET_VERSION 

19 

20 def __post_init__(self) -> None: 

21 source = self.source.strip() 

22 if not source: 

23 raise ValueError("Dataset source must be a non-empty string.") 

24 

25 symbols = tuple( 

26 sorted({symbol.strip().upper() for symbol in self.symbols if symbol.strip()}), 

27 ) 

28 if not symbols: 

29 raise ValueError("Dataset symbols must contain at least one symbol.") 

30 

31 timeframe = self.timeframe.strip() 

32 if not timeframe: 

33 raise ValueError("Dataset timeframe must be a non-empty string.") 

34 

35 adjustment = self.adjustment.strip() 

36 if not adjustment: 

37 raise ValueError("Dataset adjustment must be a non-empty string.") 

38 

39 if self.start.tzinfo is None or self.end.tzinfo is None: 

40 raise ValueError("Dataset start and end must be timezone-aware.") 

41 

42 if self.end <= self.start: 

43 raise ValueError("Dataset end must be after start.") 

44 

45 if self.version < 1: 

46 raise ValueError("Dataset version must be at least 1.") 

47 

48 source_options = tuple( 

49 sorted((key.strip(), value.strip()) for key, value in self.source_options), 

50 ) 

51 if any(not key or not value for key, value in source_options): 

52 raise ValueError("Dataset source options must contain non-empty keys and values.") 

53 

54 if len({key for key, _ in source_options}) != len(source_options): 

55 raise ValueError("Dataset source option keys must be unique.") 

56 

57 object.__setattr__(self, "source", source) 

58 object.__setattr__(self, "symbols", symbols) 

59 object.__setattr__(self, "timeframe", timeframe) 

60 object.__setattr__(self, "adjustment", adjustment) 

61 object.__setattr__(self, "source_options", source_options) 

62 

63 @property 

64 def dataset_id(self) -> str: 

65 payload = json.dumps(self.to_dict(), sort_keys=True, separators=(",", ":")) 

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

67 

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

69 return { 

70 "source": self.source, 

71 "symbols": list(self.symbols), 

72 "timeframe": self.timeframe, 

73 "start": self.start.isoformat(), 

74 "end": self.end.isoformat(), 

75 "adjustment": self.adjustment, 

76 "source_options": dict(self.source_options), 

77 "version": self.version, 

78 } 

79 

80 @classmethod 

81 def from_dict(cls, data: dict[str, object]) -> "DatasetMetadata": 

82 try: 

83 source = data["source"] 

84 symbols = data["symbols"] 

85 timeframe = data["timeframe"] 

86 start = data["start"] 

87 end = data["end"] 

88 adjustment = data["adjustment"] 

89 source_options = data["source_options"] 

90 version = data["version"] 

91 except KeyError as exc: 

92 raise ValueError(f"Dataset metadata is missing required field {exc.args[0]!r}") from exc 

93 

94 if not isinstance(source, str): 

95 raise ValueError("Dataset metadata field 'source' must be a string.") 

96 

97 if not isinstance(symbols, list) or not all(isinstance(symbol, str) for symbol in symbols): 

98 raise ValueError("Dataset metadata field 'symbols' must be a list of strings.") 

99 

100 if not isinstance(timeframe, str): 

101 raise ValueError("Dataset metadata field 'timeframe' must be a string.") 

102 

103 if not isinstance(start, str) or not isinstance(end, str): 

104 raise ValueError( 

105 "Dataset metadata fields 'start' and 'end' must be ISO datetime strings.", 

106 ) 

107 

108 if not isinstance(adjustment, str): 

109 raise ValueError("Dataset metadata field 'adjustment' must be a string.") 

110 

111 if not isinstance(source_options, dict) or not all( 111 ↛ 115line 111 didn't jump to line 115 because the condition on line 111 was never true

112 ((isinstance(key, str) and isinstance(value, str)) 

113 for key, value in source_options.items()), 

114 ): 

115 raise ValueError( 

116 "Dataset metadata field 'source_options' " 

117 "must be an object containing string values.", 

118 ) 

119 

120 if not isinstance(version, int) or isinstance(version, bool): 

121 raise ValueError("Dataset metadata field 'version' must be an integer.") 

122 

123 try: 

124 parsed_start = datetime.fromisoformat(start) 

125 parsed_end = datetime.fromisoformat(end) 

126 except ValueError as exc: 

127 raise ValueError("Dataset metadata contains an invalid ISO datetime.") from exc 

128 

129 return cls( 

130 source=source, 

131 symbols=tuple(symbols), 

132 timeframe=timeframe, 

133 start=parsed_start, 

134 end=parsed_end, 

135 adjustment=adjustment, 

136 source_options=tuple(source_options.items()), 

137 version=version, 

138 )