pandas-ta/tests/test_indicator_volume.py

# -*- coding: utf-8 -*-
from unittest import TestCase, skip
import pandas.testing as pdt
from pandas import DataFrame, Series

import talib as tal

from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD
from .context import pandas_ta


class TestVolume(TestCase):
    @classmethod
    def setUpClass(cls):
        cls.data = sample_data
        cls.data.columns = cls.data.columns.str.lower()
        cls.open = cls.data["open"]
        cls.high = cls.data["high"]
        cls.low = cls.data["low"]
        cls.close = cls.data["close"]
        if "volume" in cls.data.columns:
            cls.volume_ = cls.data["volume"]

    @classmethod
    def tearDownClass(cls):
        del cls.open
        del cls.high
        del cls.low
        del cls.close
        if hasattr(cls, "volume"):
            del cls.volume_
        del cls.data

    def setUp(self): pass
    def tearDown(self): pass


    def test_ad(self):
        """Volume: AD"""
        result = pandas_ta.ad(self.high, self.low, self.close, self.volume_, talib=False)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "AD")

        try:
            expected = tal.AD(self.high, self.low, self.close, self.volume_)
            pdt.assert_series_equal(result, expected, check_names=False)
        except AssertionError:
            try:
                corr = pandas_ta.utils.df_error_analysis(result, expected)
                self.assertGreater(corr, CORRELATION_THRESHOLD)
            except Exception as ex:
                error_analysis(result, CORRELATION, ex)

        result = pandas_ta.ad(self.high, self.low, self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "AD")

    def test_ad_open(self):
        """Volume: AD (Open)"""
        result = pandas_ta.ad(self.high, self.low, self.close, self.volume_, self.open)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "ADo")

    def test_adosc(self):
        """Volume: ADOSC"""
        result = pandas_ta.adosc(self.high, self.low, self.close, self.volume_, talib=False)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "ADOSC_3_10")

        try:
            expected = tal.ADOSC(self.high, self.low, self.close, self.volume_)
            pdt.assert_series_equal(result, expected, check_names=False)
        except AssertionError:
            try:
                corr = pandas_ta.utils.df_error_analysis(result, expected)
                self.assertGreater(corr, CORRELATION_THRESHOLD)
            except Exception as ex:
                error_analysis(result, CORRELATION, ex)

        result = pandas_ta.adosc(self.high, self.low, self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "ADOSC_3_10")

    def test_aobv(self):
        """Volume: AOBV"""
        result = pandas_ta.aobv(self.close, self.volume_)
        self.assertIsInstance(result, DataFrame)
        self.assertEqual(result.name, "AOBVe_4_12_2_2_2")

    def test_cmf(self):
        """Volume: CMF"""
        result = pandas_ta.cmf(self.high, self.low, self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "CMF_20")

    def test_efi(self):
        """Volume: EFI"""
        result = pandas_ta.efi(self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "EFI_13")

    def test_eom(self):
        """Volume: EOM"""
        result = pandas_ta.eom(self.high, self.low, self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "EOM_14_100000000")

    # @skip
    def test_kvo(self):
        """Volume: KVO"""
        result = pandas_ta.kvo(self.high, self.low, self.close, self.volume_)
        if result is not None:
            self.assertIsInstance(result, DataFrame)
            self.assertEqual(result.name, "KVO_34_55_13")

    def test_mfi(self):
        """Volume: MFI"""
        result = pandas_ta.mfi(self.high, self.low, self.close, self.volume_, talib=False)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "MFI_14")

        try:
            expected = tal.MFI(self.high, self.low, self.close, self.volume_)
            pdt.assert_series_equal(result, expected, check_names=False)
        except AssertionError:
            try:
                corr = pandas_ta.utils.df_error_analysis(result, expected)
                self.assertGreater(corr, CORRELATION_THRESHOLD)
            except Exception as ex:
                error_analysis(result, CORRELATION, ex)

        result = pandas_ta.mfi(self.high, self.low, self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "MFI_14")

    def test_nvi(self):
        """Volume: NVI"""
        result = pandas_ta.nvi(self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "NVI_1")

    def test_obv(self):
        """Volume: OBV"""
        result = pandas_ta.obv(self.close, self.volume_, talib=False)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "OBV")

        try:
            expected = tal.OBV(self.close, self.volume_)
            pdt.assert_series_equal(result, expected, check_names=False)
        except AssertionError:
            try:
                corr = pandas_ta.utils.df_error_analysis(result, expected)
                self.assertGreater(corr, CORRELATION_THRESHOLD)
            except Exception as ex:
                error_analysis(result, CORRELATION, ex)

        result = pandas_ta.obv(self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "OBV")

    def test_pvi(self):
        """Volume: PVI"""
        result = pandas_ta.pvi(self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "PVI_1")

    def test_pvol(self):
        """Volume: PVOL"""
        result = pandas_ta.pvol(self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "PVOL")

    def test_pvr(self):
        """Volume: PVR"""
        result = pandas_ta.pvr(self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "PVR")
        # sample indicator values from SPY
        # self.assertEqual(result[0], 1)
        # self.assertEqual(result[1], 3)
        # self.assertEqual(result[4], 2)
        # self.assertEqual(result[6], 4)

    def test_pvt(self):
        """Volume: PVT"""
        result = pandas_ta.pvt(self.close, self.volume_)
        self.assertIsInstance(result, Series)
        self.assertEqual(result.name, "PVT")

    def test_vp(self):
        """Volume: VP"""
        result = pandas_ta.vp(self.close, self.volume_)
        self.assertIsInstance(result, DataFrame)
        self.assertEqual(result.name, "VP_10")

    def test_wb_tsv(self):
        """Volume: WB TSV"""
        result = pandas_ta.wb_tsv(self.close, self.volume_)
        self.assertIsInstance(result, DataFrame)
        self.assertEqual(result.name, "TSV_18_10")