用python寻找W底形态的股票

前段时间招商证券出了一份关于市场底部特征的研报，被很多财经自媒体大V引用，该研报总结了市场历史大底的5个信号:

1. 超额流动性与新增社融增速的组合出现转正回升。
2. 估值水平降到历史低位。
3. 外部流动性环境出现边际改善。
4. 成交低迷，换手率明显下降。
5. K线出现类似W底的组合。并且还引用了一句俗语：“单底不是底，双底得天下”。

这份研报也讲了w底形成的原因。当有重要的会议(讲话)出来提振了市场情绪，投资者开始抄底，市场反弹，这时候形成了第1个底。但由于此时可能并未出现流动性和基本面改善的实质性信号，并且因为市场在前期大幅下跌，恐慌情绪仍未消除。部分抄底的投资者在反弹一段时间后选择获利了结头寸，并且之前大跌时未减仓的投资者也利用此次反弹选择降低仓位，从而形成了第二次探底，从K线组合上看就是一个W形态。

经典的技术分析书籍《笑傲牛熊》中也谈到了w底(双重底)， 并且指出同时出现成交量明显放大，很好地相对强度和最小的阻力区域时，极有可能预示着客观的上涨，特别的，因为w底经常出现，所以也特别重要。

w底在投资者的交流中也经常出现，尤其在股吧，投资者论坛等等.

有很多种方法可以检测出w底，本文给出一种比较简单的算法，仅仅使用pandas和numpy，不涉及像TensorFlow 或 PyTorch 这样的机器学习库。

1.首先获取价格， 无论从哪个渠道获取，都转为如下格式：

ticker_df = get_stock_price('000895.SZ').reset_index()ticker_df.head()

2.进行多项式拟合， 用图形展示则是为了方便进行调参。

x_data = ticker_df.index.tolist()      y_data = ticker_df['low']x = np.linspace(0, max(ticker_df.index.tolist()), max(ticker_df.index.tolist()) + 1)pol = np.polyfit(x_data, y_data, 17) # 注意这个参数y_pol = np.polyval(pol, x)plt.figure(figsize=(15, 2), dpi= 120, facecolor='w', edgecolor='k')# 显示股价plt.plot(x_data, y_data, 'o', markersize=1.5, color='grey', alpha=0.7)# 显示多项式拟合plt.plot(x, y_pol, '-', markersize=1.0, color='black', alpha=0.9)plt.legend(['股价', '多项式拟合'], prop=font)plt.show()

3.检测局部最小值和局部最大值。局部最小值用红色标记，局部最大值用蓝色标记。

data = y_polmin_max = np.diff(np.sign(np.diff(data))).nonzero()[0] + 1 # 局部最小 & 最大l_min = (np.diff(np.sign(np.diff(data))) > 0).nonzero()[0] + 1 # 局部最小l_max = (np.diff(np.sign(np.diff(data))) < 0).nonzero()[0] + 1 # 局部最大plt.figure(figsize=(15, 2), dpi= 120, facecolor='w', edgecolor='k')plt.rcParams['font.sans-serif'] = ['SimHei'] plt.plot(x, data, color='grey')plt.plot(x[l_min], data[l_min], 'o', label='min', color='r') # 最小plt.plot(x[l_max], data[l_max], 'o', label='max', color='b') # 最大plt.title('局部最小 & 最大')plt.show()

4.拟合函数在 x 轴上的局部极小值位置应该对应于数据中极小值的 x 轴位置。所以要进行一些处理。

delta = 10                                       # 设置范围dict_i = dict()dict_x = dict()df_len = len(ticker_df.index)                    for element in l_min:                                l_bound = element - delta                        u_bound = element + delta                        x_range = range(l_bound, u_bound + 1)            dict_x[element] = x_range                        y_loc_list = list()    for x_element in x_range:        if x_element > 0 and x_element < df_len:                            y_loc_list.append(ticker_df.low.iloc[x_element])    dict_i[element] = y_loc_list  

5.上一步找到了可疑的价格低点。但要寻找的是全局最小值，所以要设置一个阈值，比如设置为最低点的某个百分比(考虑到离群值，也可以采用几个最低值的平均值来改进)。

y_delta = 0.12 threshold = min(ticker_df['low']) * 1.15 # 设置全局低部的阈值，会有很多底部值，但是只有低于这个的才是y_dict = dict()mini = list()suspected_bottoms = list() for key in dict_i.keys(): mn = sum(dict_i[key])/len(dict_i[key]) price_min = min(dict_i[key]) mini.append(price_min) l_y = mn * (1.0 - y_delta) u_y = mn * (1.0 + y_delta) y_dict[key] = [l_y, u_y, mn, price_min] for key_i in y_dict.keys(): for key_j in y_dict.keys(): if (key_i != key_j) and (y_dict[key_i][3] < threshold): suspected_bottoms.append(key_i)

6.可视化展示最终结果。

plt.figure(figsize=(20, 10), dpi= 120, facecolor='w', edgecolor='k')plt.plot(x_data, y_data, 'o', markersize=1.5, color='magenta', alpha=0.7)plt.plot(x_data, ticker_df['high'], linestyle='-.', color='g')plt.plot(x_data, ticker_df['open'],      'o', markersize=1.5, color='grey',  alpha=0.7)plt.plot(x_data, ticker_df['close'],     'o', markersize=1.5, color='red', alpha=0.7)    plt.plot(x, y_pol, '-', markersize=1.0, color='black', alpha=0.9)for position in suspected_bottoms:    plt.axvline(x=position, linestyle='-.', color='r')    plt.axhline(threshold, linestyle='--', color='b')    #计算颈线值lowest_idx = [x for x in set(suspected_bottoms)]lowest_idx.sort()first_lowest, seconde_lowest = lowest_idx[-2:][0], lowest_idx[-2:][1]current_price = max(ticker_df.iloc[-1:]['close'])neckline_price = max((ticker_df.iloc[first_lowest:seconde_lowest])['high'])plt.axhline(max((ticker_df.iloc[first_lowest:seconde_lowest])['high']), linestyle='-.', color='g') for key in dict_x.keys():    for value in dict_x[key]:        plt.axvline(x=value, linestyle='-', color = 'lightblue', alpha=0.2)plt.show()

计算的完整代码如下:

def get_double_bottom_info(ticker_df): x_data = ticker_df.index.tolist() y_data = ticker_df['low'] x = np.linspace(0, max(ticker_df.index.tolist()), max(ticker_df.index.tolist()) + 1) # 多项式拟合 pol = np.polyfit(x_data, y_data, 17) # 注意这个参数 y_pol = np.polyval(pol, x) data = y_pol min_max = np.diff(np.sign(np.diff(data))).nonzero()[0] + 1 # 局部最小 & 最大 l_min = (np.diff(np.sign(np.diff(data))) > 0).nonzero()[0] + 1 # 局部最小 l_max = (np.diff(np.sign(np.diff(data))) < 0).nonzero()[0] + 1 # 局部最大 delta = 10 # 设置范围 dict_i = dict() dict_x = dict() df_len = len(ticker_df.index) for element in l_min: l_bound = element - delta u_bound = element + delta x_range = range(l_bound, u_bound + 1) dict_x[element] = x_range y_loc_list = list() for x_element in x_range: if x_element > 0 and x_element < df_len: y_loc_list.append(ticker_df.low.iloc[x_element]) dict_i[element] = y_loc_list y_delta = 0.12 threshold = min(ticker_df['low']) * 1.15 # 设置全局低部的阈值，会有很多底部值，但是只有低于这个的才是 y_dict = dict() mini = list() suspected_bottoms = list() for key in dict_i.keys(): mn = sum(dict_i[key])/len(dict_i[key]) price_min = min(dict_i[key]) mini.append(price_min) l_y = mn * (1.0 - y_delta) u_y = mn * (1.0 + y_delta) y_dict[key] = [l_y, u_y, mn, price_min] for key_i in y_dict.keys(): for key_j in y_dict.keys(): if (key_i != key_j) and (y_dict[key_i][3] < threshold): suspected_bottoms.append(key_i) # 移除重复值并降序返回 double_bottom_idx = [x for x in set(suspected_bottoms)] double_bottom_idx.sort() return double_bottom_idx

如果存在两个或以上的值，就说明存在w底。然后根据返回的值计算是否突破了颈线。W底只是一个形态，实际中使用还要配合成交量，基本面，动量，RPS等特征。

当然啦，w底有很多种计算方式，特别地对于机器学习来说。对于类如w底的模式检测，Kathryn Dover的Pattern Recognition in Stock Data值得看看。