Coverage for fio_wrapper/endpoints/endpoints_v1/exchange.py: 100%

1"""Access exchange information from FIO. 

2""" 

3from typing import Optional 

4from fio_wrapper.endpoints.abstracts.abstract_endpoint import AbstractEndpoint 

5from fio_wrapper.endpoints.abstracts.abstract_exchange import AbstractExchange 

6from fio_wrapper.exceptions import ExchangeTickerInvalid 

7from fio_wrapper.exceptions import ExchangeTickerNotFound 

8from fio_wrapper.exceptions import MaterialTickerInvalid 

9from fio_wrapper.models.exchange_models import ExchangeTickerFull 

10from fio_wrapper.models.exchange_models import ExchangeTickerFullList 

11from fio_wrapper.models.exchange_models import ExchangeTickerList 

12from fio_wrapper.models.exchange_models import OrderList 

13from fio_wrapper.validators import validate_company_code 

14from fio_wrapper.validators import validate_exchange_code 

15from fio_wrapper.validators import validate_ticker 

16 

17 

18class Exchange(AbstractExchange, AbstractEndpoint): 

19 def _validate_exchangeticker(self, exchange_ticker: str) -> None: 

20 """Validates an exchange ticker 

21 

22 Args: 

23 exchange_ticker (str): Exchange ticker 

24 

25 Raises: 

26 ExchangeTickerInvalid: Exchange ticker can't be None type 

27 ExchangeTickerInvalid: Exchange ticker too short or long. Must be 5 to 7 characters 

28 ExchangeTickerInvalid: Exchange ticker must be of form MaterialTicker.ExchangeCode 

29 ExchangeTickerInvalid: Material ticker part of Exchange ticker invalid 

30 """ 

31 if exchange_ticker is None: 

32 raise ExchangeTickerInvalid("Exchange ticker can't be None type") 

33 

34 # min length: 5, max length: 7 

35 if len(exchange_ticker) < 5 or len(exchange_ticker) > 7: 

36 raise ExchangeTickerInvalid( 

37 "Exchange ticker to short or long. Must be 5 to 7 characters" 

38 ) 

39 

40 # must contain . 

41 if "." not in exchange_ticker: 

42 raise ExchangeTickerInvalid( 

43 "Exchange ticker must be of form MaterialTicker.ExchangeCode" 

44 ) 

45 

46 splitted = exchange_ticker.split(".") 

47 

48 # validate Material part 

49 try: 

50 validate_ticker(material_ticker=splitted[0]) 

51 except MaterialTickerInvalid as e: 

52 raise ExchangeTickerInvalid() from e 

53 

54 # validate Exchange code 

55 validate_exchange_code(exchange_code=splitted[1]) 

56 

57 # /exchange/{ExchangeTicker} 

58 def get( 

59 self, exchange_ticker: str, timeout: Optional[float] = None 

60 ) -> ExchangeTickerFull: 

61 """Gets a single exchange ticker from FIO 

62 

63 Args: 

64 exchange_ticker (str): Exchange Ticker (e.g., "DW.AI1") 

65 timeout (float, optional): Request timeout in seconds. Defaults to None. 

66 

67 Raises: 

68 ExchangeTickerNotFound: Exchange ticker was not found 

69 

70 Returns: 

71 ExchangeTicker: Exchange ticker 

72 """ 

73 self._validate_exchangeticker(exchange_ticker=exchange_ticker) 

74 

75 (status, data) = self.adapter.get( 

76 endpoint=self.urls.exchange_get_url(exchange_ticker=exchange_ticker), 

77 err_codes=[204], 

78 timeout=timeout, 

79 ) 

80 

81 if status == 200: 

82 return ExchangeTickerFull.model_validate(data) 

83 elif status == 204: 

84 raise ExchangeTickerNotFound("Exchangeticker not found") 

85 

86 # /exchange/all 

87 def all(self, timeout: Optional[float] = None) -> ExchangeTickerList: 

88 """Gets all simple exchange ticker from FIO 

89 

90 Args: 

91 timeout (float, optional): Request timeout in seconds. Defaults to None. 

92 

93 Returns: 

94 ExchangeTickerList: Exchange ticker 

95 """ 

96 (_, data) = self.adapter.get( 

97 endpoint=self.urls.exchange_get_all_url(), timeout=timeout 

98 ) 

99 

100 return ExchangeTickerList.model_validate(data) 

101 

102 # /exchange/full 

103 def full(self, timeout: Optional[float] = None) -> ExchangeTickerFullList: 

104 """Gets a complete list of all exchange information from FIO 

105 

106 Args: 

107 timeout (float, optional): Request timeout in seconds. Defaults to None. 

108 

109 Returns: 

110 ExchangeTickerFullList: Exchange ticker full 

111 """ 

112 (_, data) = self.adapter.get( 

113 endpoint=self.urls.exchange_get_full_url(), timeout=timeout 

114 ) 

115 

116 return ExchangeTickerFullList.model_validate(data) 

117 

118 # /exchange/orders/{CompanyCode} 

119 def get_orders( 

120 self, company_code: str, timeout: Optional[float] = None 

121 ) -> OrderList: 

122 """Gets a companies order data from FIO 

123 

124 Args: 

125 company_code (str): Company code (1-4 characters) 

126 timeout (float, optional): Request timeout in seconds. Defaults to None. 

127 

128 Returns: 

129 OrderList: Orders 

130 """ 

131 # 1 to 4 character company code 

132 validate_company_code(company_code=company_code) 

133 

134 (_, data) = self.adapter.get( 

135 endpoint=self.urls.exchange_get_orders_companycode( 

136 company_code=company_code 

137 ), 

138 timeout=timeout, 

139 ) 

140 

141 return OrderList.model_validate(data) 

142 

143 # /exchange/orders/{CompanyCode}/{ExchangeCode} 

144 def get_orders_exchange( 

145 self, company_code: str, exchange_code: str, timeout: Optional[float] = None 

146 ) -> OrderList: 

147 """Gets a companies order data for a specific exchange from FIO 

148 

149 Args: 

150 company_code (str): Company code (1-4 characters) 

151 exchange_code (str): Exchange code (e.g., "AI1") 

152 timeout (float, optional): Request timeout in seconds. Defaults to None. 

153 

154 Returns: 

155 OrderList: Orders 

156 """ 

157 validate_company_code(company_code=company_code) 

158 validate_exchange_code(exchange_code=exchange_code) 

159 

160 (_, data) = self.adapter.get( 

161 endpoint=self.urls.exchange_get_orders_companycode_exchange( 

162 company_code=company_code, exchange_code=exchange_code 

163 ), 

164 timeout=timeout, 

165 ) 

166 

167 return OrderList.model_validate(data)