.\pandas-ta\pandas_ta\volatility\kc.py
# -*- coding: utf-8 -*-
# 从 pandas 库中导入 DataFrame 类
from pandas import DataFrame
# 从 .true_range 模块中导入 true_range 函数
from .true_range import true_range
# 从 pandas_ta.overlap 模块中导入 ma 函数
from pandas_ta.overlap import ma
# 从 pandas_ta.utils 模块中导入 get_offset, high_low_range, verify_series 函数
from pandas_ta.utils import get_offset, high_low_range, verify_series
# 定义函数 kc，用于计算 Keltner 通道（KC）指标
def kc(high, low, close, length=None, scalar=None, mamode=None, offset=None, **kwargs):
"""Indicator: Keltner Channels (KC)"""
# 验证参数
# 如果 length 存在且大于 0，则转换为整数，否则设置为默认值 20
length = int(length) if length and length > 0 else 20
# 如果 scalar 存在且大于 0，则转换为浮点数，否则设置为默认值 2
scalar = float(scalar) if scalar and scalar > 0 else 2
# 如果 mamode 是字符串类型，则保持不变，否则设置为默认值 "ema"
mamode = mamode if isinstance(mamode, str) else "ema"
# 验证 high、low、close 是否为有效的 Series，长度为 length
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
# 获取偏移量
offset = get_offset(offset)
# 如果 high、low、close 存在空值，则返回空值
if high is None or low is None or close is None: return
# 计算结果
# 判断是否使用 True Range（TR），默认为 True
use_tr = kwargs.pop("tr", True)
if use_tr:
range_ = true_range(high, low, close)
else:
range_ = high_low_range(high, low)
# 计算基准线和波动范围
basis = ma(mamode, close, length=length)
band = ma(mamode, range_, length=length)
lower = basis - scalar * band
upper = basis + scalar * band
# 处理偏移量
if offset != 0:
lower = lower.shift(offset)
basis = basis.shift(offset)
upper = upper.shift(offset)
# 处理填充值
if "fillna" in kwargs:
lower.fillna(kwargs["fillna"], inplace=True)
basis.fillna(kwargs["fillna"], inplace=True)
upper.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
lower.fillna(method=kwargs["fill_method"], inplace=True)
basis.fillna(method=kwargs["fill_method"], inplace=True)
upper.fillna(method=kwargs["fill_method"], inplace=True)
# 命名并分类化
_props = f"{mamode.lower()[0] if len(mamode) else ''}_{length}_{scalar}"
lower.name = f"KCL{_props}"
basis.name = f"KCB{_props}"
upper.name = f"KCU{_props}"
basis.category = upper.category = lower.category = "volatility"
# 准备返回的 DataFrame
data = {lower.name: lower, basis.name: basis, upper.name: upper}
kcdf = DataFrame(data)
kcdf.name = f"KC{_props}"
kcdf.category = basis.category
return kcdf
# 设置 kc 函数的文档字符串
kc.__doc__ = \
"""Keltner Channels (KC)
A popular volatility indicator similar to Bollinger Bands and
Donchian Channels.
Sources:
https://www.tradingview.com/wiki/Keltner_Channels_(KC)
Calculation:
Default Inputs:
length=20, scalar=2, mamode=None, tr=True
TR = True Range
SMA = Simple Moving Average
EMA = Exponential Moving Average
if tr:
RANGE = TR(high, low, close)
else:
RANGE = high - low
if mamode == "ema":
BASIS = sma(close, length)
BAND = sma(RANGE, length)
elif mamode == "sma":
BASIS = sma(close, length)
BAND = sma(RANGE, length)
LOWER = BASIS - scalar * BAND
UPPER = BASIS + scalar * BAND
Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
"""
length (int): The short period. Default: 20
scalar (float): A positive float to scale the bands. Default: 2
mamode (str): See ```help(ta.ma)```py. Default: 'ema'
offset (int): How many periods to offset the result. Default: 0
# 函数参数：
# tr (bool): 如果为 True，则使用 True Range 进行计算；如果为 False，则使用高 - 低作为范围计算。默认值为 True
# fillna (value, optional): pd.DataFrame.fillna(value) 的可选参数，用于指定填充缺失值的值
# fill_method (value, optional): 填充方法的类型
# 返回值：
# 返回一个 pandas DataFrame，包含 lower、basis、upper 列。
.\pandas-ta\pandas_ta\volatility\massi.py
# -*- coding: utf-8 -*-
# 从 pandas_ta.overlap 模块导入 ema 函数
from pandas_ta.overlap import ema
# 从 pandas_ta.utils 模块导入 get_offset、non_zero_range、verify_series 函数
from pandas_ta.utils import get_offset, non_zero_range, verify_series
# 定义一个函数 massi，用于计算 Mass Index（MASSI）
def massi(high, low, fast=None, slow=None, offset=None, **kwargs):
"""Indicator: Mass Index (MASSI)"""
# 验证参数的有效性
# 如果 fast 有值且大于 0，则将其转换为整数，否则默认为 9
fast = int(fast) if fast and fast > 0 else 9
# 如果 slow 有值且大于 0，则将其转换为整数，否则默认为 25
slow = int(slow) if slow and slow > 0 else 25
# 如果 slow 小于 fast，则交换它们的值
if slow < fast:
fast, slow = slow, fast
# 计算参数的最大值
_length = max(fast, slow)
# 验证 high 和 low 是否为有效序列
high = verify_series(high, _length)
low = verify_series(low, _length)
# 获取偏移量
offset = get_offset(offset)
# 移除 kwargs 中的 "length" 键
if "length" in kwargs: kwargs.pop("length")
# 如果 high 或 low 为 None，则返回
if high is None or low is None: return
# 计算结果
# 计算高低价范围
high_low_range = non_zero_range(high, low)
# 计算高低价范围的 EMA
hl_ema1 = ema(close=high_low_range, length=fast, **kwargs)
# 计算高低价范围的 EMA 的 EMA
hl_ema2 = ema(close=hl_ema1, length=fast, **kwargs)
# 计算 hl_ratio
hl_ratio = hl_ema1 / hl_ema2
# 计算 MASSI
massi = hl_ratio.rolling(slow, min_periods=slow).sum()
# 调整偏移量
if offset != 0:
massi = massi.shift(offset)
# 处理填充
if "fillna" in kwargs:
massi.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
massi.fillna(method=kwargs["fill_method"], inplace=True)
# 给结果命名并归类
massi.name = f"MASSI_{fast}_{slow}"
massi.category = "volatility"
return massi
# 将 massi 函数的文档字符串重新赋值，用于说明该函数的功能、计算方法以及参数等信息
massi.__doc__ = \
"""Mass Index (MASSI)
The Mass Index is a non-directional volatility indicator that utilitizes the
High-Low Range to identify trend reversals based on range expansions.
Sources:
https://stockcharts.com/school/doku.php?id=chart_school:technical_indicators:mass_index
mi = sum(ema(high - low, 9) / ema(ema(high - low, 9), 9), length)
Calculation:
Default Inputs:
fast: 9, slow: 25
EMA = Exponential Moving Average
hl = high - low
hl_ema1 = EMA(hl, fast)
hl_ema2 = EMA(hl_ema1, fast)
hl_ratio = hl_ema1 / hl_ema2
MASSI = SUM(hl_ratio, slow)
Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
fast (int): The short period. Default: 9
slow (int): The long period. Default: 25
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature generated.
"""
.\pandas-ta\pandas_ta\volatility\natr.py
# -*- coding: utf-8 -*-
# 导入必要的库和模块
from .atr import atr
from pandas_ta import Imports
from pandas_ta.utils import get_drift, get_offset, verify_series
# 定义函数，计算标准化的平均真实范围（NATR）
def natr(high, low, close, length=None, scalar=None, mamode=None, talib=None, drift=None, offset=None, **kwargs):
"""Indicator: Normalized Average True Range (NATR)"""
# 验证参数
length = int(length) if length and length > 0 else 14
mamode = mamode if isinstance(mamode, str) else "ema"
scalar = float(scalar) if scalar else 100
high = verify_series(high, length)
low = verify_series(low, length)
close = verify_series(close, length)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
# 如果输入的数据有缺失值，则返回空
if high is None or low is None or close is None: return
# 计算结果
if Imports["talib"] and mode_tal:
from talib import NATR
natr = NATR(high, low, close, length)
else:
natr = scalar / close
natr *= atr(high=high, low=low, close=close, length=length, mamode=mamode, drift=drift, offset=offset, **kwargs)
# 偏移结果
if offset != 0:
natr = natr.shift(offset)
# 处理填充值
if "fillna" in kwargs:
natr.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
natr.fillna(method=kwargs["fill_method"], inplace=True)
# 设置名称和分类
natr.name = f"NATR_{length}"
natr.category = "volatility"
return natr
# 设置函数文档字符串
natr.__doc__ = \
"""Normalized Average True Range (NATR)
Normalized Average True Range attempt to normalize the average true range.
Sources:
https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/normalized-average-true-range-natr/
Calculation:
Default Inputs:
length=20
ATR = Average True Range
NATR = (100 / close) * ATR(high, low, close)
Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
length (int): The short period. Default: 20
scalar (float): How much to magnify. Default: 100
mamode (str): See ```help(ta.ma)```py. Default: 'ema'
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature
"""
.\pandas-ta\pandas_ta\volatility\pdist.py
# -*- coding: utf-8 -*-
# 从 pandas_ta.utils 模块导入 get_drift, get_offset, non_zero_range, verify_series 函数
from pandas_ta.utils import get_drift, get_offset, non_zero_range, verify_series
# 定义函数 pdist，用于计算价格距离（PDIST）
def pdist(open_, high, low, close, drift=None, offset=None, **kwargs):
"""Indicator: Price Distance (PDIST)"""
# 验证参数的有效性，确保它们都是 pd.Series 类型
open_ = verify_series(open_)
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
# 获取漂移和偏移值，如果未提供，则使用默认值
drift = get_drift(drift)
offset = get_offset(offset)
# 计算结果
# PDIST = 2 * (high - low) - |close - open| + |open - close[drift]|
pdist = 2 * non_zero_range(high, low)
pdist += non_zero_range(open_, close.shift(drift)).abs()
pdist -= non_zero_range(close, open_).abs()
# 对结果进行偏移处理
if offset != 0:
pdist = pdist.shift(offset)
# 处理填充值
if "fillna" in kwargs:
pdist.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
pdist.fillna(method=kwargs["fill_method"], inplace=True)
# 指定结果的名称和分类
pdist.name = "PDIST"
pdist.category = "volatility"
return pdist
# 为 pdist 函数添加文档字符串
pdist.__doc__ = \
"""Price Distance (PDIST)
Measures the "distance" covered by price movements.
Sources:
https://www.prorealcode.com/prorealtime-indicators/pricedistance/
Calculation:
Default Inputs:
drift=1
PDIST = 2(high - low) - ABS(close - open) + ABS(open - close[drift])
Args:
open_ (pd.Series): Series of 'opens's
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
drift (int): The difference period. Default: 1
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature generated.
"""
.\pandas-ta\pandas_ta\volatility\rvi.py
# -*- coding: utf-8 -*-
# 从pandas_ta.overlap模块导入ma函数
from pandas_ta.overlap import ma
# 从pandas_ta.statistics模块导入stdev函数
from pandas_ta.statistics import stdev
# 从pandas_ta.utils模块导入get_drift和get_offset函数
from pandas_ta.utils import get_drift, get_offset
# 从pandas_ta.utils模块导入unsigned_differences和verify_series函数
from pandas_ta.utils import unsigned_differences, verify_series
# 定义Relative Volatility Index (RVI)指标函数
def rvi(close, high=None, low=None, length=None, scalar=None, refined=None, thirds=None, mamode=None, drift=None, offset=None, **kwargs):
"""Indicator: Relative Volatility Index (RVI)"""
# 验证参数
length = int(length) if length and length > 0 else 14 # 如果length为正整数则保留，否则默认为14
scalar = float(scalar) if scalar and scalar > 0 else 100 # 如果scalar为正浮点数则保留，否则默认为100
refined = False if refined is None else refined # 如果refined为None则默认为False
thirds = False if thirds is None else thirds # 如果thirds为None则默认为False
mamode = mamode if isinstance(mamode, str) else "ema" # 如果mamode为字符串则保留，否则默认为"ema"
close = verify_series(close, length) # 验证close是否为有效序列
drift = get_drift(drift) # 获取漂移参数
offset = get_offset(offset) # 获取偏移参数
if close is None: return
if refined or thirds:
high = verify_series(high) # 验证high是否为有效序列
low = verify_series(low) # 验证low是否为有效序列
# 计算结果
def _rvi(source, length, scalar, mode, drift):
"""RVI"""
# 计算标准差
std = stdev(source, length)
# 获取正差值和负差值
pos, neg = unsigned_differences(source, amount=drift)
# 计算正差值的标准差加权平均
pos_std = pos * std
pos_avg = ma(mode, pos_std, length=length)
# 计算负差值的标准差加权平均
neg_std = neg * std
neg_avg = ma(mode, neg_std, length=length)
# 计算RVI指标
result = scalar * pos_avg
result /= pos_avg + neg_avg
return result
_mode = ""
if refined: # 如果使用了refined模式
# 计算高价RVI
high_rvi = _rvi(high, length, scalar, mamode, drift)
# 计算低价RVI
low_rvi = _rvi(low, length, scalar, mamode, drift)
# 计算RVI
rvi = 0.5 * (high_rvi + low_rvi)
_mode = "r" # 设置模式为"r"
elif thirds: # 如果使用了thirds模式
# 计算高价RVI
high_rvi = _rvi(high, length, scalar, mamode, drift)
# 计算低价RVI
low_rvi = _rvi(low, length, scalar, mamode, drift)
# 计算收盘价RVI
close_rvi = _rvi(close, length, scalar, mamode, drift)
# 计算RVI
rvi = (high_rvi + low_rvi + close_rvi) / 3.0
_mode = "t" # 设置模式为"t"
else: # 如果未使用refined和thirds模式
# 计算RVI
rvi = _rvi(close, length, scalar, mamode, drift)
# 偏移
if offset != 0:
rvi = rvi.shift(offset)
# 处理填充
if "fillna" in kwargs:
rvi.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
rvi.fillna(method=kwargs["fill_method"], inplace=True)
# 命名和归类
rvi.name = f"RVI{_mode}_{length}"
rvi.category = "volatility"
return rvi
# 设置RVI指标的文档字符串
rvi.__doc__ = \
"""Relative Volatility Index (RVI)
The Relative Volatility Index (RVI) was created in 1993 and revised in 1995.
Instead of adding up price changes like RSI based on price direction, the RVI
adds up standard deviations based on price direction.
Sources:
https://www.tradingview.com/wiki/Keltner_Channels_(KC)
Calculation:
Default Inputs:
length=14, scalar=100, refined=None, thirds=None
EMA = Exponential Moving Average
STDEV = Standard Deviation
UP = STDEV(src, length) IF src.diff() > 0 ELSE 0
DOWN = STDEV(src, length) IF src.diff() <= 0 ELSE 0
UPSUM = EMA(UP, length)
DOWNSUM = EMA(DOWN, length
"""
# 计算相对强度指数（RSI），其计算公式为 RVI = scalar * (UPSUM / (UPSUM + DOWNSUM))
RVI = scalar * (UPSUM / (UPSUM + DOWNSUM))
# 定义一个函数，用于计算布林带指标（Bollinger Bands）
def bollinger_hband_indicator(high, low, close, length=14, scalar=100, refined=False, thirds=False, mamode='ema', offset=0, fillna=None, fill_method=None):
"""
Args:
high (pd.Series): 最高价的序列
low (pd.Series): 最低价的序列
close (pd.Series): 收盘价的序列
length (int): 短周期。默认值为14
scalar (float): 缩放带的正浮点数。默认值为100
refined (bool): 使用“精炼”计算，即 RVI(high) 和 RVI(low) 的平均值，而不是 RVI(close)。默认值为False
thirds (bool): 使用最高价、最低价和收盘价的平均值。默认值为False
mamode (str): 参见 ```help(ta.ma)```py。默认值为'ema'
offset (int): 结果的偏移周期数。默认值为0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value) 的参数
fill_method (value, optional): 填充方法的类型
Returns:
pd.DataFrame: 包含 lower、basis、upper 列的数据框
"""
.\pandas-ta\pandas_ta\volatility\thermo.py
# -*- coding: utf-8 -*-
# 从 pandas 库中导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta 库中的 overlap 模块导入 ma 函数
from pandas_ta.overlap import ma
# 从 pandas_ta 库中的 utils 模块导入 get_offset、verify_series、get_drift 函数
from pandas_ta.utils import get_offset, verify_series, get_drift
# 定义函数 thermo，计算 Elder's Thermometer 指标
def thermo(high, low, length=None, long=None, short=None, mamode=None, drift=None, offset=None, **kwargs):
"""Indicator: Elders Thermometer (THERMO)"""
# 验证参数
# 如果 length 存在且大于 0，则转换为整数，否则默认为 20
length = int(length) if length and length > 0 else 20
# 如果 long 存在且大于 0，则转换为浮点数，否则默认为 2
long = float(long) if long and long > 0 else 2
# 如果 short 存在且大于 0，则转换为浮点数，否则默认为 0.5
short = float(short) if short and short > 0 else 0.5
# 如果 mamode 是字符串，则保持不变，否则默认为 "ema"
mamode = mamode if isinstance(mamode, str) else "ema"
# 验证 high 和 low 系列数据，长度为 length
high = verify_series(high, length)
low = verify_series(low, length)
# 获取 drift 和 offset
drift = get_drift(drift)
offset = get_offset(offset)
# 从 kwargs 中弹出 asint 参数，默认为 True
asint = kwargs.pop("asint", True)
# 如果 high 或 low 为 None，则返回
if high is None or low is None: return
# 计算结果
# 计算 Elder's Thermometer 的下限
thermoL = (low.shift(drift) - low).abs()
# 计算 Elder's Thermometer 的上限
thermoH = (high - high.shift(drift)).abs()
# 取较小的值作为 Elder's Thermometer
thermo = thermoL
thermo = thermo.where(thermoH < thermoL, thermoH)
# 索引设置为 high 的索引
thermo.index = high.index
# 计算 Elder's Thermometer 的移动平均线
thermo_ma = ma(mamode, thermo, length=length)
# 生成信号
thermo_long = thermo < (thermo_ma * long)
thermo_short = thermo > (thermo_ma * short)
# 二进制输出，用于信号
if asint:
thermo_long = thermo_long.astype(int)
thermo_short = thermo_short.astype(int)
# 调整偏移量
if offset != 0:
thermo = thermo.shift(offset)
thermo_ma = thermo_ma.shift(offset)
thermo_long = thermo_long.shift(offset)
thermo_short = thermo_short.shift(offset)
# 处理填充
if "fillna" in kwargs:
thermo.fillna(kwargs["fillna"], inplace=True)
thermo_ma.fillna(kwargs["fillna"], inplace=True)
thermo_long.fillna(kwargs["fillna"], inplace=True)
thermo_short.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
thermo.fillna(method=kwargs["fill_method"], inplace=True)
thermo_ma.fillna(method=kwargs["fill_method"], inplace=True)
thermo_long.fillna(method=kwargs["fill_method"], inplace=True)
thermo_short.fillna(method=kwargs["fill_method"], inplace=True)
# 设置名称和分类
_props = f"_{length}_{long}_{short}"
thermo.name = f"THERMO{_props}"
thermo_ma.name = f"THERMOma{_props}"
thermo_long.name = f"THERMOl{_props}"
thermo_short.name = f"THERMOs{_props}"
thermo.category = thermo_ma.category = thermo_long.category = thermo_short.category = "volatility"
# 准备返回的 DataFrame
data = {
thermo.name: thermo,
thermo_ma.name: thermo_ma,
thermo_long.name: thermo_long,
thermo_short.name: thermo_short
}
df = DataFrame(data)
df.name = f"THERMO{_props}"
df.category = thermo.category
return df
# 为 thermo 函数添加文档字符串
thermo.__doc__ = \
"""Elders Thermometer (THERMO)
Elder's Thermometer measures price volatility.
Sources:
https://www.motivewave.com/studies/elders_thermometer.htm
# 导入所需的库
import requests
# 定义函数`get_tradingview_script`用于获取TradingView上的脚本内容
def get_tradingview_script(url):
# 发送GET请求获取指定URL的页面内容
response = requests.get(url)
# 返回页面内容的文本
return response.text
# 定义变量`script_url`，存储TradingView脚本的URL
script_url = "https://www.tradingview.com/script/HqvTuEMW-Elder-s-Market-Thermometer-LazyBear/"
# 调用`get_tradingview_script`函数，获取指定URL的脚本内容
script_content = get_tradingview_script(script_url)
# 计算热力指标（thermo）和相关指标
Calculation:
# 默认输入参数
length=20, drift=1, mamode=EMA, long=2, short=0.5
# EMA为指数移动平均
# 计算低价的漂移
thermoL = (low.shift(drift) - low).abs()
# 计算高价的漂移
thermoH = (high - high.shift(drift)).abs()
# 选择较大的漂移值
thermo = np.where(thermoH > thermoL, thermoH, thermoL)
# 对漂移值进行指数移动平均
thermo_ma = ema(thermo, length)
# 判断是否满足买入条件
thermo_long = thermo < (thermo_ma * long)
# 判断是否满足卖出条件
thermo_short = thermo > (thermo_ma * short)
# 将布尔值转换为整数
thermo_long = thermo_long.astype(int)
thermo_short = thermo_short.astype(int)
Args:
high (pd.Series): 'high' 的序列
low (pd.Series): 'low' 的序列
long(int): 买入因子
short(float): 卖出因子
length (int): 周期。默认值：20
mamode (str): 参见 ```help(ta.ma)```py。默认值：'ema'
drift (int): 漂移周期。默认值：1
offset (int): 结果的偏移周期数。默认值：0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): 填充方法的类型
Returns:
pd.DataFrame: 包含 thermo, thermo_ma, thermo_long, thermo_short 列的数据框
.\pandas-ta\pandas_ta\volatility\true_range.py
# -*- coding: utf-8 -*-
# 导入 numpy 库中的 nan，并将其重命名为 npNaN
from numpy import nan as npNaN
# 导入 pandas 库中的 concat 函数
from pandas import concat
# 从 pandas_ta 库中导入 Imports 模块
from pandas_ta import Imports
# 从 pandas_ta 库中导入 get_drift, get_offset, non_zero_range, verify_series 函数
from pandas_ta.utils import get_drift, get_offset, non_zero_range, verify_series
# 定义一个名为 true_range 的函数，用于计算真实波动范围
def true_range(high, low, close, talib=None, drift=None, offset=None, **kwargs):
"""Indicator: True Range"""
# 验证参数
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
drift = get_drift(drift)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
# 计算结果
if Imports["talib"] and mode_tal:
# 如果导入了 talib 库并且 mode_tal 为 True，则使用 talib 库中的 TRANGE 函数计算真实波动范围
from talib import TRANGE
true_range = TRANGE(high, low, close)
else:
# 否则，计算高低范围、前一日收盘价、以及真实波动范围
high_low_range = non_zero_range(high, low)
prev_close = close.shift(drift)
ranges = [high_low_range, high - prev_close, prev_close - low]
true_range = concat(ranges, axis=1)
true_range = true_range.abs().max(axis=1)
true_range.iloc[:drift] = npNaN
# 偏移
if offset != 0:
true_range = true_range.shift(offset)
# 处理填充
if "fillna" in kwargs:
true_range.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
true_range.fillna(method=kwargs["fill_method"], inplace=True)
# 命名和分类
true_range.name = f"TRUERANGE_{drift}"
true_range.category = "volatility"
return true_range
# 设置 true_range 函数的文档字符串
true_range.__doc__ = \
"""True Range
An method to expand a classical range (high minus low) to include
possible gap scenarios.
Sources:
https://www.macroption.com/true-range/
Calculation:
Default Inputs:
drift=1
ABS = Absolute Value
prev_close = close.shift(drift)
TRUE_RANGE = ABS([high - low, high - prev_close, low - prev_close])
Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
drift (int): The shift period. Default: 1
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature
"""
.\pandas-ta\pandas_ta\volatility\ui.py
# -*- coding: utf-8 -*-
# 从 numpy 库导入 sqrt 函数并重命名为 npsqrt
from numpy import sqrt as npsqrt
# 从 pandas_ta.overlap 模块导入 sma 函数
from pandas_ta.overlap import sma
# 从 pandas_ta.utils 模块导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series
# 定义函数 ui，计算 Ulcer Index（UI）
def ui(close, length=None, scalar=None, offset=None, **kwargs):
"""Indicator: Ulcer Index (UI)"""
# 验证参数
# 如果 length 存在且大于 0，则转换为整数，否则默认为 14
length = int(length) if length and length > 0 else 14
# 如果 scalar 存在且大于 0，则转换为浮点数，否则默认为 100
scalar = float(scalar) if scalar and scalar > 0 else 100
# 验证 close 是否为有效的时间序列，并指定长度为 length
close = verify_series(close, length)
# 获取偏移量
offset = get_offset(offset)
# 如果 close 为空，则返回空值
if close is None: return
# 计算结果
# 计算最近 length 个周期内的最高收盘价
highest_close = close.rolling(length).max()
# 计算下行波动性
downside = scalar * (close - highest_close)
downside /= highest_close
d2 = downside * downside
# 获取 everget 参数，默认为 False
everget = kwargs.pop("everget", False)
# 如果 everget 为 True，则使用 SMA 而不是 SUM 进行计算
if everget:
# 使用 SMA 对 d2 进行计算，然后应用 npsqrt 函数
ui = (sma(d2, length) / length).apply(npsqrt)
else:
# 使用 SUM 对 d2 进行计算，然后应用 npsqrt 函数
ui = (d2.rolling(length).sum() / length).apply(npsqrt)
# 偏移结果
if offset != 0:
ui = ui.shift(offset)
# 处理填充
if "fillna" in kwargs:
# 使用指定值填充空值
ui.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
# 使用指定的填充方法填充空值
ui.fillna(method=kwargs["fill_method"], inplace=True)
# 设置指标名称和分类
ui.name = f"UI{'' if not everget else 'e'}_{length}"
ui.category = "volatility"
# 返回结果
return ui
# 设置 ui 函数的文档字符串
ui.__doc__ = \
"""Ulcer Index (UI)
The Ulcer Index by Peter Martin measures the downside volatility with the use of
the Quadratic Mean, which has the effect of emphasising large drawdowns.
Sources:
https://library.tradingtechnologies.com/trade/chrt-ti-ulcer-index.html
https://en.wikipedia.org/wiki/Ulcer_index
http://www.tangotools.com/ui/ui.htm
Calculation:
Default Inputs:
length=14, scalar=100
HC = Highest Close
SMA = Simple Moving Average
HCN = HC(close, length)
DOWNSIDE = scalar * (close - HCN) / HCN
if kwargs["everget"]:
UI = SQRT(SMA(DOWNSIDE^2, length) / length)
else:
UI = SQRT(SUM(DOWNSIDE^2, length) / length)
Args:
high (pd.Series): Series of 'high's
close (pd.Series): Series of 'close's
length (int): The short period. Default: 14
scalar (float): A positive float to scale the bands. Default: 100
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
everget (value, optional): TradingView's Evergets SMA instead of SUM
calculation. Default: False
Returns:
pd.Series: New feature
"""
.\pandas-ta\pandas_ta\volatility\__init__.py
# 设置文件编码格式为 UTF-8
# 导入模块中的各种函数和类
from .aberration import aberration # 导入 aberration 模块中的函数或类
from .accbands import accbands # 导入 accbands 模块中的函数或类
from .atr import atr # 导入 atr 模块中的函数或类
from .bbands import bbands # 导入 bbands 模块中的函数或类
from .donchian import donchian # 导入 donchian 模块中的函数或类
from .hwc import hwc # 导入 hwc 模块中的函数或类
from .kc import kc # 导入 kc 模块中的函数或类
from .massi import massi # 导入 massi 模块中的函数或类
from .natr import natr # 导入 natr 模块中的函数或类
from .pdist import pdist # 导入 pdist 模块中的函数或类
from .rvi import rvi # 导入 rvi 模块中的函数或类
from .thermo import thermo # 导入 thermo 模块中的函数或类
from .true_range import true_range # 导入 true_range 模块中的函数或类
from .ui import ui # 导入 ui 模块中的函数或类
.\pandas-ta\pandas_ta\volume\ad.py
# -*- coding: utf-8 -*-
# 从 pandas_ta 库导入 Imports 模块
from pandas_ta import Imports
# 从 pandas_ta.utils 模块导入 get_offset、non_zero_range、verify_series 函数
from pandas_ta.utils import get_offset, non_zero_range, verify_series
# 定义累积/分布指标函数
def ad(high, low, close, volume, open_=None, talib=None, offset=None, **kwargs):
"""Indicator: Accumulation/Distribution (AD)"""
# 验证参数
# 将 high、low、close、volume 分别验证为 pandas.Series 类型
high = verify_series(high)
low = verify_series(low)
close = verify_series(close)
volume = verify_series(volume)
# 将 offset 转换为偏移量
offset = get_offset(offset)
# 判断是否使用 talib，若 talib 参数为布尔类型则以其值为准，否则默认为 True
mode_tal = bool(talib) if isinstance(talib, bool) else True
# 计算结果
if Imports["talib"] and mode_tal:
# 如果导入了 talib 并且 mode_tal 为真，则使用 talib 计算 AD 指标
from talib import AD
# 调用 talib 库中的 AD 函数
ad = AD(high, low, close, volume)
else:
# 如果没有使用 talib 或者 mode_tal 为假，则根据情况计算 AD 指标
if open_ is not None:
# 如果提供了 open_ 参数，则使用 close 和 open_ 计算 AD
open_ = verify_series(open_)
# 计算 AD 指标，使用 close 和 open_
ad = non_zero_range(close, open_) # AD with Open
else:
# 如果未提供 open_ 参数，则使用 high、low 和 close 计算 AD
ad = 2 * close - (high + low) # AD with High, Low, Close
# 计算 high-low 范围
high_low_range = non_zero_range(high, low)
# 根据 high-low 范围和交易量计算 AD
ad *= volume / high_low_range
# 对 AD 进行累积求和
ad = ad.cumsum()
# 偏移结果
if offset != 0:
# 如果偏移量不为零，则对 AD 进行偏移
ad = ad.shift(offset)
# 处理填充
if "fillna" in kwargs:
# 如果提供了 fillna 参数，则使用提供的值填充缺失值
ad.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
# 如果提供了 fill_method 参数，则使用提供的填充方法填充缺失值
ad.fillna(method=kwargs["fill_method"], inplace=True)
# 命名和分类
# 根据是否提供了 open_ 参数命名 AD 指标
ad.name = "AD" if open_ is None else "ADo"
# 将 AD 指标分类为“volume”
ad.category = "volume"
# 返回 AD 指标
return ad
# 设置 AD 函数的文档字符串
ad.__doc__ = \
"""Accumulation/Distribution (AD)
Accumulation/Distribution indicator utilizes the relative position
of the close to it's High-Low range with volume. Then it is cumulated.
Sources:
https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/accumulationdistribution-ad/
Calculation:
CUM = Cumulative Sum
if 'open':
AD = close - open
else:
AD = 2 * close - high - low
hl_range = high - low
AD = AD * volume / hl_range
AD = CUM(AD)
Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
volume (pd.Series): Series of 'volume's
open (pd.Series): Series of 'open's
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib
version. Default: True
offset (int): How many periods to offset the result. Default: 0
Kwargs:
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.Series: New feature generated.
"""