Coverage for functions \ flipdare \ analysis \ data \ nested \ _time_series_nested_data.py: 92%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13from abc import ABC, abstractmethod 

14from typing import override 

15from datetime import datetime, timedelta 

16from collections import defaultdict 

17from dataclasses import dataclass, field 

18from flipdare.analysis.data._time_series_protocol import TimeSeriesProtocol, TimeSeriesPlotInfo 

19from flipdare.app_types import ReportListType 

20 

21__all__ = ["TimeSeriesNestedData"] 

22 

23_DAY = timedelta(days=1) 

24 

25 

26@dataclass 

27class TimeSeriesNestedData[K, V](ABC, TimeSeriesProtocol): 

28 """ 

29 Base for time series with (datetime, K) -> V storage. 

30 

31 All datetime keys are normalised to midnight UTC via `_to_day`. 

32 `dates` fills every 1-day interval between the earliest and latest key. 

33 Subclasses must implement `headers`, `table_data`, `plot_info`, and `add`. 

34 """ 

35 

36 _rows: defaultdict[datetime, dict[K, V]] = field( 

37 default_factory=lambda: defaultdict(dict), 

38 init=False, 

39 repr=False, 

40 ) 

41 

42 _error_ct: int = field(default=0, init=False, repr=False) 

43 

44 # NOTE: these must be overridden 

45 @property 

46 @abstractmethod 

47 @override 

48 def headers(self) -> list[str]: ... 

49 

50 @abstractmethod 

51 @override 

52 def table_data(self) -> ReportListType: ... 

53 

54 @abstractmethod 

55 @override 

56 def plot_info(self) -> list[TimeSeriesPlotInfo]: ... 

57 

58 # helpers 

59 @property 

60 def error_count(self) -> int: 

61 return self._error_ct 

62 

63 def increment_error(self) -> None: 

64 self._error_ct += 1 

65 

66 @staticmethod 

67 def _to_day(dt: datetime) -> datetime: 

68 """Normalise a datetime to midnight UTC for use as a row key.""" 

69 return dt.replace(hour=0, minute=0, second=0, microsecond=0) 

70 

71 @property 

72 @override 

73 def has_data(self) -> bool: 

74 return any(bool(row) for row in self._rows.values()) 

75 

76 @property 

77 @override 

78 def count(self) -> int: 

79 return sum(len(row) for row in self._rows.values()) 

80 

81 @property 

82 @override 

83 def dates(self) -> list[datetime]: 

84 """All daily dates from earliest to latest, padding days with no data.""" 

85 keys = sorted(self._rows.keys()) 

86 if not keys: 

87 return [] 

88 result: list[datetime] = [] 

89 current = keys[0] 

90 end = keys[-1] 

91 while current <= end: 

92 result.append(current) 

93 current += _DAY 

94 return result 

95 

96 @property 

97 @override 

98 def debug_str(self) -> str: 

99 lines: list[str] = [", ".join(self.headers)] 

100 lines.extend(", ".join(str(v) for v in row) for row in self.table_data()) 

101 return "\n".join(lines)