diff --git a/ta/volume.py b/ta/volume.py index 5947c14..eabf4d2 100644 --- a/ta/volume.py +++ b/ta/volume.py @@ -6,6 +6,8 @@ """ +import typing as tp + import numpy as np import pandas as pd @@ -269,7 +271,14 @@ class VolumePriceTrendIndicator(IndicatorMixin): dropnans(bool)=False: drop nans after indicator calculated. """ - def __init__(self, close: pd.Series, volume: pd.Series, fillna: bool = False, smoothing_factor: int = None, dropnans:bool = False): + def __init__( + self, + close: pd.Series, + volume: pd.Series, + fillna: bool = False, + smoothing_factor: tp.Optional[int] = None, + dropnans: bool = False, + ): self._close = close self._volume = volume self._fillna = fillna @@ -280,9 +289,12 @@ def __init__(self, close: pd.Series, volume: pd.Series, fillna: bool = False, sm def _run(self): self._vpt = (self._close.pct_change() * self._volume).cumsum() if self._smoothing_factor: - min_periods = 0 if self._fillna else self._smoothing_factor - self._vpt = self._vpt.rolling(self._smoothing_factor, min_periods=min_periods).mean() - if self._dropnans: self._vpt = self._vpt.dropna() + min_periods = 0 if self._fillna else self._smoothing_factor + self._vpt = self._vpt.rolling( + self._smoothing_factor, min_periods=min_periods + ).mean() + if self._dropnans: + self._vpt = self._vpt.dropna() def volume_price_trend(self) -> pd.Series: """Volume-price trend (VPT) @@ -612,7 +624,9 @@ def sma_ease_of_movement(high, low, volume, window=14, fillna=False): ).sma_ease_of_movement() -def volume_price_trend(close, volume, fillna=False, smoothing_factor: int=None, dropnans: bool=False): +def volume_price_trend( + close, volume, fillna=False, smoothing_factor: tp.Optional[int] = None, dropnans: bool = False +): """Volume-price trend (VPT) Is based on a running cumulative volume that adds or substracts a multiple @@ -632,7 +646,11 @@ def volume_price_trend(close, volume, fillna=False, smoothing_factor: int=None, pandas.Series: New feature generated. """ return VolumePriceTrendIndicator( - close=close, volume=volume, fillna=fillna, smoothing_factor=smoothing_factor, dropnans=dropnans + close=close, + volume=volume, + fillna=fillna, + smoothing_factor=smoothing_factor, + dropnans=dropnans, ).volume_price_trend() diff --git a/test/__init__.py b/test/__init__.py index 8177a1c..2c477a8 100644 --- a/test/__init__.py +++ b/test/__init__.py @@ -35,8 +35,8 @@ TestForceIndexIndicator, TestMFIIndicator, TestOnBalanceVolumeIndicator, - TestVolumeWeightedAveragePrice, TestVolumePriceTrendIndicator, + TestVolumeWeightedAveragePrice, ) __all__ = [ diff --git a/test/unit/volume.py b/test/unit/volume.py index a7c18fd..7afc1ce 100644 --- a/test/unit/volume.py +++ b/test/unit/volume.py @@ -8,16 +8,16 @@ ForceIndexIndicator, MFIIndicator, OnBalanceVolumeIndicator, - VolumeWeightedAveragePrice, VolumePriceTrendIndicator, + VolumeWeightedAveragePrice, acc_dist_index, ease_of_movement, force_index, money_flow_index, on_balance_volume, sma_ease_of_movement, + volume_price_trend, volume_weighted_average_price, - volume_price_trend ) @@ -260,14 +260,16 @@ def test_vwap2(self): self._df[target].tail(), result.tail(), check_names=False ) + class TestVolumePriceTrendIndicator(unittest.TestCase): """ Original VPT: https://en.wikipedia.org/wiki/Volume%E2%80%93price_trend One more: https://www.barchart.com/education/technical-indicators/price_volume_trend According to TradingView: PVT = [((CurrentClose - PreviousClose) / PreviousClose) x Volume] + PreviousPVT - + Smoothed version (by Alex Orekhov (everget)): https://ru.tradingview.com/script/3Ah2ALck-Price-Volume-Trend/ - His script is using `pvt` (TradingView built-in variable) as described in TradingView documentation of PVT and just smoothing it with ema or sma by choice. + His script is using `pvt` (TradingView built-in variable) as described in TradingView documentation of PVT and + just smoothing it with ema or sma by choice. You can find smoothing here (13 row of script): `signal = signalType == "EMA" ? ema(pvt, signalLength) : sma(pvt, signalLength)` """ @@ -277,20 +279,24 @@ class TestVolumePriceTrendIndicator(unittest.TestCase): @classmethod def setUpClass(cls): cls._df = pd.read_csv(cls._filename, sep=",") - cls._params = dict( # default VPT params, unsmoothed - close=cls._df["Close"], - volume=cls._df["Volume"], - fillna=False, - smoothing_factor=None, - dropnans=False - ) - cls._params_smoothed = dict( # smoothed VPT params - close=cls._df["Close"], - volume=cls._df["Volume"], - fillna=False, - smoothing_factor=14, - dropnans=False - ) + + # default VPT params, unsmoothed + cls._params = { + "close": cls._df['Close'], + "volume": cls._df['Volume'], + "fillna": False, + "smoothing_factor": None, + "dropnans": False, + } + + # smoothed VPT params + cls._params_smoothed = { + "close": cls._df['Close'], + "volume": cls._df['Volume'], + "fillna": False, + "smoothing_factor": 14, + "dropnans": False, + } cls._indicator_default = VolumePriceTrendIndicator(**cls._params) cls._indicator_smoothed = VolumePriceTrendIndicator(**cls._params_smoothed) @@ -326,5 +332,6 @@ def test_vpt4(self): self._df[target].tail(), result.tail(), check_names=False ) + if __name__ == "__main__": unittest.main()