From 0731178761ef79cb984e816653991dcae1bdcfc5 Mon Sep 17 00:00:00 2001 From: BaptisteZloch Date: Sat, 25 Nov 2023 16:17:16 +0100 Subject: [PATCH] added price backtester --- quant_invest_lab/backtest.py | 472 ++++++++++++++++++++++++++++++++++- quant_invest_lab/reports.py | 42 +++- 2 files changed, 501 insertions(+), 13 deletions(-) diff --git a/quant_invest_lab/backtest.py b/quant_invest_lab/backtest.py index 6b0f73b..ca1fe82 100644 --- a/quant_invest_lab/backtest.py +++ b/quant_invest_lab/backtest.py @@ -10,7 +10,7 @@ ) from quant_invest_lab.constants import TIMEFRAMES from quant_invest_lab.types import Timeframe -from quant_invest_lab.reports import print_ohlc_backtest_report, plot_from_trade_df +from quant_invest_lab.reports import print_backtest_report, plot_from_trade_df def __ohlc_backtest_one_position_type( @@ -69,15 +69,15 @@ def __ohlc_backtest_one_position_type( ) ohlcv_df["Strategy_returns"] = 0 - for row in tqdm( - ohlcv_df[1:-1].itertuples(index=True), + for index, row in tqdm( + ohlcv_df.iloc[1:-1].iterrows(), desc="Backtesting...", total=ohlcv_df.shape[0] - 1, leave=False, ): if position_opened is False and entry_function(row, previous_row) is True: position_opened = True - current_trade["entry_date"] = row.Index + current_trade["entry_date"] = index current_trade["entry_price"] = row.Close current_trade["entry_reason"] = "Entry position triggered" timeframe_count = 1 @@ -92,7 +92,7 @@ def __ohlc_backtest_one_position_type( and current_trade["entry_price"] * (1 + stop_loss) <= row.High ): position_opened = False - current_trade["exit_date"] = row.Index + current_trade["exit_date"] = index current_trade["exit_price"] = ( current_trade["entry_price"] * (1 + take_profit) @@ -135,7 +135,7 @@ def __ohlc_backtest_one_position_type( and current_trade["entry_price"] * (1 - take_profit) >= row.Low ): position_opened = False - current_trade["exit_date"] = row.Index + current_trade["exit_date"] = index current_trade["exit_price"] = ( current_trade["entry_price"] * (1 - stop_loss) @@ -169,9 +169,7 @@ def __ohlc_backtest_one_position_type( position_opened is True and exit_function(row, previous_row, timeframe_count) is True ): - next_row_index = ( - ohlcv_df.index.get_loc(row.Index) + 1 - ) # To close on next open. + next_row_index = ohlcv_df.index.get_loc(index) + 1 # To close on next open. position_opened = False current_trade["exit_date"] = ohlcv_df.iloc[next_row_index].name @@ -330,7 +328,7 @@ def ohlc_long_only_backtester( print(f"Stop loss is set to {stop_loss*100:.2f} % of the buy price") print("Long only position") - print_ohlc_backtest_report( + print_backtest_report( trades_df=trades_df, ohlcv_df=ohlcv_df, timeframe=timeframe, @@ -444,7 +442,7 @@ def ohlc_short_only_backtester( if stop_loss != np.inf: print(f"Stop loss is set to {stop_loss*100:.2f} % of the buy price") print("Short only positions") - print_ohlc_backtest_report( + print_backtest_report( trades_df=trades_df, ohlcv_df=ohlcv_df, timeframe=timeframe, @@ -458,3 +456,455 @@ def ohlc_short_only_backtester( return trades_df if get_returns_df: return ohlcv_df + + +def __price_backtest_one_position_type( + price_df: pd.DataFrame, + entry_function: Callable[[pd.Series, pd.Series], bool], + exit_function: Callable[[pd.Series, pd.Series, int], bool], + position_type: Literal["long", "short"], + take_profit: float = np.inf, + stop_loss: float = np.inf, + maker_fees: Optional[float] = 0.001, + taker_fees: Optional[float] = 0.001, +) -> tuple[pd.DataFrame, pd.DataFrame]: + """Core backtesting function for OHLCV data. It iterate over the OHLCV records and run the entry and exit function when the conditions are satisfied. + WARNING this function shouldn't be used alone, use the long/short backtest function instead. + + Args: + ---- + price_df (pd.DataFrame): The dataframe containing the price data. + + entry_function (Callable[[pd.Series, pd.Series], bool]): The trade entry function, it should take 2 arguments, the current row and the previous row and return True or False depending on your strategy. + + exit_function (Callable[[pd.Series, pd.Series, int], bool]): The long entry function, it should take 2 arguments, the current row and the previous row and return True or False depending on your strategy. + + position_type (Literal["long", "short"]): The position type, long or short. + + take_profit (float, optional): The percent of the buy price to add to create a stop order and take the profit associated. Defaults to np.inf. + + stop_loss (float, optional): The percent of the buy price to cut to create a stop order and stop the loss associated. Defaults to np.inf. + + maker_fees (Optional[float], optional): The fees applied, here maker for limit orders (not yet implemented). Defaults to 0.001. + + taker_fees (Optional[float], optional): The fees applied, here taker for spot orders. Defaults to 0.001. + + Returns: + ----- + tuple[pd.DataFrame, pd.DataFrame]: It returns 2 dataframes, the first is the trades_df : trade summary dataframe and the second is the OHMCV dataframe that contains a new column Strategy returns. + """ + assert position_type in ["long", "short"], "position_type must be long or short" + + RETURNS_SIGNS = {"long": 1, "short": -1} + previous_row = price_df.iloc[0] + position_opened = False + timeframe_count = 0 + current_trade: dict[str, float | datetime | int | str | pd.Timestamp] = {} + + trades_df = pd.DataFrame( + columns=[ + "entry_date", + "entry_price", + "entry_reason", + "exit_date", + "exit_price", + "exit_reason", + "trade_return", + ] + ) + price_df["Strategy_returns"] = 0 + + for index, row in tqdm( + price_df[1:-1].iterrows(), + desc="Backtesting...", + total=price_df.shape[0] - 1, + leave=False, + ): + if position_opened is False and entry_function(row, previous_row) is True: + position_opened = True + current_trade["entry_date"] = index + current_trade["entry_price"] = row["Price"] + current_trade["entry_reason"] = "Entry position triggered" + timeframe_count = 1 + entry_price = row["Price"] + elif ( + position_type == "long" + and position_opened is True + and current_trade["entry_price"] * (1 + take_profit) <= row["Price"] + ) or ( + position_type == "short" + and position_opened is True + and current_trade["entry_price"] * (1 + stop_loss) <= row["Price"] + ): + position_opened = False + current_trade["exit_date"] = index + + current_trade["exit_price"] = ( + current_trade["entry_price"] * (1 + take_profit) + if position_type == "long" + else current_trade["entry_price"] * (1 + stop_loss) + ) + + current_trade["exit_reason"] = ( + "Long take profit triggered" + if position_type == "long" + else "Short stop loss triggered" + ) + + rets = ( + RETURNS_SIGNS[position_type] + * price_df.loc[ + current_trade["entry_date"] : current_trade["exit_date"] + ].Returns + ) + if isinstance(taker_fees, float) and taker_fees > 0: + rets.iloc[0] = rets.iloc[0] - taker_fees + rets.iloc[-1] = rets.iloc[-1] - taker_fees + + current_trade["trade_return"] = ((rets + 1).cumprod().iloc[-1]) - 1 # ret + + trades_df = pd.concat( + [trades_df, pd.DataFrame([current_trade])], ignore_index=True + ) + price_df.loc[rets.index[0] : rets.index[-1], "Strategy_returns"] = rets + timeframe_count = 0 + + current_trade = {} + elif ( + position_type == "long" + and position_opened is True + and current_trade["entry_price"] * (1 - stop_loss) >= row["Price"] + ) or ( + position_type == "short" + and position_opened is True + and current_trade["entry_price"] * (1 - take_profit) >= row["Price"] + ): + position_opened = False + current_trade["exit_date"] = index + + current_trade["exit_price"] = ( + current_trade["entry_price"] * (1 - stop_loss) + if position_type == "long" + else current_trade["entry_price"] * (1 - take_profit) + ) + current_trade["exit_reason"] = ( + "Long stop loss triggered" + if position_type == "long" + else "Short take profit triggered" + ) + + rets = ( + RETURNS_SIGNS[position_type] + * price_df.loc[ + current_trade["entry_date"] : current_trade["exit_date"] + ].Returns + ) + if isinstance(taker_fees, float) and taker_fees > 0: + rets.iloc[0] = rets.iloc[0] - taker_fees + rets.iloc[-1] = rets.iloc[-1] - taker_fees + current_trade["trade_return"] = ((rets + 1).cumprod().iloc[-1]) - 1 # ret + + trades_df = pd.concat( + [trades_df, pd.DataFrame([current_trade])], ignore_index=True + ) + price_df.loc[rets.index[0] : rets.index[-1], "Strategy_returns"] = rets + timeframe_count = 0 + current_trade = {} + elif ( + position_opened is True + and exit_function(row, previous_row, timeframe_count) is True + ): + next_row_index = price_df.index.get_loc(index) + 1 # To close on next open. + + position_opened = False + current_trade["exit_date"] = price_df.iloc[next_row_index].name + current_trade["exit_price"] = price_df.iloc[next_row_index]["Price"] + current_trade["exit_reason"] = "Exit position triggered" + + rets = ( + RETURNS_SIGNS[position_type] + * price_df.loc[ + current_trade["entry_date"] : current_trade["exit_date"] + ].Returns + ) + + if isinstance(taker_fees, float) and taker_fees > 0: + rets.iloc[0] = rets.iloc[0] - taker_fees + rets.iloc[-1] = rets.iloc[-1] - taker_fees + + current_trade["trade_return"] = ((rets + 1).cumprod().iloc[-1]) - 1 # ret + + trades_df = pd.concat( + [trades_df, pd.DataFrame([current_trade])], ignore_index=True + ) + price_df.loc[rets.index[0] : rets.index[-1], "Strategy_returns"] = rets + + timeframe_count = 0 + current_trade = {} + else: + timeframe_count += 1 + previous_row = row + # Close last trade + if position_opened is True: + # print("Closing trade at end of backtest") + position_opened = False + current_trade["exit_date"] = price_df.index[-1] # type: ignore + current_trade["exit_price"] = price_df.iloc[-1]["Price"] + current_trade["exit_reason"] = "Exit position triggered" + + rets = ( + RETURNS_SIGNS[position_type] + * price_df.loc[ + current_trade["entry_date"] : current_trade["exit_date"] + ].Returns + ) + + if isinstance(taker_fees, float) and taker_fees > 0: + rets.iloc[0] = rets.iloc[0] - taker_fees + rets.iloc[-1] = rets.iloc[-1] - taker_fees + + current_trade["trade_return"] = ((rets + 1).cumprod().iloc[-1]) - 1 # ret + + trades_df = pd.concat( + [trades_df, pd.DataFrame([current_trade])], ignore_index=True + ) + price_df.loc[rets.index[0] : rets.index[-1], "Strategy_returns"] = rets + + timeframe_count = 0 + current_trade = {} + return trades_df, price_df + + +def price_long_only_backtester( + df: pd.DataFrame, + price_column: str, + long_entry_function: Callable[[pd.Series, pd.Series], bool], + long_exit_function: Callable[[pd.Series, pd.Series, int], bool], + timeframe: Timeframe, + take_profit: float = np.inf, + stop_loss: float = np.inf, + initial_equity: int = 1000, + maker_fees: Optional[float] = 0.001, + taker_fees: Optional[float] = 0.001, + get_trade_df: bool = False, + get_returns_df: bool = False, + parameter_optimization: bool = False, + plot_result: bool = True, +) -> Union[None, Union[float, Union[pd.DataFrame, Tuple[pd.DataFrame, pd.DataFrame]]]]: + """Run a backtest with long only position on a price dataset. + + Args: + ---- + df (pd.DataFrame): The dataframe containing the price data and possibly the indicator used in the entry/exit functions.. + + price_column (str): The column name containing the price data. + + long_entry_function (Callable[[pd.Series, pd.Series], bool]): The long entry function, it should take 2 arguments, the current row and the previous row and return True or False depending on your strategy. + + long_exit_function (Callable[[pd.Series, pd.Series, int], bool]): The long exit function, it should take 3 arguments, the current row, the previous row and the number of timeframe count since the last entry order, and return True or False depending on your strategy. + + timeframe (Timeframe): The timeframe granularity of the dataframe. + + take_profit (float, optional): The percent of the buy price to add to create a stop order and take the profit associated. Defaults to np.inf. + + stop_loss (float, optional): The percent of the buy price to cut to create a stop order and stop the loss associated. Defaults to np.inf. + + initial_equity (int, optional): The initial capital. Defaults to 1000. + + maker_fees (Optional[float], optional): The fees applied, here maker for limit orders (not yet implemented). Defaults to 0.001. + + taker_fees (Optional[float], optional): The fees applied, here taker for spot orders. Defaults to 0.001. + + get_trade_df (bool, optional): Whether or not to return the trade dataframe (summary of trades) . Defaults to False. + + get_returns_df (bool, optional): Whether or not to return the strategy returns dataframe. Defaults to False. + + parameter_optimization (bool, optional): This parameter is useful when running fitting/optimization it prints nothing but the final strategy return. Defaults to False. + + plot_result (bool, optional): Plot equity, price, drawdown, distribution of the strategy. Defaults to True. + + Returns: + ----- + Union[None, Union[float, Union[pd.DataFrame, Tuple[pd.DataFrame, pd.DataFrame]]]]: Nothing or the strategy total return or the trade_df or the return_df or the trade_df and the return_df. + """ + assert ( + timeframe in TIMEFRAMES + ), f"timeframe must be one of the following: {','.join(TIMEFRAMES)}" + assert df.shape[0] > 1, "Dataframe must have at least 2 rows" + assert price_column in df.columns, "Dataframe must contain the price column." + + price_df = df.copy() + price_df = price_df.rename(columns={price_column: "Price"}) + price_df["Returns"] = price_df["Price"].pct_change().fillna(0) + trades_df, price_df = __price_backtest_one_position_type( + price_df, + long_entry_function, + long_exit_function, + position_type="long", + take_profit=take_profit, + stop_loss=stop_loss, + maker_fees=maker_fees, + taker_fees=taker_fees, + ) + + price_df["Strategy_cum_returns"] = cumulative_returns(price_df["Strategy_returns"]) + + if parameter_optimization is True: + if len(trades_df) > 0: + return price_df["Strategy_cum_returns"].iloc[-1] + return 0.0 + + assert len(trades_df) > 0, "No trades were generated" + + price_df["Strategy_drawdown"] = drawdown(price_df["Strategy_returns"]) + price_df["Cum_returns"] = cumulative_returns(price_df["Returns"]) + price_df["Drawdown"] = drawdown(price_df["Returns"]) + trades_df["trade_duration"] = trades_df["exit_date"] - trades_df["entry_date"] + + print(f"{' Initial informations ':-^50}") + print(f"Period: [{str(price_df.index[0])}] -> [{str(price_df.index[-1])}]") + print(f"Intial balance: {initial_equity:.2f} $") + if taker_fees is not None and maker_fees is not None: + print( + f"Taker fees: {taker_fees*100:.2f} %, Maker fees: {maker_fees*100:.2f} %, All the metrics will be calculated considering these fees" + ) + else: + print("No fees considered here.") + if take_profit != np.inf: + print(f"Take profit is set to {take_profit*100:.2f} % of the buy price") + if stop_loss != np.inf: + print(f"Stop loss is set to {stop_loss*100:.2f} % of the buy price") + print("Long only position") + + print_backtest_report( + trades_df=trades_df, + ohlcv_df=price_df, + timeframe=timeframe, + initial_equity=initial_equity, + ) + + if plot_result is True: + plot_from_trade_df(price_df) + if get_returns_df and get_trade_df: + return trades_df, price_df + if get_trade_df: + return trades_df + if get_returns_df: + return price_df + + +def price_short_only_backtester( + df: pd.DataFrame, + price_column: str, + short_entry_function: Callable[[pd.Series, pd.Series], bool], + short_exit_function: Callable[[pd.Series, pd.Series, int], bool], + timeframe: Timeframe, + take_profit: float = np.inf, + stop_loss: float = np.inf, + initial_equity: int = 1000, + maker_fees: Optional[float] = 0.001, + taker_fees: Optional[float] = 0.001, + get_trade_df: bool = False, + get_returns_df: bool = False, + parameter_optimization: bool = False, + plot_result: bool = True, +) -> Union[None, Union[float, Union[pd.DataFrame, Tuple[pd.DataFrame, pd.DataFrame]]]]: + """Run a backtest with short only position on a price dataset. + + Args: + ---- + df (pd.DataFrame): df (pd.DataFrame): The dataframe containing the price data and possibly the indicator used in the entry/exit functions.. + + price_column (str): The column name containing the price data. + + short_entry_function (Callable[[pd.Series, pd.Series], bool]): The short entry function, it should take 2 arguments, the current row and the previous row and return True or False depending on your strategy. + + short_exit_function (Callable[[pd.Series, pd.Series, int], bool]): The short exit function, it should take 3 arguments, the current row, the previous row and the number of timeframe count since the last entry order, and return True or False depending on your strategy. + + timeframe (Timeframe): The timeframe granularity of the dataframe. + + take_profit (float, optional): The percent of the buy price to add to create a stop order and take the profit associated. Defaults to np.inf. + + stop_loss (float, optional): The percent of the buy price to cut to create a stop order and stop the loss associated. Defaults to np.inf. + + initial_equity (int, optional): The initial capital. Defaults to 1000. + + maker_fees (Optional[float], optional): The fees applied, here maker for limit orders (not yet implemented). Defaults to 0.001. + + taker_fees (Optional[float], optional): The fees applied, here taker for spot orders. Defaults to 0.001. + + get_trade_df (bool, optional): Whether or not to return the trade dataframe (summary of trades) . Defaults to False. + + get_returns_df (bool, optional): Whether or not to return the strategy returns dataframe. Defaults to False. + + parameter_optimization (bool, optional): This parameter is useful when running fitting/optimization it prints nothing but the final strategy return. Defaults to False. + + plot_result (bool, optional): Plot equity, price, drawdown, distribution of the strategy. Defaults to True. + + Returns: + ----- + Union[None, Union[float, Union[pd.DataFrame, Tuple[pd.DataFrame, pd.DataFrame]]]]: Nothing or the strategy total return or the trade_df or the return_df or the trade_df and the return_df. + """ + assert ( + timeframe in TIMEFRAMES + ), f"timeframe must be one of the following: {','.join(TIMEFRAMES)}" + assert df.shape[0] > 1, "Dataframe must have at least 2 rows" + assert price_column in df.columns, "Dataframe must contain the price column." + + price_df = df.copy() + price_df = price_df.rename(columns={price_column: "Price"}) + price_df["Returns"] = price_df["Price"].pct_change().fillna(0) + trades_df, price_df = __price_backtest_one_position_type( + price_df, + short_entry_function, + short_exit_function, + position_type="short", + take_profit=take_profit, + stop_loss=stop_loss, + maker_fees=maker_fees, + taker_fees=taker_fees, + ) + + price_df["Strategy_cum_returns"] = cumulative_returns(price_df["Strategy_returns"]) + + if parameter_optimization is True: + if len(trades_df) > 0: + return price_df["Strategy_cum_returns"].iloc[-1] + return 0.0 + + assert len(trades_df) > 0, "No trades were generated" + + price_df["Strategy_drawdown"] = drawdown(price_df["Strategy_returns"]) + price_df["Cum_returns"] = cumulative_returns(price_df["Returns"]) + price_df["Drawdown"] = drawdown(price_df["Returns"]) + trades_df["trade_duration"] = trades_df["exit_date"] - trades_df["entry_date"] + + print(f"{' Initial informations ':-^50}") + print(f"Period: [{str(price_df.index[0])}] -> [{str(price_df.index[-1])}]") + print(f"Intial balance: {initial_equity:.2f} $") + if taker_fees is not None and maker_fees is not None: + print( + f"Taker fees: {taker_fees*100:.2f} %, Maker fees: {maker_fees*100:.2f} %, All the metrics will be calculated considering these fees" + ) + else: + print("No fees considered here.") + if take_profit != np.inf: + print(f"Take profit is set to {take_profit*100:.2f} % of the buy price") + if stop_loss != np.inf: + print(f"Stop loss is set to {stop_loss*100:.2f} % of the buy price") + print("Short only positions") + print_backtest_report( + trades_df=trades_df, + ohlcv_df=price_df, + timeframe=timeframe, + initial_equity=initial_equity, + ) + + if plot_result is True: + plot_from_trade_df(price_df) + if get_returns_df and get_trade_df: + return trades_df, price_df + if get_trade_df: + return trades_df + if get_returns_df: + return price_df diff --git a/quant_invest_lab/reports.py b/quant_invest_lab/reports.py index c8a8a5a..ca25927 100644 --- a/quant_invest_lab/reports.py +++ b/quant_invest_lab/reports.py @@ -333,7 +333,7 @@ def print_portfolio_strategy_report( return report_df -def print_ohlc_backtest_report( +def print_backtest_report( trades_df: pd.DataFrame, ohlcv_df: pd.DataFrame, timeframe: Timeframe, @@ -824,6 +824,37 @@ def plot_candle_stick(dataframe_with_ohlc: pd.DataFrame) -> Model: return p +def plot_price_evolution( + price: pd.Series, +) -> Model: + p = figure( + tools="pan,wheel_zoom,box_zoom,reset,save", + width=UNIT_PLOT_WIDTH, + height=UNIT_PLOT_HEIGHT, + title="Price evolution", + x_axis_label="Datetime", + x_axis_type="datetime", + y_axis_label="Price evolution", + background_fill_color=BACKGROUND_COLOR, + ) + + p.line( + x=price.index, + y=price, + color=SALMON_COLOR, + line_width=2, + legend_label="Price", + ) + p.xaxis.formatter = DT_FORMATTER + + p.legend.location = "center" + + p.add_layout(p.legend[0], "below") + p.grid.grid_line_color = GRID_COLOR + p.grid.grid_line_alpha = 1 + return p + + def plot_asset_allocation( allocation_dataframe: pd.DataFrame, min_weight: float = 0.001 ) -> Model: @@ -882,10 +913,16 @@ def plot_from_trade_df(price_df: pd.DataFrame) -> None: price_df (pd.DataFrame): The historical price dataframe. """ + output_notebook() grid = gridplot( [ [ - plot_candle_stick(price_df), + plot_candle_stick(price_df) + if set({"Open", "High", "Low", "Close", "Returns"}).issubset( + price_df.columns + ) + is True + else plot_price_evolution(price_df["Price"]), plot_returns_distribution( price_df["Strategy_returns"], price_df["Returns"] ), @@ -928,6 +965,7 @@ def plot_from_trade_df_and_ptf_optimization( benchmark_returns (pd.Series): The benchmark returns. asset_allocation_dataframe (pd.DataFrame): The allocation of each asset in the portfolio, columns are the assets and rows are the weights. """ + output_notebook() price_df = pd.DataFrame( data={"Strategy_returns": portfolio_returns, "Returns": benchmark_returns}, index=benchmark_returns.index,