Coverage for rivapy/instruments/energy_futures_specifications.py: 100%

1import numpy as np 

2import datetime as dt 

3import rivapy.tools.interfaces as interfaces 

4from rivapy.tools import SimpleSchedule 

5from rivapy.tools.factory import create as _create 

6from typing import Tuple, List 

7 

8 

9class EnergyFutureSpecifications(interfaces.FactoryObject): 

10 def __init__(self, schedule: SimpleSchedule, price: float, name: str) -> None: 

11 """Specification for an energy future contract. The delivery period is defined by the schedule. 

12 

13 Args: 

14 schedule (SimpleSchedule): Delivery period 

15 price (float): Price 

16 name (str): Name 

17 """ 

18 if isinstance(schedule, dict): 

19 self.schedule = _create(schedule) 

20 else: 

21 self.schedule = schedule 

22 self.price = price 

23 self.name = name 

24 

25 def get_schedule(self) -> np.ndarray: 

26 """Returns each delivery date time 

27 

28 Returns: 

29 np.ndarray: Numpy array containing each delivery date time 

30 """ 

31 return self.schedule.get_schedule() 

32 

33 def get_price(self) -> float: 

34 """Returns the price 

35 

36 Returns: 

37 float: Price 

38 """ 

39 return self.price 

40 

41 def get_start(self) -> dt.datetime: 

42 """Returns the delivery start. Note that this may not necessarily correspond to the first delivery. 

43 

44 Returns: 

45 dt.datetime: Start date time of the delivery scheduler 

46 """ 

47 return self.schedule.start 

48 

49 def get_end(self) -> dt.datetime: 

50 """Returns the delivery end. Note that this may not necessarily correspond to the last delivery. 

51 

52 Returns: 

53 dt.datetime: End date time of the delivery scheduler 

54 """ 

55 return self.schedule.end 

56 

57 def get_start_end(self) -> Tuple[dt.datetime, dt.datetime]: 

58 """Returns the start and end as a tuple, where (start, end) 

59 

60 Returns: 

61 Tuple[dt.datetime, dt.datetime]: Tuple containing start and end 

62 """ 

63 return (self.get_start(), self.get_end()) 

64 

65 def _to_dict(self) -> dict: 

66 return {"schedule": self.schedule.to_dict(), "price": self.price, "name": self.name} 

67 

68 @staticmethod 

69 def _create_sample(n_samples: int, seed: int = None, ref_date=None) -> List["EnergyFutureSpecifications"]: 

70 """Creates a sample of random ``EnergyFutureSpecifiactions`` objects. 

71 

72 Returns: 

73 ListEnergyFutureSpecifications]: List of sampled ``EnergyFutureSpecifiactions`` objects 

74 """ 

75 if seed is not None: 

76 np.random.seed(seed) 

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

78 result = [] 

79 for i, schedule in enumerate(schedules): 

80 price = np.random.uniform(low=50.0, high=150.0) 

81 result.append(EnergyFutureSpecifications(schedule=schedule, price=price, name=f"Contract_{i+1}")) 

82 return result