A New Day Has Come 全新的时代¶
In [1]:
# 基础库导入
from __future__ import print_function
from __future__ import division
import warnings
# warnings.filterwarnings('ignore')
# warnings.simplefilter('ignore')
import time
import moment
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import ipywidgets
%matplotlib inline
import os
import sys
# 使用insert 0即只使用github，避免交叉使用了pip安装的abupy，导致的版本不一致问题
sys.path.insert(0, os.path.abspath('../'))
import abupy
from abupy import AbuFactorAtrNStop, AbuFactorPreAtrNStop, AbuFactorCloseAtrNStop, AbuFactorBuyBreak
from abupy import AbuFactorSellBreak
from abupy import abu, EMarketTargetType, AbuMetricsBase, ABuMarketDrawing, ABuProgress, ABuSymbolPd
from abupy import EMarketTargetType, EDataCacheType, EMarketSourceType, EMarketDataFetchMode, EStoreAbu, AbuUmpMainMul
from abupy import AbuUmpMainDeg, AbuUmpMainJump, AbuUmpMainPrice, AbuUmpMainWave, feature, AbuFeatureDegExtend
from abupy import AbuUmpEdgeDeg, AbuUmpEdgePrice, AbuUmpEdgeWave, AbuUmpEdgeFull, AbuUmpEdgeMul, AbuUmpEegeDegExtend
from abupy import AbuUmpMainDegExtend, ump, Parallel, delayed, AbuMulPidProgress
pd.options.display.max_rows = 1000
from IPython.display import display
from IPython.core.display import HTML
from pprint import pprint, pformat
# 关闭沙盒数据
abupy.env.disable_example_env_ipython()
#os.path.join(ABuEnv.g_project_cache_dir, 'abu_socket_progress')
# print(os.environ.get(CACHE_ROOT, './cache'))
# for k, v in os.environ.items():
# print(f'{k}={v}')
# print(os.environ.get('CACHE_ROOT', './cache'))
In [2]:
from abupy import env
from abupy.CoreBu import ABuEnv
# print(ABuEnv._p_dir)
In [3]:
from WzFeature.WzFeatureMaDeg import WzFeatureMaDeg
from WzFeature.WzFeatureIndexMaDeg import WzFeatureIndexMaDeg
from WzFeature.WzFeatureMaSpeed import WzFeatureMaSpeed
from WzFeature.WzFeatureIndexPoly import WzFeatureIndexPoly
from WzFeature.WzFeaturePoly import WzFeaturePoly
from WzFeature.WzFeatureIndexSimilarity import WzFeatureIndexSimilarity
from WzUmp.WzUmpMainMaDeg import WzUmpMainMaDeg
from WzUmp.WzUmpMainMaSpeed import WzUmpMainMaSpeed
from WzUmp.WzUmpMainIndexMaDeg import WzUmpMainIndexMaDeg
from WzUmp.WzUmpMainIndexPoly import WzUmpMainIndexPoly
from WzUmp.WzUmpMainPoly import WzUmpMainPoly
from WzUmp.WzUmpMainMaSpeedWithIndMaDegAndSmlr import WzUmpMainMaSpeedWithIndMaDegAndSmlr
from WzUmp.WzUmpMainIndexPolyAndMaDeg import WzUmpMainIndexPolyAndMaDeg
from WzUmp.WzUmpMainPolyWithMaDeg import WzUmpMainPolyWithMaDeg
from WzUmp.WzUmpMainPriceWithMaDeg import WzUmpMainPriceWithMaDeg
# WzUmpMainMaSpeedWithIndMaDegAndSmlr.v()
from WzUmp.WzUmpEdgeIndexMaDeg import WzUmpEdgeIndexMaDeg
from WzUmp.WzUmpEdgeIndexPoly import WzUmpEdgeIndexPoly
from WzUmp.WzUmpEdgeMaDeg import WzUmpEdgeMaDeg
from WzUmp.WzUmpEdgeMaSpeed import WzUmpEdgeMaSpeed
from WzUmp.WzUmpEdgePoly import WzUmpEdgePoly
from WzUmp.WzUmpEdgeMaSpeedWithIndMaDegAndSmlr import WzUmpEdgeMaSpeedWithIndMaDegAndSmlr
from WzUmp.WzUmpEdgeIndexPolyAndMaDeg import WzUmpEdgeIndexPolyAndMaDeg
from WzUmp.WzUmpEdgePolyWithMaDeg import WzUmpEdgePolyWithMaDeg
from WzUmp.WzUmpEdgePriceWithMaDeg import WzUmpEdgePriceWithMaDeg
股池¶
股池为上证50 + 创业50 + 沪深300 + 中证500 几大指数的成分股
In [4]:
import numpy as np
import random
from stock_pool import stock_pool
def custom_symbols():
symbols = []
symbols.extend( random.sample(stock_pool.get('zh500'), 500))
symbols.extend( random.sample(stock_pool.get('hs300'), 300))
symbols.extend( random.sample(stock_pool.get('sh50'), 49))
symbols.extend( random.sample(stock_pool.get('chuang50'), 50))
print("股池总数:" , len(symbols))
symbols = np.unique(np.array(symbols))
print('unique后:', len(symbols))
return symbols
symbols = custom_symbols()
print(symbols)
策略¶
策略本身优秀与否不是重点，本次主要是研究 弱人工智能如何帮我们提高收益，是观察没有使用AI和使用AI后 的对比。
In [5]:
from picker.WzPickNonNew import WzPickNonNew
from picker.WzPickMaAng import WzPickMaAng
from abupy import AbuPtPosition
from factor.WzFactorBuyRsiV8 import WzFactorBuyRsiV8
from factor.WzFactorSellRsiV8 import WzFactorSellRsiV8
from factor.WzFactorBuyMaV6 import WzFactorBuyMaV6
from factor.WzFactorSellMaV5 import WzFactorSellMaV5
from abupy import slippage
# 开启针对非集合竞价阶段的涨停，滑点买入价格以高概率在接近涨停的价格买入
slippage.sbb.g_enable_limit_up = True
# 将集合竞价阶段的涨停买入成功概率设置为0，如果设置为0.2即20%概率成功买入
slippage.sbb.g_pre_limit_up_rate = 0
# 开启针对非集合竞价阶段的跌停，滑点卖出价格以高概率在接近跌停的价格卖出
slippage.ssb.g_enable_limit_down = True
# 将集合竞价阶段的跌停卖出成功概率设置为0, 如果设置为0.2即20%概率成功卖出
slippage.ssb.g_pre_limit_down_rate = 0
abupy.env.g_data_cache_type = EDataCacheType.E_DATA_CACHE_CSV
abupy.env.g_market_target = EMarketTargetType.E_MARKET_TARGET_CN
abupy.env.g_data_fetch_mode = EMarketDataFetchMode.E_DATA_FETCH_FORCE_LOCAL
read_cash = 10000000 #资金
o_pos_base =0.1 #abupy.beta.atr.g_atr_pos_base
print('o_pos_base:', o_pos_base)
pt_pos_base = o_pos_base / 12
print('pt_pos_base:' , pt_pos_base)
buy_factors=[{'buy_default': 'post=36.88',
'class': WzFactorBuyRsiV8,
'position': {'class': AbuPtPosition,
'past_day_cnt': 68,
'pos_base': pt_pos_base },
'rsi_timeperiod': 6,
'stock_pickers': [{'class': WzPickNonNew,
'pick_period': 'week',
'weeks': 24},
]
}]
# 卖出因子继续使用上一节使用的因子
sell_factors = [
# {'stop_loss_n': 1.0, 'stop_win_n': 3.0,
# 'class': AbuFactorAtrNStop},
# {'class': AbuFactorPreAtrNStop, 'pre_atr_n': 1.5},
{'class': WzFactorSellMaV5 ,
'ma_type': 1,
'sell_x': 'rvv=5,6,0.22,yc,1.0'},
{'class': WzFactorSellRsiV8,
'rsi_timeperiod': 6,
'sell_default':'post=1.0@78' }, #rvv=10,-,78,12,80,1.0
{'class': AbuFactorCloseAtrNStop, 'close_atr_n': 1.25},
{'class': AbuFactorAtrNStop, 'stop_loss_n': 0.65},
# {'class': WzFactorSellRsiV8,
# 'rsi_timeperiod': 6,
# 'sell_default': 'lt=2,29,1.0|lt=3,39,1.0'},
]
print("随便输出点什么表示已经执行过： ")
In [6]:
print( read_cash * 0.0083333)
策略回测 - 准备交割单数据用于机器人裁判训练¶
将股池所有股票分切成3块，拿2/3用于机器人裁判训练， 剩下1/3是测试集用途。
In [6]:
# import psutil
from warnings import simplefilter
simplefilter(action='ignore', category=FutureWarning)
simplefilter(action='ignore', category=ResourceWarning)
# g_cpu_cnt = psutil.cpu_count(logical=True) * 1
# print ("g_cpu_cnt:", g_cpu_cnt)
# 回测生成买入时刻特征
abupy.env.g_enable_ml_feature = True
# 回测开始时将symbols切割分为训练集数据和测试集两份，使用训练集进行回测
abupy.env.g_enable_train_test_split = True
# 训练：测试 = 1：1
abupy.env.g_split_tt_n_folds = 3
feature.clear_user_feature()
abupy.feature.g_price_rank_keys = [12, 25, 50, 100, 200]
feature.append_user_feature(AbuFeatureDegExtend)
feature.append_user_feature(WzFeatureMaDeg)
feature.append_user_feature(WzFeatureIndexMaDeg)
feature.append_user_feature(WzFeatureMaSpeed)
feature.append_user_feature(WzFeatureIndexPoly)
feature.append_user_feature(WzFeaturePoly)
feature.append_user_feature(WzFeatureIndexSimilarity)
abu_result_tuple = None
def run_loop_back():
global abu_result_tuple
# , symbols
# print(symbols)
abu_result_tuple, _ = abu.run_loop_back(read_cash,
buy_factors,
sell_factors,
choice_symbols = symbols,
# choice_symbols=None,
start='2016-02-02', end='2019-12-18',
# stock_picks= stock_pickers,
# n_process_kl = 8,
# n_process_pick= 8,
)
# 把运行的结果保存在本地，以便之后分析回测使用，保存回测结果数据代码如下所示
abu.store_abu_result_tuple(abu_result_tuple, n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='train_cn')
ABuProgress.clear_output()
AbuMetricsBase.show_general(*abu_result_tuple, only_show_returns=True).plot_max_draw_down()
def run_load_train():
global abu_result_tuple
abu_result_tuple = abu.load_abu_result_tuple(n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='train_cn')
AbuMetricsBase.show_general(*abu_result_tuple, only_show_returns=True)#.plot_max_draw_down()
def select(select):
if select == 'run loop back':
run_loop_back()
else:
run_load_train()
# _ = ipywidgets.interact_manual(select, select=[ 'load train data', 'run loop back'])
run_load_train()
看，原始策略很差，亏！！
训练集和测试集回顾：¶
上面的abupy.env.g_enable_train_test_split = True 和abupy.env.g_split_tt_n_folds = 3配置自动把训练集和测试集分切好了，打印出来看看。
In [6]:
from abupy import ABuFileUtil
a = ['1', '2', '3', 'a', 'b', 'c']
b = ['0','2','b','f**k']
print('a&b交集为:', set(a).intersection(b) )
train = ABuFileUtil.load_pickle("/root/abu/data/cache/market_train_symbols_hs")
test = ABuFileUtil.load_pickle("/root/abu/data/cache/market_test_symbols_hs")
print('训练集和测试集交集为:', set(train).intersection(test) )
print( len(train), "个训练集股票：\r\n", train, "\r\n")
print( len(test), "个测试集股票：\r\n", test)
In [7]:
from warnings import simplefilter
simplefilter(action='ignore', category=FutureWarning)
simplefilter(action='ignore', category=ResourceWarning)
# 需要全局设置为A股市场，在ump会根据市场类型保存读取对应的ump
abupy.env.g_market_target = EMarketTargetType.E_MARKET_TARGET_CN
ump_deg=None
ump_jump=None
ump_mul=None
ump_price=None
ump_main_deg_extend=None
ump_wave=None
# 使用训练集交易数据训练主裁
orders_pd_train_cn = abu_result_tuple.orders_pd
def train_main_ump():
print('AbuUmpMainDeg begin...') #brust_min=False,
AbuUmpMainDeg.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False, show_order=False)
print('AbuUmpMainPrice begin...')
AbuUmpMainPrice.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False, show_order=False)
print('AbuUmpMainJump begin...')
AbuUmpMainJump.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False, show_order=False)
print('AbuUmpMainWave begin...')
AbuUmpMainWave.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False, show_order=False)
print('AbuUmpMainDegExtend begin...')
AbuUmpMainDegExtend.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False, show_order=False)
print('AbuUmpMainMul begin...')
AbuUmpMainMul.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False, show_order=False)
# 依然使用load_main_ump，避免下面多进程内存拷贝过大
load_main_ump()
def load_main_ump():
global ump_deg, ump_jump, ump_mul, ump_price, ump_main_deg_extend, ump_wave
ump_deg = AbuUmpMainDeg(predict=True)
ump_jump = AbuUmpMainJump(predict=True)
ump_mul = AbuUmpMainMul(predict=True)
ump_price = AbuUmpMainPrice(predict=True)
ump_main_deg_extend = AbuUmpMainDegExtend(predict=True)
ump_wave = AbuUmpMainWave(predict=True )
print('load main ump complete!')
def select(select):
if select == 'train main ump':
train_main_ump()
else:
load_main_ump()
# _ = ipywidgets.interact_manual(select, select=['load main ump', 'train main ump'])
load_main_ump()
自定义主裁
In [8]:
from warnings import simplefilter
simplefilter(action='ignore', category=FutureWarning)
simplefilter(action='ignore', category=ResourceWarning)
abupy.env.g_market_target = EMarketTargetType.E_MARKET_TARGET_CN
ump_price_with_ma_deg = None
ump_ma_speed_with_ind_ma_deg_and_smlr = None
ump_index_poly_and_ma_deg = None
ump_poly_with_ma_deg = None
ump_index_ma_deg = None
ump_index_poly = None
ump_ma_deg = None
ump_ma_speed = None
ump_poly = None
# 使用训练集交易数据训练主裁
orders_pd_train_cn = abu_result_tuple.orders_pd
def load_main_ump_user():
global ump_price_with_ma_deg, ump_ma_speed_with_ind_ma_deg_and_smlr, ump_index_poly_and_ma_deg, ump_poly_with_ma_deg,\
ump_index_ma_deg, ump_index_poly, ump_ma_deg, ump_ma_speed, ump_poly
ump_price_with_ma_deg = WzUmpMainPriceWithMaDeg(predict=True)
ump_ma_speed_with_ind_ma_deg_and_smlr = WzUmpMainMaSpeedWithIndMaDegAndSmlr(predict=True)
ump_index_poly_and_ma_deg = WzUmpMainIndexPolyAndMaDeg(predict=True)
ump_poly_with_ma_deg = WzUmpMainPolyWithMaDeg(predict=True)
ump_index_ma_deg = WzUmpMainIndexMaDeg(predict=True)
ump_index_poly = WzUmpMainIndexPoly(predict=True)
ump_ma_deg = WzUmpMainMaDeg(predict=True)
ump_ma_speed = WzUmpMainMaSpeed(predict=True)
ump_poly = WzUmpMainPoly(predict=True)
print('load main ump user complete!')
def train_main_ump_user():
print('WzUmpMainPriceWithMaDeg begin...')
WzUmpMainPriceWithMaDeg.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False,show_order=False)
print('WzUmpMainMaSpeedWithIndMaDegAndSmlr begin...')
WzUmpMainMaSpeedWithIndMaDegAndSmlr.ump_main_clf_dump(orders_pd_train_cn,show_info=True, save_order=False,show_order=False)
print('WzUmpMainIndexPolyAndMaDeg begin...')
WzUmpMainIndexPolyAndMaDeg.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False,show_order=False)
print('WzUmpMainIndexMaDeg...')
WzUmpMainIndexMaDeg.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False,show_order=False)
# ump_ma_deg = WzUmpMainMaDeg.ump_main_clf_dump(orders_pd_train, p_ncs=slice(20, 40, 1))
# ump_ma_deg.fiter.df.head()
print('WzUmpMainIndexPoly...')
WzUmpMainIndexPoly.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False,show_order=False)
print('WzUmpMainPolyWithMaDeg begin...')
WzUmpMainPolyWithMaDeg.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False,show_order=False)
print('WzUmpMainMaDeg begin...')
WzUmpMainMaDeg.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False,show_order=False)
print('WzUmpMainMaSpeed begin...')
WzUmpMainMaSpeed.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False,show_order=False)
print('WzUmpMainPoly...')
WzUmpMainPoly.ump_main_clf_dump(orders_pd_train_cn, show_info=True, save_order=False,show_order=False)
load_main_ump_user()
print('train complete!')
def select(select):
if select == 'train main ump user':
train_main_ump_user()
else:
load_main_ump_user()
# _ = ipywidgets.interact_manual(select, select=['load main ump', 'train main ump'])
load_main_ump_user()
边裁¶
In [9]:
# from abupy import AbuUmpEdgeDeg, AbuUmpEdgePrice, AbuUmpEdgeWave, AbuUmpEdgeFull, AbuUmpEdgeMul, AbuUmpEegeDegExtend
# AbuUmpEdgeWave, AbuUmpEdgeFull, AbuUmpEdgeMul
# 需要全局设置为A股市场，在ump会根据市场类型保存读取对应的ump
abupy.env.g_market_target = EMarketTargetType.E_MARKET_TARGET_CN
# 使用训练集交易数据训练
orders_pd_train_cn = abu_result_tuple.orders_pd
print('WzUmpEdgePriceWithMaDeg begin...')
WzUmpEdgePriceWithMaDeg.ump_edge_clf_dump(orders_pd_train_cn)
edge_price_ma_deg = WzUmpEdgePriceWithMaDeg(predict=True)
print('WzUmpEdgeMaSpeedWithIndMaDegAndSmlr begin...')
WzUmpEdgeMaSpeedWithIndMaDegAndSmlr.ump_edge_clf_dump(orders_pd_train_cn)
edge_ma_speed_with_ind_ma_deg_and_smlr = WzUmpEdgeMaSpeedWithIndMaDegAndSmlr(predict=True)
print('WzUmpEdgeIndexPolyAndMaDeg begin...')
WzUmpEdgeIndexPolyAndMaDeg.ump_edge_clf_dump(orders_pd_train_cn)
edge_index_poly_and_ma_deg = WzUmpEdgeIndexPolyAndMaDeg(predict=True)
print('WzUmpEdgePolyWithMaDeg begin...')
WzUmpEdgePolyWithMaDeg.ump_edge_clf_dump(orders_pd_train_cn)
edge_poly_with_ma_deg = WzUmpEdgePolyWithMaDeg(predict=True)
print('WzUmpEdgeIndexMaDeg begin...')
WzUmpEdgeIndexMaDeg.ump_edge_clf_dump(orders_pd_train_cn)
edge_index_ma_deg = WzUmpEdgeIndexMaDeg(predict=True)
print('WzUmpEdgeIndexPoly begin...')
WzUmpEdgeIndexPoly.ump_edge_clf_dump(orders_pd_train_cn)
edge_index_poly = WzUmpEdgeIndexPoly(predict=True)
print('WzUmpEdgeMaDeg begin...')
WzUmpEdgeMaDeg.ump_edge_clf_dump(orders_pd_train_cn)
edge_ma_deg = WzUmpEdgeMaDeg(predict=True)
print('WzUmpEdgeMaSpeed begin...')
WzUmpEdgeMaSpeed.ump_edge_clf_dump(orders_pd_train_cn)
edge_ma_speed = WzUmpEdgeMaSpeed(predict=True)
print('WzUmpEdgePoly begin...')
WzUmpEdgePoly.ump_edge_clf_dump(orders_pd_train_cn)
edge_poly = WzUmpEdgePoly(predict=True)
# 内置的
print('AbuUmpEdgeDeg begin...')
AbuUmpEdgeDeg.ump_edge_clf_dump(orders_pd_train_cn)
edge_deg = AbuUmpEdgeDeg(predict=True)
print('AbuUmpEdgePrice begin...')
AbuUmpEdgePrice.ump_edge_clf_dump(orders_pd_train_cn)
edge_price = AbuUmpEdgePrice(predict=True)
print('AbuUmpEdgeMul begin...')
AbuUmpEdgeMul.ump_edge_clf_dump(orders_pd_train_cn)
edge_mul = AbuUmpEdgeMul(predict=True)
print('AbuUmpEegeDegExtend begin...')
AbuUmpEegeDegExtend.ump_edge_clf_dump(orders_pd_train_cn)
edge_deg_extend = AbuUmpEegeDegExtend(predict=True)
print('AbuUmpEdgeWave begin...')
AbuUmpEdgeWave.ump_edge_clf_dump(orders_pd_train_cn)
edge_wave = AbuUmpEdgeWave(predict=True)
print('AbuUmpEdgeFull begin...')
AbuUmpEdgeFull.ump_edge_clf_dump(orders_pd_train_cn)
edge_full = AbuUmpEdgeFull(predict=True)
print('fit edge complete!')
print(moment.utcnow().timezone("Asia/Shanghai").format("YYYYMMDD h:m:s A"))
In [7]:
from warnings import simplefilter
simplefilter(action='ignore', category=FutureWarning)
simplefilter(action='ignore', category=ResourceWarning)
# 测试集回测时依然生成买入时刻特征
# abupy.env.g_enable_ml_feature = True #True
# 回测时不重新切割训练集数据和测试集
abupy.env.g_enable_train_test_split = False
# 回测时使用切割好的测试数据
abupy.env.g_enable_last_split_test = True
# 测试集依然使用10，30，50，90，120日走势拟合角度特征AbuFeatureDegExtend，做为回测时的新的视角来录制比赛（记录回测特征）
# feature.clear_user_feature()
# feature.append_user_feature(AbuFeatureDegExtend)
read_cash = 50000000 #5000w资金
o_pos_base =0.1 #abupy.beta.atr.g_atr_pos_base
print('o_pos_base:', o_pos_base)
abupy.beta.atr.g_atr_pos_base = o_pos_base /6
print('abupy.beta.atr.g_atr_pos_base:' , abupy.beta.atr.g_atr_pos_base)
abu_result_tuple_test = None
order_has_result = None
def run_loop_back_test():
global abu_result_tuple_test, order_has_result
abu_result_tuple_test, _ = abu.run_loop_back(read_cash,
buy_factors,
sell_factors,
choice_symbols=None,
start='2016-02-02', end='2019-12-18',
n_process_kl = 1,
n_process_pick= 8,
)
# 把运行的结果保存在本地，以便之后分析回测使用，保存回测结果数据代码如下所示
abu.store_abu_result_tuple(abu_result_tuple_test, n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_cn')
ABuProgress.clear_output()
AbuMetricsBase.show_general(*abu_result_tuple_test, only_show_returns=True).plot_max_draw_down()
# 验证A股主裁是否称职
# 选取有交易结果的数据order_has_result
order_has_result = abu_result_tuple_test.orders_pd[abu_result_tuple_test.orders_pd.result != 0]
order_has_result.filter(regex='^buy(_deg_|_price_|_wave_|_jump)').head()
def run_load_test():
global abu_result_tuple_test, order_has_result
abu_result_tuple_test = abu.load_abu_result_tuple(n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_cn')
AbuMetricsBase.show_general(*abu_result_tuple_test, only_show_returns=True).plot_max_draw_down()
# 验证A股主裁是否称职
# 选取有交易结果的数据order_has_result
order_has_result = abu_result_tuple_test.orders_pd[abu_result_tuple_test.orders_pd.result != 0]
order_has_result.filter(regex='^buy(_deg_|_price_|_wave_|_jump)').head()
def select_test(select):
if select == 'run loop back':
run_loop_back_test()
else:
run_load_test()
# _ = ipywidgets.interact_manual(select_test, select=['load test data', 'run loop back',])
run_loop_back_test()
上面是没有启用机器人裁判的成绩。
以下启用机器人裁判，看看效果¶
单使用边裁
In [9]:
from warnings import simplefilter
simplefilter(action='ignore', category=FutureWarning)
# simplefilter(action='ignore', category=ResourceWarning)
abupy.env.g_data_cache_type = EDataCacheType.E_DATA_CACHE_CSV
abupy.env.g_market_target = EMarketTargetType.E_MARKET_TARGET_CN
abupy.env.g_data_fetch_mode = EMarketDataFetchMode.E_DATA_FETCH_FORCE_LOCAL
# 开启内置主裁
abupy.env.g_enable_ump_main_deg_block = False
# abupy.env.g_enable_ump_main_price_block = True
# 开启内置边裁
abupy.env.g_enable_ump_edge_deg_block = True
# abupy.env.g_enable_ump_edge_price_block = True
# 回测时需要开启特征生成，因为裁判开启需要生成特征做为输入
abupy.env.g_enable_ml_feature = True
# 回测时使用上一次切割好的测试集数据
abupy.env.g_enable_last_split_test = True
""" """
feature.clear_user_feature()
# 10，30，50，90，120日走势拟合角度特征的AbuFeatureDegExtend，做为回测时的新的视角来录制比赛
feature.append_user_feature(AbuFeatureDegExtend)
feature.append_user_feature(WzFeatureMaDeg)
feature.append_user_feature(WzFeatureIndexMaDeg)
feature.append_user_feature(WzFeatureMaSpeed)
feature.append_user_feature(WzFeatureIndexPoly)
feature.append_user_feature(WzFeaturePoly) # <- 问题嫌疑
feature.append_user_feature(WzFeatureIndexSimilarity)
# 打开使用用户自定义裁判开关
ump.manager.g_enable_user_ump = True
# 先clear一下
ump.manager.clear_user_ump()
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
ump.manager.append_user_ump(AbuUmpEegeDegExtend)
# 机器人裁判,通过对比 个股买入点前n*1,n*2,n*x天至买入当天的均线角度 和 对应大盘
# 的均线角度 得知该股是否跑赢大盘，以及跑赢百分比
ump.manager.append_user_ump(WzUmpEdgeMaSpeed)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的价格rank(相当于压力位/支撑位)和
# 均线角度
ump.manager.append_user_ump(WzUmpEdgePriceWithMaDeg)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的个股对应
# 大盘指数(上证50,上证,深证,创业)的均线角度
ump.manager.append_user_ump(WzUmpEdgeIndexMaDeg)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的个股对应大盘指数的处于
# 趋势or震荡(采用多项式拟合技术)
ump.manager.append_user_ump(WzUmpEdgeIndexPoly)
# 下面就不一样介绍了
ump.manager.append_user_ump(WzUmpEdgeMaDeg)
ump.manager.append_user_ump(WzUmpEdgePoly)
ump.manager.append_user_ump(WzUmpEdgeMaSpeedWithIndMaDegAndSmlr)
ump.manager.append_user_ump(WzUmpEdgeIndexPolyAndMaDeg)
ump.manager.append_user_ump(WzUmpEdgePolyWithMaDeg)
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
# ump.manager.append_user_ump(AbuUmpMainDegExtend)
#
# ump.manager.append_user_ump(WzUmpMainMaDeg)
# ump.manager.append_user_ump(WzUmpMainMaSpeed)
# ump.manager.append_user_ump(WzUmpMainIndexMaDeg)
# ump.manager.append_user_ump(WzUmpMainIndexPoly)
# # ump.manager.append_user_ump(WzUmpMainPoly)
# ump.manager.append_user_ump(WzUmpMainMaSpeedWithIndMaDegAndSmlr)
# ump.manager.append_user_ump(WzUmpMainIndexPolyAndMaDeg)
# ump.manager.append_user_ump(WzUmpMainPolyWithMaDeg)
read_cash = 10000000 #资金
o_pos_base =0.1 #abupy.beta.atr.g_atr_pos_base
print('o_pos_base:', o_pos_base)
pt_pos_base = o_pos_base / 6
print('pt_pos_base:' , pt_pos_base)
buy_factors=[{'buy_default': 'post=36.88',
'class': WzFactorBuyRsiV8,
'position': {'class': AbuPtPosition,
'past_day_cnt': 68,
'pos_base': pt_pos_base },
'rsi_timeperiod': 6,
'stock_pickers': [{'class': WzPickNonNew,
'pick_period': 'week',
'weeks': 24},
]
}]
abu_result_tuple_test_ump = None
def run_loop_back_ump():
global abu_result_tuple_test_ump, read_cash
abu_result_tuple_test_ump, _ = abu.run_loop_back(read_cash,
buy_factors,
sell_factors,
# choice_symbols = symbols,
choice_symbols=None,
# stock_picks= stock_pickers,
start='2016-02-02', end='2019-12-18',
n_process_kl = 8,
n_process_pick= 8,
)
# 把运行的结果保存在本地，以便之后分析回测使用，保存回测结果数据代码如下所示
abu.store_abu_result_tuple(abu_result_tuple_test_ump, n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_ump_cn')
ABuProgress.clear_output()
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def run_load_ump():
global abu_result_tuple_test_ump
abu_result_tuple_test_ump = abu.load_abu_result_tuple(n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_ump_cn')
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def select_ump(select):
if select == 'run loop back ump':
run_loop_back_ump()
else:
run_load_ump()
# _ = ipywidgets.interact_manual(select_ump, select=['load test ump data','run loop back ump'])
# print("随便输出点什么表示已经执行过： ")
run_loop_back_ump()
# order_has_result.filter(regex='^buy(_deg_|_price_|_wave_|_jump)').head()
边裁 + 3个自定义主裁
In [13]:
from warnings import simplefilter
simplefilter(action='ignore', category=FutureWarning)
# simplefilter(action='ignore', category=ResourceWarning)
abupy.env.g_data_cache_type = EDataCacheType.E_DATA_CACHE_CSV
abupy.env.g_market_target = EMarketTargetType.E_MARKET_TARGET_CN
abupy.env.g_data_fetch_mode = EMarketDataFetchMode.E_DATA_FETCH_FORCE_LOCAL
# 开启内置主裁
abupy.env.g_enable_ump_main_deg_block = False
abupy.env.g_enable_ump_main_jump_block = True
# abupy.env.g_enable_ump_main_price_block = True
# 开启内置边裁
abupy.env.g_enable_ump_edge_deg_block = True
abupy.env.g_enable_ump_edge_price_block = True
abupy.g_enable_ump_edge_wave_block = True
abupy.g_enable_ump_edge_full_block = True
# 回测时需要开启特征生成，因为裁判开启需要生成特征做为输入
abupy.env.g_enable_ml_feature = True
# 回测时使用上一次切割好的测试集数据
abupy.env.g_enable_last_split_test = True
""" """
feature.clear_user_feature()
# 10，30，50，90，120日走势拟合角度特征的AbuFeatureDegExtend，做为回测时的新的视角来录制比赛
feature.append_user_feature(AbuFeatureDegExtend)
feature.append_user_feature(WzFeatureMaDeg)
feature.append_user_feature(WzFeatureIndexMaDeg)
feature.append_user_feature(WzFeatureMaSpeed)
feature.append_user_feature(WzFeatureIndexPoly)
feature.append_user_feature(WzFeaturePoly) # <- 问题嫌疑
feature.append_user_feature(WzFeatureIndexSimilarity)
# 打开使用用户自定义裁判开关
ump.manager.g_enable_user_ump = True
# 先clear一下
ump.manager.clear_user_ump()
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
ump.manager.append_user_ump(AbuUmpEegeDegExtend)
# 机器人裁判,通过对比 个股买入点前n*1,n*2,n*x天至买入当天的均线角度 和 对应大盘
# 的均线角度 得知该股是否跑赢大盘，以及跑赢百分比
ump.manager.append_user_ump(WzUmpEdgeMaSpeed)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的价格rank(相当于压力位/支撑位)和
# 均线角度
ump.manager.append_user_ump(WzUmpEdgePriceWithMaDeg)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的个股对应
# 大盘指数(上证50,上证,深证,创业)的均线角度
ump.manager.append_user_ump(WzUmpEdgeIndexMaDeg)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的个股对应大盘指数的处于
# 趋势or震荡(采用多项式拟合技术)
ump.manager.append_user_ump(WzUmpEdgeIndexPoly)
# 下面就不一样介绍了
ump.manager.append_user_ump(WzUmpEdgeMaDeg)
ump.manager.append_user_ump(WzUmpEdgePoly)
ump.manager.append_user_ump(WzUmpEdgeMaSpeedWithIndMaDegAndSmlr)
ump.manager.append_user_ump(WzUmpEdgeIndexPolyAndMaDeg)
ump.manager.append_user_ump(WzUmpEdgePolyWithMaDeg)
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
# ump.manager.append_user_ump(AbuUmpMainDegExtend)
#
ump.manager.append_user_ump(WzUmpMainMaDeg)
# ump.manager.append_user_ump(WzUmpMainMaSpeed)
# ump.manager.append_user_ump(WzUmpMainIndexMaDeg)
# ump.manager.append_user_ump(WzUmpMainIndexPoly)
# # ump.manager.append_user_ump(WzUmpMainPoly)
# ump.manager.append_user_ump(WzUmpMainMaSpeedWithIndMaDegAndSmlr)
# ump.manager.append_user_ump(WzUmpMainIndexPolyAndMaDeg)
ump.manager.append_user_ump(WzUmpMainPolyWithMaDeg)
ump.manager.append_user_ump(WzUmpMainPriceWithMaDeg)
read_cash = 10000000 #资金
o_pos_base =0.1 #abupy.beta.atr.g_atr_pos_base
print('o_pos_base:', o_pos_base)
pt_pos_base = o_pos_base / 1.0
print('pt_pos_base:' , pt_pos_base)
buy_factors=[{'buy_default': 'post=36.88',
'class': WzFactorBuyRsiV8,
'position': {'class': AbuPtPosition,
'past_day_cnt': 68,
'pos_base': pt_pos_base },
'rsi_timeperiod': 6,
'stock_pickers': [{'class': WzPickNonNew,
'pick_period': 'week',
'weeks': 24},
]
}]
abu_result_tuple_test_ump = None
def run_loop_back_ump():
global abu_result_tuple_test_ump, read_cash
abu_result_tuple_test_ump, _ = abu.run_loop_back(read_cash,
buy_factors,
sell_factors,
# choice_symbols = symbols,
choice_symbols=None,
# stock_picks= stock_pickers,
start='2016-02-02', end='2019-12-18',
# n_process_kl = 8,
# n_process_pick= 8,
)
# 把运行的结果保存在本地，以便之后分析回测使用，保存回测结果数据代码如下所示
abu.store_abu_result_tuple(abu_result_tuple_test_ump, n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_ump_cn_')
ABuProgress.clear_output()
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def run_load_ump():
global abu_result_tuple_test_ump
abu_result_tuple_test_ump = abu.load_abu_result_tuple(n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_ump_cn_')
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def select_ump(select):
if select == 'run loop back ump':
run_loop_back_ump()
else:
run_load_ump()
# _ = ipywidgets.interact_manual(select_ump, select=['load test ump data','run loop back ump'])
# print("随便输出点什么表示已经执行过： ")
run_loop_back_ump()
# order_has_result.filter(regex='^buy(_deg_|_price_|_wave_|_jump)').head()
边裁 + 3个自定义主裁。
上面买入成交比例过小，存在运气成分。 下面降低单只股单次买入的仓位，以提高买入成交率
In [14]:
from warnings import simplefilter
simplefilter(action='ignore', category=FutureWarning)
# simplefilter(action='ignore', category=ResourceWarning)
abupy.env.g_data_cache_type = EDataCacheType.E_DATA_CACHE_CSV
abupy.env.g_market_target = EMarketTargetType.E_MARKET_TARGET_CN
abupy.env.g_data_fetch_mode = EMarketDataFetchMode.E_DATA_FETCH_FORCE_LOCAL
# 开启内置主裁
abupy.env.g_enable_ump_main_deg_block = False
abupy.env.g_enable_ump_main_jump_block = True
# abupy.env.g_enable_ump_main_price_block = True
# 开启内置边裁
abupy.env.g_enable_ump_edge_deg_block = True
abupy.env.g_enable_ump_edge_price_block = True
abupy.g_enable_ump_edge_wave_block = True
abupy.g_enable_ump_edge_full_block = True
# 回测时需要开启特征生成，因为裁判开启需要生成特征做为输入
abupy.env.g_enable_ml_feature = True
# 回测时使用上一次切割好的测试集数据
abupy.env.g_enable_last_split_test = True
""" """
feature.clear_user_feature()
# 10，30，50，90，120日走势拟合角度特征的AbuFeatureDegExtend，做为回测时的新的视角来录制比赛
feature.append_user_feature(AbuFeatureDegExtend)
feature.append_user_feature(WzFeatureMaDeg)
feature.append_user_feature(WzFeatureIndexMaDeg)
feature.append_user_feature(WzFeatureMaSpeed)
feature.append_user_feature(WzFeatureIndexPoly)
feature.append_user_feature(WzFeaturePoly) # <- 问题嫌疑
feature.append_user_feature(WzFeatureIndexSimilarity)
# 打开使用用户自定义裁判开关
ump.manager.g_enable_user_ump = True
# 先clear一下
ump.manager.clear_user_ump()
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
ump.manager.append_user_ump(AbuUmpEegeDegExtend)
# 机器人裁判,通过对比 个股买入点前n*1,n*2,n*x天至买入当天的均线角度 和 对应大盘
# 的均线角度 得知该股是否跑赢大盘，以及跑赢百分比
ump.manager.append_user_ump(WzUmpEdgeMaSpeed)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的价格rank(相当于压力位/支撑位)和
# 均线角度
ump.manager.append_user_ump(WzUmpEdgePriceWithMaDeg)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的个股对应
# 大盘指数(上证50,上证,深证,创业)的均线角度
ump.manager.append_user_ump(WzUmpEdgeIndexMaDeg)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的个股对应大盘指数的处于
# 趋势or震荡(采用多项式拟合技术)
ump.manager.append_user_ump(WzUmpEdgeIndexPoly)
# 下面就不一样介绍了
ump.manager.append_user_ump(WzUmpEdgeMaDeg)
ump.manager.append_user_ump(WzUmpEdgePoly)
ump.manager.append_user_ump(WzUmpEdgeMaSpeedWithIndMaDegAndSmlr)
ump.manager.append_user_ump(WzUmpEdgeIndexPolyAndMaDeg)
ump.manager.append_user_ump(WzUmpEdgePolyWithMaDeg)
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
# ump.manager.append_user_ump(AbuUmpMainDegExtend)
#
ump.manager.append_user_ump(WzUmpMainMaDeg)
# ump.manager.append_user_ump(WzUmpMainMaSpeed)
# ump.manager.append_user_ump(WzUmpMainIndexMaDeg)
# ump.manager.append_user_ump(WzUmpMainIndexPoly)
# # ump.manager.append_user_ump(WzUmpMainPoly)
# ump.manager.append_user_ump(WzUmpMainMaSpeedWithIndMaDegAndSmlr)
# ump.manager.append_user_ump(WzUmpMainIndexPolyAndMaDeg)
ump.manager.append_user_ump(WzUmpMainPolyWithMaDeg)
ump.manager.append_user_ump(WzUmpMainPriceWithMaDeg)
read_cash = 10000000 #资金
o_pos_base =0.1 #abupy.beta.atr.g_atr_pos_base
print('o_pos_base:', o_pos_base)
pt_pos_base = o_pos_base / 2.0
print('pt_pos_base:' , pt_pos_base)
buy_factors=[{'buy_default': 'post=36.88',
'class': WzFactorBuyRsiV8,
'position': {'class': AbuPtPosition,
'past_day_cnt': 68,
'pos_base': pt_pos_base },
'rsi_timeperiod': 6,
'stock_pickers': [{'class': WzPickNonNew,
'pick_period': 'week',
'weeks': 24},
]
}]
abu_result_tuple_test_ump = None
def run_loop_back_ump():
global abu_result_tuple_test_ump, read_cash
abu_result_tuple_test_ump, _ = abu.run_loop_back(read_cash,
buy_factors,
sell_factors,
# choice_symbols = symbols,
choice_symbols=None,
# stock_picks= stock_pickers,
start='2016-02-02', end='2019-12-18',
# n_process_kl = 8,
# n_process_pick= 8,
)
# 把运行的结果保存在本地，以便之后分析回测使用，保存回测结果数据代码如下所示
abu.store_abu_result_tuple(abu_result_tuple_test_ump, n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_ump_cn_a')
ABuProgress.clear_output()
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def run_load_ump():
global abu_result_tuple_test_ump
abu_result_tuple_test_ump = abu.load_abu_result_tuple(n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_ump_cn_a')
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def select_ump(select):
if select == 'run loop back ump':
run_loop_back_ump()
else:
run_load_ump()
# _ = ipywidgets.interact_manual(select_ump, select=['load test ump data','run loop back ump'])
# print("随便输出点什么表示已经执行过： ")
run_loop_back_ump()
# order_has_result.filter(regex='^buy(_deg_|_price_|_wave_|_jump)').head()
In [11]:
from warnings import simplefilter
simplefilter(action='ignore', category=FutureWarning)
# simplefilter(action='ignore', category=ResourceWarning)
abupy.env.g_data_cache_type = EDataCacheType.E_DATA_CACHE_CSV
abupy.env.g_market_target = EMarketTargetType.E_MARKET_TARGET_CN
abupy.env.g_data_fetch_mode = EMarketDataFetchMode.E_DATA_FETCH_FORCE_LOCAL
# 开启内置主裁
abupy.env.g_enable_ump_main_deg_block = False
abupy.env.g_enable_ump_main_jump_block = True
# abupy.env.g_enable_ump_main_price_block = True
# 开启内置边裁
abupy.env.g_enable_ump_edge_deg_block = True
abupy.env.g_enable_ump_edge_price_block = True
abupy.g_enable_ump_edge_wave_block = True
abupy.g_enable_ump_edge_full_block = True
# 回测时需要开启特征生成，因为裁判开启需要生成特征做为输入
abupy.env.g_enable_ml_feature = True
# 回测时使用上一次切割好的测试集数据
abupy.env.g_enable_last_split_test = True
""" """
feature.clear_user_feature()
# 10，30，50，90，120日走势拟合角度特征的AbuFeatureDegExtend，做为回测时的新的视角来录制比赛
feature.append_user_feature(AbuFeatureDegExtend)
feature.append_user_feature(WzFeatureMaDeg)
feature.append_user_feature(WzFeatureIndexMaDeg)
feature.append_user_feature(WzFeatureMaSpeed)
feature.append_user_feature(WzFeatureIndexPoly)
feature.append_user_feature(WzFeaturePoly) # <- 问题嫌疑
feature.append_user_feature(WzFeatureIndexSimilarity)
# 打开使用用户自定义裁判开关
ump.manager.g_enable_user_ump = True
# 先clear一下
ump.manager.clear_user_ump()
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
ump.manager.append_user_ump(AbuUmpEegeDegExtend)
# 机器人裁判,通过对比 个股买入点前n*1,n*2,n*x天至买入当天的均线角度 和 对应大盘
# 的均线角度 得知该股是否跑赢大盘，以及跑赢百分比
ump.manager.append_user_ump(WzUmpEdgeMaSpeed)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的价格rank(相当于压力位/支撑位)和
# 均线角度
ump.manager.append_user_ump(WzUmpEdgePriceWithMaDeg)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的个股对应
# 大盘指数(上证50,上证,深证,创业)的均线角度
ump.manager.append_user_ump(WzUmpEdgeIndexMaDeg)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的个股对应大盘指数的处于
# 趋势or震荡(采用多项式拟合技术)
ump.manager.append_user_ump(WzUmpEdgeIndexPoly)
# 下面就不一样介绍了
ump.manager.append_user_ump(WzUmpEdgeMaDeg)
ump.manager.append_user_ump(WzUmpEdgePoly)
ump.manager.append_user_ump(WzUmpEdgeMaSpeedWithIndMaDegAndSmlr)
ump.manager.append_user_ump(WzUmpEdgeIndexPolyAndMaDeg)
ump.manager.append_user_ump(WzUmpEdgePolyWithMaDeg)
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
# ump.manager.append_user_ump(AbuUmpMainDegExtend)
#
ump.manager.append_user_ump(WzUmpMainMaDeg)
# ump.manager.append_user_ump(WzUmpMainMaSpeed)
# ump.manager.append_user_ump(WzUmpMainIndexMaDeg)
# ump.manager.append_user_ump(WzUmpMainIndexPoly)
# # ump.manager.append_user_ump(WzUmpMainPoly)
# ump.manager.append_user_ump(WzUmpMainMaSpeedWithIndMaDegAndSmlr)
# ump.manager.append_user_ump(WzUmpMainIndexPolyAndMaDeg)
ump.manager.append_user_ump(WzUmpMainPolyWithMaDeg)
ump.manager.append_user_ump(WzUmpMainPriceWithMaDeg)
read_cash = 10000000 #资金
o_pos_base =0.1 #abupy.beta.atr.g_atr_pos_base
print('o_pos_base:', o_pos_base)
pt_pos_base = o_pos_base / 4.0
print('pt_pos_base:' , pt_pos_base)
buy_factors=[{'buy_default': 'post=36.88',
'class': WzFactorBuyRsiV8,
'position': {'class': AbuPtPosition,
'past_day_cnt': 68,
'pos_base': pt_pos_base },
'rsi_timeperiod': 6,
'stock_pickers': [{'class': WzPickNonNew,
'pick_period': 'week',
'weeks': 24},
]
}]
abu_result_tuple_test_ump = None
def run_loop_back_ump():
global abu_result_tuple_test_ump, read_cash
abu_result_tuple_test_ump, _ = abu.run_loop_back(read_cash,
buy_factors,
sell_factors,
# choice_symbols = symbols,
choice_symbols=None,
# stock_picks= stock_pickers,
start='2016-02-02', end='2019-12-18',
# n_process_kl = 8,
# n_process_pick= 8,
)
# 把运行的结果保存在本地，以便之后分析回测使用，保存回测结果数据代码如下所示
abu.store_abu_result_tuple(abu_result_tuple_test_ump, n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_ump_cn_b')
ABuProgress.clear_output()
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def run_load_ump():
global abu_result_tuple_test_ump
abu_result_tuple_test_ump = abu.load_abu_result_tuple(n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_ump_cn_b')
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def select_ump(select):
if select == 'run loop back ump':
run_loop_back_ump()
else:
run_load_ump()
# _ = ipywidgets.interact_manual(select_ump, select=['load test ump data','run loop back ump'])
# print("随便输出点什么表示已经执行过： ")
run_loop_back_ump()
# order_has_result.filter(regex='^buy(_deg_|_price_|_wave_|_jump)').head()
全市场测试¶
In [ ]:
# from warnings import simplefilter
# simplefilter(action='ignore', category=FutureWarning)
# simplefilter(action='ignore', category=ResourceWarning)
# abupy.env.g_data_cache_type = EDataCacheType.E_DATA_CACHE_CSV
# abupy.env.g_market_target = EMarketTargetType.E_MARKET_TARGET_CN
# abupy.env.g_data_fetch_mode = EMarketDataFetchMode.E_DATA_FETCH_FORCE_LOCAL
"""是否开启选股切割训练集股票数据与测试集股票数据，默认关闭False"""
abupy.env.g_enable_train_test_split = False
"""是否开启选股使用上一次切割完成的 测试集 股票数据， """
abupy.env.g_enable_last_split_test = False
"""是否开启选股使用上一次切割完成的 训练集 股票数据， """
abupy.env.g_enable_last_split_train = False
# 开启内置主裁
abupy.env.g_enable_ump_main_deg_block = True
# abupy.env.g_enable_ump_main_price_block = True
# 开启内置边裁
abupy.env.g_enable_ump_edge_deg_block = True
# abupy.env.g_enable_ump_edge_price_block = True
# 回测时需要开启特征生成，因为裁判开启需要生成特征做为输入
abupy.env.g_enable_ml_feature = True
# 回测时使用上一次切割好的测试集数据
# abupy.env.g_enable_last_split_test = True
""" """
feature.clear_user_feature()
# 10，30，50，90，120日走势拟合角度特征的AbuFeatureDegExtend，做为回测时的新的视角来录制比赛
feature.append_user_feature(AbuFeatureDegExtend)
# 打开使用用户自定义裁判开关
ump.manager.g_enable_user_ump = True
# 先clear一下
ump.manager.clear_user_ump()
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
ump.manager.append_user_ump(AbuUmpEegeDegExtend)
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
ump.manager.append_user_ump(AbuUmpMainDegExtend)
read_cash = 10000000 #500w资金
o_pos_base =0.1 #abupy.beta.atr.g_atr_pos_base
print('o_pos_base:', o_pos_base)
abupy.beta.atr.g_atr_pos_base = o_pos_base / 16
print('abupy.beta.atr.g_atr_pos_base:' , abupy.beta.atr.g_atr_pos_base)
abu_result_tuple_test_ump = None
read_cash = 5000000 #500w资金
def run_loop_back_ump():
global abu_result_tuple_test_ump, read_cash
abu_result_tuple_test_ump, _ = abu.run_loop_back(read_cash,
buy_factors,
sell_factors,
# choice_symbols = symbols,
choice_symbols=None,
# stock_picks= stock_pickers,
start='2016-02-02', end='2019-09-12',
# n_process_kl = 1,
# n_process_pick= 1,
)
# 把运行的结果保存在本地，以便之后分析回测使用，保存回测结果数据代码如下所示
abu.store_abu_result_tuple(abu_result_tuple_test_ump, n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_ump_cn_all')
ABuProgress.clear_output()
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def run_load_ump():
global abu_result_tuple_test_ump
abu_result_tuple_test_ump = abu.load_abu_result_tuple(n_folds=3, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='test_ump_cn_all')
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def select_ump(select):
if select == 'run loop back ump':
run_loop_back_ump()
else:
run_load_ump()
# _ = ipywidgets.interact_manual(select_ump, select=['load test ump data','run loop back ump'])
# print("随便输出点什么表示已经执行过： ")
run_loop_back_ump()
In [ ]:
import talib
from talib import MA_Type
from talib.abstract import *
from service.MaService import MaService
# print(type(train))
abupy.env.g_market_source = EMarketSourceType.E_MARKET_SOURCE_nt
abupy.env.g_data_cache_type = EDataCacheType.E_DATA_CACHE_CSV
abupy.env.g_data_fetch_mode = EMarketDataFetchMode.E_DATA_FETCH_FORCE_NET #E_DATA_FETCH_NORMAL #
# E_DATA_FETCH_NORMAL
ma_timeperiod = 100
missing =[]
start = '2019-01-01'
end = moment.date('tomorrow').format('YYYY-MM-DD')
os.environ['cache_start'] = '20190101'
os.environ['cache_end'] = moment.date('tomorrow').format('YYYYMMDD')
for i in symbols: #
# if str(i)[:5] == 'sz399' or str(i)[:3] == '399':
# print(i, 'continue跳过')
# continue
# print( i ,":")
kl = ABuSymbolPd.make_kl_df(i, start=start, end= end)
if kl is not None:
# print(kl.head(2))
# print(kl.tail(2))
# print(i, '\r\n', kl.head(2), '\r\n')
# print(i, '\r\n', kl.tail(2), '\r\n')
ma = MaService.calc_ma(kl.name, kl, ma_timeperiod, MA_Type.EMA, save_cache =True)
# print(i, 'head:\r\n', ma.head(10), '\r\n')
# print(i, 'tail:\r\n', ma.tail(2), '\r\n')
else:
print(i, '')
missing.append(i)
print(len(missing))
print( len(missing), '网易数据缺：', missing)
# 网易没有上证指数的行情数据，改用百度
abupy.env.g_market_source = EMarketSourceType.E_MARKET_SOURCE_bd
print(abupy.env.g_market_source)
kl = ABuSymbolPd.make_kl_df('sh000001', start=start, end=end)
print(len(missing))
print( len(missing), '网易数据缺：', missing)
In [ ]:
# 网易缺乏大盘指数数据，到百度取
abupy.env.g_market_source = EMarketSourceType.E_MARKET_SOURCE_bd
print(abupy.env.g_market_source)
end = moment.date('tomorrow').format('YYYY-MM-DD')
kl = ABuSymbolPd.make_kl_df('sh000001', start=start, end=end)
# print(kl)
实盘开始
In [12]:
from warnings import simplefilter
simplefilter(action='ignore', category=FutureWarning)
# simplefilter(action='ignore', category=ResourceWarning)
abupy.env.g_data_cache_type = EDataCacheType.E_DATA_CACHE_CSV
abupy.env.g_market_target = EMarketTargetType.E_MARKET_TARGET_CN
abupy.env.g_data_fetch_mode = EMarketDataFetchMode.E_DATA_FETCH_FORCE_LOCAL
# 回测时需要开启特征生成，因为裁判开启需要生成特征做为输入
abupy.env.g_enable_ml_feature = True
# 回测时使用上一次切割好的测试集数据
abupy.env.g_enable_last_split_test = False
abupy.env.g_enable_train_test_split = False
# 开启内置主裁
abupy.env.g_enable_ump_main_deg_block = False
abupy.env.g_enable_ump_main_jump_block = True
# abupy.env.g_enable_ump_main_price_block = True
# 开启内置边裁
abupy.env.g_enable_ump_edge_deg_block = True
abupy.env.g_enable_ump_edge_price_block = True
abupy.g_enable_ump_edge_wave_block = True
abupy.g_enable_ump_edge_full_block = True
""" """
feature.clear_user_feature()
# 10，30，50，90，120日走势拟合角度特征的AbuFeatureDegExtend，做为回测时的新的视角来录制比赛
feature.append_user_feature(AbuFeatureDegExtend)
feature.append_user_feature(WzFeatureMaDeg)
feature.append_user_feature(WzFeatureIndexMaDeg)
feature.append_user_feature(WzFeatureMaSpeed)
feature.append_user_feature(WzFeatureIndexPoly)
feature.append_user_feature(WzFeaturePoly) # <- 问题嫌疑
feature.append_user_feature(WzFeatureIndexSimilarity)
# 打开使用用户自定义裁判开关
ump.manager.g_enable_user_ump = True
# 先clear一下
ump.manager.clear_user_ump()
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
ump.manager.append_user_ump(AbuUmpEegeDegExtend)
# 机器人裁判,通过对比 个股买入点前n*1,n*2,n*x天至买入当天的均线角度 和 对应大盘
# 的均线角度 得知该股是否跑赢大盘，以及跑赢百分比
ump.manager.append_user_ump(WzUmpEdgeMaSpeed)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的价格rank(相当于压力位/支撑位)和
# 均线角度
ump.manager.append_user_ump(WzUmpEdgePriceWithMaDeg)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的个股对应
# 大盘指数(上证50,上证,深证,创业)的均线角度
ump.manager.append_user_ump(WzUmpEdgeIndexMaDeg)
# 机器人裁判,观察买入点前n*1,n*2,n*x天至买入当天的个股对应大盘指数的处于
# 趋势or震荡(采用多项式拟合技术)
ump.manager.append_user_ump(WzUmpEdgeIndexPoly)
# 下面就不一样介绍了
ump.manager.append_user_ump(WzUmpEdgeMaDeg)
ump.manager.append_user_ump(WzUmpEdgePoly)
ump.manager.append_user_ump(WzUmpEdgeMaSpeedWithIndMaDegAndSmlr)
ump.manager.append_user_ump(WzUmpEdgeIndexPolyAndMaDeg)
ump.manager.append_user_ump(WzUmpEdgePolyWithMaDeg)
# 把新的裁判AbuUmpMainDegExtend类名称使用append_user_ump添加到系统中
# ump.manager.append_user_ump(AbuUmpMainDegExtend)
#
ump.manager.append_user_ump(WzUmpMainMaDeg)
# ump.manager.append_user_ump(WzUmpMainMaSpeed)
# ump.manager.append_user_ump(WzUmpMainIndexMaDeg)
# ump.manager.append_user_ump(WzUmpMainIndexPoly)
# # ump.manager.append_user_ump(WzUmpMainPoly)
# ump.manager.append_user_ump(WzUmpMainMaSpeedWithIndMaDegAndSmlr)
# ump.manager.append_user_ump(WzUmpMainIndexPolyAndMaDeg)
ump.manager.append_user_ump(WzUmpMainPolyWithMaDeg)
ump.manager.append_user_ump(WzUmpMainPriceWithMaDeg)
read_cash = 5000000 #资金 500w
o_pos_base =0.1 #abupy.beta.atr.g_atr_pos_base
print('o_pos_base:', o_pos_base)
pt_pos_base = o_pos_base / 8
print('pt_pos_base:' , pt_pos_base)
buy_factors=[{'buy_default': 'post=36.88',
'class': WzFactorBuyRsiV8,
'position': {'class': AbuPtPosition,
'past_day_cnt': 68,
'pos_base': pt_pos_base },
'rsi_timeperiod': 6,
'stock_pickers': [{'class': WzPickNonNew,
'pick_period': 'week',
'weeks': 24},
]
}]
abu_result_tuple_test_ump = None
start = '2019-01-01'
end = moment.date('tomorrow').format('YYYY-MM-DD')
def run_loop_back_ump():
global abu_result_tuple_test_ump, read_cash
abu_result_tuple_test_ump, _ = abu.run_loop_back(read_cash,
buy_factors,
sell_factors,
choice_symbols = symbols,
# choice_symbols=None,
# stock_picks= stock_pickers,
start=start, end=end,
# n_process_kl = 6,
# n_process_pick= 6,
)
# 把运行的结果保存在本地，以便之后分析回测使用，保存回测结果数据代码如下所示
abu.store_abu_result_tuple(abu_result_tuple_test_ump, n_folds=1, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='live')
ABuProgress.clear_output()
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def run_load_ump():
global abu_result_tuple_test_ump
abu_result_tuple_test_ump = abu.load_abu_result_tuple(n_folds=1, store_type=EStoreAbu.E_STORE_CUSTOM_NAME,
custom_name='live')
AbuMetricsBase.show_general(*abu_result_tuple_test_ump).plot_max_draw_down() #, returns_cmp=True
def select_ump(select):
if select == 'run loop back ump':
run_loop_back_ump()
else:
run_load_ump()
# _ = ipywidgets.interact_manual(select_ump, select=['load test ump data','run loop back ump'])
# print("随便输出点什么表示已经执行过： ")
run_loop_back_ump()
In [26]:
display(HTML('<h3 id="latest-actions">上周一至今日的交易信号 </h3>'))
action_pd_local = abu_result_tuple_test_ump.action_pd
date_offset = moment.date("monday").format('YYYYMMDD')
display(action_pd_local[action_pd_local.Date >= int(date_offset)])
display(HTML('<b>您的系统的今天日期是{}，请检查系统时间/日期设置是否正确，日期不正确会造成信号失误。</b>'
.format( moment.now().format('YYYY-MM-DD') )))
display(HTML('<p>推荐买入deal为true的股票，deal为false的交易意思是符合买入条件，但资金不够了系统未买入。</p>'))
orders_pd = abu_result_tuple_test_ump.orders_pd
display(HTML('<h3 id="latest-orders">最近交易单 </h3>')) # 'buy_factor', 'sell_type_extra',
cols = ['symbol','buy_date','buy_price','buy_cnt', 'sell_date', 'sell_price',
'sell_type', 'keep_days', 'ml_features', 'profit_cg','result' ]
display(orders_pd.tail(26)[cols])
display(HTML('<h3 id="keep">今日持仓 </h3>')) # 'buy_factor',
cols_keep =['symbol','buy_date','buy_price','buy_cnt','buy_pos', 'buy_type_str',
'ml_features', 'keep_days'] # 'key',
order_keep_pd = orders_pd[orders_pd.sell_type=='keep']
display(order_keep_pd[cols_keep])
print('当前一共持有{}个单，共{}只股'.format(len(order_keep_pd), len(set(order_keep_pd.symbol))) )
# abu_result_tuple_test_ump
In [ ]: