PandasTA 源码解析(二十一)(4)

本文涉及的产品
函数计算FC,每月15万CU 3个月
简介: PandasTA 源码解析(二十一)

PandasTA 源码解析(二十一)(3)https://developer.aliyun.com/article/1506317

.\pandas-ta\tests\test_indicator_volume.py

# 从.config中导入错误分析、示例数据、相关性、相关性阈值、详细模式
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
# 从.context中导入pandas_ta
from .context import pandas_ta
# 导入TestCase和skip
from unittest import TestCase, skip
# 导入pandas测试工具
import pandas.testing as pdt
# 导入DataFrame和Series
from pandas import DataFrame, Series
# 导入talib库,并重命名为tal
import talib as tal
# 定义测试Volume的测试类
class TestVolume(TestCase):
    # 设置测试类的一些初始属性
    @classmethod
    def setUpClass(cls):
        cls.data = sample_data
        # 将列名转换为小写
        cls.data.columns = cls.data.columns.str.lower()
        # 设置测试数据的open、high、low、close列
        cls.open = cls.data["open"]
        cls.high = cls.data["high"]
        cls.low = cls.data["low"]
        cls.close = cls.data["close"]
        # 如果数据中包含volume列,则设置volume_
        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
        # 如果存在volume属性,则删除
        if hasattr(cls, "volume"):
            del cls.volume_
        del cls.data
    # 设置测试方法的setUp方法
    def setUp(self): pass
    # 设置测试方法的tearDown方法
    def tearDown(self): pass
    # 测试ad方法
    def test_ad(self):
        # 调用pandas_ta中的ad方法,不使用talib
        result = pandas_ta.ad(self.high, self.low, self.close, self.volume_, talib=False)
        # 检查返回结果是否为Series类型
        self.assertIsInstance(result, Series)
        # 检查返回结果的名称是否为"AD"
        self.assertEqual(result.name, "AD")
        # 尝试使用talib计算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, col=CORRELATION)
                # 检查相关性是否大于相关性阈值
                self.assertGreater(corr, CORRELATION_THRESHOLD)
            except Exception as ex:
                # 如果出现异常,则进行错误分析
                error_analysis(result, CORRELATION, ex)
        # 再次调用pandas_ta中的ad方法,不使用talib
        result = pandas_ta.ad(self.high, self.low, self.close, self.volume_)
        # 检查返回结果是否为Series类型
        self.assertIsInstance(result, Series)
        # 检查返回结果的名称是否为"AD"
    # 测试ad_open方法
    def test_ad_open(self):
        # 调用pandas_ta中的ad方法,不使用talib
        result = pandas_ta.ad(self.high, self.low, self.close, self.volume_, self.open)
        # 检查返回结果是否为Series类型
        self.assertIsInstance(result, Series)
        # 检查返回结果的名称是否为"ADo"
    # 测试adosc方法
    def test_adosc(self):
        # 调用pandas_ta中的adosc方法,不使用talib
        result = pandas_ta.adosc(self.high, self.low, self.close, self.volume_, talib=False)
        # 检查返回结果是否为Series类型
        self.assertIsInstance(result, Series)
        # 检查返回结果的名称是否为"ADOSC_3_10"
        # 尝试使用talib计算ADOSC指标并检查结果是否一致,不检查名称
        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, col=CORRELATION)
                # 检查相关性是否大于相关性阈值
                self.assertGreater(corr, CORRELATION_THRESHOLD)
            except Exception as ex:
                # 如果出现异常,则进行错误分析
                error_analysis(result, CORRELATION, ex)
        # 再次调用pandas_ta中的adosc方法,不使用talib
        result = pandas_ta.adosc(self.high, self.low, self.close, self.volume_)
        # 检查返回结果是否为Series类型
        self.assertIsInstance(result, Series)
        # 检查返回结果的名称是否为"ADOSC_3_10"
    # 测试aobv方法
    def test_aobv(self):
        # 调用pandas_ta中的aobv方法
        result = pandas_ta.aobv(self.close, self.volume_)
        # 检查返回结果是否为DataFrame类型
        self.assertIsInstance(result, DataFrame)
        # 检查返回结果的名称是否为"AOBVe_4_12_2_2_2"
    # 测试 CMF 指标计算函数
    def test_cmf(self):
        # 调用 pandas_ta 库的 CMF 函数计算结果
        result = pandas_ta.cmf(self.high, self.low, self.close, self.volume_)
        # 断言结果类型为 Series
        self.assertIsInstance(result, Series)
        # 断言结果的名称为 "CMF_20"
        self.assertEqual(result.name, "CMF_20")
    # 测试 EFI 指标计算函数
    def test_efi(self):
        # 调用 pandas_ta 库的 EFI 函数计算结果
        result = pandas_ta.efi(self.close, self.volume_)
        # 断言结果类型为 Series
        self.assertIsInstance(result, Series)
        # 断言结果的名称为 "EFI_13"
        self.assertEqual(result.name, "EFI_13")
    # 测试 EOM 指标计算函数
    def test_eom(self):
        # 调用 pandas_ta 库的 EOM 函数计算结果
        result = pandas_ta.eom(self.high, self.low, self.close, self.volume_)
        # 断言结果类型为 Series
        self.assertIsInstance(result, Series)
        # 断言结果的名称为 "EOM_14_100000000"
        self.assertEqual(result.name, "EOM_14_100000000")
    # 测试 KVO 指标计算函数
    def test_kvo(self):
        # 调用 pandas_ta 库的 KVO 函数计算结果
        result = pandas_ta.kvo(self.high, self.low, self.close, self.volume_)
        # 断言结果类型为 DataFrame
        self.assertIsInstance(result, DataFrame)
        # 断言结果的名称为 "KVO_34_55_13"
        self.assertEqual(result.name, "KVO_34_55_13")
    # 测试 MFI 指标计算函数
    def test_mfi(self):
        # 调用 pandas_ta 库的 MFI 函数计算结果,指定不使用 talib
        result = pandas_ta.mfi(self.high, self.low, self.close, self.volume_, talib=False)
        # 断言结果类型为 Series
        self.assertIsInstance(result, Series)
        # 断言结果的名称为 "MFI_14"
        self.assertEqual(result.name, "MFI_14")
        try:
            # 尝试使用 talib 计算 MFI,并与 pandas_ta 计算结果进行比较
            expected = tal.MFI(self.high, self.low, self.close, self.volume_)
            # 检查两个 Series 是否相等
            pdt.assert_series_equal(result, expected, check_names=False)
        except AssertionError:
            try:
                # 如果计算结果不相等,则进行错误分析并检查相关性
                corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
                self.assertGreater(corr, CORRELATION_THRESHOLD)
            except Exception as ex:
                # 如果出现异常,则进行错误分析
                error_analysis(result, CORRELATION, ex)
        # 重新使用 pandas_ta 计算 MFI 指标
        result = pandas_ta.mfi(self.high, self.low, self.close, self.volume_)
        # 断言结果类型为 Series
        self.assertIsInstance(result, Series)
        # 断言结果的名称为 "MFI_14"
        self.assertEqual(result.name, "MFI_14")
    # 测试 NVI 指标计算函数
    def test_nvi(self):
        # 调用 pandas_ta 库的 NVI 函数计算结果
        result = pandas_ta.nvi(self.close, self.volume_)
        # 断言结果类型为 Series
        self.assertIsInstance(result, Series)
        # 断言结果的名称为 "NVI_1"
        self.assertEqual(result.name, "NVI_1")
    # 测试 OBV 指标计算函数
    def test_obv(self):
        # 调用 pandas_ta 库的 OBV 函数计算结果,指定不使用 talib
        result = pandas_ta.obv(self.close, self.volume_, talib=False)
        # 断言结果类型为 Series
        self.assertIsInstance(result, Series)
        # 断言结果的名称为 "OBV"
        self.assertEqual(result.name, "OBV")
        try:
            # 尝试使用 talib 计算 OBV,并与 pandas_ta 计算结果进行比较
            expected = tal.OBV(self.close, self.volume_)
            # 检查两个 Series 是否相等
            pdt.assert_series_equal(result, expected, check_names=False)
        except AssertionError:
            try:
                # 如果计算结果不相等,则进行错误分析并检查相关性
                corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
                self.assertGreater(corr, CORRELATION_THRESHOLD)
            except Exception as ex:
                # 如果出现异常,则进行错误分析
                error_analysis(result, CORRELATION, ex)
        # 重新使用 pandas_ta 计算 OBV 指标
        result = pandas_ta.obv(self.close, self.volume_)
        # 断言结果类型为 Series
        self.assertIsInstance(result, Series)
        # 断言结果的名称为 "OBV"
        self.assertEqual(result.name, "OBV")
    # 测试 PVI 指标计算函数
    def test_pvi(self):
        # 调用 pandas_ta 库的 PVI 函数计算结果
        result = pandas_ta.pvi(self.close, self.volume_)
        # 断言结果类型为 Series
        self.assertIsInstance(result, Series)
        # 断言结果的名称为 "PVI_1"
        self.assertEqual(result.name, "PVI_1")
    # 测试 PVOL 指标计算函数
    def test_pvol(self):
        # 调用 pandas_ta 库的 PVOL 函数计算结果
        result = pandas_ta.pvol(self.close, self.volume_)
        # 断言结果类型为 Series
        self.assertIsInstance(result, Series)
        # 断言结果的名称为 "PVOL"
        self.assertEqual(result.name, "PVOL")
    # 测试 Price Volume Ratio (PVR) 指标函数
    def test_pvr(self):
        # 计算 PVR 指标
        result = pandas_ta.pvr(self.close, self.volume_)
        # 确保返回结果为 Series 类型
        self.assertIsInstance(result, Series)
        # 确保返回结果的名称为 "PVR"
        self.assertEqual(result.name, "PVR")
        # 样本指标值来自于 SPY
        self.assertEqual(result[0], 1)
        self.assertEqual(result[1], 3)
        self.assertEqual(result[4], 2)
        self.assertEqual(result[6], 4)
    # 测试 Price Volume Trend (PVT) 指标函数
    def test_pvt(self):
        # 计算 PVT 指标
        result = pandas_ta.pvt(self.close, self.volume_)
        # 确保返回结果为 Series 类型
        self.assertIsInstance(result, Series)
        # 确保返回结果的名称为 "PVT"
        self.assertEqual(result.name, "PVT")
    # 测试 Volume Price (VP) 指标函数
    def test_vp(self):
        # 计算 VP 指标
        result = pandas_ta.vp(self.close, self.volume_)
        # 确保返回结果为 DataFrame 类型
        self.assertIsInstance(result, DataFrame)
        # 确保返回结果的名称为 "VP_10"
        self.assertEqual(result.name, "VP_10")


目录
打赏
0
1
1
0
265
分享
相关文章
分片上传技术全解析:原理、优势与应用(含简单实现源码)
分片上传通过将大文件分割成多个小的片段或块,然后并行或顺序地上传这些片段,从而提高上传效率和可靠性,特别适用于大文件的上传场景,尤其是在网络环境不佳时,分片上传能有效提高上传体验。 博客不应该只有代码和解决方案,重点应该在于给出解决方案的同时分享思维模式,只有思维才能可持续地解决问题,只有思维才是真正值得学习和分享的核心要素。如果这篇博客能给您带来一点帮助,麻烦您点个赞支持一下,还可以收藏起来以备不时之需,有疑问和错误欢迎在评论区指出~
JS数组操作方法全景图,全网最全构建完整知识网络!js数组操作方法全集(实现筛选转换、随机排序洗牌算法、复杂数据处理统计等情景详解,附大量源码和易错点解析)
这些方法提供了对数组的全面操作,包括搜索、遍历、转换和聚合等。通过分为原地操作方法、非原地操作方法和其他方法便于您理解和记忆,并熟悉他们各自的使用方法与使用范围。详细的案例与进阶使用,方便您理解数组操作的底层原理。链式调用的几个案例,让您玩转数组操作。 只有锻炼思维才能可持续地解决问题,只有思维才是真正值得学习和分享的核心要素。如果这篇博客能给您带来一点帮助,麻烦您点个赞支持一下,还可以收藏起来以备不时之需,有疑问和错误欢迎在评论区指出~
深入理解HTTP/2:nghttp2库源码解析及客户端实现示例
通过解析nghttp2库的源码和实现一个简单的HTTP/2客户端示例,本文详细介绍了HTTP/2的关键特性和nghttp2的核心实现。了解这些内容可以帮助开发者更好地理解HTTP/2协议,提高Web应用的性能和用户体验。对于实际开发中的应用,可以根据需要进一步优化和扩展代码,以满足具体需求。
405 29
在线教育网课系统源码开发指南:功能设计与技术实现深度解析
在线教育网课系统是近年来发展迅猛的教育形式的核心载体,具备用户管理、课程管理、教学互动、学习评估等功能。本文从功能和技术两方面解析其源码开发,涵盖前端(HTML5、CSS3、JavaScript等)、后端(Java、Python等)、流媒体及云计算技术,并强调安全性、稳定性和用户体验的重要性。
从入门到精通:H5游戏源码开发技术全解析与未来趋势洞察
H5游戏凭借其跨平台、易传播和开发成本低的优势,近年来发展迅猛。接下来,让我们深入了解 H5 游戏源码开发的技术教程以及未来的发展趋势。
mindspeed-llm源码解析(一)preprocess_data
mindspeed-llm是昇腾模型套件代码仓,原来叫"modelLink"。这篇文章带大家阅读一下数据处理脚本preprocess_data.py(基于1.0.0分支),数据处理是模型训练的第一步,经常会用到。
183 0
【23种设计模式·全精解析 | 行为型模式篇】11种行为型模式的结构概述、案例实现、优缺点、扩展对比、使用场景、源码解析
行为型模式用于描述程序在运行时复杂的流程控制,即描述多个类或对象之间怎样相互协作共同完成单个对象都无法单独完成的任务,它涉及算法与对象间职责的分配。行为型模式分为类行为模式和对象行为模式,前者采用继承机制来在类间分派行为,后者采用组合或聚合在对象间分配行为。由于组合关系或聚合关系比继承关系耦合度低,满足“合成复用原则”,所以对象行为模式比类行为模式具有更大的灵活性。 行为型模式分为: • 模板方法模式 • 策略模式 • 命令模式 • 职责链模式 • 状态模式 • 观察者模式 • 中介者模式 • 迭代器模式 • 访问者模式 • 备忘录模式 • 解释器模式
【23种设计模式·全精解析 | 行为型模式篇】11种行为型模式的结构概述、案例实现、优缺点、扩展对比、使用场景、源码解析
【23种设计模式·全精解析 | 创建型模式篇】5种创建型模式的结构概述、实现、优缺点、扩展、使用场景、源码解析
结构型模式描述如何将类或对象按某种布局组成更大的结构。它分为类结构型模式和对象结构型模式,前者采用继承机制来组织接口和类,后者釆用组合或聚合来组合对象。由于组合关系或聚合关系比继承关系耦合度低,满足“合成复用原则”,所以对象结构型模式比类结构型模式具有更大的灵活性。 结构型模式分为以下 7 种: • 代理模式 • 适配器模式 • 装饰者模式 • 桥接模式 • 外观模式 • 组合模式 • 享元模式
【23种设计模式·全精解析 | 创建型模式篇】5种创建型模式的结构概述、实现、优缺点、扩展、使用场景、源码解析

热门文章

最新文章

推荐镜像

更多
AI助理

你好,我是AI助理

可以解答问题、推荐解决方案等

登录插画

登录以查看您的控制台资源

管理云资源
状态一览
快捷访问