Untitled
unknown
plain_text
a year ago
4.9 kB
9
Indexable
def run_single_exp_simplified_new(df, asset, resample_time, roll_d, exqu_d, roll, exqu, tl, rk, rw, weight1):
df = df.rename(columns={asset+'close':'close', asset+'open':'open', asset+'high':'high', asset+'low':'low'})
df = df.dropna()
hourly_data = df.resample(str(resample_time)+'T', closed='right', label='right').agg({'close': ['first', 'last', 'max', 'min']})
# Rename the columns to match your desired output
hourly_data.columns = ['open', 'close', 'high', 'low']
hourly_data = hourly_data.loc[~hourly_data.index.duplicated(), :]
hourly_data = wick(hourly_data)
hourly_data['wick_abs'] = swingfilter_d(hourly_data, resample_time, roll_d, exqu_d)
hourly_data = swing(hourly_data)
hourly_data = fvg_filter(hourly_data, resample_time, roll, exqu)
hourly_data = fvg(hourly_data, 100)
hourly_data = hourly_data[1:-1]
#ffill must be done after calculation of entry fvg.
hourly_data[['fair_value_gap', 'entry_1','entry_2','entry_3', 'SL']] = hourly_data[['fair_value_gap','entry_1','entry_2','entry_3', 'SL']].shift(2)
hourly_data['fair_value_gap'] = hourly_data['fair_value_gap'].replace(0, method='ffill')
hourly_data['SL'] = hourly_data['SL'].replace(0, method='ffill')
#hourly_data['SL'] = hourly_data['SL'].ffill()
df = df.merge(hourly_data[['fair_value_gap', 'entry_1', 'entry_2', 'swing_signal', 'SL']], left_index = True, right_index=True, how = 'left')
df[['entry_1','entry_2']] = df[['entry_1','entry_2']].replace(0, method='ffill')
df[['entry_1','entry_2']] = df[['entry_1','entry_2']].ffill()
df['fair_value_gap'] = df['fair_value_gap'].replace(0, method='ffill')
df['fair_value_gap'] = df['fair_value_gap'].ffill()
df['SL'] = df['SL'].replace(0, method ='ffill')
df['SL'] = df['SL'].ffill()
confidence_interval = 0.0
conditions1 = [
(
(df['close'] == df['entry_1']) |
(
(df['high'] >= df['entry_1'] + confidence_interval) &
(df['low'] <= df['entry_1'] - confidence_interval)
)
)
]
conditions2 = [
(
(df['close'] == df['entry_2']) |
(
(df['high'] >= df['entry_2'] + confidence_interval) &
(df['low'] <= df['entry_2'] - confidence_interval)
)
)
]
# Define the corresponding values to assign for each condition
values = [np.where(df['fair_value_gap'] > 0, 1, -1)]
# Use np.select to apply the conditions and assign values accordingly
df['LevelEntry1'] = np.select(conditions1, values, default=0)
df['LevelEntry2'] = np.select(conditions2, values, default=0)
#---------------------------------------------------------------------------------------------
initial_nav = 1e6
df['signalClose'] = df['close'].shift(1)
df = df.reset_index()
df = df.set_index('Timestamp')
df['new_fvg'] = (df['fair_value_gap'] != df['fair_value_gap'].shift(1))*1
minutely_df = df.copy()
minutely_df = minutely_df.rename(columns = {'open': asset+'minutely_open', 'high': asset+'minutely_high', 'close':asset+'minutely_close', 'low':asset+'minutely_low'})
minutely_df = minutely_df[[asset+'minutely_close', 'signalClose', asset+'minutely_high', asset+'minutely_low', 'fair_value_gap', 'entry_1', 'entry_2', 'swing_signal', 'LevelEntry1','LevelEntry2', 'SL', 'new_fvg']]
temp1 = minutely_df.copy()
temp1 = temp1.dropna(subset=['fair_value_gap', 'entry_1', 'entry_2', 'SL'])
minutely_df = minutely_df.dropna(subset=['fair_value_gap', 'entry_1', 'entry_2', 'SL'])
equitylist = ['ES', 'NQ', 'RTY', 'YM']
lotsize=definelotsize(asset)
#swing, LevelEntry1, entry_1, entry_2, c, SL, newfvg, lotsize, timelimit, \nrisk, reward, weight1, symbol, equitylist, counter, initial_nav
res = backtest_goldenbullet_simplified(*temp1[['swing_signal', 'LevelEntry1', 'entry_1', 'entry_2', asset+'minutely_close', 'SL', 'new_fvg']].values.T, lotsize, tl, rk, rw, weight1, asset, np.array(equitylist), resample_time, initial_nav)
res = pd.DataFrame(res, index=minutely_df.index)
backtest_res = pd.concat([minutely_df, res], axis=1)
df2 = backtest_res.copy()
df2['return'] = (df2['total_nav'] - df2['total_nav'].shift(1))/(df2['total_nav'].shift(1))
if df2['total_nav'].min() < 0:
df2.loc[df2.index > df2['total_nav'][df2['total_nav']<0].index[0], 'return'] = 0
temp = df2[['return']].copy()
temp['return'] = np.log(temp['return'] + 1)
temp = temp.resample('D').sum()
temp['return'] = np.exp(temp['return']) - 1
temp4 = temp.copy()
temp4['final_return'] = temp['return']
return temp4['final_return']Editor is loading...
Leave a Comment