Merge pull request #131 from bukosabino/develop

Develop

ta/momentum.py

@@ -51,69 +51,6 @@ def rsi(self) -> pd.Series:
         return pd.Series(rsi, name='rsi')
 
 
-class MFIIndicator(IndicatorMixin):
-    """Money Flow Index (MFI)
-
-    Uses both price and volume to measure buying and selling pressure. It is
-    positive when the typical price rises (buying pressure) and negative when
-    the typical price declines (selling pressure). A ratio of positive and
-    negative money flow is then plugged into an RSI formula to create an
-    oscillator that moves between zero and one hundred.
-
-    http://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:money_flow_index_mfi
-
-    Args:
-        high(pandas.Series): dataset 'High' column.
-        low(pandas.Series): dataset 'Low' column.
-        close(pandas.Series): dataset 'Close' column.
-        volume(pandas.Series): dataset 'Volume' column.
-        n(int): n period.
-        fillna(bool): if True, fill nan values.
-    """
-
-    def __init__(self,
-                 high: pd.Series,
-                 low: pd.Series,
-                 close: pd.Series,
-                 volume: pd.Series,
-                 n: int = 14,
-                 fillna: bool = False):
-        self._high = high
-        self._low = low
-        self._close = close
-        self._volume = volume
-        self._n = n
-        self._fillna = fillna
-        self._run()
-
-    def _run(self):
-        # 1 typical price
-        tp = (self._high + self._low + self._close) / 3.0
-
-        # 2 up or down column
-        up_down = np.where(tp > tp.shift(1), 1, np.where(tp < tp.shift(1), -1, 0))
-
-        # 3 money flow
-        mf = tp * self._volume * up_down
-
-        # 4 positive and negative money flow with n periods
-        n_positive_mf = mf.rolling(self._n).apply(lambda x: np.sum(np.where(x >= 0.0, x, 0.0)), raw=True)
-        n_negative_mf = abs(mf.rolling(self._n).apply(lambda x: np.sum(np.where(x < 0.0, x, 0.0)), raw=True))
-
-        # 5 money flow index
-        mr = n_positive_mf / n_negative_mf
-        self._mr = (100 - (100 / (1 + mr)))
-
-    def money_flow_index(self) -> pd.Series:
-        """Money Flow Index (MFI)
-
-        Returns:
-            pandas.Series: New feature generated.
-        """
-        mr = self._check_fillna(self._mr, value=50)
-        return pd.Series(mr, name=f'mfi_{self._n}')
-
-
 class TSIIndicator(IndicatorMixin):
     """True strength index (TSI)
 
@@ -530,33 +467,6 @@ def rsi(close, n=14, fillna=False):
     return RSIIndicator(close=close, n=n, fillna=fillna).rsi()
 
 
-def money_flow_index(high, low, close, volume, n=14, fillna=False):
-    """Money Flow Index (MFI)
-
-    Uses both price and volume to measure buying and selling pressure. It is
-    positive when the typical price rises (buying pressure) and negative when
-    the typical price declines (selling pressure). A ratio of positive and
-    negative money flow is then plugged into an RSI formula to create an
-    oscillator that moves between zero and one hundred.
-
-    http://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:money_flow_index_mfi
-
-    Args:
-        high(pandas.Series): dataset 'High' column.
-        low(pandas.Series): dataset 'Low' column.
-        close(pandas.Series): dataset 'Close' column.
-        volume(pandas.Series): dataset 'Volume' column.
-        n(int): n period.
-        fillna(bool): if True, fill nan values.
-
-    Returns:
-        pandas.Series: New feature generated.
-
-    """
-    indicator = MFIIndicator(high=high, low=low, close=close, volume=volume, n=n, fillna=fillna)
-    return indicator.money_flow_index()
-
-
 def tsi(close, r=25, s=13, fillna=False):
     """True strength index (TSI)
 

ta/tests/__init__.py

@@ -1,13 +1,13 @@
-from ta.tests.momentum import (TestKAMAIndicator, TestMFIIndicator,
-                               TestRateOfChangeIndicator, TestRSIIndicator,
-                               TestStochasticOscillator, TestTSIIndicator,
-                               TestUltimateOscillator, TestWilliamsRIndicator)
+from ta.tests.momentum import (TestKAMAIndicator, TestRateOfChangeIndicator,
+                               TestRSIIndicator, TestStochasticOscillator,
+                               TestTSIIndicator, TestUltimateOscillator,
+                               TestWilliamsRIndicator)
 from ta.tests.trend import (TestADXIndicator, TestCCIIndicator,
                             TestMACDIndicator, TestPSARIndicator,
                             TestVortexIndicator)
 from ta.tests.utils import TestGeneral
 from ta.tests.volatility import TestAverageTrueRange, TestBollingerBands
 from ta.tests.volume import (TestAccDistIndexIndicator,
                              TestEaseOfMovementIndicator,
-                             TestForceIndexIndicator,
+                             TestForceIndexIndicator, TestMFIIndicator,
                              TestOnBalanceVolumeIndicator)

ta/tests/momentum.py

@@ -2,8 +2,8 @@
 
 import pandas as pd
 
-from ta.momentum import (KAMAIndicator, MFIIndicator, ROCIndicator,
-                         RSIIndicator, StochasticOscillator, TSIIndicator,
+from ta.momentum import (KAMAIndicator, ROCIndicator, RSIIndicator,
+                         StochasticOscillator, TSIIndicator,
                          UltimateOscillator, WilliamsRIndicator, roc)
 from ta.tests.utils import TestIndicator
 
@@ -47,28 +47,6 @@ def test_rsi(self):
         pd.testing.assert_series_equal(self._df[target].tail(), result.tail(), check_names=False)
 
 
-class TestMFIIndicator(unittest.TestCase):
-    """
-    https://school.stockcharts.com/doku.php?id=technical_indicators:money_flow_index_mfi
-    """
-
-    _filename = 'ta/tests/data/cs-mfi.csv'
-
-    def setUp(self):
-        self._df = pd.read_csv(self._filename, sep=',')
-        self._indicator = MFIIndicator(
-            high=self._df['High'], low=self._df['Low'], close=self._df['Close'], volume=self._df['Volume'], n=14,
-            fillna=False)
-
-    def tearDown(self):
-        del(self._df)
-
-    def test_mfi(self):
-        target = 'MFI'
-        result = self._indicator.money_flow_index()
-        pd.testing.assert_series_equal(self._df[target].tail(), result.tail(), check_names=False)
-
-
 class TestUltimateOscillator(unittest.TestCase):
     """
     https://school.stockcharts.com/doku.php?id=technical_indicators:ultimate_oscillator

ta/tests/utils.py

@@ -23,7 +23,7 @@ def test_general(self):
         df = ta.utils.dropna(self._df)
 
         # Add all ta features filling nans values
-        ta.add_all_ta_features(self._df, "Open", "High", "Low", "Close", "Volume_BTC", fillna=True)
+        ta.add_all_ta_features(df, "Open", "High", "Low", "Close", "Volume_BTC", fillna=True)
 
         # Add all ta features not filling nans values
-        df = ta.add_all_ta_features(self._df, "Open", "High", "Low", "Close", "Volume_BTC", fillna=False)
+        df = ta.add_all_ta_features(df, "Open", "High", "Low", "Close", "Volume_BTC", fillna=False)

ta/tests/volume.py

@@ -1,9 +1,12 @@
+import unittest
+
 import pandas as pd
 
 from ta.tests.utils import TestIndicator
 from ta.volume import (AccDistIndexIndicator, EaseOfMovementIndicator,
-                       ForceIndexIndicator, OnBalanceVolumeIndicator,
-                       acc_dist_index, ease_of_movement, force_index,
+                       ForceIndexIndicator, MFIIndicator,
+                       OnBalanceVolumeIndicator, acc_dist_index,
+                       ease_of_movement, force_index, money_flow_index,
                        on_balance_volume, sma_ease_of_movement)
 
 
@@ -99,3 +102,32 @@ def test_adl2(self):
             high=self._df['High'], low=self._df['Low'], close=self._df['Close'], volume=self._df['Volume'],
             fillna=False).acc_dist_index()
         pd.testing.assert_series_equal(self._df[target].tail(), result.tail(), check_names=False)
+
+
+class TestMFIIndicator(unittest.TestCase):
+    """
+    https://school.stockcharts.com/doku.php?id=technical_indicators:money_flow_index_mfi
+    """
+
+    _filename = 'ta/tests/data/cs-mfi.csv'
+
+    def setUp(self):
+        self._df = pd.read_csv(self._filename, sep=',')
+        self._indicator = MFIIndicator(
+            high=self._df['High'], low=self._df['Low'], close=self._df['Close'], volume=self._df['Volume'], n=14,
+            fillna=False)
+
+    def tearDown(self):
+        del(self._df)
+
+    def test_mfi(self):
+        target = 'MFI'
+        result = self._indicator.money_flow_index()
+        pd.testing.assert_series_equal(self._df[target].tail(), result.tail(), check_names=False)
+
+    def test_mfi2(self):
+        target = 'MFI'
+        result = money_flow_index(
+            high=self._df['High'], low=self._df['Low'], close=self._df['Close'], volume=self._df['Volume'], n=14,
+            fillna=False)
+        pd.testing.assert_series_equal(self._df[target].tail(), result.tail(), check_names=False)

ta/trend.py

@@ -496,7 +496,7 @@ def __init__(self, high: pd.Series, low: pd.Series, close: pd.Series, n: int = 1
         self._run()
 
     def _run(self):
-        assert self._n is not 0, "N may not be 0 and is %r" % n
+        assert self._n != 0, "N may not be 0 and is %r" % n
 
         cs = self._close.shift(1)
         pdm = get_min_max(self._high, cs, 'max')

ta/utils.py

@@ -30,6 +30,7 @@ def _check_fillna(self, serie: pd.Series, value: int = 0):
 def dropna(df):
     """Drop rows with "Nans" values
     """
+    df = df.copy()
     number_cols = df.select_dtypes('number').columns.to_list()
     df[number_cols] = df[number_cols][df[number_cols] < math.exp(709)]  # big number
     df[number_cols] = df[number_cols][df[number_cols] != 0.0]

ta/volatility.py

@@ -171,24 +171,38 @@ class KeltnerChannel(IndicatorMixin):
         close(pandas.Series): dataset 'Close' column.
         n(int): n period.
         fillna(bool): if True, fill nan values.
+        ov(bool): if True, use original version as the centerline (SMA of typical price)
+            if False, use EMA of close as the centerline. More info:
+            https://school.stockcharts.com/doku.php?id=technical_indicators:keltner_channels
     """
 
-    def __init__(self, high: pd.Series, low: pd.Series, close: pd.Series, n: int = 14, fillna: bool = False):
+    def __init__(
+            self, high: pd.Series, low: pd.Series, close: pd.Series, n: int = 14, fillna: bool = False,
+            ov: bool = True):
         self._high = high
         self._low = low
         self._close = close
         self._n = n
         self._fillna = fillna
+        self._ov = ov
         self._run()
 
     def _run(self):
-        self._tp = ((self._high + self._low + self._close) / 3.0).rolling(self._n, min_periods=0).mean()
-        self._tp_high = (((4 * self._high) - (2 * self._low) + self._close) / 3.0).rolling(
-            self._n, min_periods=0).mean()
-        self._tp_low = (((-2 * self._high) + (4 * self._low) + self._close) / 3.0).rolling(
-            self._n, min_periods=0).mean()
-
-    def keltner_channel_central(self) -> pd.Series:
+        if self._ov:
+            self._tp = ((self._high + self._low + self._close) / 3.0).rolling(self._n, min_periods=0).mean()
+            self._tp_high = (((4 * self._high) - (2 * self._low) + self._close) / 3.0).rolling(
+                self._n, min_periods=0).mean()
+            self._tp_low = (((-2 * self._high) + (4 * self._low) + self._close) / 3.0).rolling(
+                self._n, min_periods=0).mean()
+        else:
+            self._tp = self._close.ewm(span=self._n, min_periods=0, adjust=False).mean()
+            atr = AverageTrueRange(
+                close=self._close, high=self._high, low=self._high, n=10, fillna=self._fillna
+            ).average_true_range()
+            self._tp_high = self._tp + (2*atr)
+            self._tp_low = self._tp - (2*atr)
+
+    def keltner_channel_mband(self) -> pd.Series:
         """Keltner Channel Middle Band
 
         Returns:
@@ -203,7 +217,7 @@ def keltner_channel_hband(self) -> pd.Series:
         Returns:
             pandas.Series: New feature generated.
         """
-        tp = self._check_fillna(self._tp, value=-1)
+        tp = self._check_fillna(self._tp_high, value=-1)
         return pd.Series(tp, name='kc_hband')
 
     def keltner_channel_lband(self) -> pd.Series:
@@ -215,6 +229,26 @@ def keltner_channel_lband(self) -> pd.Series:
         tp_low = self._check_fillna(self._tp_low, value=-1)
         return pd.Series(tp_low, name='kc_lband')
 
+    def keltner_channel_wband(self) -> pd.Series:
+        """Keltner Channel Band Width
+
+        Returns:
+            pandas.Series: New feature generated.
+        """
+        wband = ((self._tp_high - self._tp_low) / self._tp) * 100
+        wband = self._check_fillna(wband, value=0)
+        return pd.Series(wband, name='bbiwband')
+
+    def keltner_channel_pband(self) -> pd.Series:
+        """Keltner Channel Percentage Band
+
+        Returns:
+            pandas.Series: New feature generated.
+        """
+        pband = (self._close - self._tp_low) / (self._tp_high - self._tp_low)
+        pband = self._check_fillna(pband, value=0)
+        return pd.Series(pband, name='bbipband')
+
     def keltner_channel_hband_indicator(self) -> pd.Series:
         """Keltner Channel Indicator Crossing High Band (binary)
 
@@ -429,7 +463,7 @@ def bollinger_lband_indicator(close, n=20, ndev=2, fillna=False):
     return indicator.bollinger_hband_indicator()
 
 
-def keltner_channel_central(high, low, close, n=10, fillna=False):
+def keltner_channel_mband(high, low, close, n=10, fillna=False):
     """Keltner channel (KC)
 
     Showing a simple moving average line (central) of typical price.
@@ -447,7 +481,7 @@ def keltner_channel_central(high, low, close, n=10, fillna=False):
         pandas.Series: New feature generated.
     """
     indicator = KeltnerChannel(high=high, low=low, close=close, n=n, fillna=False)
-    return indicator.keltner_channel_central()
+    return indicator.keltner_channel_mband()
 
 
 def keltner_channel_hband(high, low, close, n=10, fillna=False):