Coverage for tests/test_ppa_hedge.py: 12%

1import sys 

2 

3try: 

4 import tensorflow as tf # exit this test if no tensorflow is installed 

5 import unittest 

6 import datetime as dt 

7 import numpy as np 

8 from rivapy.models.residual_demand_model import MultiRegionWindForecastModel, WindPowerForecastModel, OrnsteinUhlenbeck, ResidualDemandForwardModel, SmoothstepSupplyCurve 

9 from rivapy.instruments.ppa_specification import GreenPPASpecification 

10 from rivapy.pricing.green_ppa_pricing import price, DeepHedgeModel 

11 

12 class DeepHedger(unittest.TestCase): 

13 def test_simple(self): 

14 spots = {} 

15 np.random.seed(42) 

16 n_sims = 100 

17 timegrid = np.linspace(0.0,1.0,12, endpoint=True) 

18 payoff = None 

19 for i in range(2): 

20 model = OrnsteinUhlenbeck(1.0,0.2,1.0) 

21 rnd = np.random.normal(size=(timegrid.shape[0], n_sims)) 

22 s = model.simulate(timegrid, start_value=1.0, rnd=rnd) 

23 spots['Asset'+str(i)] = s 

24 if payoff is None: 

25 payoff = np.maximum(s[-1,:]-1.0,0.0) 

26 else: 

27 payoff = np.maximum(np.maximum(s[-1,:]-1.0,0.0), payoff) 

28 model = DeepHedgeModel([k for k in spots.keys()], None, timegrid=timegrid, 

29 regularization=0.0, depth=3, n_neurons=32) 

30 lr_schedule = tf.keras.optimizers.schedules.ExponentialDecay( 

31 initial_learning_rate=0.1,#1e-3, 

32 decay_steps=100, 

33 decay_rate=0.9) 

34 model.train(spots, payoff, lr_schedule, 5, 10) 

35 

36 class GreenPPAHedger(unittest.TestCase): 

37 def test_hedging(self): 

38 """Simple test with a perfect forecast  

39 """ 

40 val_date = dt.datetime(2023,1,1) 

41 days = 2 

42 timegrid = np.linspace(0.0, days*1.0/365.0, days*24) 

43 #forecast_points = [i for i in range(len(timegrid)) if i%8==0] 

44 forward_expiries = [timegrid[-1]] 

45 n_sims = 1_000 

46 

47 regions = [ MultiRegionWindForecastModel.Region( 

48 WindPowerForecastModel(speed_of_mean_reversion=0.5, 

49 volatility=0.05, 

50 expiries=forward_expiries, 

51 forecasts = [0.8],#*len(forward_expiries) 

52 region = 'Onshore' 

53 ), 

54 capacity=100.0, 

55 rnd_weights=[1.0,0.0] 

56 ), 

57 MultiRegionWindForecastModel.Region( 

58 WindPowerForecastModel(speed_of_mean_reversion=0.5, 

59 volatility=1.80, 

60 expiries=forward_expiries, 

61 forecasts = [0.8],#*len(forward_expiries) 

62 region = 'Offshore' 

63 ), 

64 capacity=100.0, 

65 rnd_weights=[1.0,0.0] 

66 ) 

67 

68 ] 

69 multi_region_wind_foecast_model = MultiRegionWindForecastModel(regions) 

70 

71 highest_price = OrnsteinUhlenbeck(1.0, 0.01, mean_reversion_level=1.0) 

72 supply_curve = SmoothstepSupplyCurve(1.0, 0) 

73 rdm = ResidualDemandForwardModel( 

74 #wind_forecast_model,  

75 multi_region_wind_foecast_model, 

76 highest_price, 

77 supply_curve, 

78 max_price = 1.0, 

79 forecast_hours=[6, 10, 14, 18], 

80 ) 

81 

82 strike = 0.4#fwd_prices[:,-1].mean() 

83 spec = GreenPPASpecification(technology = 'Wind', 

84 location = 'Onshore', 

85 udl = 'Power', 

86 schedule = [val_date + dt.timedelta(days=2)], 

87 fixed_price=strike, 

88 max_capacity = 1.0) 

89 result = price(val_date, spec, rdm , 

90 depth=3, nb_neurons=64, 

91 n_sims = n_sims, regularization= 10.0, 

92 epochs = 10, 

93 verbose=1, 

94 initial_lr = 1e-2, 

95 batch_size=400, 

96 decay_rate=0.6, 

97 tensorboard_logdir=None 

98 ) 

99 t = 0 

100 #delta = result.hedge_model.model.predict([fwd_prices[:,t], forecasts[:,t], np.array([timegrid[t]]*forecasts.shape[0])]).reshape((-1)) 

101 delta = result.hedge_model.compute_delta(result.fwd_prices, result.forecasts, t) 

102 #self.assertAlmostEqual(delta[0], result.forecasts[spec.location][0,0], 1e-3) 

103 

104except: 

105 pass 

106 

107if __name__ == '__main__': 

108 unittest.main()