在交易信号发出后，我们需要一些程序化的流程，来验证信号的有效性，其中信号发出后N日的涨跌幅就是一个比较常见的任务
布林带交易策略
我们以布林带（BOLL）交易策略为示例：
- 中轨线 = N日移动平均线
- 上轨线 = 中轨线 + k 标准差
- 下轨线 = 中轨线 - k 标准差
交易信号：
- 买点：收盘价下穿布林带下方
- 卖点：收盘价上穿布林带上方
我们在python中，使用talib.BBANDS()来计算，参数包含：
- 中轨线 = {timeperiod} 日移动平均线，移动平均线计算规则为 {matype}
- 上轨线 = 中轨线 + {nbdevup} 标准差
- 下轨线 = 中轨线 - {nbdevdn} 标准差
其中 matype值对应的算法为：0=SMA, 1=EMA, 2=WMA, 3=DEMA, 4=TEMA, 5=TRIMA, 6=KAMA, 7=MAMA, 8=T3
N日涨跌幅测算
统计信号发出后，计算N日的涨跌幅：
- 首先有个对比，即全时间段的N日涨跌幅
- 计算信号发出后N日的涨跌幅
- 由于有些信号存在连续性，因此统计跳过一些连续的交易日信号后，重新统计N日涨跌幅
def n_day_rise_fall(stock_df, n=10, skip_n=None):
"""计算未来N日的涨跌幅
:param stock_df: 统计涨跌幅的dataframe
:param n: 未来N日的涨跌幅
:param skip_n: 跳过一段时间，统计未来N日的涨跌幅
"""
stock_df.reset_index(inplace=True, drop=True) # 重置index
stock_df.sort_values("date", inplace=True) # 按时间，从小到大排序
collect_dict = {}
# =========== 以每一天为参考，统计未来N日涨跌幅 ==============
pct_series = stock_df['close'].pct_change(n)
collect_dict['全部涨跌幅'] = (pct_series.dropna().values * 100).tolist()
# =========== 统计买点之后未来N日涨跌幅 ==============
buy_index = stock_df[stock_df['BUY'] == True].index
n_after_buy_list = []
for _buy_i in buy_index:
close_values = stock_df.loc[_buy_i:(_buy_i + n), :]['close'].values
range_value = (close_values[-1] - close_values[0]) / close_values[0]
n_after_buy_list.append(range_value * 100)
collect_dict[f"未来{n}日涨跌幅"] = n_after_buy_list
# ======== 跳过买点的连续性，统计未来N日的涨跌幅 ==========
if skip_n:
skip_n_after_buy_list = []
record_buy_index = -skip_n
for _buy_i in buy_index:
if _buy_i < record_buy_index + skip_n:
continue
close_values = stock_df.loc[_buy_i:(_buy_i + n), :]['close'].values
range_value = (close_values[-1] - close_values[0]) / close_values[0]
skip_n_after_buy_list.append(range_value * 100)
record_buy_index = deepcopy(_buy_i)
collect_dict[f'跳过{skip_n}的未来{n}日涨跌幅'] = skip_n_after_buy_list
return collect_dict
示例
其中from data_utils import total_stock_market_data，这个是博主自己用的数据源，大家可以换成自己的
import pandas as pd
from data_utils import total_stock_market_data
from copy import deepcopy
import talib
import numpy as np
def n_day_rise_fall(stock_df, n=10, skip_n=None):
"""计算未来N日的涨跌幅
:param stock_df: 统计涨跌幅的dataframe
:param n: 未来N日的涨跌幅
:param skip_n: 跳过一段时间，统计未来N日的涨跌幅
"""
stock_df.reset_index(inplace=True, drop=True) # 重置index
stock_df.sort_values("date", inplace=True) # 按时间，从小到大排序
collect_dict = {}
# =========== 以每一天为参考，统计未来N日涨跌幅 ==============
pct_series = stock_df['close'].pct_change(n)
collect_dict['全部涨跌幅'] = (pct_series.dropna().values * 100).tolist()
# =========== 统计买点之后未来N日涨跌幅 ==============
buy_index = stock_df[stock_df['BUY'] == True].index
n_after_buy_list = []
for _buy_i in buy_index:
close_values = stock_df.loc[_buy_i:(_buy_i + n), :]['close'].values
range_value = (close_values[-1] - close_values[0]) / close_values[0]
n_after_buy_list.append(range_value * 100)
collect_dict[f"未来{n}日涨跌幅"] = n_after_buy_list
# ======== 跳过买点的连续性，统计未来N日的涨跌幅 ==========
if skip_n:
skip_n_after_buy_list = []
record_buy_index = -skip_n
for _buy_i in buy_index:
if _buy_i < record_buy_index + skip_n:
continue
close_values = stock_df.loc[_buy_i:(_buy_i + n), :]['close'].values
range_value = (close_values[-1] - close_values[0]) / close_values[0]
skip_n_after_buy_list.append(range_value * 100)
record_buy_index = deepcopy(_buy_i)
collect_dict[f'跳过{skip_n}的未来{n}日涨跌幅'] = skip_n_after_buy_list
return collect_dict
def cal_describe(data_dict):
"""统计每一个个股的情况"""
des_dict = {}
for _key, _value in data_dict.items():
_value = np.array(_value)
des_dict[_key] = {
"num": len(_value),
"mean": _value.mean(),
"std": _value.std(),
"1/4分位": np.percentile(_value, 25),
"中位数": np.percentile(_value, 50),
"3/4分位": np.percentile(_value, 75),
}
return des_dict
def add_boll_feature(df: pd.DataFrame):
"""计算指标[BUY][SELL]"""
upper, middle, lower = talib.BBANDS(df["close"], timeperiod=14) # 计算布林带指标
df['boll_upper'] = upper
df['boll_middle'] = middle
df['boll_lower'] = lower
# 判断买点与卖点
df.dropna(inplace=True)
df['BUY'] = None
df['SELL'] = None
for _index, series in df.iterrows():
if series['close'] <= series['boll_lower']:
df.loc[_index, "BUY"] = True
if series['close'] >= series['boll_upper']:
df.loc[_index, "SELL"] = True
return df
def main():
stock_dict = total_stock_market_data(limit=3, start_date="2020-01-01", end_date="2023-01-01")
describe_collect_dict = {}
for _key, _stock_df in stock_dict.items():
_stock_df = _stock_df[["date", "open", "close", "high", 'low']]
_stock_df.sort_values("date", inplace=True)
stock_df = deepcopy(_stock_df)
stock_df = add_boll_feature(stock_df)
_coll_dict = n_day_rise_fall(stock_df, skip_n=5)
describe_dict = cal_describe(_coll_dict)
describe_collect_dict[_key] = describe_dict
if __name__ == '__main__':
main()
比如数据源是：
最终得到的：
{
'全部涨跌幅': {
'num': 705,
'mean': 0.048260603236591246,
'std': 7.074501373770825,
'1/4分位': -4.712887188691961,
'中位数': -0.509262075967043,
'3/4分位': 4.109585014510864},
'未来10日涨跌幅': {
'num': 34,
'mean': 0.9538746236276364,
'std': 6.347669735599827,
'1/4分位': -3.5691535297657966,
'中位数': -0.6329642309802745,
'3/4分位': 5.306936708754319},
'跳过5的未来10日涨跌幅': {
'num': 19,
'mean': -0.7768875022887046,
'std': 5.618893194327587,
'1/4分位': -4.332845170390464,
'中位数': -1.5776713355815666,
'3/4分位': 1.9155965506750263}
}
