PandasTA 源码解析（十一）（2）https://developer.aliyun.com/article/1506213
.\pandas-ta\pandas_ta\statistics\mad.py
# -*- coding: utf-8 -*- # 从 numpy 中导入 fabs 函数并重命名为 npfabs from numpy import fabs as npfabs # 从 pandas_ta.utils 中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义函数：均值绝对偏差 def mad(close, length=None, offset=None, **kwargs): """Indicator: Mean Absolute Deviation""" # 验证参数 # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 30 length = int(length) if length and length > 0 else 30 # 如果 kwargs 中存在 "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 为空，则返回空值 if close is None: return # 计算结果 def mad_(series): """Mean Absolute Deviation""" # 计算序列与其均值的绝对差值的均值 return npfabs(series - series.mean()).mean() # 使用 rolling 函数计算滚动均值绝对偏差 mad = close.rolling(length, min_periods=min_periods).apply(mad_, raw=True) # 偏移 if offset != 0: mad = mad.shift(offset) # 处理填充 # 如果 kwargs 中存在 "fillna"，则使用该值填充空值 if "fillna" in kwargs: mad.fillna(kwargs["fillna"], inplace=True) # 如果 kwargs 中存在 "fill_method"，则使用指定的填充方法 if "fill_method" in kwargs: mad.fillna(method=kwargs["fill_method"], inplace=True) # 设定指标的名称和类别 mad.name = f"MAD_{length}" mad.category = "statistics" return mad # 设置函数文档字符串 mad.__doc__ = \ """Rolling Mean Absolute Deviation Sources: Calculation: Default Inputs: length=30 mad = close.rolling(length).mad() 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. """
.\pandas-ta\pandas_ta\statistics\median.py
# -*- coding: utf-8 -*- # 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义一个名为 median 的函数，用于计算中位数指标 def median(close, length=None, offset=None, **kwargs): """Indicator: Median""" # 验证参数 # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 30 length = int(length) if length and length > 0 else 30 # 如果 kwargs 中存在 "min_periods"，且其值不为 None，则将其转换为整数，否则设为 length 的值 min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length # 验证 close 是否为有效的 Series，并设定最小长度为 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 median = close.rolling(length, min_periods=min_periods).median() # 偏移结果 if offset != 0: median = median.shift(offset) # 处理填充值 if "fillna" in kwargs: median.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: median.fillna(method=kwargs["fill_method"], inplace=True) # 设置指标名称和类别 median.name = f"MEDIAN_{length}" median.category = "statistics" return median # 设置 median 函数的文档字符串 median.__doc__ = \ """Rolling Median Rolling Median of over 'n' periods. Sibling of a Simple Moving Average. Sources: https://www.incrediblecharts.com/indicators/median_price.php Calculation: Default Inputs: length=30 MEDIAN = close.rolling(length).median() 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. """
.\pandas-ta\pandas_ta\statistics\quantile.py
# -*- coding: utf-8 -*- # 从 pandas_ta.utils 导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义一个名为 quantile 的函数，用于计算滚动分位数 def quantile(close, length=None, q=None, offset=None, **kwargs): """Indicator: Quantile""" # 函数文档字符串，指示 quantile 函数的作用 # 验证参数 # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 30 length = int(length) if length and length > 0 else 30 # 如果 kwargs 中存在 "min_periods" 键且其值不为 None，则将其转换为整数，否则设为 length 的值 min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length # 如果 q 存在且大于 0 且小于 1，则将其转换为浮点数，否则设为默认值 0.5 q = float(q) if q and q > 0 and q < 1 else 0.5 # 验证 close 序列，确保其长度不小于 length 和 min_periods 中的较大值 close = verify_series(close, max(length, min_periods)) # 获取偏移量 offset = get_offset(offset) # 如果 close 为 None，则返回 None if close is None: return # 计算结果 # 使用 close 序列进行滚动窗口计算分位数，窗口长度为 length，最小周期数为 min_periods quantile = close.rolling(length, min_periods=min_periods).quantile(q) # 偏移结果 # 如果偏移量不为 0，则对 quantile 序列进行偏移 if offset != 0: quantile = quantile.shift(offset) # 处理填充值 # 如果 kwargs 中存在 "fillna" 键，则使用给定值填充 quantile 序列的缺失值 if "fillna" in kwargs: quantile.fillna(kwargs["fillna"], inplace=True) # 如果 kwargs 中存在 "fill_method" 键，则使用指定的填充方法填充 quantile 序列的缺失值 if "fill_method" in kwargs: quantile.fillna(method=kwargs["fill_method"], inplace=True) # 序列命名和分类 # 设置 quantile 序列的名称为 "QTL_{length}_{q}" quantile.name = f"QTL_{length}_{q}" # 设置 quantile 序列的分类为 "statistics" quantile.category = "statistics" return quantile # 返回 quantile 序列 # 为 quantile 函数添加文档字符串，说明其作用、源以及参数等信息 quantile.__doc__ = \ """Rolling Quantile Sources: Calculation: Default Inputs: length=30, q=0.5 QUANTILE = close.rolling(length).quantile(q) Args: close (pd.Series): Series of 'close's length (int): It's period. Default: 30 q (float): The quantile. Default: 0.5 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\skew.py
# -*- coding: utf-8 -*- # 从 pandas_ta.utils 中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series def skew(close, length=None, offset=None, **kwargs): """Indicator: Skew""" # 验证参数 # 如果 length 存在且大于0，则将其转换为整数，否则默认为30 length = int(length) if length and length > 0 else 30 # 如果 kwargs 中包含 "min_periods" 并且其值不为 None，则将其转换为整数，否则默认为 length min_periods = int(kwargs["min_periods"]) if "min_periods" in kwargs and kwargs["min_periods"] is not None else length # 验证 close 是否为 pd.Series 类型，并保证其长度不小于 length 和 min_periods 中的较大值 close = verify_series(close, max(length, min_periods)) # 获取偏移量 offset = get_offset(offset) # 如果 close 为 None，则返回空值 if close is None: return # 计算结果 # 计算 close 的滚动 skewness（偏度） skew = close.rolling(length, min_periods=min_periods).skew() # 偏移结果 # 如果偏移量不为零，则对 skew 进行偏移 if offset != 0: skew = skew.shift(offset) # 处理填充值 # 如果 kwargs 中包含 "fillna"，则使用指定值填充缺失值 if "fillna" in kwargs: skew.fillna(kwargs["fillna"], inplace=True) # 如果 kwargs 中包含 "fill_method"，则使用指定的填充方法填充缺失值 if "fill_method" in kwargs: skew.fillna(method=kwargs["fill_method"], inplace=True) # 名称和类别 # 设置 skew 的名称为 "SKEW_长度" skew.name = f"SKEW_{length}" # 设置 skew 的类别为 "statistics" skew.category = "statistics" # 返回结果 return skew # 更新 skew 函数的文档字符串 skew.__doc__ = \ """Rolling Skew Sources: Calculation: Default Inputs: length=30 SKEW = close.rolling(length).skew() 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. """
