【Quant102】 经典技术指标 Pandas 实现（第一部分）（4）https://developer.aliyun.com/article/1524459
CCI 策略
假设你是个高级程序员和量化研究员，编写函数实现CCI策略。函数接受数据帧
df，CCI列名称cci_col，inplace参数控制是否原地更新df。买卖信号应保存在signal列中。最后返回df。
def cci_strategy(df, cci_col='cci', inplace=False): """ 实现CCI策略，生成买卖信号。 参数: df : DataFrame 包含CCI指标数据的DataFrame，必须包含'cci_col'列。 cci_col : str CCI指标的列名。 inplace : bool, optional 是否在原地更新DataFrame，默认为False。 返回: df : DataFrame 原始DataFrame，增加了一个名为'signal'的列，包含买卖信号。 """ # 复制DataFrame以避免修改原始数据 if not inplace: df = df.copy() # 初始化信号列，默认为无操作 df['signal'] = 0 # 产生买入信号：CCI低于-100 df.loc[df[cci_col] < -100, 'signal'] = 1 # 产生卖出信号：CCI高于+100 df.loc[df[cci_col] > 100, 'signal'] = -1 return df
OBV 指标
假设你是个高级程序员和量化研究员，编写函数实现 OBV 指标。函数接受数据帧
df，inplace参数控制是否原地更新df。df包含四个列：open开盘价、high最高价、low最低价和close收盘价。所有指标都应当保存到df中，最后返回df。
def obv(df, obv_col='obv', inplace=False): """ 计算OBV指标，并更新DataFrame。 参数: df : DataFrame 包含价格数据的DataFrame，必须包含'open', 'high', 'low', 'close'列。 inplace : bool, optional 是否在原地更新DataFrame，默认为False。 返回: df : DataFrame 包含OBV指标的新DataFrame（如果inplace为False）。 """ # 复制DataFrame以避免修改原始数据 if not inplace: df = df.copy() # 计算收盘价差异的符号 chg = df['close'].diff() sgn = np.sign(chg) sgn.iloc[0] = 0 # 计算OBV df[obv_col] = (df['volume'] * sgn).cumsum() return df
OBV 策略
假设你是个高级程序员和量化研究员，编写函数实现OBV策略。函数接受数据帧
df，OBV列名称obv_col，inplace参数控制是否原地更新df。买卖信号应保存在signal列中。最后返回df。
def obv_strategy(df, obv_col, inplace=False): """ 实现OBV策略，生成买卖信号。 参数: df : DataFrame 包含OBV指标数据的DataFrame，必须包含'obv_col'列。 obv_col : str OBV指标的列名。 inplace : bool, optional 是否在原地更新DataFrame，默认为False。 返回: df : DataFrame 原始DataFrame，增加了一个名为'signal'的列，包含买卖信号。 """ # 复制DataFrame以避免修改原始数据 if not inplace: df = df.copy() # 初始化信号列，默认为无操作 df['signal'] = 0 # 产生买入信号：OBV连续上升 df.loc[df[obv_col].rolling(window=3).sum() > 0, 'signal'] = 1 # 产生卖出信号：OBV连续下降 df.loc[df[obv_col].rolling(window=3).sum() < 0, 'signal'] = -1 return df
ADX 指标
假设你是个高级程序员和量化研究员，编写函数实现 ADX 指标。函数接受数据帧
df，inplace参数控制是否原地更新df。df包含四个列：open开盘价、high最高价、low最低价和close收盘价。所有指标都应当保存到df中，最后返回df。
def adx(df, window=14, inplace=False): """ 计算ADX指标，并更新DataFrame。 参数: df : DataFrame 包含价格数据的DataFrame，必须包含'open', 'high', 'low', 'close'列。 window : int, optional ADX指标的时间窗口，默认为14。 inplace : bool, optional 是否在原地更新DataFrame，默认为False。 返回: df : DataFrame 包含ADX指标的新DataFrame（如果inplace为False）。 """ # 复制DataFrame以避免修改原始数据 if not inplace: df = df.copy() # 计算最高价和最低价之间的差异 df['range'] = df['high'] - df['low'] # 计算动量（Momentum） plus_dm = np.where(df['close'] - df['open'] > 0, df['range'], 0) minus_dm = np.where(df['open'] - df['close'] > 0, -df['range'], 0) # 计算ADM和ADN plus_dm_mean = plus_dm.rolling(window=window).mean() minus_dm_mean = minus_dm.rolling(window=window).mean() # 计算DX dx = (plus_dm_mean - minus_dm_mean) / (plus_dm_mean + minus_dm_mean) dx_mean = dx.rolling(window=window).mean() # 计算ADX df['adx'] = (100 - 100 / (1 + np.sqrt(dx_mean))) return df
ADX 策略
假设你是个高级程序员和量化研究员，编写函数实现ADX策略。函数接受数据帧
df，ADX列名称adx_col，inplace参数控制是否原地更新df。买卖信号应保存在signal列中。最后返回df。
def adx_strategy(df, adx_col='adx', inplace=False): """ 实现ADX策略，生成买卖信号。 参数: df : DataFrame 包含ADX指标数据的DataFrame，必须包含'adx_col'列。 adx_col : str ADX指标的列名。 inplace : bool, optional 是否在原地更新DataFrame，默认为False。 返回: df : DataFrame 原始DataFrame，增加了一个名为'signal'的列，包含买卖信号。 """ # 复制DataFrame以避免修改原始数据 if not inplace: df = df.copy() # 初始化信号列，默认为无操作 df['signal'] = 0 # 产生买入信号：ADX超过25 df.loc[df[adx_col] > 25, 'signal'] = 1 # 产生卖出信号：ADX低于25 df.loc[df[adx_col] < 25, 'signal'] = -1 return df
VWAP 指标
假设你是个高级程序员和量化研究员，编写函数实现 VWAP 指标。函数接受数据帧
df，inplace参数控制是否原地更新df。df包含四个列：open开盘价、high最高价、low最低价和close收盘价。所有指标都应当保存到df中，最后返回df。
def vwap(df, vwap_col='vwap', inplace=False): """ 计算VWAP指标，并更新DataFrame。 参数: df : DataFrame 包含价格数据的DataFrame，必须包含'open', 'high', 'low', 'close', 'volume'列。 window : int, optional 滚动窗口的大小，默认为1。 inplace : bool, optional 是否在原地更新DataFrame，默认为False。 返回: df : DataFrame 包含VWAP指标的新DataFrame（如果inplace为False）。 """ # 复制DataFrame以避免修改原始数据 if not inplace: df = df.copy() # 计算VWAP df['vwap'] = (df['open'] + df['high'] + df['low'] + df['close']) / 4 * df['volume'] return df
VWAP 策略
假设你是个高级程序员和量化研究员，编写函数实现VWAP策略。函数接受数据帧
df，VWAP列名称vwap_col，inplace参数控制是否原地更新df。买卖信号应保存在signal列中。最后返回df。
def vwap_strategy(df, vwap_col='vwap', inplace=False): """ 实现VWAP策略，生成买卖信号。 参数: df : DataFrame 包含VWAP指标数据的DataFrame，必须包含'vwap_col'列。 vwap_col : str VWAP指标的列名。 inplace : bool, optional 是否在原地更新DataFrame，默认为False。 返回: df : DataFrame 原始DataFrame，增加了一个名为'signal'的列，包含买卖信号。 """ # 复制DataFrame以避免修改原始数据 if not inplace: df = df.copy() # 初始化信号列，默认为无操作 df['signal'] = 0 # 产生买入信号：价格突破VWAP df.loc[df['close'] > df[vwap_col], 'signal'] = 1 # 产生卖出信号：价格跌破VWAP df.loc[df['close'] < df[vwap_col], 'signal'] = -1 return df
