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, SimpleCategoricalRegression, CategoricalFourierShaper 

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 for regression_obj in [CategoricalRegression, SimpleCategoricalRegression, CategoricalFourierShaper]: 

177 pfc_shaper = regression_obj(spot_prices=spot_prices, holiday_calendar=holiday_calendar) 

178 pfc_shaper.calibrate() 

179 pfc_fit = pfc_shaper.apply(spot_prices.index) 

180 

181 self.assertLess(np.sum((spot_prices.values / np.mean(spot_prices.values) - pfc_fit.to_numpy())) ** 2, 10 ** (-5)) 

182 

183 pfc_rollout = pfc_shaper.apply(apply_schedule=apply_schedule.get_schedule()) 

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

185 

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

187 

188 def test_normalization_with_config(self): 

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

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

191 

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

193 

194 for regression_obj in [CategoricalRegression, SimpleCategoricalRegression, CategoricalFourierShaper]: 

195 pfc_shaper = regression_obj( 

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

197 ) 

198 pfc_shaper.calibrate() 

199 pfc_shape = pfc_shaper.apply(apply_schedule=apply_schedule.get_schedule()) 

200 

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

202 

203 self.assertLess( 

204 np.abs( 

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

206 ), 

207 10 ** (-5), 

208 ) 

209 

210 self.assertLess( 

211 np.abs( 

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

213 ), 

214 10 ** (-5), 

215 ) 

216 

217 self.assertLess( 

218 np.abs( 

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

220 ), 

221 10 ** (-5), 

222 ) 

223 

224 self.assertLess( 

225 np.abs( 

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

227 ), 

228 10 ** (-5), 

229 ) 

230 

231 self.assertLess( 

232 np.abs( 

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

234 ), 

235 10 ** (-5), 

236 ) 

237 

238 def test_normalization_without_config(self): 

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

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

241 for regression_obj in [CategoricalRegression, SimpleCategoricalRegression, CategoricalFourierShaper]: 

242 pfc_shaper = regression_obj(spot_prices=self.example_spot_price_data, holiday_calendar=holiday_calendar) 

243 pfc_shaper.calibrate() 

244 pfc_shape = pfc_shaper.apply(apply_schedule=apply_schedule.get_schedule()) 

245 

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

247 

248 self.assertLess( 

249 np.abs( 

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

251 ), 

252 10 ** (-5), 

253 ) 

254 

255 

256class TestEnergyPriceForwardCurve(unittest.TestCase): 

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

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

259 

260 self.example_spot_price_data = pd.read_excel( 

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

262 ) 

263 self.parameter_dict = { 

264 "spot_price_level": 100, 

265 "peak_price_level": 10, 

266 "solar_price_level": 8, 

267 "weekend_price_level": 10, 

268 "winter_price_level": 20, 

269 "epsilon_mean": 0, 

270 "epsilon_var": 5, 

271 } 

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

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

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

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

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

277 

278 df_spot = self.df.copy() 

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

280 

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

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

283 

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

285 contracts = defaultdict(dict) 

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

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

288 tg = schedule.get_schedule() 

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

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

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

292 

293 def test_from_shape(self): 

294 contracts_schedules = { 

295 "base": { 

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

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

298 }, 

299 "peak": { 

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

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

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

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

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

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

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

307 }, 

308 } 

309 contracts_list = self.__get_contracts_list(contracts_schedules=contracts_schedules) 

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

311 

312 pfc = pfc_obj.get_pfc() 

313 

314 for contract in contracts_list: 

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

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

317 

318 def test_from_scratch(self): 

319 contracts_schedules = { 

320 "base": { 

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

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

323 }, 

324 "peak": { 

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

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

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

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

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

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

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

332 }, 

333 } 

334 contracts_list = self.__get_contracts_list(contracts_schedules=contracts_schedules) 

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

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

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

338 

339 pfc_shaper = CategoricalRegression( 

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

341 ) 

342 # raise ValueError due to different frequencies in the contracts 

343 with self.assertRaises(ValueError): 

344 pfc_obj = EnergyPriceForwardCurve.from_scratch( 

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

346 ) 

347 

348 def test_existing_pfc(self): 

349 # create shape without DateTimeIndex 

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

351 with self.assertRaises(TypeError): 

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

353 

354 # create shape which is not a pd.DataFrame 

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

356 with self.assertRaises(TypeError): 

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

358 

359 with self.assertRaises(ValueError): 

360 EnergyPriceForwardCurve(id=None, refdate=None) 

361 

362 

363if __name__ == "__main__": 

364 unittest.main()