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

1 

2import abc 

3from typing import List, Tuple 

4import datetime as dt 

5import numpy as np 

6import json 

7import hashlib 

8from rivapy.tools.datetime_grid import DateTimeGrid 

9 

10class DateTimeFunction(abc.ABC): 

11 @abc.abstractmethod 

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

13 pass 

14 

15class _JSONDecoder(json.JSONDecoder): 

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

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

18 

19 def object_hook(self, obj): 

20 ret = {} 

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

22 if key in {'timestamp', 'whatever'}: 

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

24 else: 

25 ret[key] = value 

26 return ret 

27 

28class _JSONEncoder(json.JSONEncoder): 

29 def default(self, obj): 

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

31 return obj.isoformat() 

32 return json.JSONEncoder.default(obj) 

33 

34class FactoryObject(abc.ABC): 

35 

36 def to_dict(self): 

37 result = self._to_dict() 

38 result['cls'] = type(self).__name__ 

39 return result 

40 

41 def to_json(self): 

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

43 

44 @classmethod 

45 def from_json(cls, json_str: str): 

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

47 return cls.from_dict(tmp) 

48 

49 @staticmethod 

50 def hash_for_dict(data: dict): 

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

52 

53 

54 def hash(self): 

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

56 

57 @abc.abstractmethod 

58 def _to_dict(self)->dict: 

59 pass 

60 

61 @classmethod 

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

63 return cls(**{k:v for k,v in data.items() if k != 'cls'}) 

64 

65class BaseDatedCurve(abc.ABC): 

66 @abc.abstractmethod 

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

68 pass 

69 

70 

71class HasExpectedCashflows(abc.ABC): 

72 @abc.abstractmethod 

73 def expected_cashflows(self)->List[Tuple[dt.datetime, float]]: 

74 pass