PandasTA 源码解析（十八）（2）

# 定义模块名称
name = "pandas_ta"
"""
.. moduleauthor:: Kevin Johnson
"""
# 导入模块
from importlib.util import find_spec
from pathlib import Path
from pkg_resources import get_distribution, DistributionNotFound
# 获取已安装的模块分布对象
_dist = get_distribution("pandas_ta")
try:
    # 获取模块所在路径
    here = Path(_dist.location) / __file__
    # 检查文件是否存在
    if not here.exists():
        # 如果未安装，但存在其他已安装版本
        raise DistributionNotFound
except DistributionNotFound:
    # 如果未找到分布对象，则提示安装该项目
    __version__ = "Please install this project with setup.py"
# 获取模块版本号
version = __version__ = _dist.version
# 检查导入的模块是否存在
Imports = {
    "alphaVantage-api": find_spec("alphaVantageAPI") is not None,
    "matplotlib": find_spec("matplotlib") is not None,
    "mplfinance": find_spec("mplfinance") is not None,
    "numba": find_spec("numba") is not None,
    "yaml": find_spec("yaml") is not None,
    "scipy": find_spec("scipy") is not None,
    "sklearn": find_spec("sklearn") is not None,
    "statsmodels": find_spec("statsmodels") is not None,
    "stochastic": find_spec("stochastic") is not None,
    "talib": find_spec("talib") is not None,
    "tqdm": find_spec("tqdm") is not None,
    "vectorbt": find_spec("vectorbt") is not None,
    "yfinance": find_spec("yfinance") is not None,
}
# 不是最理想的，也不是动态的，但它可以工作。
# 之后会找到一个动态的解决方案。
Category = {
    # 蜡烛图
    "candles": [
        "cdl_pattern", "cdl_z", "ha"
    ],
    # 周期
    "cycles": ["ebsw"],
    # 动量
    "momentum": [
        "ao", "apo", "bias", "bop", "brar", "cci", "cfo", "cg", "cmo",
        "coppock", "cti", "er", "eri", "fisher", "inertia", "kdj", "kst", "macd",
        "mom", "pgo", "ppo", "psl", "pvo", "qqe", "roc", "rsi", "rsx", "rvgi",
        "slope", "smi", "squeeze", "squeeze_pro", "stc", "stoch", "stochrsi", "td_seq", "trix",
        "tsi", "uo", "willr"
    ],
    # 重叠
    "overlap": [
        "alma", "dema", "ema", "fwma", "hilo", "hl2", "hlc3", "hma", "ichimoku",
        "jma", "kama", "linreg", "mcgd", "midpoint", "midprice", "ohlc4",
        "pwma", "rma", "sinwma", "sma", "ssf", "supertrend", "swma", "t3",
        "tema", "trima", "vidya", "vwap", "vwma", "wcp", "wma", "zlma"
    ],
    # 性能
    "performance": ["log_return", "percent_return"],
    # 统计
    "statistics": [
        "entropy", "kurtosis", "mad", "median", "quantile", "skew", "stdev",
        "tos_stdevall", "variance", "zscore"
    ],
    # 趋势
    "trend": [
        "adx", "amat", "aroon", "chop", "cksp", "decay", "decreasing", "dpo",
        "increasing", "long_run", "psar", "qstick", "short_run", "tsignals",
        "ttm_trend", "vhf", "vortex", "xsignals"
    ],
    # 波动性
    "volatility": [
        "aberration", "accbands", "atr", "bbands", "donchian", "hwc", "kc", "massi",
        "natr", "pdist", "rvi", "thermo", "true_range", "ui"
    ],
    # 交易量，"vp" 或 "Volume Profile" 是独特的
}
    # "volume" 键对应的值是一个列表，包含了各种技术指标的名称
    "volume": [
        "ad",    # AD 指标，积累/分配线
        "adosc", # AD 指标，震荡指标
        "aobv",  # 指标，累积/派发线
        "cmf",   # CMF 指标，资金流量指标
        "efi",   # EFI 指标，振荡器
        "eom",   # EOM 指标，指标
        "kvo",   # Klinger Oscillator（克林格震荡器）指标
        "mfi",   # MFI 指标，资金流指标
        "nvi",   # NVI 指标，价值线
        "obv",   # OBV 指标，累积/分配线
        "pvi",   # PVI 指标，价值线
        "pvol",  # PVO 指标，价值线
        "pvr",   # PVR 指标，价值线
        "pvt"    # PVT 指标，价值线
    ],
# 字典，用于指定聚合函数的方式，对于开盘价、最高价、最低价、收盘价和成交量分别指定了不同的聚合方式
CANGLE_AGG = {
    "open": "first",    # 开盘价取第一个值
    "high": "max",      # 最高价取最大值
    "low": "min",       # 最低价取最小值
    "close": "last",    # 收盘价取最后一个值
    "volume": "sum"     # 成交量取总和
}
# 字典，用于记录各个交易所的时区偏移
EXCHANGE_TZ = {
    "NZSX": 12,         # 新西兰股票交易所，时区偏移为+12
    "ASX": 11,          # 澳大利亚证券交易所，时区偏移为+11
    "TSE": 9,           # 东京证券交易所，时区偏移为+9
    "HKE": 8,           # 香港证券交易所，时区偏移为+8
    "SSE": 8,           # 上海证券交易所，时区偏移为+8
    "SGX": 8,           # 新加坡证券交易所，时区偏移为+8
    "NSE": 5.5,         # 印度证券交易所，时区偏移为+5.5
    "DIFX": 4,          # 迪拜金融市场，时区偏移为+4
    "RTS": 3,           # 莫斯科证券交易所，时区偏移为+3
    "JSE": 2,           # 南非证券交易所，时区偏移为+2
    "FWB": 1,           # 法兰克福证券交易所，时区偏移为+1
    "LSE": 1,           # 伦敦证券交易所，时区偏移为+1
    "BMF": -2,          # 巴西商品与期货交易所，时区偏移为-2
    "NYSE": -4,         # 纽约证券交易所，时区偏移为-4
    "TSX": -4           # 多伦多证券交易所，时区偏移为-4
}
# 字典，用于定义一些时间相关的常量
RATE = {
    "DAYS_PER_MONTH": 21,              # 每月交易日数
    "MINUTES_PER_HOUR": 60,            # 每小时分钟数
    "MONTHS_PER_YEAR": 12,             # 每年月份数
    "QUARTERS_PER_YEAR": 4,            # 每年季度数
    "TRADING_DAYS_PER_YEAR": 252,      # 每年交易日数，保持为偶数
    "TRADING_HOURS_PER_DAY": 6.5,      # 每日交易小时数
    "WEEKS_PER_YEAR": 52,              # 每年周数
    "YEARLY": 1                         # 年度
}
# 从 pandas_ta.core 模块导入所有内容
from pandas_ta.core import *

# -*- coding: utf-8 -*-
# 导入 setup 函数，用于设置 Python 包的元数据和安装信息
from distutils.core import setup
# 定义长描述信息
long_description = "An easy to use Python 3 Pandas Extension with 130+ Technical Analysis Indicators. Can be called from a Pandas DataFrame or standalone like TA-Lib. Correlation tested with TA-Lib."
# 设置函数调用，用于设置 Python 包的元数据和安装信息
setup(
    # 包的名称
    name="pandas_ta",
    # 包含的子包列表
    packages=[
        "pandas_ta",
        "pandas_ta.candles",
        "pandas_ta.cycles",
        "pandas_ta.momentum",
        "pandas_ta.overlap",
        "pandas_ta.performance",
        "pandas_ta.statistics",
        "pandas_ta.trend",
        "pandas_ta.utils",
        "pandas_ta.utils.data",
        "pandas_ta.volatility",
        "pandas_ta.volume"
    ],
    # 版本号
    version=".".join(("0", "3", "14b")),
    # 简要描述
    description=long_description,
    # 长描述
    long_description=long_description,
    # 作者
    author="Kevin Johnson",
    # 作者邮箱
    author_email="appliedmathkj@gmail.com",
    # 项目 URL
    url="https://github.com/twopirllc/pandas-ta",
    # 维护者
    maintainer="Kevin Johnson",
    # 维护者邮箱
    maintainer_email="appliedmathkj@gmail.com",
    # 下载 URL
    download_url="https://github.com/twopirllc/pandas-ta.git",
    # 关键字列表
    keywords=["technical analysis", "trading", "python3", "pandas"],
    # 许可证
    license="The MIT License (MIT)",
    # 分类信息列表
    classifiers=[
        "Development Status :: 4 - Beta",
        "Programming Language :: Python :: 3.6",
        "Programming Language :: Python :: 3.7",
        "Programming Language :: Python :: 3.8",
        "Programming Language :: Python :: 3.9",
        "Operating System :: OS Independent",
        "License :: OSI Approved :: MIT License",
        "Natural Language :: English",
        "Intended Audience :: Developers",
        "Intended Audience :: Financial and Insurance Industry",
        "Intended Audience :: Science/Research",
        "Topic :: Office/Business :: Financial",
        "Topic :: Office/Business :: Financial :: Investment",
        "Topic :: Scientific/Engineering",
        "Topic :: Scientific/Engineering :: Information Analysis",
    ],
    # 包数据配置
    package_data={
        "data": ["data/*.csv"],
    },
    # 安装依赖项
    install_requires=["pandas"],
    # 列出额外的依赖组（例如开发依赖）
    extras_require={
        "dev": [
            "alphaVantage-api", "matplotlib", "mplfinance", "scipy",
            "sklearn", "statsmodels", "stochastic",
            "talib", "tqdm", "vectorbt", "yfinance",
        ],
        "test": ["ta-lib"],
    },
)

# 导入 os 模块，用于操作系统相关功能
import os
# 导入 pandas 模块中的 DatetimeIndex 和 read_csv 函数
from pandas import DatetimeIndex, read_csv
# 设定是否显示详细信息的标志
VERBOSE = True
# 设定警报和信息提示的图标
ALERT = f"[!]"
INFO = f"[i]"
# 设定相关性分析方法，这里选择使用 'corr'，也可以选择 'sem'
CORRELATION = "corr"  # "sem"
# 设定相关性阈值，小于 0.99 视为不理想
CORRELATION_THRESHOLD = 0.99  # Less than 0.99 is undesirable
# 读取样本数据集，使用 pandas 的 read_csv 函数
sample_data = read_csv(
    f"data/SPY_D.csv",  # 文件路径
    index_col=0,  # 以第一列作为索引
    parse_dates=True,  # 解析日期
    infer_datetime_format=True,  # 推断日期格式
    keep_date_col=True,  # 保留日期列
)
# 将日期列设置为索引，并丢弃原始日期列
sample_data.set_index(DatetimeIndex(sample_data["date"]), inplace=True, drop=True)
sample_data.drop("date", axis=1, inplace=True)
# 定义错误分析函数，用于输出错误信息
def error_analysis(df, kind, msg, icon=INFO, newline=True):
    if VERBOSE:  # 如果 VERBOSE 为 True，则输出信息
        s = f"{icon} {df.name}['{kind}']: {msg}"  # 构造输出字符串
        if newline:  # 如果需要换行
            s = f"\n{s}"  # 在字符串前添加换行符
        print(s)  # 打印信息

# `.\pandas-ta\tests\context.py`
```py
# 导入 os 模块，提供对操作系统功能的访问
import os
# 导入 sys 模块，提供对 Python 解释器的访问和控制
import sys
# 将当前文件所在目录的父目录添加到 Python 模块搜索路径中，以便导入自定义模块
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
# 导入 pandas_ta 模块，该模块提供了一系列用于技术分析的函数和指标
import pandas_ta

# 从config模块中导入sample_data
from .config import sample_data
# 从context模块中导入pandas_ta
from .context import pandas_ta
# 从unittest模块中导入TestCase和skip
from unittest import TestCase, skip
# 从pandas模块中导入DataFrame
from pandas import DataFrame
# 定义测试类TestCandleExtension，继承自TestCase类
class TestCandleExtension(TestCase):
    # 设置类方法setUpClass，在所有测试用例执行前执行一次
    @classmethod
    def setUpClass(cls):
        cls.data = sample_data
    # 设置类方法tearDownClass，在所有测试用例执行后执行一次
    @classmethod
    def tearDownClass(cls):
        del cls.data
    # 定义实例方法setUp，在每个测试用例执行前执行一次
    def setUp(self): pass
    # 定义实例方法tearDown，在每个测试用例执行后执行一次
    def tearDown(self): pass
    # 测试CDL_DOJI_10_0.1模式的扩展
    def test_cdl_doji_ext(self):
        self.data.ta.cdl_pattern("doji", append=True)
        # 断言self.data是DataFrame类型
        self.assertIsInstance(self.data, DataFrame)
        # 断言self.data的最后一列的列名为"CDL_DOJI_10_0.1"
        self.assertEqual(self.data.columns[-1], "CDL_DOJI_10_0.1")
    # 测试CDL_INSIDE模式的扩展
    def test_cdl_inside_ext(self):
        self.data.ta.cdl_pattern("inside", append=True)
        # 断言self.data是DataFrame类型
        self.assertIsInstance(self.data, DataFrame)
        # 断言self.data的最后一列的列名为"CDL_INSIDE"
        self.assertEqual(self.data.columns[-1], "CDL_INSIDE")
    # 测试CDL_Z指标的扩展
    def test_cdl_z_ext(self):
        self.data.ta.cdl_z(append=True)
        # 断言self.data是DataFrame类型
        self.assertIsInstance(self.data, DataFrame)
        # 断言self.data的倒数第四列到最后一列的列名为["open_Z_30_1", "high_Z_30_1", "low_Z_30_1", "close_Z_30_1"]
        self.assertEqual(list(self.data.columns[-4:]), ["open_Z_30_1", "high_Z_30_1", "low_Z_30_1", "close_Z_30_1"])
    # 测试HA指标的扩展
    def test_ha_ext(self):
        self.data.ta.ha(append=True)
        # 断言self.data是DataFrame类型
        self.assertIsInstance(self.data, DataFrame)
        # 断言self.data的倒数第四列到最后一列的列名为["HA_open", "HA_high", "HA_low", "HA_close"]
        self.assertEqual(list(self.data.columns[-4:]), ["HA_open", "HA_high", "HA_low", "HA_close"])

# 从 pandas 库的 core.series 模块中导入 Series 类
from pandas.core.series import Series
# 从当前目录中的 config 模块中导入 sample_data 变量
from .config import sample_data
# 从当前目录中的 context 模块中导入 pandas_ta 模块
from .context import pandas_ta
# 从 unittest 模块中导入 TestCase 类
from unittest import TestCase
# 从 pandas 库中导入 DataFrame 类
from pandas import DataFrame
# 定义测试类 TestCylesExtension，继承自 TestCase 类
class TestCylesExtension(TestCase):
    # 在整个测试类运行之前执行的方法，设置测试数据
    @classmethod
    def setUpClass(cls):
        cls.data = sample_data
    # 在整个测试类运行之后执行的方法，清理测试数据
    @classmethod
    def tearDownClass(cls):
        del cls.data
    # 在每个测试方法运行之前执行的方法
    def setUp(self): pass
    # 在每个测试方法运行之后执行的方法
    def tearDown(self): pass
    # 定义测试方法 test_ebsw_ext，测试 EBSW 扩展函数
    def test_ebsw_ext(self):
        # 调用数据框对象的 ta 属性中的 ebsw 方法，并将结果追加到原数据框中
        self.data.ta.ebsw(append=True)
        # 断言数据框对象是 DataFrame 类型
        self.assertIsInstance(self.data, DataFrame)
        # 断言数据框对象的最后一列的列名为 "EBSW_40_10"
        self.assertEqual(self.data.columns[-1], "EBSW_40_10")