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

1# indicators/moving_volume_profile_indicator.py 

2 

3# global imports 

4import pandas as pd 

5 

6# local imports 

7from source.indicators import VolumeProfileIndicatorHandler 

8 

9class MovingVolumeProfileIndicatorHandler(VolumeProfileIndicatorHandler): 

10 """ 

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

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

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

14 """ 

15 

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

17 """ 

18 Class constructor. 

19 

20 Parameters: 

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

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

23 while creating volume profile for certain window. 

24 """ 

25 

26 super().__init__(number_of_steps) 

27 self.window_size = window_size 

28 

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

30 """ 

31 Calculates moving volume profile indicator values for given data. 

32 

33 Parameters: 

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

35 

36 Returns: 

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

38 """ 

39 

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

41 

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

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

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

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

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

47 

48 return moving_volume_price_df