Coverage for rivapy/instruments/ppa_specification.py: 80%

1from typing import Union, Tuple, Iterable, Set, List 

2import abc 

3import numpy as np 

4import pandas as pd 

5import datetime as dt 

6import rivapy.tools.interfaces as interfaces 

7 

8# from rivapy.instruments.factory import create as _create 

9from rivapy.tools.factory import create as _create 

10from rivapy.tools.datetime_grid import DateTimeGrid 

11from rivapy.tools import SimpleSchedule 

12 

13 

14class PPASpecification(interfaces.FactoryObject): 

15 def __init__( 

16 self, udl: str, amount: Union[float, np.ndarray], schedule: Union[SimpleSchedule, List[dt.datetime]], fixed_price: float, id: str = None 

17 ): 

18 """Specification for a simple power purchase agreement (PPA). 

19 

20 Args: 

21 udl (str): Name of underlying (power) that is delivered (just use for consistency checking within pricing against simulated model values). 

22 amount (Union[None, float, np.ndarray]): Amount of power delivered at each timepoint/period. Either a single value s.t. all volumes delivered are constant or a load table. If None, a non-constant amount (e.g. by production from renewables) is assumed. 

23 schedule (Union[SimpleSchedule, List[dt.datetime]): Schedule describing when power is delivered. 

24 fixed_price (float): The fixed price paif for the power. 

25 id (str): Simple id of the specification. If None, a uuid will be generated. Defaults to None. 

26 """ 

27 self.id = id 

28 self.udl = udl 

29 if id is None: 

30 self.id = type(self).__name__ + "/" + str(dt.datetime.now()) 

31 self.amount = amount 

32 if isinstance(schedule, dict): # if schedule is a dict we try to create it from factory 

33 self.schedule = _create(schedule) 

34 else: 

35 self.schedule = schedule 

36 self.fixed_price = fixed_price 

37 if isinstance(schedule, list): 

38 self._schedule_df = pd.DataFrame({"dates": self.schedule}).reset_index() 

39 else: 

40 self._schedule_df = self.schedule.get_df().set_index(["dates"]).sort_index() 

41 self._schedule_df["amount"] = amount 

42 self._schedule_df["flow"] = None 

43 

44 @staticmethod 

45 def _create_sample(n_samples: int, seed: int = None, ref_date=None): 

46 schedules = SimpleSchedule._create_sample(n_samples, seed, ref_date) 

47 result = [] 

48 for schedule in schedules: 

49 amount = np.random.uniform(low=50.0, high=100.0) 

50 fixed_price = np.random.uniform(low=0.5, high=1.5) 

51 result.append(PPASpecification(udl="Power", amount=amount, schedule=schedule, fixed_price=fixed_price)) 

52 return result 

53 

54 def get_schedule(self) -> List[dt.datetime]: 

55 if not isinstance(self.schedule, list): 

56 return self.schedule.get_schedule() 

57 return self.schedule 

58 

59 def _to_dict(self) -> dict: 

60 try: # if isinstance(self.schedule, interfaces.FactoryObject): 

61 schedule = self.schedule.to_dict() 

62 except Exception as e: 

63 schedule = self.schedule 

64 return {"udl": self.udl, "id": self.id, "amount": self.amount, "schedule": schedule, "fixed_price": self.fixed_price} 

65 

66 def set_amount(self, amount): 

67 self.amount = amount 

68 self._schedule_df["amount"] = amount 

69 

70 def n_deliveries(self): 

71 return self._schedule_df.shape[0] 

72 

73 def compute_flows(self, refdate: dt.datetime, pfc, result: pd.DataFrame = None, result_col=None) -> pd.DataFrame: 

74 df = pfc.get_df() 

75 if result is None: 

76 self._schedule_df["flow"] = self._schedule_df["amount"] * (df.loc[self._schedule_df.index]["values"] - self.fixed_price) 

77 return self._schedule_df 

78 result[result_col] = self._schedule_df["amount"] * (df.loc[self._schedule_df.index]["values"] - self.fixed_price) 

79 

80 

81class GreenPPASpecification(PPASpecification): 

82 def __init__( 

83 self, 

84 schedule: Union[SimpleSchedule, List[dt.datetime]], 

85 fixed_price: float, 

86 max_capacity: float, 

87 technology: str, 

88 udl: str, 

89 location: str = None, 

90 id: str = None, 

91 ): 

92 """:term:`Specification` for a green power purchase agreement. 

93 

94 In contrast to a normal PPA the quantities of this PPA are related to some kind of 

95 renewable energy such as wind or solar, i.e. the quantity is related to some uncertain production. 

96 

97 Args: 

98 schedule (Union[SimpleSchedule, List[dt.datetime]]): Delivery schedule. 

99 fixed_price (float): Fixed price paid for the power. 

100 max_capacity (float): The absolute maximal capacity of the renewable energy source. This is used to derive the production amount of the plant by multiplying forecasts with the factor max_capacity/total_capacity (where total capacity may be time dependent). 

101 technology (str): Identifier for the technology. This is used to retrieve the simulated values for production of this technology from a model 

102 location (str, optional): Identifier for the location. This is used to retrieve the simulated values for production of this technology at this location from a model that supports this feature. Defaults to None. 

103 udl (str, optional): Name of underlying (power) that is delivered (just use for consistency checking within pricing against simulated model values). It is used within pricing when the respective simulated price must be retrieved from a model's simulation results. 

104 id (str, optional): Unique identifier of this contract. Defaults to None. 

105 """ 

106 super().__init__(udl, None, schedule, fixed_price, id) 

107 self.technology = technology 

108 self.max_capacity = max_capacity 

109 self.location = location 

110 

111 @staticmethod 

112 def _create_sample(n_samples: int, seed: int = None, ref_date=None): 

113 schedules = SimpleSchedule._create_sample(n_samples, seed, ref_date) 

114 result = [] 

115 for schedule in schedules: 

116 max_capacity = np.random.uniform(low=50.0, high=100.0) 

117 fixed_price = np.random.uniform(low=0.5, high=1.5) 

118 result.append( 

119 GreenPPASpecification( 

120 udl="Power", technology="Wind", location="Onshore", fixed_price=fixed_price, max_capacity=max_capacity, schedule=schedule 

121 ) 

122 ) 

123 return result 

124 

125 def _to_dict(self) -> dict: 

126 result = super()._to_dict() 

127 del result["amount"] 

128 result["technology"] = self.technology 

129 result["max_capacity"] = self.max_capacity 

130 result["location"] = self.location 

131 return result 

132 

133 def compute_flows(self, refdate: dt.datetime, pfc, forecast_amount: np.ndarray, result: pd.DataFrame = None, result_col=None) -> pd.DataFrame: 

134 self.set_amount(forecast_amount) 

135 return super().compute_flows(refdate, pfc, result, result_col)