Coverage for rivapy / tools / interfaces.py: 65%

1import abc 

2import enum 

3from typing import List, Tuple 

4import datetime as dt 

5import numpy as np 

6import json 

7import hashlib 

8from rivapy.tools.datetime_grid import DateTimeGrid 

9from rivapy.tools.holidays_compat import ECB, UnitedStates, Germany 

10 

11 

12class DateTimeFunction(abc.ABC): 

13 @abc.abstractmethod 

14 def compute(self, ref_date: dt.datetime, dt_grid: DateTimeGrid) -> np.ndarray: 

15 pass 

16 

17 

18class _JSONDecoder(json.JSONDecoder): 

19 def __init__(self, *args, **kwargs): 

20 json.JSONDecoder.__init__(self, object_hook=self.object_hook, *args, **kwargs) 

21 

22 def object_hook(self, obj): 

23 ret = {} 

24 for key, value in obj.items(): 

25 if key in {"timestamp", "whatever"}: 

26 ret[key] = dt.fromisoformat(value) 

27 else: 

28 ret[key] = value 

29 return ret 

30 

31 

32class _JSONEncoder(json.JSONEncoder): 

33 def default(self, obj): 

34 if isinstance(obj, (dt.date, dt.datetime)): # , pd.Timestamp)): 

35 return obj.isoformat() 

36 if isinstance(obj, enum.Enum): 

37 return obj.value 

38 # return json.JSONEncoder.default(obj) 

39 return super().default(obj) 

40 

41 

42class FactoryObject(abc.ABC): 

43 

44 def to_dict(self): 

45 result = self._to_dict() 

46 result["cls"] = type(self).__name__ 

47 return result 

48 

49 def to_json(self): 

50 return json.dumps(self.to_dict(), cls=_JSONEncoder).encode() 

51 

52 @classmethod 

53 def from_json(cls, json_str: str): 

54 tmp = json.loads(json_str, cls=_JSONDecoder) 

55 return cls.from_dict(tmp) 

56 

57 @staticmethod 

58 def hash_for_dict(data: dict): 

59 return hashlib.sha1(json.dumps(data, cls=_JSONEncoder).encode()).hexdigest() 

60 

61 def hash(self): 

62 return FactoryObject.hash_for_dict(self.to_dict()) 

63 

64 @abc.abstractmethod 

65 def _to_dict(self) -> dict: 

66 pass 

67 

68 @classmethod 

69 def from_dict(cls, data: dict) -> object: 

70 from datetime import datetime, date 

71 

72 def parse_date(val): 

73 if isinstance(val, str): 

74 try: 

75 # Try parsing as datetime first 

76 return datetime.fromisoformat(val) 

77 except ValueError: 

78 return val 

79 return val 

80 

81 CALENDAR_REGISTRY = { 

82 "ECB": ECB, 

83 "UnitedStates": UnitedStates, 

84 "Germany": Germany, 

85 # Add more as needed 

86 } 

87 

88 def calendar_from_name(name): 

89 cls = CALENDAR_REGISTRY.get(name) 

90 if cls: 

91 return cls() 

92 raise ValueError(f"Unknown calendar: {name}") 

93 

94 # List of keys that may be date/datetime fields (customize as needed) 

95 date_keys = ["issue_date", "trade_date", "maturity_date", "start_date", "end_date", "rate_start_date", "rate_end_date", "fixing_date"] 

96 parsed_data = {} 

97 for k, v in data.items(): 

98 if k != "cls" and k in date_keys: 

99 parsed_data[k] = parse_date(v) 

100 elif k == "calendar": 

101 parsed_data[k] = calendar_from_name(v) 

102 elif k != "cls": 

103 parsed_data[k] = v 

104 return cls(**parsed_data) 

105 

106 

107class BaseDatedCurve(abc.ABC): 

108 @abc.abstractmethod 

109 def value(self, ref_date: dt.datetime, d: dt.datetime) -> np.ndarray: # , dt_grid: DateTimeGrid)->np.ndarray: 

110 pass