标签：断言 PandasTA Series self 源码 result 解析 pandas ta
.\pandas-ta\tests\test_indicator_cycles.py
# 从config模块中导入error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE变量
# 从context模块中导入pandas_ta
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
from .context import pandas_ta
# 从unittest模块中导入TestCase类和skip函数
# 导入pandas.testing模块并重命名为pdt
import pandas.testing as pdt
# 从pandas模块中导入DataFrame和Series类
from pandas import DataFrame, Series
# 导入talib库并重命名为tal
import talib as tal
# 定义TestCycles类，继承自TestCase类
class TestCycles(TestCase):
# 设置类方法setUpClass，在测试类开始时执行
@classmethod
def setUpClass(cls):
# 设置类属性data为sample_data
cls.data = sample_data
# 将data的列名转换为小写
cls.data.columns = cls.data.columns.str.lower()
# 设置类属性open为data的"open"列
cls.open = cls.data["open"]
# 设置类属性high为data的"high"列
cls.high = cls.data["high"]
# 设置类属性low为data的"low"列
cls.low = cls.data["low"]
# 设置类属性close为data的"close"列
cls.close = cls.data["close"]
# 如果"data"的列中包含"volume"列，设置类属性volume为"data"的"volume"列
if "volume" in cls.data.columns:
cls.volume = cls.data["volume"]
# 设置类方法tearDownClass，在测试类结束时执行
@classmethod
def tearDownClass(cls):
# 删除类属性open
del cls.open
# 删除类属性high
del cls.high
# 删除类属性low
del cls.low
# 删除类属性close
del cls.close
# 如果类有volume属性，删除类属性volume
if hasattr(cls, "volume"):
del cls.volume
# 删除类属性data
del cls.data
# 设置实例方法setUp，在每个测试方法执行前执行
def setUp(self): pass
# 设置实例方法tearDown，在每个测试方法执行后执行
def tearDown(self): pass
# 定义测试方法test_ebsw
def test_ebsw(self):
# 调用pandas_ta模块中的ebsw函数，并传入close列，将结果赋给result变量
result = pandas_ta.ebsw(self.close)
# 断言result的类型为Series
self.assertIsInstance(result, Series)
# 断言result的名称为"EBSW_40_10"
self.assertEqual(result.name, "EBSW_40_10")
.\pandas-ta\tests\test_indicator_momentum.py
# 从config模块中导入error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE变量
# 从context模块中导入pandas_ta模块
from .config import error_analysis, sample_data, CORRELATION, CORRELATION_THRESHOLD, VERBOSE
from .context import pandas_ta
# 从unittest模块中导入TestCase类和skip函数
# 从pandas.testing模块中导入pdt别名
# 从pandas模块中导入DataFrame, Series类
# 导入talib模块并使用tal别名
from unittest import TestCase, skip
import pandas.testing as pdt
from pandas import DataFrame, Series
import talib as tal
# 定义TestMomentum类，继承自TestCase类
class TestMomentum(TestCase):
# 类方法setUpClass，用于设置测试类的初始状态
@classmethod
def setUpClass(cls):
# 初始化sample_data数据
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"]
# 类方法tearDownClass，用于清理测试类的状态
@classmethod
def tearDownClass(cls):
# 删除open, high, low, close数据列
del cls.open
del cls.high
del cls.low
del cls.close
# 如果存在volume数据列，则删除volume数据列
if hasattr(cls, "volume"):
del cls.volume
# 删除数据
del cls.data
# setUp方法，用于设置每个测试方法的初始状态
def setUp(self): pass
# tearDown方法，用于清理每个测试方法的状态
def tearDown(self): pass
# 测试方法test_datetime_ordered
def test_datetime_ordered(self):
# 测试datetime64索引是否有序
result = self.data.ta.datetime_ordered
self.assertTrue(result)
# 测试索引是否无序
original = self.data.copy()
reversal = original.ta.reverse
result = reversal.ta.datetime_ordered
self.assertFalse(result)
# 测试非datetime64索引
original = self.data.copy()
original.reset_index(inplace=True)
result = original.ta.datetime_ordered
self.assertFalse(result)
# 测试方法test_reverse
def test_reverse(self):
original = self.data.copy()
result = original.ta.reverse
# 检查第一个和最后一个时间是否被颠倒
self.assertEqual(result.index[-1], original.index[0])
self.assertEqual(result.index[0], original.index[-1])
# 测试方法test_ao
def test_ao(self):
result = pandas_ta.ao(self.high, self.low)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "AO_5_34")
# 测试方法test_apo
def test_apo(self):
result = pandas_ta.apo(self.close, talib=False)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "APO_12_26")
try:
expected = tal.APO(self.close)
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)
result = pandas_ta.apo(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "APO_12_26")
# 测试方法test_bias
def test_bias(self):
result = pandas_ta.bias(self.close)
self.assertIsInstance(result, Series)
self.assertEqual(result.name, "BIAS_SMA_26")
# 测试 Balance of Power 指标计算函数
def test_bop(self):
# 使用 pandas_ta 库计算 Balance of Power 指标，不使用 TA-Lib
result = pandas_ta.bop(self.open, self.high, self.low, self.close, talib=False)
# 断言结果为 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "BOP"
self.assertEqual(result.name, "BOP")
try:
# 使用 TA-Lib 计算 Balance of Power 指标
expected = tal.BOP(self.open, self.high, self.low, self.close)
# 断言计算结果与预期结果相等，忽略名称检查
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 库计算 Balance of Power 指标，使用 TA-Lib
result = pandas_ta.bop(self.open, self.high, self.low, self.close)
# 断言结果为 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "BOP"
self.assertEqual(result.name, "BOP")
# 测试 BRAR 指标计算函数
def test_brar(self):
# 使用 pandas_ta 库计算 BRAR 指标
result = pandas_ta.brar(self.open, self.high, self.low, self.close)
# 断言结果为 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果 DataFrame 的名称为 "BRAR_26"
self.assertEqual(result.name, "BRAR_26")
# 测试 CCI 指标计算函数
def test_cci(self):
# 使用 pandas_ta 库计算 CCI 指标，不使用 TA-Lib
result = pandas_ta.cci(self.high, self.low, self.close, talib=False)
# 断言结果为 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "CCI_14_0.015"
self.assertEqual(result.name, "CCI_14_0.015")
try:
# 使用 TA-Lib 计算 CCI 指标
expected = tal.CCI(self.high, self.low, self.close)
# 断言计算结果与预期结果相等，忽略名称检查
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 库计算 CCI 指标，使用 TA-Lib
result = pandas_ta.cci(self.high, self.low, self.close)
# 断言结果为 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "CCI_14_0.015"
self.assertEqual(result.name, "CCI_14_0.015")
# 测试 CFO 指标计算函数
def test_cfo(self):
# 使用 pandas_ta 库计算 CFO 指标
result = pandas_ta.cfo(self.close)
# 断言结果为 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "CFO_9"
self.assertEqual(result.name, "CFO_9")
# 测试 CG 指标计算函数
def test_cg(self):
# 使用 pandas_ta 库计算 CG 指标
result = pandas_ta.cg(self.close)
# 断言结果为 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "CG_10"
self.assertEqual(result.name, "CG_10")
# 测试 CMO 指标计算函数
def test_cmo(self):
# 使用 pandas_ta 库计算 CMO 指标
result = pandas_ta.cmo(self.close)
# 断言结果为 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "CMO_14"
self.assertEqual(result.name, "CMO_14")
try:
# 使用 TA-Lib 计算 CMO 指标
expected = tal.CMO(self.close)
# 断言计算结果与预期结果相等，忽略名称检查
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 库计算 CMO 指标，不使用 TA-Lib
result = pandas_ta.cmo(self.close, talib=False)
# 断言结果为 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "CMO_14"
self.assertEqual(result.name, "CMO_14")
# 测试 Coppock 指标计算函数
def test_coppock(self):
# 调用 coppock 函数计算结果
result = pandas_ta.coppock(self.close)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果的名称为 "COPC_11_14_10"
# 测试 CTI 指标计算函数
def test_cti(self):
# 调用 cti 函数计算结果
result = pandas_ta.cti(self.close)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果的名称为 "CTI_12"
# 测试 ER 指标计算函数
def test_er(self):
# 调用 er 函数计算结果
result = pandas_ta.er(self.close)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果的名称为 "ER_10"
# 测试 DM 指标计算函数
def test_dm(self):
# 调用 dm 函数计算结果
result = pandas_ta.dm(self.high, self.low, talib=False)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "DM_14"
try:
# 使用 tal.PLUS_DM 和 tal.MINUS_DM 计算期望结果
expected_pos = tal.PLUS_DM(self.high, self.low)
expected_neg = tal.MINUS_DM(self.high, self.low)
expecteddf = DataFrame({"DMP_14": expected_pos, "DMN_14": expected_neg})
# 比较结果和期望结果
pdt.assert_frame_equal(result, expecteddf)
except AssertionError:
try:
# 分析结果和期望结果的相关性
dmp = pandas_ta.utils.df_error_analysis(result.iloc[:,0], expecteddf.iloc[:,0], col=CORRELATION)
self.assertGreater(dmp, CORRELATION_THRESHOLD)
except Exception as ex:
# 处理异常情况
error_analysis(result, CORRELATION, ex)
try:
# 分析结果和期望结果的相关性
dmn = pandas_ta.utils.df_error_analysis(result.iloc[:,1], expecteddf.iloc[:,1], col=CORRELATION)
self.assertGreater(dmn, CORRELATION_THRESHOLD)
except Exception as ex:
# 处理异常情况
error_analysis(result, CORRELATION, ex)
# 重新调用 dm 函数计算结果
result = pandas_ta.dm(self.high, self.low)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "DM_14"
# 测试 ERI 指标计算函数
def test_eri(self):
# 调用 eri 函数计算结果
result = pandas_ta.eri(self.high, self.low, self.close)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "ERI_13"
# 测试 Fisher 指标计算函数
def test_fisher(self):
# 调用 fisher 函数计算结果
result = pandas_ta.fisher(self.high, self.low)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "FISHERT_9_1"
# 测试 Inertia 指标计算函数
def test_inertia(self):
# 调用 inertia 函数计算结果
result = pandas_ta.inertia(self.close)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果的名称为 "INERTIA_20_14"
# 调用 inertia 函数计算结果，使用高、低价数据，开启 refined 参数
result = pandas_ta.inertia(self.close, self.high, self.low, refined=True)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断���结果的名称为 "INERTIAr_20_14"
# 调用 inertia 函数计算结果，使用高、低价数据，开启 thirds 参数
result = pandas_ta.inertia(self.close, self.high, self.low, thirds=True)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果的名称为 "INERTIAt_20_14"
# 测试 KDJ 指标计算函数
def test_kdj(self):
# 调用 kdj 函数计算结果
result = pandas_ta.kdj(self.high, self.low, self.close)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "KDJ_9_3"
# 测试 KST 指标计算函数
def test_kst(self):
# 调用 kst 函数计算结果
result = pandas_ta.kst(self.close)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "KST_10_15_20_30_10_10_10_15_9"
# 测试 MACD 指标计算函数
def test_macd(self):
# 使用 pandas_ta 库计算 MACD 指标
result = pandas_ta.macd(self.close, talib=False)
# 断言结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果 DataFrame 的名称是 "MACD_12_26_9"
self.assertEqual(result.name, "MACD_12_26_9")
# 尝试使用 talib 库计算 MACD 指标，并与 pandas_ta 的结果比较
try:
# 使用 talib 库计算 MACD 指标
expected = tal.MACD(self.close)
# 将 talib 计算的 MACD 结果转换为 DataFrame
expecteddf = DataFrame({"MACD_12_26_9": expected[0], "MACDh_12_26_9": expected[2], "MACDs_12_26_9": expected[1]})
# 断言 pandas_ta 计算的结果与 talib 计算的结果相等
pdt.assert_frame_equal(result, expecteddf)
except AssertionError:
# 如果结果不相等，则进行进一步分析
try:
# 计算 MACD 指标数据的相关性
macd_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 0], expecteddf.iloc[:, 0], col=CORRELATION)
# 断言相关性大于预设阈值
self.assertGreater(macd_corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 如果出现异常，则进行错误分析
error_analysis(result.iloc[:, 0], CORRELATION, ex)
try:
# 分析历史数据的相关性
history_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 1], expecteddf.iloc[:, 1], col=CORRELATION)
# 断言相关性大于预设阈值
self.assertGreater(history_corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 如果出现异常，则进行错误分析
error_analysis(result.iloc[:, 1], CORRELATION, ex, newline=False)
try:
# 分析信号数据的相关性
signal_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 2], expecteddf.iloc[:, 2], col=CORRELATION)
# 断言相关性大于预设阈值
self.assertGreater(signal_corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 如果出现异常，则进行错误分析
error_analysis(result.iloc[:, 2], CORRELATION, ex, newline=False)
# 重新使用 pandas_ta 库计算 MACD 指标
result = pandas_ta.macd(self.close)
# 断言结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果 DataFrame 的名称是 "MACD_12_26_9"
self.assertEqual(result.name, "MACD_12_26_9")
# 测试 MACD 指标计算函数（带 asmode 参数）
def test_macdas(self):
# 使用 pandas_ta 库计算 MACD 指标（带 asmode 参数）
result = pandas_ta.macd(self.close, asmode=True)
# 断言结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果 DataFrame 的名称是 "MACDAS_12_26_9"
self.assertEqual(result.name, "MACDAS_12_26_9")
# 测试动量指标计算函数
def test_mom(self):
# 使用 pandas_ta 库计算动量指标
result = pandas_ta.mom(self.close, talib=False)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称是 "MOM_10"
self.assertEqual(result.name, "MOM_10")
# 尝试使用 talib 库计算动量指标，并与 pandas_ta 的结果比较
try:
# 使用 talib 库计算动量指标
expected = tal.MOM(self.close)
# 断言 pandas_ta 计算的结果与 talib 计算的结果相等
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 库计算动量指标
result = pandas_ta.mom(self.close)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称是 "MOM_10"
self.assertEqual(result.name, "MOM_10")
# 测试价格振荡器指标计算函数
def test_pgo(self):
# 使用 pandas_ta 库计算价格振荡器指标
result = pandas_ta.pgo(self.high, self.low, self.close)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称是 "PGO_14"
self.assertEqual(result.name, "PGO_14")
# 测试基于价格的振荡器（Price Percentage Oscillator，PPO），设置 talib=False 表示不使用 talib
def test_ppo(self):
# 调用 pandas_ta 库中的 PPO 函数，计算 PPO 指标
result = pandas_ta.ppo(self.close, talib=False)
# 断言返回结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果 DataFrame 的名称为 "PPO_12_26_9"
self.assertEqual(result.name, "PPO_12_26_9")
try:
# 尝试使用 talib 计算 PPO 指标
expected = tal.PPO(self.close)
# 对比结果与 talib 计算结果
pdt.assert_series_equal(result["PPO_12_26_9"], expected, check_names=False)
except AssertionError:
try:
# 若对比失败，则进行误差分析
corr = pandas_ta.utils.df_error_analysis(result["PPO_12_26_9"], expected, col=CORRELATION)
# 断言相关性大于阈值
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 若出现异常，则输出错误分析
error_analysis(result["PPO_12_26_9"], CORRELATION, ex)
# 重新计算 PPO 指标（使用 pandas_ta 默认设置）
result = pandas_ta.ppo(self.close)
# 断言返回结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果 DataFrame 的名称为 "PPO_12_26_9"
self.assertEqual(result.name, "PPO_12_26_9")
# 测试趋势状态线（Price Speed and Length，PSL）指标
def test_psl(self):
# 调用 pandas_ta 库中的 PSL 函数，计算 PSL 指标
result = pandas_ta.psl(self.close)
# 断言返回结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "PSL_12"
self.assertEqual(result.name, "PSL_12")
# 测试量价震荡器（Price Volume Oscillator，PVO）指标
def test_pvo(self):
# 调用 pandas_ta 库中的 PVO 函数，计算 PVO 指标
result = pandas_ta.pvo(self.volume)
# 断言返回结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果 DataFrame 的名称为 "PVO_12_26_9"
self.assertEqual(result.name, "PVO_12_26_9")
# 测试 QQE 指标
def test_qqe(self):
# 调用 pandas_ta 库中的 QQE 函数，计算 QQE 指标
result = pandas_ta.qqe(self.close)
# 断言返回结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果 DataFrame 的名称为 "QQE_14_5_4.236"
self.assertEqual(result.name, "QQE_14_5_4.236")
# 测试变动率指标（Rate of Change，ROC）
def test_roc(self):
# 测试不使用 talib 计算 ROC 指标
result = pandas_ta.roc(self.close, talib=False)
# 断言返回结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "ROC_10"
self.assertEqual(result.name, "ROC_10")
try:
# 尝试使用 talib 计算 ROC 指标
expected = tal.ROC(self.close)
# 对比结果与 talib 计算结果
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)
# 重新计算 ROC 指标（使用 pandas_ta 默认设置）
result = pandas_ta.roc(self.close)
# 断言返回结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "ROC_10"
self.assertEqual(result.name, "ROC_10")
# 测试相对强弱指标（Relative Strength Index，RSI）
def test_rsi(self):
# 测试不使用 talib 计算 RSI 指标
result = pandas_ta.rsi(self.close, talib=False)
# 断言返回结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "RSI_14"
self.assertEqual(result.name, "RSI_14")
try:
# 尝试使用 talib 计算 RSI 指标
expected = tal.RSI(self.close)
# 对比结果与 talib 计算结果
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)
# 重新计算 RSI 指标（使用 pandas_ta 默认设置）
result = pandas_ta.rsi(self.close)
# 断言返回结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "RSI_14"
self.assertEqual(result.name, "RSI_14")
# 测试相对强弱指数（Relative Strength Index Smoothed，RSX）
def test_rsx(self):
# 调用 pandas_ta 库中的 RSX 函数，计算 RSX 指标
result = pandas_ta.rsx(self.close)
# 断言返回结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "RSX_14"
self.assertEqual(result.name, "RSX_14")
# 测试 RVGI 指标计算函数
def test_rvgi(self):
# 调用 rvgi 函数计算结果
result = pandas_ta.rvgi(self.open, self.high, self.low, self.close)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "RVGI_14_4"
self.assertEqual(result.name, "RVGI_14_4")
# 测试斜率指标计算函数
def test_slope(self):
# 调用 slope 函数计算结果
result = pandas_ta.slope(self.close)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果名称为 "SLOPE_1"
self.assertEqual(result.name, "SLOPE_1")
# 测试斜率指标计算函数，返回角度值
def test_slope_as_angle(self):
# 调用 slope 函数计算结果，返回角度值
result = pandas_ta.slope(self.close, as_angle=True)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果名称为 "ANGLEr_1"
self.assertEqual(result.name, "ANGLEr_1")
# 测试斜率指标计算函数，返回角度值并转换为度数
def test_slope_as_angle_to_degrees(self):
# 调用 slope 函数计算结果，返回角度值并转换为度数
result = pandas_ta.slope(self.close, as_angle=True, to_degrees=True)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果名称为 "ANGLEd_1"
self.assertEqual(result.name, "ANGLEd_1")
# 测试 SMI 指标计算函数
def test_smi(self):
# 调用 smi 函数计算结果
result = pandas_ta.smi(self.close)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "SMI_5_20_5"
self.assertEqual(result.name, "SMI_5_20_5")
# 断言结果列数为 3
self.assertEqual(len(result.columns), 3)
# 测试 SMI 指标计算函数，设置 scalar 参数
def test_smi_scalar(self):
# 调用 smi 函数计算结果，设置 scalar 参数为 10
result = pandas_ta.smi(self.close, scalar=10)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "SMI_5_20_5_10.0"
self.assertEqual(result.name, "SMI_5_20_5_10.0")
# 断言结果列数为 3
self.assertEqual(len(result.columns), 3)
# 测试 Squeeze 指标计算函数
def test_squeeze(self):
# 调用 squeeze 函数计算结果
result = pandas_ta.squeeze(self.high, self.low, self.close)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "SQZ_20_2.0_20_1.5"
# 调用 squeeze 函数计算结果，设置 tr 参数为 False
result = pandas_ta.squeeze(self.high, self.low, self.close, tr=False)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "SQZhlr_20_2.0_20_1.5"
# 调用 squeeze 函数计算结果，设置 lazybear 参数为 True
result = pandas_ta.squeeze(self.high, self.low, self.close, lazybear=True)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "SQZ_20_2.0_20_1.5_LB"
# 调用 squeeze 函数计算结果，设置 tr 和 lazybear 参数为 True
result = pandas_ta.squeeze(self.high, self.low, self.close, tr=False, lazybear=True)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "SQZhlr_20_2.0_20_1.5_LB"
# 测试 Squeeze Pro 指标计算函数
def test_squeeze_pro(self):
# 调用 squeeze_pro 函数计算结果
result = pandas_ta.squeeze_pro(self.high, self.low, self.close)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "SQZPRO_20_2.0_20_2_1.5_1"
# 调用 squeeze_pro 函数计算结果，设置 tr 参数为 False
result = pandas_ta.squeeze_pro(self.high, self.low, self.close, tr=False)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "SQZPROhlr_20_2.0_20_2_1.5_1"
# 调用 squeeze_pro 函数计算结果，设置各参数值
result = pandas_ta.squeeze_pro(self.high, self.low, self.close, 20, 2, 20, 3, 2, 1)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "SQZPRO_20_2.0_20_3.0_2.0_1.0"
# 调用 squeeze_pro 函数计算结果，设置各参数值和 tr 参数为 False
result = pandas_ta.squeeze_pro(self.high, self.low, self.close, 20, 2, 20, 3, 2, 1, tr=False)
# 断言结果类型为 DataFrame
self.assertIsInstance(result, DataFrame)
# 断言结果名称为 "SQZPROhlr_20_2.0_20_3.0_2.0_1.0"
# 测试 Smoothed Triple Exponential Moving Average (STC) 函数
def test_stc(self):
# 使用 pandas_ta 库中的 stc 函数计算结果
result = pandas_ta.stc(self.close)
# 断言结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "STC_10_12_26_0.5"
self.assertEqual(result.name, "STC_10_12_26_0.5")
# 测试 Stochastic Oscillator (STOCH) 函数
def test_stoch(self):
# TV Correlation
# 使用 pandas_ta 库中的 stoch 函数计算结果
result = pandas_ta.stoch(self.high, self.low, self.close)
# 断言结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "STOCH_14_3_3"
self.assertEqual(result.name, "STOCH_14_3_3")
try:
# 使用 talib 库中的 STOCH 函数计算预期结果
expected = tal.STOCH(self.high, self.low, self.close, 14, 3, 0, 3, 0)
# 构建预期结果的 DataFrame
expecteddf = DataFrame({"STOCHk_14_3_0_3_0": expected[0], "STOCHd_14_3_0_3": expected[1]})
# 断言结果与预期结果相等
pdt.assert_frame_equal(result, expecteddf)
except AssertionError:
try:
# 计算结果与预期结果的相关性
stochk_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 0], expecteddf.iloc[:, 0], col=CORRELATION)
# 断言相关性大于阈值
self.assertGreater(stochk_corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 如果相关性不符合要求，进行错误分析
error_analysis(result.iloc[:, 0], CORRELATION, ex)
try:
# 计算结果与预期结果的相关性
stochd_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 1], expecteddf.iloc[:, 1], col=CORRELATION)
# 断言相关性大于阈值
self.assertGreater(stochd_corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 如果相关性不符合要求，进行错误分析
error_analysis(result.iloc[:, 1], CORRELATION, ex, newline=False)
# 测试 Stochastic RSI (STOCHRSI) 函数
def test_stochrsi(self):
# TV Correlation
# 使用 pandas_ta 库中的 stochrsi 函数计算结果
result = pandas_ta.stochrsi(self.close)
# 断言结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "STOCHRSI_14_14_3_3"
self.assertEqual(result.name, "STOCHRSI_14_14_3_3")
try:
# 使用 talib 库中的 STOCHRSI 函数计算预期结果
expected = tal.STOCHRSI(self.close, 14, 14, 3, 0)
# 构建预期结果的 DataFrame
expecteddf = DataFrame({"STOCHRSIk_14_14_0_3": expected[0], "STOCHRSId_14_14_3_0": expected[1]})
# 断言结果与预期结果相等
pdt.assert_frame_equal(result, expecteddf)
except AssertionError:
try:
# 计算结果与预期结果的相关性
stochrsid_corr = pandas_ta.utils.df_error_analysis(result.iloc[:, 0], expecteddf.iloc[:, 1], col=CORRELATION)
# 断言相关性大于阈值
self.assertGreater(stochrsid_corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 如果相关性不符合要求，进行错误分析
error_analysis(result.iloc[:, 0], CORRELATION, ex, newline=False)
# 跳过测试 TS Sequential 函数
@skip
def test_td_seq(self):
"""TS Sequential: Working but SLOW implementation"""
# 使用 pandas_ta 库中的 td_seq 函数计算结果（已跳过）
result = pandas_ta.td_seq(self.close)
# 断言结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "TD_SEQ"
# 测试 Triple Exponential Moving Average (TRIX) 函数
def test_trix(self):
# 使用 pandas_ta 库中的 trix 函数计算结果
result = pandas_ta.trix(self.close)
# 断言结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "TRIX_30_9"
# 测试 True Strength Index (TSI) 函数
def test_tsi(self):
# 使用 pandas_ta 库中的 tsi 函数计算结果
result = pandas_ta.tsi(self.close)
# 断言结果是 DataFrame 类型
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "TSI_13_25_13"
# 测试 `uo` 函数的单元测试
def test_uo(self):
# 使用 pandas_ta 库的 UO 函数计算结果
result = pandas_ta.uo(self.high, self.low, self.close, talib=False)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果的名称为 "UO_7_14_28"
self.assertEqual(result.name, "UO_7_14_28")
try:
# 使用 TA-Lib 库的 ULTOSC 函数计算预期结果
expected = tal.ULTOSC(self.high, self.low, self.close)
# 比较结果和预期结果，不检查名称
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 库的 UO 函数计算结果（默认情况下使用 TA-Lib）
result = pandas_ta.uo(self.high, self.low, self.close)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果的名称为 "UO_7_14_28"
self.assertEqual(result.name, "UO_7_14_28")
# 测试 `willr` 函数的单元测试
def test_willr(self):
# 使用 pandas_ta 库的 WILLR 函数计算结果
result = pandas_ta.willr(self.high, self.low, self.close, talib=False)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果的名称为 "WILLR_14"
self.assertEqual(result.name, "WILLR_14")
try:
# 使用 TA-Lib 库的 WILLR 函数计算预期结果
expected = tal.WILLR(self.high, self.low, self.close)
# 比较结果和预期结果，不检查名称
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 库的 WILLR 函数计算结果（默认情况下使用 TA-Lib）
result = pandas_ta.willr(self.high, self.low, self.close)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果的名称为 "WILLR_14"
self.assertEqual(result.name, "WILLR_14")
.\pandas-ta\tests\test_indicator_overlap.py
# 从当前目录的 config 模块中导入 CORRELATION, CORRELATION_THRESHOLD, error_analysis, sample_data, VERBOSE 变量
from .config import CORRELATION, CORRELATION_THRESHOLD, error_analysis, sample_data, VERBOSE
# 从当前目录的 context 模块中导入 pandas_ta 模块
from .context import pandas_ta
# 导入 TestCase 类
from unittest import TestCase
# 导入 pandas.testing 模块，并重命名为 pdt
import pandas.testing as pdt
# 导入 DataFrame, Series 类
from pandas import DataFrame, Series
# 导入 talib 库，并重命名为 tal
import talib as tal
# 定义测试类 TestOverlap，继承自 TestCase 类
class TestOverlap(TestCase):
# 设置类方法 setUpClass，用于设置测试类的数据
@classmethod
def setUpClass(cls):
# 设置类属性 data 为 sample_data
cls.data = sample_data
# 将数据列名转换为小写
cls.data.columns = cls.data.columns.str.lower()
# 设置类属性 open 为 data 中的 "open" 列
cls.open = cls.data["open"]
# 设置类属性 high 为 data 中的 "high" 列
cls.high = cls.data["high"]
# 设置类属性 low 为 data 中的 "low" 列
cls.low = cls.data["low"]
# 设置类属性 close 为 data 中的 "close" 列
cls.close = cls.data["close"]
# 如果数据中包含 "volume" 列，则设置类属性 volume 为 data 中的 "volume" 列
if "volume" in cls.data.columns:
cls.volume = cls.data["volume"]
# 设置类方法 tearDownClass，用于清理测试类的数据
@classmethod
def tearDownClass(cls):
# 删除类属性 open
del cls.open
# 删除类属性 high
del cls.high
# 删除类属性 low
del cls.low
# 删除类属性 close
del cls.close
# 如果类属性中存在 volume，则删除 volume
if hasattr(cls, "volume"):
del cls.volume
# 删除类属性 data
del cls.data
# 设置实例方法 setUp，用于测试方法的初始化
def setUp(self): pass
# 设置实例方法 tearDown，用于测试方法的清理
def tearDown(self): pass
# 定义测试方法 test_alma，测试 alma 函数
def test_alma(self):
# 调用 pandas_ta.alma 函数，传入 close 列作为参数
result = pandas_ta.alma(self.close)# , length=None, sigma=None, distribution_offset=)
# 断言 result 是 Series 类型
self.assertIsInstance(result, Series)
# 断言 result 的名称为 "ALMA_10_6.0_0.85"
self.assertEqual(result.name, "ALMA_10_6.0_0.85")
# 定义测试方法 test_dema，测试 dema 函数
def test_dema(self):
# 调用 pandas_ta.dema 函数，传入 close 列和 talib=False 参数
result = pandas_ta.dema(self.close, talib=False)
# 断言 result 是 Series 类型
self.assertIsInstance(result, Series)
# 断言 result 的名称为 "DEMA_10"
self.assertEqual(result.name, "DEMA_10")
try:
# 使用 talib 计算预期值
expected = tal.DEMA(self.close, 10)
# 使用 pandas.testing 模块的 assert_series_equal 函数比较 result 和 expected，不检查名称
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError:
try:
# 调用 pandas_ta.utils.df_error_analysis 函数计算 result 和 expected 之间的相关性
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
# 断言相关性大于 CORRELATION_THRESHOLD
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 如果计算相关性时出现异常，则调用 error_analysis 函数记录异常信息
error_analysis(result, CORRELATION, ex)
# 再次调用 pandas_ta.dema 函数，传入 close 列，默认参数
result = pandas_ta.dema(self.close)
# 断言 result 是 Series 类型
self.assertIsInstance(result, Series)
# 断言 result 的名称为 "DEMA_10"
self.assertEqual(result.name, "DEMA_10")
# 测试指数移动平均值函数
def test_ema(self):
# 调用指数移动平均值函数，计算不带SMA的EMA
result = pandas_ta.ema(self.close, presma=False)
# 断言结果为Series类型
self.assertIsInstance(result, Series)
# 断言结果的名称为"EMA_10"
self.assertEqual(result.name, "EMA_10")
# 尝试使用talib库计算EMA，并进行结果对比
try:
# 期望值通过talib库计算
expected = tal.EMA(self.close, 10)
# 断言两个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)
# 调用指数移动平均值函数，计算不使用talib的EMA
result = pandas_ta.ema(self.close, talib=False)
# 断言结果为Series类型
self.assertIsInstance(result, Series)
# 断言结果的名称为"EMA_10"
self.assertEqual(result.name, "EMA_10")
# 尝试断言两个Series相等，忽略名称检查
try:
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)
# 调用指数移动平均值函数，计算默认的EMA
result = pandas_ta.ema(self.close)
# 断言结果为Series类型
self.assertIsInstance(result, Series)
# 断言结果的名称为"EMA_10"
self.assertEqual(result.name, "EMA_10")
# 测试前加权移动平均值函数
def test_fwma(self):
# 调用前加权移动平均值函数
result = pandas_ta.fwma(self.close)
# 断言结果为Series类型
self.assertIsInstance(result, Series)
# 断言结果的名称为"FWMA_10"
self.assertEqual(result.name, "FWMA_10")
# 测试高低价通道指标函数
def test_hilo(self):
# 调用高低价通道指标函数
result = pandas_ta.hilo(self.high, self.low, self.close)
# 断言结果为DataFrame类型
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为"HILO_13_21"
self.assertEqual(result.name, "HILO_13_21")
# 测试高低价中值函数
def test_hl2(self):
# 调用高低价中值函数
result = pandas_ta.hl2(self.high, self.low)
# 断言结果为Series类型
self.assertIsInstance(result, Series)
# 断言结果的名称为"HL2"
self.assertEqual(result.name, "HL2")
# 测试高低收盘价中值函数
def test_hlc3(self):
# 调用高低收盘价中值函数，不使用talib
result = pandas_ta.hlc3(self.high, self.low, self.close, talib=False)
# 断言结果为Series类型
self.assertIsInstance(result, Series)
# 断言结果的名称为"HLC3"
# 尝试使用talib库计算TYPPRICE，并进行结果对比
try:
# 期望值通过talib库计算
expected = tal.TYPPRICE(self.high, self.low, self.close)
# 断言两个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)
# 调用高低收盘价中值函数，使用talib
result = pandas_ta.hlc3(self.high, self.low, self.close)
# 断言结果为Series类型
self.assertIsInstance(result, Series)
# 断言结果的名称为"HLC3"
# 测试Hull移动平均值函数
def test_hma(self):
# 调用Hull移动平均值函数
result = pandas_ta.hma(self.close)
# 断言结果为Series类型
self.assertIsInstance(result, Series)
# 断言结果的名称为"HMA_10"
# 测试Hull加权移动平均值函数
def test_hwma(self):
# 调用Hull加权移动平均值函数
result = pandas_ta.hwma(self.close)
# 断言结果为Series类型
self.assertIsInstance(result, Series)
# 断言结果的名称为"HWMA_0.2_0.1_0.1"
# 测试 KAMA 指标计算函数
def test_kama(self):
# 调用 pandas_ta 库中的 kama 函数计算
result = pandas_ta.kama(self.close)
# 断言结果为 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "KAMA_10_2_30"
self.assertEqual(result.name, "KAMA_10_2_30")
# 测试 JMA 指标计算函数
def test_jma(self):
# 调用 pandas_ta 库中的 jma 函数计算
result = pandas_ta.jma(self.close)
# 断言结果为 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "JMA_7_0"
self.assertEqual(result.name, "JMA_7_0")
# 测试 Ichimoku 指标计算函数
def test_ichimoku(self):
# 调用 pandas_ta 库中的 ichimoku 函数计算
ichimoku, span = pandas_ta.ichimoku(self.high, self.low, self.close)
# 断言 ichimoku 结果为 DataFrame 对象
self.assertIsInstance(ichimoku, DataFrame)
# 断言 span 结果为 DataFrame 对象
self.assertIsInstance(span, DataFrame)
# 断言 ichimoku 结果的名称为 "ICHIMOKU_9_26_52"
self.assertEqual(ichimoku.name, "ICHIMOKU_9_26_52")
# 断言 span 结果的名称为 "ICHISPAN_9_26"
self.assertEqual(span.name, "ICHISPAN_9_26")
# 测试 LinReg 指标计算函数
def test_linreg(self):
# 调用 pandas_ta 库中的 linreg 函数计算，不使用 talib
result = pandas_ta.linreg(self.close, talib=False)
# 断言结果为 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "LR_14"
self.assertEqual(result.name, "LR_14")
try:
# 尝试使用 talib 进行结果比较
expected = tal.LINEARREG(self.close)
# 使用 pandas 的 assert_series_equal 检查结果
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)
# 再次调用 linreg 函数，使用 talib
result = pandas_ta.linreg(self.close)
# 断言结果为 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "LR_14"
self.assertEqual(result.name, "LR_14")
# 测试 LinReg Angle 指标计算函数
def test_linreg_angle(self):
# 调用 pandas_ta 库中的 linreg 函数计算，包括角度计算，不使用 talib
result = pandas_ta.linreg(self.close, angle=True, talib=False)
# 断言结果为 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "LRa_14"
self.assertEqual(result.name, "LRa_14")
try:
# 尝试使用 talib 进行结果比较
expected = tal.LINEARREG_ANGLE(self.close)
# 使用 pandas 的 assert_series_equal 检查结果
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)
# 再次调用 linreg 函数，包括角度计算，使用 talib
result = pandas_ta.linreg(self.close, angle=True)
# 断言结果为 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "LRa_14"
self.assertEqual(result.name, "LRa_14")
# 测试 LinReg Intercept 指标计算函数
def test_linreg_intercept(self):
# 调用 pandas_ta 库中的 linreg 函数计算，包括截距计算，不使用 talib
result = pandas_ta.linreg(self.close, intercept=True, talib=False)
# 断言结果为 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "LRb_14"
self.assertEqual(result.name, "LRb_14")
try:
# 尝试使用 talib 进行结果比较
expected = tal.LINEARREG_INTERCEPT(self.close)
# 使用 pandas 的 assert_series_equal 检查结果
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)
# 再次调用 linreg 函数，包括截距计算，使用 talib
result = pandas_ta.linreg(self.close, intercept=True)
# 断言结果为 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "LRb_14"
self.assertEqual(result.name, "LRb_14")
# 测试线性回归指标的随机性（r）计算是否正确
def test_linreg_r(self):
# 计算线性回归指标的随机性（r）
result = pandas_ta.linreg(self.close, r=True)
# 确保返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 确保返回结果的名称为 "LRr_14"
self.assertEqual(result.name, "LRr_14")
# 测试线性回归指标的斜率（slope）计算是否正确
def test_linreg_slope(self):
# 计算线性回归指标的斜率
result = pandas_ta.linreg(self.close, slope=True, talib=False)
# 确保返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 确保返回结果的名称为 "LRm_14"
self.assertEqual(result.name, "LRm_14")
try:
# 尝试使用 talib 计算线性回归指标的斜率
expected = tal.LINEARREG_SLOPE(self.close)
# 对比结果与预期结果是否相等，不检查名称
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError:
try:
# 计算结果与预期结果之间的相关性
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
# 确保相关性大于 CORRELATION_THRESHOLD
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 在出现异常时执行错误分析
error_analysis(result, CORRELATION, ex)
# 使用默认设置计算线性回归指标的斜率
result = pandas_ta.linreg(self.close, slope=True)
# 确保返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 确保返回结果的名称为 "LRm_14"
self.assertEqual(result.name, "LRm_14")
# 测试移动平均（ma）指标计算是否正确
def test_ma(self):
# 计算简单移动平均（SMA）指标
result = pandas_ta.ma()
# 确保返回结果是一个列表
self.assertIsInstance(result, list)
# 确保返回结果长度大于 0
self.assertGreater(len(result), 0)
# 计算指定类型的移动平均（EMA）指标
result = pandas_ta.ma("ema", self.close)
# 确保返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 确保返回结果的名称为 "EMA_10"
self.assertEqual(result.name, "EMA_10")
# 计算指定类型和长度的移动平均（FWMA）指标
result = pandas_ta.ma("fwma", self.close, length=15)
# 确保返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 确保返回结果的名称为 "FWMA_15"
self.assertEqual(result.name, "FWMA_15")
# 测试 MACD 指标计算是否正确
def test_mcgd(self):
# 计算 MACD 指标
result = pandas_ta.mcgd(self.close)
# 确保返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 确保返回结果的名称为 "MCGD_10"
# 测试中点（midpoint）指标计算是否正确
def test_midpoint(self):
# 计算中点（midpoint）指标
result = pandas_ta.midpoint(self.close, talib=False)
# 确保返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 确保返回结果的名称为 "MIDPOINT_2"
try:
# 尝试使用 talib 计算中点（midpoint）指标
expected = tal.MIDPOINT(self.close, 2)
# 对比结果与预期结果是否相等，不检查名称
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError:
try:
# 计算结果与预期结果之间的相关性
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
# 确保相关性大于 CORRELATION_THRESHOLD
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 在出现异常时执行错误分析
error_analysis(result, CORRELATION, ex)
# 使用默认设置计算中点（midpoint）指标
result = pandas_ta.midpoint(self.close)
# 确保返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 确保返回结果的名称为 "MIDPOINT_2"
self.assertEqual(result.name, "MIDPOINT_2")
# 测试中位数价格指标函数
def test_midprice(self):
# 调用 midprice 函数计算中位数价格
result = pandas_ta.midprice(self.high, self.low, talib=False)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "MIDPRICE_2"
self.assertEqual(result.name, "MIDPRICE_2")
try:
# 使用 Talib 库计算期望结果
expected = tal.MIDPRICE(self.high, self.low, 2)
# 断言结果与期望结果相等
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError:
try:
# 分析结果与期望结果之间的相关性
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
# 断言相关性大于 CORRELATION_THRESHOLD
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 在出现异常时进行错误分析
error_analysis(result, CORRELATION, ex)
# 未指定时间周期调用 midprice 函数
result = pandas_ta.midprice(self.high, self.low)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "MIDPRICE_2"
self.assertEqual(result.name, "MIDPRICE_2")
# 测试 OHLC4 函数
def test_ohlc4(self):
# 调用 ohlc4 函数计算 OHLC4
result = pandas_ta.ohlc4(self.open, self.high, self.low, self.close)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "OHLC4"
self.assertEqual(result.name, "OHLC4")
# 测试 PWMA 函数
def test_pwma(self):
# 调用 pwma 函数计算 PWMA
result = pandas_ta.pwma(self.close)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "PWMA_10"
self.assertEqual(result.name, "PWMA_10")
# 测试 RMA 函数
def test_rma(self):
# 调用 rma 函数计算 RMA
result = pandas_ta.rma(self.close)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "RMA_10"
self.assertEqual(result.name, "RMA_10")
# 测试 SINWMA 函数
def test_sinwma(self):
# 调用 sinwma 函数计算 SINWMA
result = pandas_ta.sinwma(self.close)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "SINWMA_14"
self.assertEqual(result.name, "SINWMA_14")
# 测试 SMA 函数
def test_sma(self):
# 不使用 Talib 库调用 sma 函数计算 SMA
result = pandas_ta.sma(self.close, talib=False)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "SMA_10"
self.assertEqual(result.name, "SMA_10")
try:
# 使用 Talib 库计算期望结果
expected = tal.SMA(self.close, 10)
# 断言结果与期望结果相等
pdt.assert_series_equal(result, expected, check_names=False)
except AssertionError:
try:
# 分析结果与期望结果之间的相关性
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
# 断言相关性大于 CORRELATION_THRESHOLD
self.assertGreater(corr, CORRELATION_THRESHOLD)
except Exception as ex:
# 在出现异常时进行错误分析
error_analysis(result, CORRELATION, ex)
# 未指定时间周期调用 sma 函数
result = pandas_ta.sma(self.close)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "SMA_10"
self.assertEqual(result.name, "SMA_10")
# 测试 SSF 函数
def test_ssf(self):
# 调用 ssf 函数计算 SSF
result = pandas_ta.ssf(self.close, poles=2)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "SSF_10_2"
self.assertEqual(result.name, "SSF_10_2")
# 再次调用 ssf 函数计算 SSF
result = pandas_ta.ssf(self.close, poles=3)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "SSF_10_3"
self.assertEqual(result.name, "SSF_10_3")
# 测试 SWMA 函数
def test_swma(self):
# 调用 swma 函数计算 SWMA
result = pandas_ta.swma(self.close)
# 断言返回结果是一个 Series 对象
self.assertIsInstance(result, Series)
# 断言结果的名称为 "SWMA_10"
self.assertEqual(result.name, "SWMA_10")
# 测试 SuperTrend 函数
def test_supertrend(self):
# 调用 supertrend 函数计算 SuperTrend
result = pandas_ta.supertrend(self.high, self.low, self.close)
# 断言返回结果是一个 DataFrame 对象
self.assertIsInstance(result, DataFrame)
# 断言结果的名称为 "SUPERT_7_3.0"
self.assertEqual(result.name, "SUPERT_7_3.0")
# 测试 T3 指标计算函数
def test_t3(self):
# 使用 pandas_ta 库计算 T3 指标，关闭使用 talib
result = pandas_ta.t3(self.close, talib=False)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "T3_10_0.7"
self.assertEqual(result.name, "T3_10_0.7")
# 尝试使用 talib 计算 T3 指标，并与预期结果比较
try:
expected = tal.T3(self.close, 10)
# 使用 pandas.testing.assert_series_equal 函数比较两个 Series，关闭名称检查
pdt.assert_series_equal(result, expected, check_names=False)
# 如果断言失败
except AssertionError:
# 尝试进行误差分析并检查相关性
try:
# 调用 pandas_ta.utils.df_error_analysis 函数分析误差
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
# 断言相关性大于指定阈值
self.assertGreater(corr, CORRELATION_THRESHOLD)
# 如果发生异常
except Exception as ex:
# 调用 error_analysis 函数处理异常
error_analysis(result, CORRELATION, ex)
# 重新使用 pandas_ta 库计算 T3 指标
result = pandas_ta.t3(self.close)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "T3_10_0.7"
self.assertEqual(result.name, "T3_10_0.7")
# 测试 TEMA 指标计算函数
def test_tema(self):
# 使用 pandas_ta 库计算 TEMA 指标，关闭使用 talib
result = pandas_ta.tema(self.close, talib=False)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "TEMA_10"
self.assertEqual(result.name, "TEMA_10")
# 尝试使用 talib 计算 TEMA 指标，并与预期结果比较
try:
expected = tal.TEMA(self.close, 10)
# 使用 pandas.testing.assert_series_equal 函数比较两个 Series，关闭名称检查
pdt.assert_series_equal(result, expected, check_names=False)
# 如果断言失败
except AssertionError:
# 尝试进行误差分析并检查相关性
try:
# 调用 pandas_ta.utils.df_error_analysis 函数分析误差
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
# 断言相关性大于指定阈值
self.assertGreater(corr, CORRELATION_THRESHOLD)
# 如果发生异常
except Exception as ex:
# 调用 error_analysis 函数处理异常
error_analysis(result, CORRELATION, ex)
# 重新使用 pandas_ta 库计算 TEMA 指标
result = pandas_ta.tema(self.close)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "TEMA_10"
self.assertEqual(result.name, "TEMA_10")
# 测试 TRIMA 指标计算函数
def test_trima(self):
# 使用 pandas_ta 库计算 TRIMA 指标，关闭使用 talib
result = pandas_ta.trima(self.close, talib=False)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "TRIMA_10"
self.assertEqual(result.name, "TRIMA_10")
# 尝试使用 talib 计算 TRIMA 指标，并与预期结果比较
try:
expected = tal.TRIMA(self.close, 10)
# 使用 pandas.testing.assert_series_equal 函数比较两个 Series，关闭名称检查
pdt.assert_series_equal(result, expected, check_names=False)
# 如果断言失败
except AssertionError:
# 尝试进行误差分析并检查相关性
try:
# 调用 pandas_ta.utils.df_error_analysis 函数分析误差
corr = pandas_ta.utils.df_error_analysis(result, expected, col=CORRELATION)
# 断言相关性大于指定阈值
self.assertGreater(corr, CORRELATION_THRESHOLD)
# 如果发生异常
except Exception as ex:
# 调用 error_analysis 函数处理异常
error_analysis(result, CORRELATION, ex)
# 重新使用 pandas_ta 库计算 TRIMA 指标
result = pandas_ta.trima(self.close)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "TRIMA_10"
self.assertEqual(result.name, "TRIMA_10")
# 测试 VIDYA 指标计算函数
def test_vidya(self):
# 使用 pandas_ta 库计算 VIDYA 指标
result = pandas_ta.vidya(self.close)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "VIDYA_14"
self.assertEqual(result.name, "VIDYA_14")
# 测试 VWAP 指标计算函数
def test_vwap(self):
# 使用 pandas_ta 库计算 VWAP 指标
result = pandas_ta.vwap(self.high, self.low, self.close, self.volume)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "VWAP_D"
self.assertEqual(result.name, "VWAP_D")
# 测试 VWMA 指标计算函数
def test_vwma(self):
# 使用 pandas_ta 库计算 VWMA 指标
result = pandas_ta.vwma(self.close, self.volume)
# 断言结果类型为 Series
self.assertIsInstance(result, Series)
# 断言结果 Series 的名称为 "VWMA_10"
self.assertEqual(result.name, "VWMA_10")
# 测试 Weighted Close Price (WCP) 函数
def test_wcp(self):
# 使用 pandas_ta 库中的 wcp 函数计算结果
result = pandas_ta.wcp(self.high, self.low, self.close, talib=False)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果的名称为 "WCP"
self.assertEqual(result.name, "WCP")
# 尝试使用 Talib 库中的 WCLPRICE 函数计算期望结果
try:
expected = tal.WCLPRICE(self.high, self.low, self.close)
# 使用 pandas 测试工具 pdt 断言两个 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 库中的 wcp 函数计算结果
result = pandas_ta.wcp(self.high, self.low, self.close)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果的名称为 "WCP"
self.assertEqual(result.name, "WCP")
# 测试 Weighted Moving Average (WMA) 函数
def test_wma(self):
# 使用 pandas_ta 库中的 wma 函数计算结果
result = pandas_ta.wma(self.close, talib=False)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果的名称为 "WMA_10"
self.assertEqual(result.name, "WMA_10")
# 尝试使用 Talib 库中的 WMA 函数计算期望结果
try:
expected = tal.WMA(self.close, 10)
# 使用 pandas 测试工具 pdt 断言两个 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 库中的 wma 函数计算结果
result = pandas_ta.wma(self.close)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果的名称为 "WMA_10"
self.assertEqual(result.name, "WMA_10")
# 测试 Zero-Lag Exponential Moving Average (ZLEMA) 函数
def test_zlma(self):
# 使用 pandas_ta 库中的 zlma 函数计算结果
result = pandas_ta.zlma(self.close)
# 断言结果是 Series 类型
self.assertIsInstance(result, Series)
# 断言结果的名称为 "ZL_EMA_10"
self.assertEqual(result.name, "ZL_EMA_10")
标签：断言,
PandasTA,
Series,
self,
源码,
result,
解析,
pandas,
ta
From： https://www.cnblogs.com/apachecn/p/18135801