Coverage for rivapy / marketdata / fixing_table.py: 54%

1# 2025.07.17 Hans Nguyen 

2# Fixing table class for Interest Rate Swap Pricer for IR bootstrapping implementation 

3 

4 

5# Modules 

6from typing import List as _List, Union as _Union, Tuple, Dict, Any 

7from datetime import datetime, date 

8import bisect 

9 

10 

11# Class 

12class FixingTable: 

13 """Container for historical fixings.""" 

14 

15 def __init__(self, id: str = None, fixings: Dict[str, Tuple[_List[datetime], _List[float]]] = None): 

16 # id: Optional[str] = None, 

17 # fixings: Optional[Dict[str, Tuple[List[datetime], List[float]]]] = None): 

18 """Constructor for FixingTable. Creates 'empty' instance if no parameters passed. 

19 

20 Args: 

21 id (str): identifier for the table. Defaults to None. 

22 fixings (Dict[str, Tuple[List[datetime], List[float]]]): A dictionary mapping underlying IDs to 

23 pairs of fixing dates and values. Defaults to None. 

24 """ 

25 

26 self.id = id 

27 self.fixings: Dict[str, Tuple[_List[datetime], _List[float]]] = fixings if fixings else {} 

28 

29 if fixings: 

30 expected_len = None 

31 for udl, (dates, values) in fixings.items(): 

32 if expected_len is None: 

33 expected_len = len(dates) 

34 if len(dates) != expected_len: 

35 raise ValueError(f"Inconsistent number of dates for {udl}: {len(dates)} != {expected_len}") 

36 if len(values) != expected_len: 

37 raise ValueError(f"Inconsistent number of fixings for {udl}: {len(values)} != {expected_len}") 

38 

39 def get_object_type(self) -> str: 

40 return "FIXING_TABLE" 

41 

42 def get_fixing(self, udl_id: str, fixing_date: datetime) -> float: 

43 """ 

44 Return fixing for a given underlying and date. Raises error if not found. 

45 

46 

47 Args: 

48 udl_id (str): The underlying ID. 

49 fixing_date (datetime): The date of the fixing 

50 

51 Raises: 

52 ValueError: no fixings for underlying 

53 ValueError: no fixing for given date 

54 

55 Returns: 

56 float: The fixing value if found, otherwise raise error 

57 """ 

58 

59 if udl_id not in self.fixings: 

60 raise ValueError(f"No fixings for underlying '{udl_id}'") 

61 

62 dates, values = self.fixings[udl_id] 

63 for i, date in enumerate(dates): 

64 if date == fixing_date: 

65 return values[i] 

66 

67 raise ValueError(f"No fixing found for '{udl_id}' on {fixing_date.isoformat()}") 

68 

69 def get_num_underlyings(self) -> int: 

70 """Return number of underlyings 

71 

72 Returns: 

73 int: _description_ 

74 """ 

75 return len(self.fixings) 

76 

77 def add(self, key: str, fixing_date: datetime, value: float) -> None: 

78 """ 

79 Add a fixing and keep entries sorted by date. 

80 

81 Args: 

82 key (str): The underlying ID key to map to fixing. 

83 fixing_date (datetime): The date of the fixing. 

84 value (float): The fixing value. 

85 """ 

86 

87 if key not in self.fixings: 

88 self.fixings[key] = ([], []) 

89 

90 dates, values = self.fixings[key] 

91 

92 # Insert while keeping both lists sorted to preserve date order 

93 index = bisect.bisect_left(dates, fixing_date) 

94 dates.insert(index, fixing_date) 

95 values.insert(index, value) 

96 

97 def get(self, udl_id: str) -> Tuple[_List[datetime], _List[float]]: 

98 """ 

99 Get Tuple of all fixings for a given underlying. 

100 

101 Args: 

102 udl_id (str): key of underlying ID 

103 

104 Returns: 

105 Tuple[List[datetime], List[float]]: Tuple of dates and fixings. 

106 """ 

107 if udl_id not in self.fixings: 

108 return [], [] # TODO or riase an error? 

109 dates, values = self.fixings[udl_id] 

110 

111 return dates[:], values[:] 

112 

113 def _to_dict(self): 

114 # Serializes the fixings dictionary: keys are underlying IDs, 

115 # values are tuples of (list of dates, list of floats) 

116 result = {} 

117 for udl_id, (dates, values) in self.fixings.items(): 

118 # Convert all dates to ISO strings 

119 dates_serialized = [d.isoformat() if hasattr(d, "isoformat") else str(d) for d in dates] 

120 result[udl_id] = {"dates": dates_serialized, "values": values} 

121 return {"id": self.id, "fixings": result} 

122 

123 

124# Functions 

125 

126 

127if __name__ == "__main__": 

128 pass