标签：PandasTA signal kwargs length 源码 offset close fillna 解析
.\pandas-ta\pandas_ta\momentum\kst.py
# -*- coding: utf-8 -*-
# 导入DataFrame类
from pandas import DataFrame
# 导入roc函数
from .roc import roc
# 导入验证序列函数、获取漂移和偏移的函数
from pandas_ta.utils import get_drift, get_offset, verify_series
# 定义函数：'Know Sure Thing' (KST)
def kst(close, roc1=None, roc2=None, roc3=None, roc4=None, sma1=None, sma2=None, sma3=None, sma4=None, signal=None, drift=None, offset=None, **kwargs):
"""Indicator: 'Know Sure Thing' (KST)"""
# 验证参数
# 若roc1参数存在且大于0，则将其转换为整数，否则使用默认值10
roc1 = int(roc1) if roc1 and roc1 > 0 else 10
# 若roc2参数存在且大于0，则将其转换为整数，否则使用默认值15
roc2 = int(roc2) if roc2 and roc2 > 0 else 15
# 若roc3参数存在且大于0，则将其转换为整数，否则使用默认值20
roc3 = int(roc3) if roc3 and roc3 > 0 else 20
# 若roc4参数存在且大于0，则将其转换为整数，否则使用默认值30
roc4 = int(roc4) if roc4 and roc4 > 0 else 30
# 若sma1参数存在且大于0，则将其转换为整数，否则使用默认值10
sma1 = int(sma1) if sma1 and sma1 > 0 else 10
# 若sma2参数存在且大于0，则将其转换为整数，否则使用默认值10
sma2 = int(sma2) if sma2 and sma2 > 0 else 10
# 若sma3参数存在且大于0，则将其转换为整数，否则使用默认值10
sma3 = int(sma3) if sma3 and sma3 > 0 else 10
# 若sma4参数存在且大于0，则将其转换为整数，否则使用默认值15
sma4 = int(sma4) if sma4 and sma4 > 0 else 15
# 若signal参数存在且大于0，则将其转换为整数，否则使用默认值9
signal = int(signal) if signal and signal > 0 else 9
# 计算参数的最大值
_length = max(roc1, roc2, roc3, roc4, sma1, sma2, sma3, sma4, signal)
# 验证序列
close = verify_series(close, _length)
# 获取漂移和偏移
drift = get_drift(drift)
offset = get_offset(offset)
# 若close为空，则返回空值
if close is None: return
# 计算结果
# 计算第一个ROC的移动平均值
rocma1 = roc(close, roc1).rolling(sma1).mean()
# 计算第二个ROC的移动平均值
rocma2 = roc(close, roc2).rolling(sma2).mean()
# 计算第三个ROC的移动平均值
rocma3 = roc(close, roc3).rolling(sma3).mean()
# 计算第四个ROC的移动平均值
rocma4 = roc(close, roc4).rolling(sma4).mean()
# 计算KST值
kst = 100 * (rocma1 + 2 * rocma2 + 3 * rocma3 + 4 * rocma4)
# 计算KST信号线
kst_signal = kst.rolling(signal).mean()
# 偏移
if offset != 0:
kst = kst.shift(offset)
kst_signal = kst_signal.shift(offset)
# 处理填充值
if "fillna" in kwargs:
kst.fillna(kwargs["fillna"], inplace=True)
kst_signal.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
kst.fillna(method=kwargs["fill_method"], inplace=True)
kst_signal.fillna(method=kwargs["fill_method"], inplace=True)
# 命名和分类
# 命名KST指标
kst.name = f"KST_{roc1}_{roc2}_{roc3}_{roc4}_{sma1}_{sma2}_{sma3}_{sma4}"
# 命名KST信号线
kst_signal.name = f"KSTs_{signal}"
# 分类为动量类指标
kst.category = kst_signal.category = "momentum"
# 准备返回的DataFrame
data = {kst.name: kst, kst_signal.name: kst_signal}
kstdf = DataFrame(data)
# 命名DataFrame
kstdf.name = f"KST_{roc1}_{roc2}_{roc3}_{roc4}_{sma1}_{sma2}_{sma3}_{sma4}_{signal}"
# 分类为动量类指标
kstdf.category = "momentum"
return kstdf
# 设置函数文档字符串
kst.__doc__ = \
"""'Know Sure Thing' (KST)
The 'Know Sure Thing' is a momentum based oscillator and based on ROC.
Sources:
https://www.tradingview.com/wiki/Know_Sure_Thing_(KST)
https://www.incrediblecharts.com/indicators/kst.php
Calculation:
Default Inputs:
roc1=10, roc2=15, roc3=20, roc4=30,
sma1=10, sma2=10, sma3=10, sma4=15, signal=9, drift=1
ROC = Rate of Change
SMA = Simple Moving Average
rocsma1 = SMA(ROC(close, roc1), sma1)
rocsma2 = SMA(ROC(close, roc2), sma2)
rocsma3 = SMA(ROC(close, roc3), sma3)
rocsma4 = SMA(ROC(close, roc4), sma4)
KST = 100 * (rocsma1 + 2 * rocsma2 + 3 * rocsma3 +
# 'close'是一个 pd.Series 对象，代表了股票的收盘价数据
close (pd.Series): Series of 'close's
# roc1 是 ROC 指标的第一个周期，缺省值为 10
roc1 (int): ROC 1 period. Default: 10
# roc2 是 ROC 指标的第二个周期，缺省值为 15
roc2 (int): ROC 2 period. Default: 15
# roc3 是 ROC 指标的第三个周期，缺省值为 20
roc3 (int): ROC 3 period. Default: 20
# roc4 是 ROC 指标的第四个周期，缺省值为 30
roc4 (int): ROC 4 period. Default: 30
# sma1 是简单移动平均线（SMA）的第一个周期，缺省值为 10
sma1 (int): SMA 1 period. Default: 10
# sma2 是简单移动平均线（SMA）的第二个周期，缺省值为 10
sma2 (int): SMA 2 period. Default: 10
# sma3 是简单移动平均线（SMA）的第三个周期，缺省值为 10
sma3 (int): SMA 3 period. Default: 10
# sma4 是简单移动平均线（SMA）的第四个周期，缺省值为 15
sma4 (int): SMA 4 period. Default: 15
# signal 是信号线的周期，缺省值为 9
signal (int): It's period. Default: 9
# drift 是差分周期，缺省值为 1
drift (int): The difference period. Default: 1
# offset 是结果的偏移周期数，缺省值为 0
offset (int): How many periods to offset the result. Default: 0
Kwargs:
# fillna参数用于填充缺失值，其值将传递给pd.DataFrame.fillna()方法
fillna (value, optional): pd.DataFrame.fillna(value)
# fill_method参数用于指定填充方法的类型
fill_method (value, optional): Type of fill method
Returns:
# 返回一个Pandas DataFrame对象，包含kst和kst_signal两列
pd.DataFrame: kst and kst_signal columns
.\pandas-ta\pandas_ta\momentum\macd.py
# -*- coding: utf-8 -*-
# 从 pandas 库中导入 concat 和 DataFrame 函数
from pandas import concat, DataFrame
# 从 pandas_ta 库中导入 Imports 模块
from pandas_ta import Imports
# 从 pandas_ta.overlap 模块中导入 ema 函数
from pandas_ta.overlap import ema
# 从 pandas_ta.utils 模块中导入 get_offset, verify_series, signals 函数
from pandas_ta.utils import get_offset, verify_series, signals
# 定义 MACD 指标函数，参数包括 close（收盘价）、fast（快速线周期）、slow（慢速线周期）、signal（信号线周期）、talib（是否使用 talib 库计算）、offset（偏移量）等
def macd(close, fast=None, slow=None, signal=None, talib=None, offset=None, **kwargs):
"""Indicator: Moving Average, Convergence/Divergence (MACD)"""
# 验证参数
fast = int(fast) if fast and fast > 0 else 12
slow = int(slow) if slow and slow > 0 else 26
signal = int(signal) if signal and signal > 0 else 9
if slow < fast:
fast, slow = slow, fast
close = verify_series(close, max(fast, slow, signal))
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
as_mode = kwargs.setdefault("asmode", False)
# 计算结果
if Imports["talib"] and mode_tal:
from talib import MACD
macd, signalma, histogram = MACD(close, fast, slow, signal)
else:
fastma = ema(close, length=fast)
slowma = ema(close, length=slow)
macd = fastma - slowma
signalma = ema(close=macd.loc[macd.first_valid_index():,], length=signal)
histogram = macd - signalma
if as_mode:
macd = macd - signalma
signalma = ema(close=macd.loc[macd.first_valid_index():,], length=signal)
histogram = macd - signalma
# 偏移
if offset != 0:
macd = macd.shift(offset)
histogram = histogram.shift(offset)
signalma = signalma.shift(offset)
# 处理填充
if "fillna" in kwargs:
macd.fillna(kwargs["fillna"], inplace=True)
histogram.fillna(kwargs["fillna"], inplace=True)
signalma.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
macd.fillna(method=kwargs["fill_method"], inplace=True)
histogram.fillna(method=kwargs["fill_method"], inplace=True)
signalma.fillna(method=kwargs["fill_method"], inplace=True)
# 命名和分类
_asmode = "AS" if as_mode else ""
_props = f"_{fast}_{slow}_{signal}"
macd.name = f"MACD{_asmode}{_props}"
histogram.name = f"MACD{_asmode}h{_props}"
signalma.name = f"MACD{_asmode}s{_props}"
macd.category = histogram.category = signalma.category = "momentum"
# 准备返回的 DataFrame
data = {macd.name: macd, histogram.name: histogram, signalma.name: signalma}
df = DataFrame(data)
df.name = f"MACD{_asmode}{_props}"
df.category = macd.category
signal_indicators = kwargs.pop("signal_indicators", False)
# 如果信号指标存在
if signal_indicators:
# 将数据框与直方图信号和 MACD 信号合并
signalsdf = concat(
[
df,
signals(
indicator=histogram,
xa=kwargs.pop("xa", 0),
xb=kwargs.pop("xb", None),
xserie=kwargs.pop("xserie", None),
xserie_a=kwargs.pop("xserie_a", None),
xserie_b=kwargs.pop("xserie_b", None),
cross_values=kwargs.pop("cross_values", True),
cross_series=kwargs.pop("cross_series", True),
offset=offset,
),
signals(
indicator=macd,
xa=kwargs.pop("xa", 0),
xb=kwargs.pop("xb", None),
xserie=kwargs.pop("xserie", None),
xserie_a=kwargs.pop("xserie_a", None),
xserie_b=kwargs.pop("xserie_b", None),
cross_values=kwargs.pop("cross_values", False),
cross_series=kwargs.pop("cross_series", True),
offset=offset,
),
],
axis=1,
)
# 返回合并后的数据框
return signalsdf
else:
# 如果信号指标不存在，直接返回原始数据框
return df
macd.__doc__ = \
"""Moving Average Convergence Divergence (MACD)
The MACD is a popular indicator to that is used to identify a security's trend.
While APO and MACD are the same calculation, MACD also returns two more series
called Signal and Histogram. The Signal is an EMA of MACD and the Histogram is
the difference of MACD and Signal.
Sources:
https://www.tradingview.com/wiki/MACD_(Moving_Average_Convergence/Divergence)
AS Mode: https://tr.tradingview.com/script/YFlKXHnP/
Calculation:
Default Inputs:
fast=12, slow=26, signal=9
EMA = Exponential Moving Average
MACD = EMA(close, fast) - EMA(close, slow)
Signal = EMA(MACD, signal)
Histogram = MACD - Signal
if asmode:
MACD = MACD - Signal
Signal = EMA(MACD, signal)
Histogram = MACD - Signal
Args:
close (pd.Series): Series of 'close's
fast (int): The short period. Default: 12
slow (int): The long period. Default: 26
signal (int): The signal period. Default: 9
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:
asmode (value, optional): When True, enables AS version of MACD.
Default: False
fillna (value, optional): pd.DataFrame.fillna(value)
fill_method (value, optional): Type of fill method
Returns:
pd.DataFrame: macd, histogram, signal columns.
"""
.\pandas-ta\pandas_ta\momentum\mom.py
# -*- coding: utf-8 -*-
# 从 pandas_ta 模块导入 Imports
from pandas_ta import Imports
# 从 pandas_ta.utils 模块导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series
# 定义动量指标（Momentum）函数
def mom(close, length=None, talib=None, offset=None, **kwargs):
"""Indicator: Momentum (MOM)"""
# 验证参数
# 如果 length 参数存在且大于 0，则将其转换为整数，否则设为默认值 10
length = int(length) if length and length > 0 else 10
# 验证 close 是否为有效的数据序列，并设置长度
close = verify_series(close, length)
# 获取偏移量
offset = get_offset(offset)
# 设置是否使用 talib 模式
mode_tal = bool(talib) if isinstance(talib, bool) else True
# 如果 close 为空，则返回 None
if close is None: return
# 计算结果
# 如果 Imports["talib"] 为真且 mode_tal 为真，则使用 TA Lib 中的 MOM 函数计算动量
if Imports["talib"] and mode_tal:
from talib import MOM
mom = MOM(close, length)
# 否则，使用差分计算动量
else:
mom = close.diff(length)
# 偏移结果
if offset != 0:
mom = mom.shift(offset)
# 处理填充值
# 如果 kwargs 中有 "fillna" 参数，则使用指定值填充空值
if "fillna" in kwargs:
mom.fillna(kwargs["fillna"], inplace=True)
# 如果 kwargs 中有 "fill_method" 参数，则使用指定填充方法填充空值
if "fill_method" in kwargs:
mom.fillna(method=kwargs["fill_method"], inplace=True)
# 设置名称和分类
# 设置动量指标的名称为 "MOM_长度"，设置分类为 "momentum"
mom.name = f"MOM_{length}"
mom.category = "momentum"
# 返回动量指标结果
return mom
# 设置动量指标函数的文档字符串
mom.__doc__ = \
"""Momentum (MOM)
Momentum is an indicator used to measure a security's speed (or strength) of
movement. Or simply the change in price.
Sources:
http://www.onlinetradingconcepts.com/TechnicalAnalysis/Momentum.html
Calculation:
Default Inputs:
length=1
MOM = close.diff(length)
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 1
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.
"""
.\pandas-ta\pandas_ta\momentum\pgo.py
# 设置文件编码格式为 UTF-8
# -*- coding: utf-8 -*-
# 导入所需模块和函数
from pandas_ta.overlap import ema, sma
from pandas_ta.volatility import atr
from pandas_ta.utils import get_offset, verify_series
# 定义 Pretty Good Oscillator (PGO) 指标函数
def pgo(high, low, close, length=None, offset=None, **kwargs):
"""Indicator: Pretty Good Oscillator (PGO)"""
# 验证参数
# 如果 length 有值并且大于 0，将其转换为整数，否则设为默认值 14
length = int(length) if length and length > 0 else 14
# 验证 high、low、close 参数并限制长度为 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
# 计算 PGO 指标
# PGO = (close - SMA(close, length)) / EMA(ATR(high, low, close, length), length)
pgo = close - sma(close, length)
pgo /= ema(atr(high, low, close, length), length)
# 偏移结果
if offset != 0:
pgo = pgo.shift(offset)
# 处理填充
# 如果 kwargs 中含有 "fillna"，使用指定的填充值进行填充
if "fillna" in kwargs:
pgo.fillna(kwargs["fillna"], inplace=True)
# 如果 kwargs 中含有 "fill_method"，使用指定的填充方法进行填充
if "fill_method" in kwargs:
pgo.fillna(method=kwargs["fill_method"], inplace=True)
# 设置指标名称和分类
pgo.name = f"PGO_{length}"
pgo.category = "momentum"
return pgo
# 为 PGO 函数添加文档字符串
pgo.__doc__ = \
"""Pretty Good Oscillator (PGO)
The Pretty Good Oscillator indicator was created by Mark Johnson to measure the distance of the current close from its N-day Simple Moving Average, expressed in terms of an average true range over a similar period. Johnson's approach was to
use it as a breakout system for longer term trades. Long if greater than 3.0 and
short if less than -3.0.
Sources:
https://library.tradingtechnologies.com/trade/chrt-ti-pretty-good-oscillator.html
Calculation:
Default Inputs:
length=14
ATR = Average True Range
SMA = Simple Moving Average
EMA = Exponential Moving Average
PGO = (close - SMA(close, length)) / EMA(ATR(high, low, close, length), length)
Args:
high (pd.Series): Series of 'high's
low (pd.Series): Series of 'low's
close (pd.Series): Series of 'close's
length (int): It's period. Default: 14
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\momentum\ppo.py
# -*- coding: utf-8 -*-
# 导入所需的库和模块
from pandas import DataFrame
from pandas_ta import Imports
from pandas_ta.overlap import ma
from pandas_ta.utils import get_offset, tal_ma, verify_series
# 定义 PPO 函数，计算百分比价格振荡器（PPO）
def ppo(close, fast=None, slow=None, signal=None, scalar=None, mamode=None, talib=None, offset=None, **kwargs):
"""Indicator: Percentage Price Oscillator (PPO)"""
# 验证参数
# 设置默认值并确保参数为整数或浮点数
fast = int(fast) if fast and fast > 0 else 12
slow = int(slow) if slow and slow > 0 else 26
signal = int(signal) if signal and signal > 0 else 9
scalar = float(scalar) if scalar else 100
mamode = mamode if isinstance(mamode, str) else "sma"
# 如果 slow 小于 fast，则交换它们的值
if slow < fast:
fast, slow = slow, fast
# 验证 close 数据，并获取偏移量
close = verify_series(close, max(fast, slow, signal))
offset = get_offset(offset)
# 判断是否使用 talib 库
mode_tal = bool(talib) if isinstance(talib, bool) else True
# 如果 close 为空，则返回
if close is None: return
# 计算结果
if Imports["talib"] and mode_tal:
# 使用 talib 库计算 PPO
from talib import PPO
ppo = PPO(close, fast, slow, tal_ma(mamode))
else:
# 使用自定义函数计算 PPO
fastma = ma(mamode, close, length=fast)
slowma = ma(mamode, close, length=slow)
ppo = scalar * (fastma - slowma)
ppo /= slowma
# 计算信号线和直方图
signalma = ma("ema", ppo, length=signal)
histogram = ppo - signalma
# 处理偏移
if offset != 0:
ppo = ppo.shift(offset)
histogram = histogram.shift(offset)
signalma = signalma.shift(offset)
# 处理填充值
if "fillna" in kwargs:
ppo.fillna(kwargs["fillna"], inplace=True)
histogram.fillna(kwargs["fillna"], inplace=True)
signalma.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
ppo.fillna(method=kwargs["fill_method"], inplace=True)
histogram.fillna(method=kwargs["fill_method"], inplace=True)
signalma.fillna(method=kwargs["fill_method"], inplace=True)
# 设置名称和分类
_props = f"_{fast}_{slow}_{signal}"
ppo.name = f"PPO{_props}"
histogram.name = f"PPOh{_props}"
signalma.name = f"PPOs{_props}"
ppo.category = histogram.category = signalma.category = "momentum"
# 准备返回的 DataFrame
data = {ppo.name: ppo, histogram.name: histogram, signalma.name: signalma}
df = DataFrame(data)
df.name = f"PPO{_props}"
df.category = ppo.category
return df
# 设置 PPO 函数的文档字符串
ppo.__doc__ = \
"""Percentage Price Oscillator (PPO)
The Percentage Price Oscillator is similar to MACD in measuring momentum.
Sources:
https://www.tradingview.com/wiki/MACD_(Moving_Average_Convergence/Divergence)
Calculation:
Default Inputs:
fast=12, slow=26
SMA = Simple Moving Average
EMA = Exponential Moving Average
fast_sma = SMA(close, fast)
slow_sma = SMA(close, slow)
PPO = 100 * (fast_sma - slow_sma) / slow_sma
Signal = EMA(PPO, signal)
Histogram = PPO - Signal
Args:
close(pandas.Series): Series of 'close's
fast(int): The short period. Default: 12
slow(int): The long period. Default: 26
"""
# 定义一个函数参数 signal，表示信号周期，默认值为 9
signal(int): The signal period. Default: 9
# 定义一个函数参数 scalar，表示放大倍数，默认值为 100
scalar (float): How much to magnify. Default: 100
# 定义一个函数参数 mamode，表示移动平均模式，查看可用模式的说明可调用 help(ta.ma)，默认值为 'sma'
mamode (str): See ```help(ta.ma)```py. Default: 'sma'
# 定义一个函数参数 talib，表示是否使用 TA Lib，如果 TA Lib 已安装且 talib 为 True，则返回 TA Lib 版本，默认值为 True
talib (bool): If TA Lib is installed and talib is True, Returns the TA Lib version. Default: True
# 定义一个函数参数 offset，表示结果偏移多少周期，默认值为 0
offset(int): How many periods to offset the result. Default: 0
# 定义函数的参数及其作用
Kwargs:
# 填充缺失值的数值或方法，用于填充 DataFrame 的缺失值
fillna (value, optional): pd.DataFrame.fillna(value)
# 填充方法的类型，用于指定填充缺失值的方法
fill_method (value, optional): Type of fill method
# 返回值说明
Returns:
# 返回一个包含 ppo、histogram 和 signal 列的 pandas DataFrame
pd.DataFrame: ppo, histogram, signal columns
.\pandas-ta\pandas_ta\momentum\psl.py
# -*- coding: utf-8 -*-
# 从 numpy 导入 sign 函数并命名为 npSign
from numpy import sign as npSign
# 从 pandas_ta.utils 模块导入 get_drift, get_offset, verify_series 函数
from pandas_ta.utils import get_drift, get_offset, verify_series
# 定义函数 psl，计算心理线指标
def psl(close, open_=None, length=None, scalar=None, drift=None, offset=None, **kwargs):
"""Indicator: Psychological Line (PSL)"""
# 验证参数
# 将长度转换为整数，如果长度大于 0 则取参数值，否则默认为 12
length = int(length) if length and length > 0 else 12
# 将标量转换为浮点数，如果标量大于 0 则取参数值，否则默认为 100
scalar = float(scalar) if scalar and scalar > 0 else 100
# 验证 close 数据类型，并将其调整为指定长度
close = verify_series(close, length)
# 获取漂移值
drift = get_drift(drift)
# 获取偏移值
offset = get_offset(offset)
# 如果 close 为空，则返回空值
if close is None: return
# 计算结果
# 如果存在 open_ 参数
if open_ is not None:
# 验证 open_ 数据类型
open_ = verify_series(open_)
# 计算 close 与 open_ 之间的差异，并取其符号
diff = npSign(close - open_)
else:
# 计算 close 在漂移期内的差异，并取其符号
diff = npSign(close.diff(drift))
# 将缺失值填充为 0
diff.fillna(0, inplace=True)
# 将小于等于 0 的值设置为 0
diff[diff <= 0] = 0 # Zero negative values
# 计算心理线值
psl = scalar * diff.rolling(length).sum()
psl /= length
# 偏移
if offset != 0:
psl = psl.shift(offset)
# 填充缺失值
if "fillna" in kwargs:
psl.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
psl.fillna(method=kwargs["fill_method"], inplace=True)
# 命名和分类
_props = f"_{length}"
psl.name = f"PSL{_props}"
psl.category = "momentum"
return psl
# 设置函数 psl 的文档字符串
psl.__doc__ = \
"""Psychological Line (PSL)
The Psychological Line is an oscillator-type indicator that compares the
number of the rising periods to the total number of periods. In other
words, it is the percentage of bars that close above the previous
bar over a given period.
Sources:
https://www.quantshare.com/item-851-psychological-line
Calculation:
Default Inputs:
length=12, scalar=100, drift=1
IF NOT open:
DIFF = SIGN(close - close[drift])
ELSE:
DIFF = SIGN(close - open)
DIFF.fillna(0)
DIFF[DIFF <= 0] = 0
PSL = scalar * SUM(DIFF, length) / length
Args:
close (pd.Series): Series of 'close's
open_ (pd.Series, optional): Series of 'open's
length (int): It's period. Default: 12
scalar (float): How much to magnify. Default: 100
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\momentum\pvo.py
# -*- coding: utf-8 -*-
# 从 pandas 库中导入 DataFrame 类
from pandas import DataFrame
# 从 pandas_ta.overlap 模块中导入 ema 函数
from pandas_ta.overlap import ema
# 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series
# 定义 Percentage Volume Oscillator (PVO) 函数
def pvo(volume, fast=None, slow=None, signal=None, scalar=None, offset=None, **kwargs):
"""Indicator: Percentage Volume Oscillator (PVO)"""
# 验证参数
fast = int(fast) if fast and fast > 0 else 12
slow = int(slow) if slow and slow > 0 else 26
signal = int(signal) if signal and signal > 0 else 9
scalar = float(scalar) if scalar else 100
if slow < fast:
fast, slow = slow, fast
volume = verify_series(volume, max(fast, slow, signal))
offset = get_offset(offset)
if volume is None: return
# 计算结果
fastma = ema(volume, length=fast)
slowma = ema(volume, length=slow)
pvo = scalar * (fastma - slowma)
pvo /= slowma
signalma = ema(pvo, length=signal)
histogram = pvo - signalma
# 偏移
if offset != 0:
pvo = pvo.shift(offset)
histogram = histogram.shift(offset)
signalma = signalma.shift(offset)
# 处理填充
if "fillna" in kwargs:
pvo.fillna(kwargs["fillna"], inplace=True)
histogram.fillna(kwargs["fillna"], inplace=True)
signalma.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
pvo.fillna(method=kwargs["fill_method"], inplace=True)
histogram.fillna(method=kwargs["fill_method"], inplace=True)
signalma.fillna(method=kwargs["fill_method"], inplace=True)
# 命名和分类
_props = f"_{fast}_{slow}_{signal}"
pvo.name = f"PVO{_props}"
histogram.name = f"PVOh{_props}"
signalma.name = f"PVOs{_props}"
pvo.category = histogram.category = signalma.category = "momentum"
# 创建 DataFrame 对象
data = {pvo.name: pvo, histogram.name: histogram, signalma.name: signalma}
df = DataFrame(data)
df.name = pvo.name
df.category = pvo.category
return df
# 设置函数文档字符串
pvo.__doc__ = \
"""Percentage Volume Oscillator (PVO)
Percentage Volume Oscillator is a Momentum Oscillator for Volume.
Sources:
https://www.fmlabs.com/reference/default.htm?url=PVO.htm
Calculation:
Default Inputs:
fast=12, slow=26, signal=9
EMA = Exponential Moving Average
PVO = (EMA(volume, fast) - EMA(volume, slow)) / EMA(volume, slow)
Signal = EMA(PVO, signal)
Histogram = PVO - Signal
Args:
volume (pd.Series): Series of 'volume's
fast (int): The short period. Default: 12
slow (int): The long period. Default: 26
signal (int): The signal period. Default: 9
scalar (float): How much to magnify. 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
Returns:
pd.DataFrame: pvo, histogram, signal columns.
"""
.\pandas-ta\pandas_ta\momentum\qqe.py
# -*- coding: utf-8 -*-
# 从 numpy 库中导入 maximum 函数并重命名为 npMaximum
from numpy import maximum as npMaximum
# 从 numpy 库中导入 minimum 函数并重命名为 npMinimum
from numpy import minimum as npMinimum
# 从 numpy 库中导入 nan 常量并重命名为 npNaN
from numpy import nan as npNaN
# 从 pandas 库中导入 DataFrame 和 Series 类
from pandas import DataFrame, Series
# 从 rsi 模块中导入 rsi 函数
from .rsi import rsi
# 从 pandas_ta.overlap 模块中导入 ma 函数
from pandas_ta.overlap import ma
# 从 pandas_ta.utils 模块中导入 get_drift, get_offset, verify_series 函数
# 定义函数 qqe，计算 Quantitative Qualitative Estimation (QQE) 指标
def qqe(close, length=None, smooth=None, factor=None, mamode=None, drift=None, offset=None, **kwargs):
"""Indicator: Quantitative Qualitative Estimation (QQE)"""
# 验证参数
length = int(length) if length and length > 0 else 14
smooth = int(smooth) if smooth and smooth > 0 else 5
factor = float(factor) if factor else 4.236
wilders_length = 2 * length - 1
mamode = mamode if isinstance(mamode, str) else "ema"
# 验证 close 序列，并设置最大长度为 length、smooth 和 wilders_length 中的最大值
close = verify_series(close, max(length, smooth, wilders_length))
drift = get_drift(drift)
offset = get_offset(offset)
# 如果 close 为空，则返回空值
if close is None: return
# 计算结果
rsi_ = rsi(close, length)
_mode = mamode.lower()[0] if mamode != "ema" else ""
rsi_ma = ma(mamode, rsi_, length=smooth)
# 计算 RSI MA True Range
rsi_ma_tr = rsi_ma.diff(drift).abs()
# 使用 Wilder's Length 和默认宽度 4.236 双重平滑 RSI MA True Range
smoothed_rsi_tr_ma = ma("ema", rsi_ma_tr, length=wilders_length)
dar = factor * ma("ema", smoothed_rsi_tr_ma, length=wilders_length)
# 创建围绕 RSI MA 的上下轨带
upperband = rsi_ma + dar
lowerband = rsi_ma - dar
m = close.size
# 创建 Series 对象 long、short、trend、qqe、qqe_long 和 qqe_short
long = Series(0, index=close.index)
short = Series(0, index=close.index)
trend = Series(1, index=close.index)
qqe = Series(rsi_ma.iloc[0], index=close.index)
qqe_long = Series(npNaN, index=close.index)
qqe_short = Series(npNaN, index=close.index)
# 遍历范围为1到m-1的整数
for i in range(1, m):
# 获取当前和前一个 RSI_MA 值
c_rsi, p_rsi = rsi_ma.iloc[i], rsi_ma.iloc[i - 1]
# 获取当前和前一个 Long Line 值
c_long, p_long = long.iloc[i - 1], long.iloc[i - 2]
# 获取当前和前一个 Short Line 值
c_short, p_short = short.iloc[i - 1], short.iloc[i - 2]
# Long Line
# 如果前一个 RSI_MA 大于当前 Long Line 并且当前 RSI_MA 大于当前 Long Line
if p_rsi > c_long and c_rsi > c_long:
long.iloc[i] = npMaximum(c_long, lowerband.iloc[i])
else:
long.iloc[i] = lowerband.iloc[i]
# Short Line
# 如果前一个 RSI_MA 小于当前 Short Line 并且当前 RSI_MA 小于当前 Short Line
if p_rsi < c_short and c_rsi < c_short:
short.iloc[i] = npMinimum(c_short, upperband.iloc[i])
else:
short.iloc[i] = upperband.iloc[i]
# Trend & QQE Calculation
# Long: 当前 RSI_MA 值穿过前一个 Short Line 值
# Short: 当前 RSI_MA 值穿过前一个 Long Line 值
if (c_rsi > c_short and p_rsi < p_short) or (c_rsi <= c_short and p_rsi >= p_short):
trend.iloc[i] = 1
qqe.iloc[i] = qqe_long.iloc[i] = long.iloc[i]
elif (c_rsi > c_long and p_rsi < p_long) or (c_rsi <= c_long and p_rsi >= p_long):
trend.iloc[i] = -1
qqe.iloc[i] = qqe_short.iloc[i] = short.iloc[i]
else:
trend.iloc[i] = trend.iloc[i - 1]
if trend.iloc[i] == 1:
qqe.iloc[i] = qqe_long.iloc[i] = long.iloc[i]
else:
qqe.iloc[i] = qqe_short.iloc[i] = short.iloc[i]
# Offset
# 如果偏移量不为0，则对数据进行偏移
if offset != 0:
rsi_ma = rsi_ma.shift(offset)
qqe = qqe.shift(offset)
long = long.shift(offset)
short = short.shift(offset)
# Handle fills
# 处理填充值
if "fillna" in kwargs:
rsi_ma.fillna(kwargs["fillna"], inplace=True)
qqe.fillna(kwargs["fillna"], inplace=True)
qqe_long.fillna(kwargs["fillna"], inplace=True)
qqe_short.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
rsi_ma.fillna(method=kwargs["fill_method"], inplace=True)
qqe.fillna(method=kwargs["fill_method"], inplace=True)
qqe_long.fillna(method=kwargs["fill_method"], inplace=True)
qqe_short.fillna(method=kwargs["fill_method"], inplace=True)
# Name and Categorize it
# 设置名称和分类
_props = f"{_mode}_{length}_{smooth}_{factor}"
qqe.name = f"QQE{_props}"
rsi_ma.name = f"QQE{_props}_RSI{_mode.upper()}MA"
qqe_long.name = f"QQEl{_props}"
qqe_short.name = f"QQEs{_props}"
qqe.category = rsi_ma.category = "momentum"
qqe_long.category = qqe_short.category = qqe.category
# Prepare DataFrame to return
# 准备要返回的 DataFrame
data = {
qqe.name: qqe, rsi_ma.name: rsi_ma,
# long.name: long, short.name: short
qqe_long.name: qqe_long, qqe_short.name: qqe_short
}
df = DataFrame(data)
df.name = f"QQE{_props}"
df.category = qqe.category
return df
# 设置 qqe 对象的文档字符串，描述了 Quantitative Qualitative Estimation (QQE) 指标的计算方法和用法
qqe.__doc__ = \
"""Quantitative Qualitative Estimation (QQE)
The Quantitative Qualitative Estimation (QQE) is similar to SuperTrend but uses a Smoothed RSI with an upper and lower bands. The band width is a combination of a one period True Range of the Smoothed RSI which is double smoothed using Wilder's smoothing length (2 * rsiLength - 1) and multiplied by the default factor of 4.236. A Long trend is determined when the Smoothed RSI crosses the previous upperband and a Short trend when the Smoothed RSI crosses the previous lowerband.
Based on QQE.mq5 by EarnForex Copyright © 2010, based on version by Tim Hyder (2008), based on version by Roman Ignatov (2006)
Sources:
https://www.tradingview.com/script/IYfA9R2k-QQE-MT4/
https://www.tradingpedia.com/forex-trading-indicators/quantitative-qualitative-estimation
https://www.prorealcode.com/prorealtime-indicators/qqe-quantitative-qualitative-estimation/
Calculation:
Default Inputs:
length=14, smooth=5, factor=4.236, mamode="ema", drift=1
Args:
close (pd.Series): Series of 'close's
length (int): RSI period. Default: 14
smooth (int): RSI smoothing period. Default: 5
factor (float): QQE Factor. Default: 4.236
mamode (str): See ```help(ta.ma)```py. Default: 'sma'
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.DataFrame: QQE, RSI_MA (basis), QQEl (long), and QQEs (short) columns.
"""
.\pandas-ta\pandas_ta\momentum\roc.py
# -*- coding: utf-8 -*-
# 从 mom 模块中导入 mom 函数
from .mom import mom
# 从 pandas_ta 模块中导入 Imports
from pandas_ta import Imports
# 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数
from pandas_ta.utils import get_offset, verify_series
# 定义 Rate of Change (ROC) 函数
def roc(close, length=None, scalar=None, talib=None, offset=None, **kwargs):
"""Indicator: Rate of Change (ROC)"""
# 验证参数
length = int(length) if length and length > 0 else 10
scalar = float(scalar) if scalar and scalar > 0 else 100
close = verify_series(close, length)
offset = get_offset(offset)
mode_tal = bool(talib) if isinstance(talib, bool) else True
if close is None: return
# 计算结果
if Imports["talib"] and mode_tal:
from talib import ROC
roc = ROC(close, length)
else:
roc = scalar * mom(close=close, length=length) / close.shift(length)
# 偏移
if offset != 0:
roc = roc.shift(offset)
# 处理填充
if "fillna" in kwargs:
roc.fillna(kwargs["fillna"], inplace=True)
if "fill_method" in kwargs:
roc.fillna(method=kwargs["fill_method"], inplace=True)
# 命名和分类
roc.name = f"ROC_{length}"
roc.category = "momentum"
return roc
# 设置 ROC 函数的文档字符串
roc.__doc__ = \
"""Rate of Change (ROC)
Rate of Change is an indicator is also referred to as Momentum (yeah, confusingly).
It is a pure momentum oscillator that measures the percent change in price with the
previous price 'n' (or length) periods ago.
Sources:
https://www.tradingview.com/wiki/Rate_of_Change_(ROC)
Calculation:
Default Inputs:
length=1
MOM = Momentum
ROC = 100 * MOM(close, length) / close.shift(length)
Args:
close (pd.Series): Series of 'close's
length (int): It's period. Default: 1
scalar (float): How much to magnify. Default: 100
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.
"""
标签：PandasTA,
signal,
kwargs,
length,
源码,
offset,
close,
fillna,
解析
From： https://www.cnblogs.com/apachecn/p/18135771