219 lines
8.9 KiB
Python
219 lines
8.9 KiB
Python
import pandas as pd
|
|
import numpy as np
|
|
|
|
from cycles.supertrend import Supertrends
|
|
from cycles.market_fees import MarketFees
|
|
|
|
class Backtest:
|
|
@staticmethod
|
|
def run(min1_df, df, initial_usd, stop_loss_pct, debug=False):
|
|
"""
|
|
Backtest a simple strategy using the meta supertrend (all three supertrends agree).
|
|
Buys when meta supertrend is positive, sells when negative, applies a percentage stop loss.
|
|
|
|
Parameters:
|
|
- min1_df: pandas DataFrame, 1-minute timeframe data for more accurate stop loss checking (optional)
|
|
- initial_usd: float, starting USD amount
|
|
- stop_loss_pct: float, stop loss as a fraction (e.g. 0.05 for 5%)
|
|
- debug: bool, whether to print debug info
|
|
"""
|
|
_df = df.copy().reset_index(drop=True)
|
|
_df['timestamp'] = pd.to_datetime(_df['timestamp'])
|
|
|
|
supertrends = Supertrends(_df, verbose=False)
|
|
|
|
supertrend_results_list = supertrends.calculate_supertrend_indicators()
|
|
trends = [st['results']['trend'] for st in supertrend_results_list]
|
|
trends_arr = np.stack(trends, axis=1)
|
|
meta_trend = np.where((trends_arr[:,0] == trends_arr[:,1]) & (trends_arr[:,1] == trends_arr[:,2]),
|
|
trends_arr[:,0], 0)
|
|
# Shift meta_trend by one to avoid lookahead bias
|
|
meta_trend_signal = np.roll(meta_trend, 1)
|
|
meta_trend_signal[0] = 0 # or np.nan, but 0 means 'no signal' for first bar
|
|
|
|
position = 0 # 0 = no position, 1 = long
|
|
entry_price = 0
|
|
usd = initial_usd
|
|
coin = 0
|
|
trade_log = []
|
|
max_balance = initial_usd
|
|
drawdowns = []
|
|
trades = []
|
|
entry_time = None
|
|
current_trade_min1_start_idx = None
|
|
|
|
min1_df['timestamp'] = pd.to_datetime(min1_df.index)
|
|
|
|
for i in range(1, len(_df)):
|
|
price_open = _df['open'].iloc[i]
|
|
price_close = _df['close'].iloc[i]
|
|
date = _df['timestamp'].iloc[i]
|
|
prev_mt = meta_trend_signal[i-1]
|
|
curr_mt = meta_trend_signal[i]
|
|
|
|
# Check stop loss if in position
|
|
if position == 1:
|
|
stop_loss_result = Backtest.check_stop_loss(
|
|
min1_df,
|
|
entry_time,
|
|
date,
|
|
entry_price,
|
|
stop_loss_pct,
|
|
coin,
|
|
usd,
|
|
debug,
|
|
current_trade_min1_start_idx
|
|
)
|
|
if stop_loss_result is not None:
|
|
trade_log_entry, current_trade_min1_start_idx, position, coin, entry_price = stop_loss_result
|
|
trade_log.append(trade_log_entry)
|
|
continue
|
|
# Update the start index for next check
|
|
current_trade_min1_start_idx = min1_df.index[min1_df.index <= date][-1]
|
|
|
|
# Entry: only if not in position and signal changes to 1
|
|
if position == 0 and prev_mt != 1 and curr_mt == 1:
|
|
entry_result = Backtest.handle_entry(usd, price_open, date)
|
|
coin, entry_price, entry_time, usd, position, trade_log_entry = entry_result
|
|
trade_log.append(trade_log_entry)
|
|
|
|
# Exit: only if in position and signal changes from 1 to -1
|
|
elif position == 1 and prev_mt == 1 and curr_mt == -1:
|
|
exit_result = Backtest.handle_exit(coin, price_open, entry_price, entry_time, date)
|
|
usd, coin, position, entry_price, trade_log_entry = exit_result
|
|
trade_log.append(trade_log_entry)
|
|
|
|
# Track drawdown
|
|
balance = usd if position == 0 else coin * price_close
|
|
if balance > max_balance:
|
|
max_balance = balance
|
|
drawdown = (max_balance - balance) / max_balance
|
|
drawdowns.append(drawdown)
|
|
|
|
# If still in position at end, sell at last close
|
|
if position == 1:
|
|
exit_result = Backtest.handle_exit(coin, _df['close'].iloc[-1], entry_price, entry_time, _df['timestamp'].iloc[-1])
|
|
usd, coin, position, entry_price, trade_log_entry = exit_result
|
|
trade_log.append(trade_log_entry)
|
|
|
|
# Calculate statistics
|
|
final_balance = usd
|
|
n_trades = len(trade_log)
|
|
wins = [1 for t in trade_log if t['exit'] is not None and t['exit'] > t['entry']]
|
|
win_rate = len(wins) / n_trades if n_trades > 0 else 0
|
|
max_drawdown = max(drawdowns) if drawdowns else 0
|
|
avg_trade = np.mean([t['exit']/t['entry']-1 for t in trade_log if t['exit'] is not None]) if trade_log else 0
|
|
|
|
trades = []
|
|
total_fees_usd = 0.0
|
|
for trade in trade_log:
|
|
if trade['exit'] is not None:
|
|
profit_pct = (trade['exit'] - trade['entry']) / trade['entry']
|
|
else:
|
|
profit_pct = 0.0
|
|
trades.append({
|
|
'entry_time': trade['entry_time'],
|
|
'exit_time': trade['exit_time'],
|
|
'entry': trade['entry'],
|
|
'exit': trade['exit'],
|
|
'profit_pct': profit_pct,
|
|
'type': trade.get('type', 'SELL'),
|
|
'fee_usd': trade.get('fee_usd')
|
|
})
|
|
fee_usd = trade.get('fee_usd')
|
|
total_fees_usd += fee_usd
|
|
|
|
results = {
|
|
"initial_usd": initial_usd,
|
|
"final_usd": final_balance,
|
|
"n_trades": n_trades,
|
|
"win_rate": win_rate,
|
|
"max_drawdown": max_drawdown,
|
|
"avg_trade": avg_trade,
|
|
"trade_log": trade_log,
|
|
"trades": trades,
|
|
"total_fees_usd": total_fees_usd,
|
|
}
|
|
if n_trades > 0:
|
|
results["first_trade"] = {
|
|
"entry_time": trade_log[0]['entry_time'],
|
|
"entry": trade_log[0]['entry']
|
|
}
|
|
results["last_trade"] = {
|
|
"exit_time": trade_log[-1]['exit_time'],
|
|
"exit": trade_log[-1]['exit']
|
|
}
|
|
return results
|
|
|
|
@staticmethod
|
|
def check_stop_loss(min1_df, entry_time, date, entry_price, stop_loss_pct, coin, usd, debug, current_trade_min1_start_idx):
|
|
stop_price = entry_price * (1 - stop_loss_pct)
|
|
|
|
if current_trade_min1_start_idx is None:
|
|
current_trade_min1_start_idx = min1_df.index[min1_df.index >= entry_time][0]
|
|
current_min1_end_idx = min1_df.index[min1_df.index <= date][-1]
|
|
|
|
# Check all 1-minute candles in between for stop loss
|
|
min1_slice = min1_df.loc[current_trade_min1_start_idx:current_min1_end_idx]
|
|
if (min1_slice['low'] <= stop_price).any():
|
|
# Stop loss triggered, find the exact candle
|
|
stop_candle = min1_slice[min1_slice['low'] <= stop_price].iloc[0]
|
|
# More realistic fill: if open < stop, fill at open, else at stop
|
|
if stop_candle['open'] < stop_price:
|
|
sell_price = stop_candle['open']
|
|
else:
|
|
sell_price = stop_price
|
|
if debug:
|
|
print(f"STOP LOSS triggered: entry={entry_price}, stop={stop_price}, sell_price={sell_price}, entry_time={entry_time}, stop_time={stop_candle.name}")
|
|
btc_to_sell = coin
|
|
usd_gross = btc_to_sell * sell_price
|
|
exit_fee = MarketFees.calculate_okx_taker_maker_fee(usd_gross, is_maker=False)
|
|
trade_log_entry = {
|
|
'type': 'STOP',
|
|
'entry': entry_price,
|
|
'exit': sell_price,
|
|
'entry_time': entry_time,
|
|
'exit_time': stop_candle.name,
|
|
'fee_usd': exit_fee
|
|
}
|
|
# After stop loss, reset position and entry
|
|
return trade_log_entry, None, 0, 0, 0
|
|
return None
|
|
|
|
@staticmethod
|
|
def handle_entry(usd, price_open, date):
|
|
entry_fee = MarketFees.calculate_okx_taker_maker_fee(usd, is_maker=False)
|
|
usd_after_fee = usd - entry_fee
|
|
coin = usd_after_fee / price_open
|
|
entry_price = price_open
|
|
entry_time = date
|
|
usd = 0
|
|
position = 1
|
|
trade_log_entry = {
|
|
'type': 'BUY',
|
|
'entry': entry_price,
|
|
'exit': None,
|
|
'entry_time': entry_time,
|
|
'exit_time': None,
|
|
'fee_usd': entry_fee
|
|
}
|
|
return coin, entry_price, entry_time, usd, position, trade_log_entry
|
|
|
|
@staticmethod
|
|
def handle_exit(coin, price_open, entry_price, entry_time, date):
|
|
btc_to_sell = coin
|
|
usd_gross = btc_to_sell * price_open
|
|
exit_fee = MarketFees.calculate_okx_taker_maker_fee(usd_gross, is_maker=False)
|
|
usd = usd_gross - exit_fee
|
|
trade_log_entry = {
|
|
'type': 'SELL',
|
|
'entry': entry_price,
|
|
'exit': price_open,
|
|
'entry_time': entry_time,
|
|
'exit_time': date,
|
|
'fee_usd': exit_fee
|
|
}
|
|
coin = 0
|
|
position = 0
|
|
entry_price = 0
|
|
return usd, coin, position, entry_price, trade_log_entry |