# -*- coding: utf-8 -*-
"""strat_sz100iw.py
一种趋势策略的尝试, 针对深证100指数周线数据
"""
import itertools
import os, pathlib, datetime; type(os); type(pathlib);type(datetime)
import pandas as pd; type(pd)
import numpy as np; type(np)
import matplotlib.pyplot as plt
from copy import deepcopy;type(deepcopy)
from collections import OrderedDict ; type(OrderedDict)
import toolkit.myDataIO as mdio
import util.ttr as ttr
import strat_sz100iw_report as wr; type(wr)
import imp; type(imp)
#imp.reload(mdio)
#imp.reload(ttr)
global g
g=ttr.GlobalEnv()
g.params=OrderedDict(
atr_n=14, delta=0.5, sig_filter_n=3,
start_year=None,
code='399330',
maxloss_pct=0.12,
with_plot=False,
)
#%%
#def simulation_LineMode(atr_n, delta, sig_filter_n, start_year=None, code='399330'):
def simulation_LineMode(**kwargs):
'''
Python里的可变参数(*和**字头的参数):
在函数应用中(被定义或者被调用)，
有时候我们不确定调用的时候会传递多少个参数(不传参也可以)。
此时，可用包裹(packing)位置参数/包裹型位置参数，或者包裹型关键字参数，
来进行参数传递， 会显得非常方便灵活。
关键字参数(双星号字头的参数)允许你传入0个或任意个含参数名的参数,
这些关键字参数在函数内部会自动组装为一个dict.
>>> df_line=simulation_LineMode(**g.params)
atr14_pct=ttr.atr(df_line,14)/df_line.close *100
atr14_pct.quantile() # 4.85%
atr14_pct.describe()
count 716.000000
mean 5.476807
std 2.481463
min 2.134225
25% 3.923625
50% 4.851406
75% 6.285523
max 15.688897
所以: stoploss的选择: 应该 > mean (or 1/2 quantile)
'''
global g
# 解包裹, 获取本函数所需的每个参数
atr_n =kwargs['atr_n']
delta =kwargs['delta']
sig_filter_n=kwargs['sig_filter_n']
start_year =kwargs['start_year']
code =kwargs['code']
with_plot =kwargs['with_plot']
fname = 'd:/new_haitong/T0002/export/{}.txt'.format(code)
with open(fname, encoding='gbk') as f:
first_line_data=f.readline()
asset_name=first_line_data.strip()
title_='标的资产的代码和名称是: {}'.format(asset_name)
print(title_)
print('+'*(len(title_) + ttr.cchar(title_)))
g.title_info=OrderedDict(
fname=fname,
asset_name=asset_name,
)
ohlc_ = mdio.read_tdxExport_txtFile(fname)
ohlc=ohlc_[0]
ohlc=ohlc.iloc[:, 0:5]
if start_year:
ohlc = ohlc[start_year:]
out, perf_dict = ttr.strat_sz100iw_lm(ohlc, atr_n, delta, sig_filter_n, start_year)
if with_plot:
fig, ax=plt.subplots(1,1)
ohlc.close.plot(ax=ax)
out.bprice.plot(ax=ax)
out.sprice.plot(ax=ax)
out.equity.plot()
print('
End equity : {:12.2f}
'.format(out.equity[-1] ,))
[print('{0:{2:}s} : {1:}'.format(k, v, (18-ttr.cchar(k)))) for k, v in perf_dict.items()]
print()
return out
#%%
def simu_with_stoploss(atr_n=14, delta=0.5, sig_filter_n=3,
start_year=None,
code='399330',
signal_type=1, # exRemoved, 0=original
maxloss_pct=0.11,
stoploss_mode=1, # (0, 1)分别表示当日尾盘止损, 次日尾盘止损
trx_timing='closing',
with_plot=False,
):
'''
>>> reset -f
>>> pos,cagr,mdd,sharpe = simu_with_stoploss(maxloss_pct=0.13, signal_type=1, stoploss_mode=1, with_plot=False, trx_timing='closing') #### 不错的止损值
>>> pos,cagr,mdd,sharpe = simu_with_stoploss(maxloss_pct=0.13, signal_type=0, stoploss_mode=1, with_plot=False, trx_timing='closing') #### 不错的止损值
>>> pos,cagr,mdd,sharpe = simu_with_stoploss(maxloss_pct=0.13, stoploss_mode=1, with_plot=False, trx_timing='opening') #
>>> pos,cagr,mdd,sharpe = simu_with_stoploss(maxloss_pct=0.13, stoploss_mode=0, with_plot=False, trx_timing='closing') #
>>> pos,cagr,mdd,sharpe = simu_with_stoploss(maxloss_pct=0.13, stoploss_mode=0, with_plot=False, trx_timing='opening') #
>>> pos,cagr,mdd,sharpe = simu_with_stoploss(maxloss_pct=0.11, stoploss_mode=0, with_plot=True) #### 不错的止损值
>>> pos,cagr,mdd,sharpe=simu_with_stoploss(maxloss_pct=0.12, trx_timing='opening') #### 不错的止损值
参数: OrderedDict(
[('atr_n', 14), ('delta', 0.5), ('sig_filter_n', 3), ('start_year', None), ('code', '399330'),
('maxloss_pct', 0.12), ('trx_timing', 'closing'), ('with_plot', False)])
下单指令集合: {'满仓寻出', '止损', '开仓', '空仓寻进', '清仓'}
dk 的次数: 49
dp 的次数: 45
gr 的次数: 4
hold_equity 的次数: 264
hold_cash 的次数: 326
end_equity : 11541.354 if trx_timing=='opening'
end_nav : 12.003 7.996
cagr : 0.19 0.156
mdd : 0.235 0.274
sharpe : 1.151
annu_volatility : 0.165
净值的标准差 : 0.023
'''
global g
#打包所需的关键字参数为: g.params, 方便调用函数时使用.
g.params=OrderedDict(
atr_n=atr_n, delta=delta, sig_filter_n=sig_filter_n,
start_year=start_year,
code=code,
signal_type=signal_type,
trx_timing=trx_timing,
maxloss_pct=maxloss_pct,
stoploss_mode=stoploss_mode,
with_plot=with_plot,
)
df = simulation_LineMode(**g.params)
# print(df.columns)
# Index(['open', 'high', 'low', 'close', 'trend',
# 'b_sig', 's_sig', 'bprice', 'sprice', 'buybars', 'sellbars',
# 'pos', 'equity', 'b22_sig', 's22_sig'],
pos = ttr.simulation_BarMode(df, **g.params)
pos = pos.set_index('dt', drop=False)
print('
参数: {}'.format(g.params))
ttr.print_orders_count(pos)
pos_=pos.reindex(df.index)
pos_.position.fillna(0.0, inplace=True)
#print(set(pos_.position))
roc1=df.close.pct_change().fillna(0)
#交割时机的把握逻辑:
#收盘时刻观察信号, 按收盘价成交:
if trx_timing=='closing':
eq = (1+roc1*pos_.position).cumprod() * df.close[0]
elif trx_timing=='opening':
# 校正持仓第一日的roc1: 应该为: 该日收盘价 to 该日开盘价
#roc1_B = ttr.IF((df.b_sig.shift(1)>0, (df.open/df.close.shift(1)-1), roc1))
roc1_B = ttr.IF((df.b_sig.shift(1)>0, (df.close/df.open-1), roc1))
# 校正持仓最后日的roc1: 应该为: 次日开盘价 to 昨日收盘价
# 下述方法仅保证了正常卖出时的收益率的正确性, 止损时的情形需要完善.
#roc1_C = ttr.IF((df.s_sig>0, (df.open.shift(-1)/df.close.shift(1)-1), roc1_B))
exit_case = ttr.downCross(pos_.position, 0.5).shift(-1)
roc1_C = ttr.IF( (exit_case>0,
(df.open.shift(-1)/df.close.shift(1)-1),
roc1_B))
eq = (1+roc1_C * pos_.position).cumprod() * df.close[0]
pos_['open']=df.open
pos_['close']=df.close
pos_['equity']=eq
perf_dict=ttr.perf(pos_)
print()
#[print('{:15s} : {}'.format(k, v)) for k, v in perf_dict.items()]
for k, v in perf_dict.items():
pwidth = 18 - ttr.cchar(k)
print('{1:{0:}s} : {2:}'.format(pwidth, k, v))
print()
if with_plot:
fig, ax=plt.subplots(1,1)
pos_.close.plot(ax=ax)
#out.bprice.plot(ax=ax)
#out.sprice.plot(ax=ax)
pos_.equity.plot()
return pos_, perf_dict['cagr'], perf_dict['mdd'], perf_dict['sharpe']
#%%
def opt(trx_timing='closing', stoploss_mode=1, signal_type=1,):
'''参数寻优
最佳参数大概为:
14 atr_n,
13% maxloss_pct,
'closing' trx_timing
1 stoploss_mode (次日止损)
绩效指标:
CAGR MDD SHARPE
19.5% 23.5% 1.180
>>> opt_df=opt(trx_timing='closing', stoploss_mode=1, )
>>> opt_df=opt(trx_timing='closing', stoploss_mode=1, signal_type=0) #用了原始信号后, 结果较差
>>> opt_df=opt(trx_timing='opening', stoploss_mode=1)
>>> opt_df=opt(trx_timing='closing', stoploss_mode=0, )
>>> opt_df=opt(trx_timing='opening', stoploss_mode=0, signal_type=0) #更差
stoploss_mode trx_timing atr_n maxloss_pct cagr mdd sharpe
0 1 closing 13 0.05 0.173 0.359 1.107
1 1 closing 13 0.08 0.171 0.395 1.043
2 1 closing 13 0.11 0.181 0.395 1.098
3 1 closing 13 0.14 0.187 0.395 1.093
4 1 closing 14 0.05 0.160 0.248 1.058
5 1 closing 14 0.08 0.167 0.318 1.047
6 1 closing 14 0.11 0.184 0.235 1.142
7 1 closing 14 0.14 0.189 0.235 1.135
8 1 closing 15 0.05 0.152 0.391 0.987
9 1 closing 15 0.08 0.142 0.426 0.886
10 1 closing 15 0.11 0.150 0.445 0.923
11 1 closing 15 0.14 0.161 0.407 0.950
if 次日开盘成交: 那么结果如下:
成交时机选择当日尾盘略微好一点点(cagr能有3%的提升)
但是没有明显的优势
stoploss_mode trx_timing atr_n maxloss_pct cagr mdd sharpe
0 1 opening 13 0.05 0.161 0.376 1.037
1 1 opening 13 0.08 0.165 0.376 1.010
2 1 opening 13 0.11 0.185 0.376 1.114
3 1 opening 13 0.14 0.176 0.392 1.022
4 1 opening 14 0.05 0.153 0.245 1.019
5 1 opening 14 0.08 0.160 0.310 1.013
6 1 opening 14 0.11 0.187 0.222 1.159
7 1 opening 14 0.14 0.180 0.222 1.076
8 1 opening 15 0.05 0.139 0.378 0.908
9 1 opening 15 0.08 0.134 0.378 0.842
10 1 opening 15 0.11 0.154 0.378 0.942
11 1 opening 15 0.14 0.151 0.383 0.879
'''
global g
#把参数字典, 通过笛卡尔乘积, 转换为参数数据框
par_dict = OrderedDict(
atr_n=range(13,16),
#maxloss_pct=np.arange(5,17,1)/100,
maxloss_pct=np.arange(5,17,1)/100,
)
prod = itertools.product(*tuple(par_dict.values())) # 传递包裹型位置参数
par_df = pd.DataFrame(list(prod), columns=par_dict.keys())
out=pd.DataFrame()
opt_summary=pd.DataFrame()
for (i, atr_n, maxloss_pct ) in par_df.itertuples():
pos, cagr, mdd, sharpe = simu_with_stoploss(
atr_n=atr_n,
maxloss_pct=maxloss_pct,
signal_type=signal_type,
trx_timing=trx_timing,
stoploss_mode=stoploss_mode)
out['equity{}'.format(i)]=pos.equity
opt_summary = opt_summary.append(
pd.DataFrame(
OrderedDict(
atr_n=atr_n,
maxloss_pct=maxloss_pct,
stoploss_mode=g.params['stoploss_mode'],
trx_timing=g.params['trx_timing'],
cagr=cagr, mdd=mdd, sharpe=sharpe
),
index=[1],
),
ignore_index=True,
)
print(opt_summary)
out['close']=pos.close
# out.plot()
return out
#%%
#wr.write_report_md()
#%%
if __name__=="__main__":
#atr_n=int(input('Please enter parameter: atr_n: '))
#for atr_n in range(12, 18):
pass