Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

PVI update to match Investopedia definition and fix for issue 625 #779

Merged
merged 2 commits into from
Jun 11, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions pandas_ta/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -1806,10 +1806,10 @@ def obv(self, offset=None, **kwargs: DictLike):
result = obv(close=close, volume=volume, offset=offset, **kwargs)
return self._post_process(result, **kwargs)

def pvi(self, length=None, initial=None, signed=True, offset=None, **kwargs: DictLike):
def pvi(self, length=None, initial=None, mamode=None, offset=None, **kwargs: DictLike):
close = self._get_column(kwargs.pop("close", "close"))
volume = self._get_column(kwargs.pop("volume", "volume"))
result = pvi(close=close, volume=volume, length=length, initial=initial, signed=signed, offset=offset, **kwargs)
result = pvi(close=close, volume=volume, length=length, initial=initial, mamode=mamode, offset=offset, **kwargs)
return self._post_process(result, **kwargs)

def pvo(self, fast=None, slow=None, signal=None, scalar=None, offset=None, **kwargs: DictLike):
Expand Down
69 changes: 42 additions & 27 deletions pandas_ta/volume/pvi.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,15 @@
# -*- coding: utf-8 -*-
import pandas as pd
import numpy as np
from pandas import Series
from pandas_ta._typing import DictLike, Int
from pandas_ta.momentum import roc
from pandas_ta.utils import signed_series, v_offset, v_pos_default, v_series

from pandas_ta.ma import ma
from pandas_ta.utils import v_offset, v_mamode, v_pos_default, v_series

def pvi(
close: Series, volume: Series, length: Int = None, initial: Int = None,
offset: Int = None, **kwargs: DictLike
) -> Series:
mamode: str = None, offset: Int = None, **kwargs: DictLike
) -> pd.DataFrame:
"""Positive Volume Index (PVI)

The Positive Volume Index is a cumulative indicator that uses volume
Expand All @@ -21,48 +22,62 @@ def pvi(
Args:
close (pd.Series): Series of 'close's
volume (pd.Series): Series of 'volume's
length (int): The short period. Default: 13
initial (int): The short period. Default: 1000
length (int): The short period. Default: 255
initial (int): The short period. Default: 100
mamode (str): See ``help(ta.ma)``. Default: 'ema'
offset (int): How many periods to offset the result. Default: 0

Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method

Returns:
pd.Series: New feature generated.
pd.DataFrame: New DataFrame with ['PVI_1', 'PVIs_<length>']
"""

# Validate
length = v_pos_default(length, 1)
mamode = v_mamode(mamode, "ema")
length = v_pos_default(length, 255)
close = v_series(close, length + 1)
volume = v_series(volume, length + 1)
initial = v_pos_default(initial, 100)
offset = v_offset(offset)

if close is None or volume is None:
return

initial = v_pos_default(initial, 1000)
offset = v_offset(offset)
# Get numpy arrays of the data
close_prices = close.to_numpy()
volumes = volume.to_numpy()
pvis = np.empty(len(close_prices))

# Set the first value from from initial
pvis[0] = initial

# Calculate
signed_volume = signed_series(volume, 1)
_roc = roc(close=close, length=length)
pvi = _roc * signed_volume[signed_volume > 0].abs()
pvi.fillna(0, inplace=True)
pvi.iloc[0] = initial
pvi = pvi.cumsum()

# Offset
for i in range(1, len(close_prices)):
if volumes[i] > volumes[i-1]:
# PVI = Yesterday’s PVI + [[(Close – Yesterday’s Close) / Yesterday’s Close] * Yesterday’s PVI
pvis[i] = pvis[i-1] + (((close_prices[i] - close_prices[i-1]) / close_prices[i-1]) * pvis[i-1])
else:
# PVI = Yesterday’s PVI
pvis[i] = pvis[i-1]

data = {
'PVI_1': pvis,
}
df = pd.DataFrame(data, index=close.index)

if offset != 0:
pvi = pvi.shift(offset)
df['PVI_1'] = df['PVI_1'].shift(offset)

sig_series = ma(mamode, df['PVI_1'], length=length)
df[f'PVIs_{length}'] = sig_series

# Fill
if "fillna" in kwargs:
pvi.fillna(kwargs["fillna"], inplace=True)
df.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
pvi.fillna(method=kwargs["fill_method"], inplace=True)

# Name and Category
pvi.name = f"PVI_{length}"
pvi.category = "volume"
df.fillna(method=kwargs["fill_method"], inplace=True)

return pvi
return df
4 changes: 2 additions & 2 deletions tests/test_indicator_volume.py
Original file line number Diff line number Diff line change
Expand Up @@ -138,7 +138,7 @@ def test_obv(df):

def test_pvi(df):
result = ta.pvi(df.close, df.volume)
assert isinstance(result, Series)
assert isinstance(result, DataFrame)
assert result.name == "PVI_1"


Expand Down Expand Up @@ -253,7 +253,7 @@ def test_ext_nvi(df):

def test_ext_pvi(df):
df.ta.pvi(append=True)
assert df.columns[-1] == "PVI_1"
assert list(df.columns[-2:]) == ["PVI_1", "PVIs_255"]


def test_ext_pvol(df):
Expand Down
Loading