Coverage for source/indicators/moving_volume_profile_indicator.py: 100%

1# indicators/moving_volume_profile_indicator.py 

2 

3from .volume_profile_indicator import * 

4 

5class MovingVolumeProfileIndicatorHandler(VolumeProfileIndicatorHandler): 

6 """ 

7 Implements moving volume profile indicator. Uses normal volume profile indicator 

8 to calculate volume profile for the given window size. It denotes volume traded 

9 at certain price over the period given by window size. 

10 """ 

11 

12 def __init__(self, window_size: int = 14, number_of_steps: int = 40) -> None: 

13 """ 

14 Class constructor. 

15 

16 Parameters: 

17 window_size (int): Length of window that indicator should be applied over. 

18 number_of_steps (int): Number of bins that price should be put into 

19 while creating volume profile for certain window. 

20 """ 

21 

22 super().__init__(number_of_steps) 

23 self.window_size = window_size 

24 

25 def calculate(self, data: pd.DataFrame) -> pd.DataFrame: 

26 """ 

27 Calculates moving volume profile indicator values for given data. 

28 

29 Parameters: 

30 data (pd.DataFrame): Data frame with input data. 

31 

32 Returns: 

33 (pd.DataFrame): Output data with calculated moving volume profile values. 

34 """ 

35 

36 moving_volume_price_df = pd.DataFrame(index = data.index) 

37 

38 for i, (index, row) in enumerate(data.iterrows()): 

39 volume_profiles_data = super().calculate(data[max(0, i - self.window_size) : i + 1]) 

40 current_average_price = (row['low'] + row['high']) / 2 

41 volume_profiles_connected_to_lower_prices = volume_profiles_data['price'] <= current_average_price 

42 moving_volume_price_df.loc[index, 'moving_volume_profile'] = volume_profiles_data[volume_profiles_connected_to_lower_prices]['volume'].iloc[-1] 

43 

44 return moving_volume_price_df