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

1# indicators/volume_profile_indicator.py 

2 

3from .indicator_base import * 

4from collections import defaultdict 

5 

6class VolumeProfileIndicatorHandler(IndicatorHandlerBase): 

7 """ 

8 Implements static volume profile indicator. It denotes volume traded at 

9 certain price levels. Calculated data can not be directly mapped to input 

10 data and should be treated as the separate chart. 

11 """ 

12 

13 def __init__(self, number_of_steps: int = 40) -> None: 

14 """ 

15 Class constructor. 

16 

17 Parameters: 

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

19 while creating volume profile. 

20 """ 

21 

22 self.number_of_steps = number_of_steps 

23 

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

25 """ 

26 Calculates static volume profile indicator values for given data. 

27 

28 Parameters: 

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

30 

31 Returns: 

32 (pd.DataFrame): Output data with calculated static volume profile values. 

33 """ 

34 

35 volume_profile = defaultdict(float) 

36 data_min = data['low'].min() 

37 data_max = data['high'].max() 

38 step = (data_max - data_min) / (self.number_of_steps - 1) 

39 

40 for _, row in data.iterrows(): 

41 equalized_low = row['low'] // step * step 

42 equalized_high = row['high'] // step * step 

43 price_range = np.linspace(equalized_low, equalized_high, num = int(round((equalized_high - equalized_low) / step + 1, 0))) 

44 price_range = np.round(price_range, 6 - int(np.floor(np.log10(data_min)))) 

45 volume_per_step = row['volume'] / len(price_range) 

46 

47 for price in price_range: 

48 volume_profile[price] += volume_per_step 

49 

50 profile_df = pd.DataFrame(list(volume_profile.items()), columns = ['price', 'volume']) 

51 profile_df.sort_values(by='price', inplace=True) 

52 

53 return profile_df