# -*- 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)指标defkc(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 >0else20# 如果 scalar 存在且大于 0,则转换为浮点数,否则设置为默认值 2
scalar =float(scalar)if scalar and scalar >0else2# 如果 mamode 是字符串类型,则保持不变,否则设置为默认值 "ema"
mamode = mamode ifisinstance(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 isNoneor low isNoneor close isNone: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]iflen(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)defmassi(high, low, fast=None, slow=None, offset=None,**kwargs):"""Indicator: Mass Index (MASSI)"""# 验证参数的有效性# 如果 fast 有值且大于 0,则将其转换为整数,否则默认为 9
fast =int(fast)if fast and fast >0else9# 如果 slow 有值且大于 0,则将其转换为整数,否则默认为 25
slow =int(slow)if slow and slow >0else25# 如果 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 isNoneor low isNone: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)defnatr(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 >0else14
mamode = mamode ifisinstance(mamode,str)else"ema"
scalar =float(scalar)if scalar else100
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)ifisinstance(talib,bool)elseTrue# 如果输入的数据有缺失值,则返回空if high isNoneor low isNoneor close isNone: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)defpdist(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.
"""
# -*- 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
# 定义累积/分布指标函数defad(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)ifisinstance(talib,bool)elseTrue# 计算结果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_ isnotNone:# 如果提供了 open_ 参数,则使用 close 和 open_ 计算 AD
open_ = verify_series(open_)# 计算 AD 指标,使用 close 和 open_
ad = non_zero_range(close, open_)# AD with Openelse:# 如果未提供 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_ isNoneelse"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.
"""