1from datetime import datetime 

2from typing import Union 

3import pandas as pd 

4import modin.pandas as mpd 

5import numpy as np 

6import numbers 

7 

8from cached_property import cached_property 

9 

10from .location import Location 

11from .files import cached_download_cpi 

12from .dates import convert_date 

13 

14 

15class CPI: 

16 """A class to manage the Australian Consumer Index (CPI) data.""" 

17 

18 @cached_property 

19 def latest_cpi_df(self) -> pd.DataFrame: 

20 """ 

21 Returns a Pandas DataFrame with the latest Australian Consumer Price Index (CPI) data. 

22 

23 Returns: 

24 pd.DataFrame: The latest Australian Consumer Price Index (CPI) data. The index of the series is the relevant date for each row. 

25 """ 

26 local_path = cached_download_cpi() 

27 excel_file = pd.ExcelFile(local_path) 

28 df = excel_file.parse("Data1") 

29 

30 # Get rid of extra headers 

31 df = df.iloc[9:] 

32 

33 df = df.rename(columns={"Unnamed: 0": "Date"}) 

34 df.index = df["Date"] 

35 

36 return df 

37 

38 def column_name(self, location: Union[Location, str] = Location.AUSTRALIA): 

39 return f"Index Numbers ; All groups CPI ; {str(location).title()} ;" 

40 

41 def cpi_series(self, location: Union[Location, str] = Location.AUSTRALIA) -> pd.Series: 

42 """ 

43 Returns a Pandas Series with the Australian CPI (Consumer Price Index) per quarter. 

44 

45 This is taken from the latest spreadsheet from the Australian Bureau of Statistics (file id '640101'). 

46 The index of the series is the relevant date. 

47 

48 Args: 

49 location (Union[Location, str], optional): The location for calculating the CPI. 

50 Options are 'Australia', 'Sydney', 'Melbourne', 'Brisbane', 'Adelaide', 'Perth', 'Hobart', 'Darwin', and 'Canberra'. 

51 Default is 'Australia'. 

52 

53 Returns: 

54 pd.Series: The CPIs per quarter. 

55 """ 

56 df = self.latest_cpi_df 

57 

58 return df[self.column_name(location)] 

59 

60 def cpi_at( 

61 self, date: Union[datetime, str, pd.Series, np.ndarray], location: Union[Location, str] = Location.AUSTRALIA 

62 ) -> Union[float, np.ndarray]: 

63 """ 

64 Returns the CPI (Consumer Price Index) for a date (or a number of dates). 

65 

66 If `date` is a string then it is converted to a datetime using dateutil.parser. 

67 If `date` is a vector then it returns a vector otherwise it returns a single scalar value. 

68 If `date` is before the earliest reference date (i.e. 1948-09-01) then it returns a NaN. 

69 

70 Args: 

71 date (Union[datetime, str, pd.Series, np.ndarray]): The date(s) to get the CPI(s) for. 

72 location (Union[Location, str], optional): The location for calculating the CPI. 

73 Options are 'Australia', 'Sydney', 'Melbourne', 'Brisbane', 'Adelaide', 'Perth', 'Hobart', 'Darwin', and 'Canberra'. 

74 Default is 'Australia'. 

75 

76 Returns: 

77 Union[float, np.ndarray]: The CPI value(s). 

78 """ 

79 date = convert_date(date) 

80 

81 cpi_series = self.cpi_series(location=location) 

82 

83 min_date = cpi_series.index.min() 

84 cpis = np.array( 

85 cpi_series[np.searchsorted(cpi_series.index, date, side="right") - 1], 

86 dtype=float, 

87 ) 

88 cpis[date < min_date] = np.nan 

89 

90 # TODO check if the date difference is greater than 3 months 

91 

92 if cpis.size == 1: 

93 return cpis.item() 

94 

95 return cpis 

96 

97 def calc_inflation( 

98 self, 

99 value: Union[numbers.Number, np.ndarray, pd.Series], 

100 original_date: Union[datetime, str], 

101 evaluation_date: Union[datetime, str, None] = None, 

102 location: Union[Location, str] = Location.AUSTRALIA, 

103 ): 

104 """ 

105 Adjusts a value (or list of values) for inflation. 

106 

107 Args: 

108 value (Union[numbers.Number, np.ndarray, pd.Series]): The value to be converted. 

109 original_date (Union[datetime, str]): The date that the value is in relation to. 

110 evaluation_date (Union[datetime, str], optional): The date to adjust the value to. Defaults to the current date. 

111 location (Union[Location, str], optional): The location for calculating the CPI. 

112 Options are 'Australia', 'Sydney', 'Melbourne', 'Brisbane', 'Adelaide', 'Perth', 'Hobart', 'Darwin', and 'Canberra'. 

113 Default is 'Australia'. 

114 

115 Returns: 

116 Union[float, np.ndarray]: The adjusted value. 

117 """ 

118 if evaluation_date is None: 

119 evaluation_date = datetime.now() 

120 

121 original_cpi = self.cpi_at(original_date, location=location) 

122 evaluation_cpi = self.cpi_at(evaluation_date, location=location) 

123 return value * evaluation_cpi / original_cpi 

124 

125 def calc_inflation_timeseries( 

126 self, 

127 compare_date: Union[datetime, str], 

128 start_date: Union[datetime, str, None] = None, 

129 end_date: Union[datetime, str, None] = None, 

130 value=1, 

131 location: Union[Location, str] = Location.AUSTRALIA, 

132 ) -> pd.Series: 

133 compare_date = convert_date(compare_date) 

134 if start_date != None: 

135 start_date = convert_date(start_date).item() 

136 if end_date != None: 

137 end_date = convert_date(end_date).item() 

138 

139 cpi_ts = self.cpi_series(location=location)[start_date:end_date].copy() 

140 evaluation_cpi = self.cpi_at(compare_date, location=location) 

141 inflation = value * (evaluation_cpi / cpi_ts) 

142 return inflation 

143 

144 

145_cpi = CPI() 

146 

147 

148def calc_inflation( 

149 value: Union[numbers.Number, np.ndarray, pd.Series], 

150 original_date: Union[datetime, str], 

151 evaluation_date: Union[datetime, str] = None, 

152 location: Union[Location, str] = Location.AUSTRALIA, 

153) -> Union[float, np.ndarray]: 

154 """ 

155 Adjusts a value (or list of values) for inflation. 

156 

157 Args: 

158 value (numbers.Number, np.ndarray, pd.Series): The value to be converted. 

159 original_date (datetime, str): The date that the value is in relation to. 

160 evaluation_date (datetime, str, optional): The date to adjust the value to. Defaults to the current date. 

161 location (Location, str, optional): The location for calculating the CPI. 

162 Options are 'Australia', 'Sydney', 'Melbourne', 'Brisbane', 'Adelaide', 'Perth', 'Hobart', 'Darwin', and 'Canberra'. 

163 Default is 'Australia'. 

164 

165 Returns: 

166 Union[float, np.ndarray]: The adjusted value. 

167 """ 

168 return _cpi.calc_inflation( 

169 value, 

170 original_date=original_date, 

171 evaluation_date=evaluation_date, 

172 location=location, 

173 ) 

174 

175 

176def latest_cpi_df() -> pd.DataFrame: 

177 """ 

178 Returns a pandas DataFrame with the latest CPI data from the Australian Bureau of Statistics. 

179 

180 This will contain the data for each quarter going back to 1948. 

181 

182 Returns: 

183 pandas.DataFrame: The latest dataframe with the CPI data. 

184 

185 """ 

186 return _cpi.latest_cpi_df