写在前面：
1. 本文中提到的“K线形态查看工具”的具体使用操作请查看该博文；
2. K线形体所处背景，诸如处在上升趋势、下降趋势、盘整等，背景内容在K线形态策略代码中没有体现；
3. 文中知识内容来自书籍《K线技术分析》by邱立波。
目录
解说
技术特征
技术含义
K线形态策略代码
结果
解说
空方尖兵指出现在下跌行情中，由若干向下倾斜、收在第一根中阴线或大阴线长下影线下方的K线组成。
技术特征
1）出现在下跌行情中。
2）由若干K线组合而成。
3）在拉出一根中阴线或大阴线时，留下一根较长的下影线。
4）股价反弹后不久又跌至下影线的下方。
技术含义
空方尖兵是卖出信号，后市看跌。
第一根阴线的长下影线可以看做是空方大举进攻前向多方发动的一次试探性冲锋，如同深入敌方阵地的尖兵，所以称为空方尖兵。
空方尖兵表明多方的反击有心无力，时间短暂得如昙花一现，后市下跌的概率很大。交易者见到空方尖兵，应及时做空，离场观望。
K线形态策略代码
def excute_strategy(daily_file_path):
'''
名称：空方尖兵
识别：
1. 若干向下倾斜、收在第一根中阴线或大阴线长下影线下方的K线组合
自定义：
1. 长下影线 =》影线长度大于实体长度
2. 影线下方 =》实体中心点一下，影线最低点往上
3. 至少5根
前置条件：计算时间区间 2021-01-01 到 2022-01-01
:param daily_file_path: 股票日数据文件路径
:return:
'''
import pandas as pd
import os
start_date_str = '2013-01-01'
end_date_str = '2014-01-01'
df = pd.read_csv(daily_file_path,encoding='utf-8')
# 删除停牌的数据
df = df.loc[df['openPrice'] > 0].copy()
df['o_date'] = df['tradeDate']
df['o_date'] = pd.to_datetime(df['o_date'])
df = df.loc[(df['o_date'] >= start_date_str) & (df['o_date']<=end_date_str)].copy()
# 保存未复权收盘价数据
df['close'] = df['closePrice']
# 计算前复权数据
df['openPrice'] = df['openPrice'] * df['accumAdjFactor']
df['closePrice'] = df['closePrice'] * df['accumAdjFactor']
df['highestPrice'] = df['highestPrice'] * df['accumAdjFactor']
df['lowestPrice'] = df['lowestPrice'] * df['accumAdjFactor']
# 开始计算
df['type'] = 0
df.loc[df['closePrice'] > df['openPrice'], 'type'] = 1
df.loc[df['closePrice'] < df['openPrice'], 'type'] = -1
df['body_length'] = abs(df['closePrice']-df['openPrice'])
df['big_yeah'] = 0
df.loc[(df['type']==-1) & (df['body_length']/df['closePrice'].shift(1)>0.02) & (df['closePrice']-df['lowestPrice']>df['body_length']),'big_yeah'] = 1
df.reset_index(inplace=True)
df.at[len(df)-1,'big_yeah'] = 1
df['i_row'] = [i for i in range(len(df))]
df_target = df.loc[df['big_yeah']==1].copy()
target_list = df_target['i_row'].values.tolist()
s_list = []
e_list = []
for i in range(len(target_list)-1):
s_list.append(target_list[i])
e_list.append(target_list[i+1])
df['signal'] = 0
df['signal_name'] = ''
for s,e in zip(s_list,e_list):
if e-s < 5:
continue
temp_e = None
up_point = df.iloc[s]['closePrice'] + df.iloc[s]['body_length']*0.5
down_point = df.iloc[s]['lowestPrice']
for i in range(s+1,e):
if df.iloc[i]['highestPrice']<=up_point and df.iloc[i]['lowestPrice']>=down_point:
temp_e = i
else:
break
if temp_e is None or temp_e-s < 5:
continue
df.loc[(df['i_row']>=s) & (df['i_row']<=temp_e),'signal'] = 1
df.loc[(df['i_row']>=s) & (df['i_row']<=temp_e),'signal_name'] = str(temp_e-s)
pass
file_name = os.path.basename(daily_file_path)
title_str = file_name.split('.')[0]
line_data = {
'title_str':title_str,
'whole_header':['日期','收','开','高','低'],
'whole_df':df,
'whole_pd_header':['tradeDate','closePrice','openPrice','highestPrice','lowestPrice'],
'start_date_str':start_date_str,
'end_date_str':end_date_str,
'signal_type':'duration_detail',
'duration_len':[],
'temp':len(df.loc[df['signal']==1])
}
return line_data
结果
