Untitled
unknown
plain_text
a year ago
4.9 kB
6
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