Coverage for tests/test_pfc.py: 99%

1import unittest 

2import holidays 

3import itertools 

4import pandas as pd 

5import numpy as np 

6import datetime as dt 

7from rivapy.tools.scheduler import SimpleSchedule, OffPeakSchedule, PeakSchedule, GasSchedule, BaseSchedule 

8from rivapy.marketdata_tools import PFCShifter 

9from rivapy.marketdata_tools.pfc_shaper import PFCShaper, CategoricalRegression 

10from rivapy.instruments.energy_futures_specifications import EnergyFutureSpecifications 

11from rivapy.sample_data.dummy_power_spot_price import spot_price_model 

12from rivapy.marketdata.curves import EnergyPriceForwardCurve 

13from typing import Dict 

14from collections import defaultdict 

15 

16 

17class TestPFCShifter(unittest.TestCase): 

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

19 super(TestPFCShifter, self).__init__(*args, **kwargs) 

20 self.parameter_dict = { 

21 "spot_price_level": 100, 

22 "peak_price_level": 10, 

23 "solar_price_level": 8, 

24 "weekend_price_level": 10, 

25 "winter_price_level": 20, 

26 "epsilon_mean": 0, 

27 "epsilon_var": 5, 

28 } 

29 self.date_range = pd.date_range(start="1/1/2023", end="1/1/2025", freq="h", inclusive="left") 

30 self.spot_prices = list(map(lambda x: spot_price_model(x, **self.parameter_dict), self.date_range)) 

31 self.df = pd.DataFrame(data=self.spot_prices, index=self.date_range, columns=["Spot"]) 

32 base_y = self.df.resample("YE").mean() 

33 base_y.index = base_y.index.strftime("%Y") 

34 

35 df_spot = self.df.copy() 

36 df_spot.index = df_spot.index.strftime("%Y") 

37 

38 shape = df_spot.divide(base_y, axis="index") 

39 self.shape_df = pd.DataFrame(data=shape["Spot"].tolist(), index=self.date_range, columns=["shape"]) 

40 

41 def __get_contracts_dict(self, contracts_schedules: Dict[str, Dict[str, SimpleSchedule]]) -> Dict[str, Dict[str, EnergyFutureSpecifications]]: 

42 contracts = defaultdict(dict) 

43 for contract_type, contracts_dict in contracts_schedules.items(): 

44 for contract_name, schedule in contracts_dict.items(): 

45 tg = schedule.get_schedule() 

46 price = self.df.loc[tg, :].mean().iloc[0] 

47 contracts[contract_type][contract_name] = EnergyFutureSpecifications(schedule=schedule, price=price, name=contract_name) 

48 return dict(contracts) 

49 

50 def test_pfc_shifter(self): 

51 contracts_schedules = { 

52 "off_peak": { 

53 "Cal23_OffPeak": OffPeakSchedule(dt.datetime(2023, 1, 1), dt.datetime(2024, 1, 1)), 

54 "Q2/23_OffPeak": OffPeakSchedule(dt.datetime(2023, 4, 1), dt.datetime(2023, 7, 1)), 

55 }, 

56 "peak": { 

57 "Cal23_Peak": PeakSchedule(dt.datetime(2023, 1, 1), dt.datetime(2024, 1, 1)), 

58 "Cal24_Peak": PeakSchedule(dt.datetime(2024, 1, 1), dt.datetime(2025, 1, 1)), 

59 "Q1/23_Peak": PeakSchedule(dt.datetime(2023, 1, 1), dt.datetime(2023, 4, 1)), 

60 "Q2/23_Peak": PeakSchedule(dt.datetime(2023, 4, 1), dt.datetime(2023, 7, 1)), 

61 "Q3/23_Peak": PeakSchedule(dt.datetime(2023, 7, 1), dt.datetime(2023, 10, 1)), 

62 "Q4/23_Peak": PeakSchedule(dt.datetime(2023, 10, 1), dt.datetime(2024, 1, 1)), 

63 "Q2/24_Peak": PeakSchedule(dt.datetime(2024, 4, 1), dt.datetime(2024, 7, 1)), 

64 }, 

65 } 

66 contracts = self.__get_contracts_dict(contracts_schedules=contracts_schedules) 

67 data_dict = { 

68 "off_peak": self.shape_df.loc[ 

69 SimpleSchedule( 

70 dt.datetime(2023, 1, 1), 

71 dt.datetime(2024, 1, 1), 

72 freq="h", 

73 hours=[0, 1, 2, 3, 4, 5, 6, 7, 20, 21, 22, 23], 

74 ignore_hours_for_weekdays=[5, 6], 

75 ).get_schedule() 

76 ], 

77 "peak": self.shape_df.loc[ 

78 SimpleSchedule( 

79 dt.datetime(2023, 1, 1), 

80 dt.datetime(2025, 1, 1), 

81 freq="h", 

82 hours=[8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19], 

83 weekdays=[0, 1, 2, 3, 4], 

84 ).get_schedule() 

85 ], 

86 } 

87 result = [] 

88 for contracts_type, data in data_dict.items(): 

89 pfc_shifter = PFCShifter(shape=data, contracts=list(contracts[contracts_type].values())) 

90 result.append(pfc_shifter.compute()) 

91 

92 df_shifted = pd.concat(result, axis=0) 

93 df_shifted.sort_index(inplace=True, ascending=True) 

94 

95 contracts_all = self.__get_contracts_dict(contracts_schedules=contracts_schedules) 

96 for contracts_type, contracts_dict in contracts_all.items(): 

97 for contract_name, contract_spec in contracts_dict.items(): 

98 shift_price = df_shifted.loc[contract_spec.get_schedule(), :].mean().iloc[0] 

99 self.assertAlmostEqual(contract_spec.price, shift_price, delta=10 ** (-10)) 

100 

101 def test_only_overlapping_contracts(self): 

102 contracts_schedules = { 

103 "GasBase": { 

104 "Cal23_GasBase": GasSchedule(dt.datetime(2023, 1, 1), dt.datetime(2024, 1, 1)), 

105 "Q1/23_GasBase": GasSchedule(dt.datetime(2023, 1, 1), dt.datetime(2023, 4, 1)), 

106 "Q2/23_GasBase": GasSchedule(dt.datetime(2023, 4, 1), dt.datetime(2023, 7, 1)), 

107 "Q3/23_GasBase": GasSchedule(dt.datetime(2023, 7, 1), dt.datetime(2023, 10, 1)), 

108 "Q4/23_GasBase": GasSchedule(dt.datetime(2023, 10, 1), dt.datetime(2024, 1, 1)), 

109 }, 

110 } 

111 contracts = self.__get_contracts_dict(contracts_schedules=contracts_schedules) 

112 data_dict = { 

113 "GasBase": self.shape_df.loc[GasSchedule(dt.datetime(2023, 1, 1), dt.datetime(2024, 1, 1)).get_schedule()], 

114 } 

115 result = [] 

116 for contracts_type, data in data_dict.items(): 

117 pfc_shifter = PFCShifter(shape=data, contracts=list(contracts[contracts_type].values())) 

118 result.append(pfc_shifter.compute()) 

119 

120 df_shifted = pd.concat(result, axis=0) 

121 df_shifted.sort_index(inplace=True, ascending=True) 

122 

123 contracts_all = self.__get_contracts_dict(contracts_schedules=contracts_schedules) 

124 for contracts_type, contracts_dict in contracts_all.items(): 

125 for contract_name, contract_spec in contracts_dict.items(): 

126 shift_price = df_shifted.loc[contract_spec.get_schedule(), :].mean().iloc[0] 

127 self.assertAlmostEqual(contract_spec.price, shift_price, delta=10 ** (-10)) 

128 

129 def test_non_coverage_of_shape(self): 

130 contracts_schedules = { 

131 "Peak": { 

132 "Cal23_GasBase": PeakSchedule(dt.datetime(2023, 1, 1), dt.datetime(2024, 1, 1)), 

133 "Q1/23_GasBase": PeakSchedule(dt.datetime(2023, 1, 1), dt.datetime(2023, 4, 1)), 

134 "Q2/23_GasBase": PeakSchedule(dt.datetime(2023, 4, 1), dt.datetime(2023, 7, 1)), 

135 "Q3/23_GasBase": PeakSchedule(dt.datetime(2023, 7, 1), dt.datetime(2023, 10, 1)), 

136 "Q4/23_GasBase": PeakSchedule(dt.datetime(2023, 10, 1), dt.datetime(2024, 1, 1)), 

137 }, 

138 } 

139 contracts = self.__get_contracts_dict(contracts_schedules=contracts_schedules) 

140 data_dict = { 

141 "Peak": self.shape_df.loc[SimpleSchedule(dt.datetime(2023, 1, 1), dt.datetime(2024, 1, 1)).get_schedule()], 

142 } 

143 result = [] 

144 with self.assertRaises(ValueError): 

145 for contracts_type, data in data_dict.items(): 

146 pfc_shifter = PFCShifter(shape=data, contracts=list(contracts[contracts_type].values())) 

147 result.append(pfc_shifter()) 

148 

149 

150class TestPFCShaper(unittest.TestCase): 

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

152 super(TestPFCShaper, self).__init__(*args, **kwargs) 

153 

154 self.example_spot_price_data = pd.read_excel( 

155 "./tests/data/hpfc_test.xlsx", parse_dates=["Date"], date_format="%d.%m.%Y %H:%M", index_col="Date" 

156 ) 

157 

158 def test_categoricalregression(self): 

159 parameter_dict = { 

160 "spot_price_level": 100, 

161 "peak_price_level": 10, 

162 "solar_price_level": 8, 

163 "weekend_price_level": 10, 

164 "winter_price_level": 20, 

165 "epsilon_mean": 0, 

166 "epsilon_var": 5, 

167 } 

168 date_range = pd.date_range(start="1/1/2024", end="1/1/2025", freq="h", inclusive="left") 

169 spot_prices = list(map(lambda x: spot_price_model(x, **parameter_dict), date_range)) 

170 spot_prices = pd.DataFrame(data=spot_prices, index=date_range, columns=["Spot"]) 

171 

172 holiday_calendar = holidays.country_holidays("DE", years=[2024]) 

173 

174 apply_schedule = SimpleSchedule(start=dt.datetime(2025, 1, 1), end=dt.datetime(2026, 1, 1), freq="h") 

175 

176 pfc_shaper = CategoricalRegression(spot_prices=spot_prices, holiday_calendar=holiday_calendar) 

177 pfc_fit = pfc_shaper.calibrate() 

178 

179 self.assertLess(np.sum((spot_prices.values - pfc_fit)) ** 2, 10 ** (-5)) 

180 

181 pfc_rollout = pfc_shaper.apply(apply_schedule=apply_schedule) 

182 spot_prices_rollout = np.array(list(map(lambda x: spot_price_model(x, **parameter_dict), apply_schedule.get_schedule()))) 

183 

184 self.assertLess(np.sum((spot_prices_rollout / np.mean(spot_prices_rollout) - pfc_rollout.values)) ** 2, 10 ** (-5)) 

185 

186 def test_normalization_with_config(self): 

187 normalization_config = {"D": 2, "W": 2, "ME": 1} 

188 holiday_calendar = holidays.country_holidays("DE", years=[2024, 2025, 2026]) 

189 

190 pfc_shaper = CategoricalRegression( 

191 spot_prices=self.example_spot_price_data, holiday_calendar=holiday_calendar, normalization_config=normalization_config 

192 ) 

193 

194 apply_schedule = BaseSchedule(start=dt.datetime(2025, 12, 31), end=dt.datetime(2027, 1, 1)) 

195 

196 pfc_spot = pfc_shaper.calibrate() 

197 pfc_shape = pfc_shaper.apply(apply_schedule=apply_schedule) 

198 

199 self.assertLess(np.abs(pfc_shape.loc[pfc_shape.index < dt.datetime(2026, 1, 1), :].mean().iloc[0] - 1.0), 10 ** (-5)) 

200 

201 self.assertLess( 

202 np.abs(pfc_shape.loc[(dt.datetime(2026, 1, 1) <= pfc_shape.index) & (pfc_shape.index < dt.datetime(2026, 1, 2)), :].mean().iloc[0] - 1.0), 

203 10 ** (-5), 

204 ) 

205 

206 self.assertLess( 

207 np.abs(pfc_shape.loc[(dt.datetime(2026, 1, 2) <= pfc_shape.index) & (pfc_shape.index < dt.datetime(2026, 1, 5)), :].mean().iloc[0] - 1.0), 

208 10 ** (-5), 

209 ) 

210 

211 self.assertLess( 

212 np.abs( 

213 pfc_shape.loc[(dt.datetime(2026, 1, 5) <= pfc_shape.index) & (pfc_shape.index < dt.datetime(2026, 1, 12)), :].mean().iloc[0] - 1.0 

214 ), 

215 10 ** (-5), 

216 ) 

217 

218 self.assertLess( 

219 np.abs( 

220 pfc_shape.loc[(dt.datetime(2026, 1, 12) <= pfc_shape.index) & (pfc_shape.index < dt.datetime(2026, 2, 1)), :].mean().iloc[0] - 1.0 

221 ), 

222 10 ** (-5), 

223 ) 

224 

225 self.assertLess( 

226 np.abs(pfc_shape.loc[(dt.datetime(2026, 2, 1) <= pfc_shape.index) & (pfc_shape.index < dt.datetime(2027, 1, 1)), :].mean().iloc[0] - 1.0), 

227 10 ** (-5), 

228 ) 

229 

230 def test_normalization_without_config(self): 

231 

232 holiday_calendar = holidays.country_holidays("DE", years=[2024, 2025, 2026]) 

233 

234 pfc_shaper = CategoricalRegression(spot_prices=self.example_spot_price_data, holiday_calendar=holiday_calendar) 

235 

236 apply_schedule = BaseSchedule(start=dt.datetime(2025, 12, 31), end=dt.datetime(2027, 1, 1)) 

237 

238 pfc_spot = pfc_shaper.calibrate() 

239 pfc_shape = pfc_shaper.apply(apply_schedule=apply_schedule) 

240 

241 self.assertLess(np.abs(pfc_shape.loc[pfc_shape.index < dt.datetime(2026, 1, 1), :].mean().iloc[0] - 1.0), 10 ** (-5)) 

242 

243 self.assertLess( 

244 np.abs(pfc_shape.loc[(dt.datetime(2026, 1, 1) <= pfc_shape.index) & (pfc_shape.index < dt.datetime(2027, 1, 1)), :].mean().iloc[0] - 1.0), 

245 10 ** (-5), 

246 ) 

247 

248 

249class TestEnergyPriceForwardCurve(unittest.TestCase): 

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

251 super(TestEnergyPriceForwardCurve, self).__init__(*args, **kwargs) 

252 

253 self.example_spot_price_data = pd.read_excel( 

254 "./tests/data/hpfc_test.xlsx", parse_dates=["Date"], date_format="%d.%m.%Y %H:%M", index_col="Date" 

255 ) 

256 self.parameter_dict = { 

257 "spot_price_level": 100, 

258 "peak_price_level": 10, 

259 "solar_price_level": 8, 

260 "weekend_price_level": 10, 

261 "winter_price_level": 20, 

262 "epsilon_mean": 0, 

263 "epsilon_var": 5, 

264 } 

265 self.date_range = pd.date_range(start="1/1/2023", end="1/1/2025", freq="h", inclusive="left") 

266 self.spot_prices = list(map(lambda x: spot_price_model(x, **self.parameter_dict), self.date_range)) 

267 self.df = pd.DataFrame(data=self.spot_prices, index=self.date_range, columns=["Spot"]) 

268 base_y = self.df.resample("YE").mean() 

269 base_y.index = base_y.index.strftime("%Y") 

270 

271 df_spot = self.df.copy() 

272 df_spot.index = df_spot.index.strftime("%Y") 

273 

274 shape = df_spot.divide(base_y, axis="index") 

275 self.shape_df = pd.DataFrame(data=shape["Spot"].tolist(), index=self.date_range, columns=["shape"]) 

276 

277 def __get_contracts_list(self, contracts_schedules: Dict[str, Dict[str, SimpleSchedule]]) -> Dict[str, Dict[str, EnergyFutureSpecifications]]: 

278 contracts = defaultdict(dict) 

279 for contract_type, contracts_dict in contracts_schedules.items(): 

280 for contract_name, schedule in contracts_dict.items(): 

281 tg = schedule.get_schedule() 

282 price = self.df.loc[tg, :].mean().iloc[0] 

283 contracts[contract_type][contract_name] = EnergyFutureSpecifications(schedule=schedule, price=price, name=contract_name) 

284 return list(itertools.chain(*[list(constracts_spec.values()) for constracts_spec in contracts.values()])) 

285 

286 def test_from_shape(self): 

287 contracts_schedules = { 

288 "base": { 

289 "Cal23_Base": BaseSchedule(dt.datetime(2023, 1, 1), dt.datetime(2024, 1, 1)), 

290 "Q2/23_Base": BaseSchedule(dt.datetime(2023, 4, 1), dt.datetime(2023, 7, 1)), 

291 }, 

292 "peak": { 

293 "Cal23_Peak": PeakSchedule(dt.datetime(2023, 1, 1), dt.datetime(2024, 1, 1)), 

294 "Cal24_Peak": PeakSchedule(dt.datetime(2024, 1, 1), dt.datetime(2025, 1, 1)), 

295 "Q1/23_Peak": PeakSchedule(dt.datetime(2023, 1, 1), dt.datetime(2023, 4, 1)), 

296 "Q2/23_Peak": PeakSchedule(dt.datetime(2023, 4, 1), dt.datetime(2023, 7, 1)), 

297 "Q3/23_Peak": PeakSchedule(dt.datetime(2023, 7, 1), dt.datetime(2023, 10, 1)), 

298 "Q4/23_Peak": PeakSchedule(dt.datetime(2023, 10, 1), dt.datetime(2024, 1, 1)), 

299 "Q2/24_Peak": PeakSchedule(dt.datetime(2024, 4, 1), dt.datetime(2024, 7, 1)), 

300 }, 

301 } 

302 contracts_list = self.__get_contracts_list(contracts_schedules=contracts_schedules) 

303 pfc_obj = EnergyPriceForwardCurve.from_existing_shape(id=None, refdate=None, pfc_shape=self.shape_df, future_contracts=contracts_list) 

304 

305 pfc = pfc_obj.get_pfc() 

306 

307 for contract in contracts_list: 

308 pfc_price = np.mean(pfc.loc[contract.get_schedule(), :]) 

309 self.assertAlmostEqual(pfc_price, contract.get_price(), delta=10 ** (-10)) 

310 

311 def test_from_scratch(self): 

312 contracts_schedules = { 

313 "base": { 

314 "Cal23_Base": SimpleSchedule(dt.datetime(2023, 1, 1), dt.datetime(2024, 1, 1), freq="D"), 

315 "Q2/23_Base": BaseSchedule(dt.datetime(2023, 4, 1), dt.datetime(2023, 7, 1)), 

316 }, 

317 "peak": { 

318 "Cal23_Peak": PeakSchedule(dt.datetime(2023, 1, 1), dt.datetime(2024, 1, 1)), 

319 "Cal24_Peak": PeakSchedule(dt.datetime(2024, 1, 1), dt.datetime(2025, 1, 1)), 

320 "Q1/23_Peak": PeakSchedule(dt.datetime(2023, 1, 1), dt.datetime(2023, 4, 1)), 

321 "Q2/23_Peak": PeakSchedule(dt.datetime(2023, 4, 1), dt.datetime(2023, 7, 1)), 

322 "Q3/23_Peak": PeakSchedule(dt.datetime(2023, 7, 1), dt.datetime(2023, 10, 1)), 

323 "Q4/23_Peak": PeakSchedule(dt.datetime(2023, 10, 1), dt.datetime(2024, 1, 1)), 

324 "Q2/24_Peak": PeakSchedule(dt.datetime(2024, 4, 1), dt.datetime(2024, 7, 1)), 

325 }, 

326 } 

327 contracts_list = self.__get_contracts_list(contracts_schedules=contracts_schedules) 

328 normalization_config = {"D": 2, "W": 2, "ME": 1} 

329 holiday_calendar = holidays.country_holidays("DE", years=[2024, 2025, 2026]) 

330 apply_schedule = BaseSchedule(start=dt.datetime(2025, 12, 31), end=dt.datetime(2027, 1, 1)) 

331 

332 pfc_shaper = CategoricalRegression( 

333 spot_prices=self.example_spot_price_data, holiday_calendar=holiday_calendar, normalization_config=normalization_config 

334 ) 

335 # raise ValueError due to different frequencies in the contracts 

336 with self.assertRaises(ValueError): 

337 pfc_obj = EnergyPriceForwardCurve.from_scratch( 

338 id=None, refdate=None, apply_schedule=apply_schedule, pfc_shaper=pfc_shaper, future_contracts=contracts_list 

339 ) 

340 

341 def test_existing_pfc(self): 

342 # create shape without DateTimeIndex 

343 not_acceptable_shape = pd.DataFrame(data=np.array([1, 2, 3, 4, 5, 6, 7, 8, 9])) 

344 with self.assertRaises(TypeError): 

345 pfc_obj = EnergyPriceForwardCurve.from_existing_pfc(id=None, refdate=None, pfc=not_acceptable_shape) 

346 

347 # create shape which is not a pd.DataFrame 

348 not_acceptable_shape = [1, 2, 3, 4, 5, 6, 7, 8, 9] 

349 with self.assertRaises(TypeError): 

350 pfc_obj = EnergyPriceForwardCurve.from_existing_pfc(id=None, refdate=None, pfc=not_acceptable_shape) 

351 

352 with self.assertRaises(ValueError): 

353 EnergyPriceForwardCurve(id=None, refdate=None) 

354 

355 

356if __name__ == "__main__": 

357 unittest.main()