def dual_thrust_sim( mdf, config):
close_daily = config['close_daily']
offset = config['offset']
k = config['param'][0]
win = config['param'][1]
proc_func = config['proc_func']
proc_args = config['proc_args']
chan_func = config['chan_func']
chan_high = eval(chan_func['high']['func'])
chan_low = eval(chan_func['low']['func'])
tcost = config['trans_cost']
min_rng = config['min_range']
chan = config['chan']
xdf = proc_func(mdf, **proc_args)
tr= pd.concat([pd.rolling_max(xdf.high, win) - pd.rolling_min(xdf.close, win),
pd.rolling_max(xdf.close, win) - pd.rolling_min(xdf.low, win)],
join='outer', axis=1).max(axis=1)
xdf['tr'] = tr
xdf['chan_h'] = chan_high(xdf, chan, **chan_func['high']['args'])
xdf['chan_l'] = chan_low(xdf, chan, **chan_func['low']['args'])
#sar_param = config['sar_param']
#sar = dh.SAR(xdf, **sar_param)
#sar_signal = pd.Series(0, index = sar.index)
#sar_signal[(sar < xdf['close'])] = 1
#sar_signal[(sar > xdf['close'])] = -1
#xdf['atr'] = dh.ATR(xdf, chan)
xdata = pd.concat([xdf['tr'].shift(1), xdf['chan_h'].shift(1), xdf['chan_l'].shift(1),
xdf['open']], axis=1, keys=['tr','chan_h', 'chan_l', 'xopen'])
mdf = mdf.join(xdata, how = 'left').fillna(method='ffill')
rng = pd.DataFrame([min_rng * mdf['xopen'], k * mdf['tr']]).max()
long_signal = pd.Series(np.nan, index = mdf.index)
long_signal[(mdf['high'] >= mdf['xopen'] + rng) & (mdf['high'] >= mdf['chan_h']) ] = 1
long_signal[(mdf['low'] <= mdf['xopen'] - rng) | (mdf['low'] <= mdf['chan_l'])] = 0
if close_daily:
long_signal[(mdf['min_id']>=config['exit_min'])] = 0
long_signal = long_signal.shift(1)
long_signal = long_signal.fillna(method='ffill').fillna(0)
short_signal = pd.Series(np.nan, index = mdf.index)
short_signal[(mdf['low'] <= mdf['xopen'] - rng) & (mdf['low'] <= mdf['chan_l']) ] = -1
short_signal[(mdf['high'] >= mdf['xopen'] + rng) | (mdf['high'] >= mdf['chan_h'])] = 0
if close_daily:
short_signal[(mdf['min_id']>=config['exit_min'])] = 0
short_signal = short_signal.shift(1)
short_signal = short_signal.fillna(method='ffill').fillna(0)
if len(mdf[(long_signal>0)&(short_signal<0)]) > 0:
print "Warning: long and short signal happen at the same time"
mdf['pos'] = long_signal + short_signal
mdf.ix[-3:, 'pos'] = 0
mdf['cost'] = abs(mdf['pos'] - mdf['pos'].shift(1)) * (offset + mdf['open'] * tcost)
mdf['cost'] = mdf['cost'].fillna(0.0)
mdf['traded_price'] = mdf.open + (mdf['pos'] - mdf['pos'].shift(1)) * offset
closed_trades = backtest.simdf_to_trades1(mdf, slippage = offset )
return (mdf, closed_trades)
评论列表
文章目录
