PandasTA 源码解析（十一）（1）https://developer.aliyun.com/article/1506211
.\pandas-ta\pandas_ta\performance\log_return.py
# -*- coding: utf-8 -*- # 从 numpy 中导入 log 函数并重命名为 nplog from numpy import log as nplog # 从 pandas_ta.utils 中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series def log_return(close, length=None, cumulative=None, offset=None, **kwargs): """Indicator: Log Return""" # 验证参数 # 如果 length 存在且大于 0，则将其转换为整数，否则设为 1 length = int(length) if length and length > 0 else 1 # 如果 cumulative 存在且为真，则设为 True，否则设为 False cumulative = bool(cumulative) if cumulative is not None and cumulative else False # 验证 close 是否为有效的 Series，并设置长度 close = verify_series(close, length) # 获取偏移量 offset = get_offset(offset) # 如果 close 为空，则返回 None if close is None: return # 计算结果 if cumulative: # 计算累积对数收益率 log_return = nplog(close / close.iloc[0]) else: # 计算对数收益率 log_return = nplog(close / close.shift(length)) # 偏移结果 if offset != 0: log_return = log_return.shift(offset) # 处理填充值 if "fillna" in kwargs: log_return.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: log_return.fillna(method=kwargs["fill_method"], inplace=True) # 设置名称和类别 log_return.name = f"{'CUM' if cumulative else ''}LOGRET_{length}" log_return.category = "performance" return log_return # 设置 log_return 函数的文档字符串 log_return.__doc__ = \ """Log Return Calculates the logarithmic return of a Series. See also: help(df.ta.log_return) for additional **kwargs a valid 'df'. Sources: https://stackoverflow.com/questions/31287552/logarithmic-returns-in-pandas-dataframe Calculation: Default Inputs: length=1, cumulative=False LOGRET = log( close.diff(periods=length) ) CUMLOGRET = LOGRET.cumsum() if cumulative Args: close (pd.Series): Series of 'close's length (int): It's period. Default: 20 cumulative (bool): If True, returns the cumulative returns. Default: False 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\performance\percent_return.py
# -*- coding: utf-8 -*- # 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义一个名为 percent_return 的函数，用于计算百分比收益率 def percent_return(close, length=None, cumulative=None, offset=None, **kwargs): """Indicator: Percent Return""" # 验证参数 length = int(length) if length and length > 0 else 1 cumulative = bool(cumulative) if cumulative is not None and cumulative else False close = verify_series(close, length) offset = get_offset(offset) if close is None: return # 计算结果 if cumulative: pct_return = (close / close.iloc[0]) - 1 else: pct_return = close.pct_change(length) # (close / close.shift(length)) - 1 # 偏移 if offset != 0: pct_return = pct_return.shift(offset) # 处理填充 if "fillna" in kwargs: pct_return.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: pct_return.fillna(method=kwargs["fill_method"], inplace=True) # 设置名称和类别 pct_return.name = f"{'CUM' if cumulative else ''}PCTRET_{length}" pct_return.category = "performance" return pct_return # 设置 percent_return 函数的文档字符串 percent_return.__doc__ = \ """Percent Return Calculates the percent return of a Series. See also: help(df.ta.percent_return) for additional **kwargs a valid 'df'. Sources: https://stackoverflow.com/questions/31287552/logarithmic-returns-in-pandas-dataframe Calculation: Default Inputs: length=1, cumulative=False PCTRET = close.pct_change(length) CUMPCTRET = PCTRET.cumsum() if cumulative Args: close (pd.Series): Series of 'close's length (int): It's period. Default: 20 cumulative (bool): If True, returns the cumulative returns. Default: False 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\performance\__init__.py
# 设置文件编码为 UTF-8 # 导入 drawdown 模块中的 drawdown 函数 # 导入 log_return 模块中的 log_return 函数 # 导入 percent_return 模块中的 percent_return 函数
.\pandas-ta\pandas_ta\statistics\entropy.py
# -*- coding: utf-8 -*- # 导入 log 函数并将其命名为 npLog from numpy import log as npLog # 导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义熵指标函数，接受收盘价、周期、对数的基数、偏移量和其他参数 def entropy(close, length=None, base=None, offset=None, **kwargs): """Indicator: Entropy (ENTP)""" # 验证参数 # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 10 length = int(length) if length and length > 0 else 10 # 如果 base 存在且大于 0，则将其转换为浮点数，否则设为默认值 2.0 base = float(base) if base and base > 0 else 2.0 # 验证收盘价是否是一个有效的 Series，如果不是则返回 None close = verify_series(close, length) # 获取偏移量 offset = get_offset(offset) # 如果收盘价为 None，则返回 None if close is None: return # 计算结果 # 计算每个价格占总和的比例 p = close / close.rolling(length).sum() # 计算熵 entropy = (-p * npLog(p) / npLog(base)).rolling(length).sum() # 偏移结果 if offset != 0: entropy = entropy.shift(offset) # 处理填充 # 如果参数中包含 "fillna"，则使用指定值填充空值 if "fillna" in kwargs: entropy.fillna(kwargs["fillna"], inplace=True) # 如果参数中包含 "fill_method"，则使用指定的填充方法填充空值 if "fill_method" in kwargs: entropy.fillna(method=kwargs["fill_method"], inplace=True) # 设置名称和类别 entropy.name = f"ENTP_{length}" entropy.category = "statistics" return entropy # 设置熵指标函数的文档字符串 entropy.__doc__ = \ """Entropy (ENTP) Introduced by Claude Shannon in 1948, entropy measures the unpredictability of the data, or equivalently, of its average information. A die has higher entropy (p=1/6) versus a coin (p=1/2). Sources: https://en.wikipedia.org/wiki/Entropy_(information_theory) Calculation: Default Inputs: length=10, base=2 P = close / SUM(close, length) E = SUM(-P * npLog(P) / npLog(base), length) Args: close (pd.Series): Series of 'close's length (int): It's period. Default: 10 base (float): Logarithmic Base. Default: 2 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\statistics\kurtosis.py
# -*- coding: utf-8 -*- # 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义函数 kurtosis，计算某个时间段内的峰度 def kurtosis(close, length=None, offset=None, **kwargs): """Indicator: Kurtosis""" # 验证参数 # 如果 length 存在且大于 0，则将其转换为整数；否则将 length 设为默认值 30 length = int(length) if length and length > 0 else 30 # 如果 kwargs 中有 "min_periods" 参数且不为 None，则将其转换为整数；否则将 min_periods 设为 length 的值 min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length # 验证 close 参数，并确保其长度不小于 length 和 min_periods 的最大值 close = verify_series(close, max(length, min_periods)) # 获取偏移量 offset = get_offset(offset) # 如果 close 为 None，则返回空值 if close is None: return # 计算结果 # 计算 close 的滚动窗口长度为 length 的峰度，并使用 min_periods 参数指定最小期数 kurtosis = close.rolling(length, min_periods=min_periods).kurt() # 处理偏移 if offset != 0: # 对计算的峰度结果进行偏移 kurtosis = kurtosis.shift(offset) # 处理填充 if "fillna" in kwargs: # 使用指定的值填充缺失值 kurtosis.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: # 使用指定的填充方法填充缺失值 kurtosis.fillna(method=kwargs["fill_method"], inplace=True) # 设定指标的名称和类别 kurtosis.name = f"KURT_{length}" kurtosis.category = "statistics" # 返回计算结果 return kurtosis # 为函数 kurtosis 添加文档字符串 kurtosis.__doc__ = \ """Rolling Kurtosis Sources: Calculation: Default Inputs: length=30 KURTOSIS = close.rolling(length).kurt() Args: close (pd.Series): Series of 'close's length (int): It's period. Default: 30 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 源码解析（十一）（3）https://developer.aliyun.com/article/1506214
