PandasTA 源码解析（八）（2）https://developer.aliyun.com/article/1506130
.\pandas-ta\pandas_ta\overlap\hma.py
# -*- coding: utf-8 -*- # 从 numpy 库中导入 sqrt 函数并重命名为 npSqrt from numpy import sqrt as npSqrt # 从当前目录下的 wma 模块中导入 wma 函数 from .wma import wma # 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义 Hull Moving Average (HMA) 指标函数 def hma(close, length=None, offset=None, **kwargs): """Indicator: Hull Moving Average (HMA)""" # 验证参数 # 如果 length 存在且大于 0，则将其转换为整数，否则设为默认值 10 length = int(length) if length and length > 0 else 10 # 验证 close 数据为 pd.Series 类型，并且长度符合要求 close = verify_series(close, length) # 获取 offset 值 offset = get_offset(offset) # 如果 close 为空，则返回空值 if close is None: return # 计算结果 half_length = int(length / 2) sqrt_length = int(npSqrt(length)) # 计算 wmaf 和 wmas wmaf = wma(close=close, length=half_length) wmas = wma(close=close, length=length) # 计算 HMA hma = wma(close=2 * wmaf - wmas, length=sqrt_length) # 调整偏移量 if offset != 0: hma = hma.shift(offset) # 处理填充值 if "fillna" in kwargs: hma.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: hma.fillna(method=kwargs["fill_method"], inplace=True) # 设置指标名称和类别 hma.name = f"HMA_{length}" hma.category = "overlap" return hma # 设置 HMA 函数的文档字符串 hma.__doc__ = \ """Hull Moving Average (HMA) The Hull Exponential Moving Average attempts to reduce or remove lag in moving averages. Sources: https://alanhull.com/hull-moving-average Calculation: Default Inputs: length=10 WMA = Weighted Moving Average half_length = int(0.5 * length) sqrt_length = int(sqrt(length)) wmaf = WMA(close, half_length) wmas = WMA(close, length) HMA = WMA(2 * wmaf - wmas, sqrt_length) Args: close (pd.Series): Series of 'close's length (int): It's period. Default: 10 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\overlap\hwma.py
# -*- coding: utf-8 -*- # 从 pandas 库中导入 Series 类 from pandas import Series # 从 pandas_ta.utils 模块中导入 get_offset 和 verify_series 函数 from pandas_ta.utils import get_offset, verify_series # 定义函数 hwma，计算 Holt-Winter 移动平均值 def hwma(close, na=None, nb=None, nc=None, offset=None, **kwargs): """Indicator: Holt-Winter Moving Average""" # 验证参数有效性并设置默认值 na = float(na) if na and na > 0 and na < 1 else 0.2 nb = float(nb) if nb and nb > 0 and nb < 1 else 0.1 nc = float(nc) if nc and nc > 0 and nc < 1 else 0.1 # 验证 close 参数并转换为 pd.Series 类型 close = verify_series(close) # 获取偏移量 offset = get_offset(offset) # 计算结果 last_a = last_v = 0 last_f = close.iloc[0] result = [] m = close.size for i in range(m): # 计算 F F = (1.0 - na) * (last_f + last_v + 0.5 * last_a) + na * close.iloc[i] # 计算 V V = (1.0 - nb) * (last_v + last_a) + nb * (F - last_f) # 计算 A A = (1.0 - nc) * last_a + nc * (V - last_v) result.append((F + V + 0.5 * A)) # 更新变量值 last_a, last_f, last_v = A, F, V # 创建 Series 对象 hwma = Series(result, index=close.index) # 处理偏移 if offset != 0: hwma = hwma.shift(offset) # 处理填充值 if "fillna" in kwargs: hwma.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: hwma.fillna(method=kwargs["fill_method"], inplace=True) # 设置名称和分类 suffix = f"{na}_{nb}_{nc}" hwma.name = f"HWMA_{suffix}" hwma.category = "overlap" return hwma # 为 hwma 函数添加文档字符串 hwma.__doc__ = \ """HWMA (Holt-Winter Moving Average) Indicator HWMA (Holt-Winter Moving Average) is a three-parameter moving average by the Holt-Winter method; the three parameters should be selected to obtain a forecast. This version has been implemented for Pandas TA by rengel8 based on a publication for MetaTrader 5. Sources: https://www.mql5.com/en/code/20856 Calculation: HWMA[i] = F[i] + V[i] + 0.5 * A[i] where.. F[i] = (1-na) * (F[i-1] + V[i-1] + 0.5 * A[i-1]) + na * Price[i] V[i] = (1-nb) * (V[i-1] + A[i-1]) + nb * (F[i] - F[i-1]) A[i] = (1-nc) * A[i-1] + nc * (V[i] - V[i-1]) Args: close (pd.Series): Series of 'close's na (float): Smoothed series parameter (from 0 to 1). Default: 0.2 nb (float): Trend parameter (from 0 to 1). Default: 0.1 nc (float): Seasonality parameter (from 0 to 1). Default: 0.1 close (pd.Series): Series of 'close's Kwargs: fillna (value, optional): pd.DataFrame.fillna(value) fill_method (value, optional): Type of fill method Returns: pd.Series: hwma """
.\pandas-ta\pandas_ta\overlap\ichimoku.py
# -*- coding: utf-8 -*- # 导入所需的库 from pandas import date_range, DataFrame, RangeIndex, Timedelta from .midprice import midprice from pandas_ta.utils import get_offset, verify_series # 定义 Ichimoku 函数，计算 Ichimoku Kinkō Hyō 指标 def ichimoku(high, low, close, tenkan=None, kijun=None, senkou=None, include_chikou=True, offset=None, **kwargs): """Indicator: Ichimoku Kinkō Hyō (Ichimoku)""" # 设置默认的参数值 tenkan = int(tenkan) if tenkan and tenkan > 0 else 9 kijun = int(kijun) if kijun and kijun > 0 else 26 senkou = int(senkou) if senkou and senkou > 0 else 52 _length = max(tenkan, kijun, senkou) # 验证输入的数据 high = verify_series(high, _length) low = verify_series(low, _length) close = verify_series(close, _length) offset = get_offset(offset) # 根据参数设置是否包含未来数据 if not kwargs.get("lookahead", True): include_chikou = False # 如果输入数据有缺失，则返回空值 if high is None or low is None or close is None: return None, None # 计算 Ichimoku 指标的各个线 tenkan_sen = midprice(high=high, low=low, length=tenkan) kijun_sen = midprice(high=high, low=low, length=kijun) span_a = 0.5 * (tenkan_sen + kijun_sen) span_b = midprice(high=high, low=low, length=senkou) # 复制 Span A 和 Span B 在移动之前的值 _span_a = span_a[-kijun:].copy() _span_b = span_b[-kijun:].copy() # 移动 Span A 和 Span B 的值 span_a = span_a.shift(kijun) span_b = span_b.shift(kijun) chikou_span = close.shift(-kijun) # 根据偏移量对数据进行偏移 if offset != 0: tenkan_sen = tenkan_sen.shift(offset) kijun_sen = kijun_sen.shift(offset) span_a = span_a.shift(offset) span_b = span_b.shift(offset) chikou_span = chikou_span.shift(offset) # 处理缺失值 if "fillna" in kwargs: span_a.fillna(kwargs["fillna"], inplace=True) span_b.fillna(kwargs["fillna"], inplace=True) chikou_span.fillna(kwargs["fillna"], inplace=True) if "fill_method" in kwargs: span_a.fillna(method=kwargs["fill_method"], inplace=True) span_b.fillna(method=kwargs["fill_method"], inplace=True) chikou_span.fillna(method=kwargs["fill_method"], inplace=True) # 设置各个线的名称和类别 span_a.name = f"ISA_{tenkan}" span_b.name = f"ISB_{kijun}" tenkan_sen.name = f"ITS_{tenkan}" kijun_sen.name = f"IKS_{kijun}" chikou_span.name = f"ICS_{kijun}" chikou_span.category = kijun_sen.category = tenkan_sen.category = "trend" span_b.category = span_a.category = chikou_span # 准备 Ichimoku DataFrame data = { span_a.name: span_a, span_b.name: span_b, tenkan_sen.name: tenkan_sen, kijun_sen.name: kijun_sen, } if include_chikou: data[chikou_span.name] = chikou_span ichimokudf = DataFrame(data) ichimokudf.name = f"ICHIMOKU_{tenkan}_{kijun}_{senkou}" ichimokudf.category = "overlap" # 准备 Span DataFrame last = close.index[-1] # 如果收盘价索引的数据类型为 "int64"，执行以下操作 if close.index.dtype == "int64": # 创建一个新的范围索引，起始于 last + 1，结束于 last + kijun + 1 ext_index = RangeIndex(start=last + 1, stop=last + kijun + 1) # 创建一个空的 DataFrame，索引为 ext_index，列为 span_a.name 和 span_b.name spandf = DataFrame(index=ext_index, columns=[span_a.name, span_b.name]) # 将 _span_a 和 _span_b 的索引设置为 ext_index _span_a.index = _span_b.index = ext_index # 如果收盘价索引的数据类型不为 "int64"，执行以下操作 else: # 统计收盘价索引中各值的频次，并取出出现频次最多的值 df_freq = close.index.value_counts().mode()[0] # 创建一个时间增量对象，时间间隔为 df_freq 天 tdelta = Timedelta(df_freq, unit="d") # 创建一个新的日期范围，起始日期为 last + tdelta，包含 kijun 个工作日 new_dt = date_range(start=last + tdelta, periods=kijun, freq="B") # 创建一个空的 DataFrame，索引为 new_dt，列为 span_a.name 和 span_b.name spandf = DataFrame(index=new_dt, columns=[span_a.name, span_b.name]) # 将 _span_a 和 _span_b 的索引设置为 new_dt spandf[span_a.name] = _span_a spandf[span_b.name] = _span_b # 设置 spandf 的名称为特定字符串，包含 tenkan 和 kijun 的值 spandf.name = f"ICHISPAN_{tenkan}_{kijun}" # 设置 spandf 的类别为 "overlap" spandf.category = "overlap" # 返回 ichimokudf 和 spandf return ichimokudf, spandf # 将 ichimoku.__doc__ 的值设为字符串，用于描述 Ichimoku Kinkō Hyō（一种用于金融市场预测的模型）的计算方法和参数 ichimoku.__doc__ = \ """Ichimoku Kinkō Hyō (ichimoku) Developed Pre WWII as a forecasting model for financial markets. Sources: https://www.tradingtechnologies.com/help/x-study/technical-indicator-definitions/ichimoku-ich/ Calculation: Default Inputs: tenkan=9, kijun=26, senkou=52 MIDPRICE = Midprice TENKAN_SEN = MIDPRICE(high, low, close, length=tenkan) KIJUN_SEN = MIDPRICE(high, low, close, length=kijun) CHIKOU_SPAN = close.shift(-kijun) SPAN_A = 0.5 * (TENKAN_SEN + KIJUN_SEN) SPAN_A = SPAN_A.shift(kijun) SPAN_B = MIDPRICE(high, low, close, length=senkou) SPAN_B = SPAN_B.shift(kijun) Args: high (pd.Series): Series of 'high's # high 数据序列 low (pd.Series): Series of 'low's # low 数据序列 close (pd.Series): Series of 'close's # close 数据序列 tenkan (int): Tenkan period. Default: 9 # Tenkan 周期，默认为 9 kijun (int): Kijun period. Default: 26 # Kijun 周期，默认为 26 senkou (int): Senkou period. Default: 52 # Senkou 周期，默认为 52 include_chikou (bool): Whether to include chikou component. Default: True # 是否包含 chikou 组件，默认为 True offset (int): How many periods to offset the result. Default: 0 # 结果偏移的周期数，默认为 0 Kwargs: fillna (value, optional): pd.DataFrame.fillna(value) # fillna 方法的参数，用于填充缺失值 fill_method (value, optional): Type of fill method # 填充方法的类型 Returns: pd.DataFrame: Two DataFrames. # 返回两个 DataFrame For the visible period: spanA, spanB, tenkan_sen, kijun_sen, # 可见期间的 DataFrame，包含 spanA、spanB、tenkan_sen、kijun_sen and chikou_span columns # 以及 chikou_span 列 For the forward looking period: spanA and spanB columns # 未来观察期间的 DataFrame，包含 spanA 和 spanB 列 """
