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) 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[i-1] curr_mt = meta_trend[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