Refactor returns calculation with leverage

stratestic/backtesting/_mixin.py

@@ -263,7 +263,8 @@ def maximum_leverage(self, margin_threshold=None):
         if margin_threshold is not None:
             self.set_margin_threshold(margin_threshold)
 
-        self._load_symbol_data()
+        self.include_margin = True
+        self._load_leverage_brackets()
 
         left_limit, right_limit = self.leverage_limits
 
@@ -374,6 +375,9 @@ def _adapt_optimization_input(self, params):
 
     def _sanitize_equity(self, df, trades):
 
+        if len(trades) == 0:
+            return df
+
         trades_df = pd.DataFrame(trades)
 
         # Bring equity to 0 if a trade has gotten to zero equity
@@ -401,6 +405,8 @@ def _sanitize_trades(data, trades):
 
         trades_df = trades_df[trades_df.index < no_equity[0]].copy()
 
+        trades_df["pnl"] = np.where(trades_df["pnl"] < -1, -1, trades_df["pnl"])
+
         return [Trade(**row) for _, row in trades_df.reset_index().iterrows()]
 
     @staticmethod
@@ -413,6 +419,13 @@ def _sanitize_margin_ratio(df):
 
         return df
 
+    def _calculate_strategy_returns(self, df):
+        df[STRATEGY_RETURNS_TC] = np.log(df['equity'] / df['equity'].shift(1)).fillna(0)
+        df[STRATEGY_RETURNS] = df[STRATEGY_RETURNS_TC] + df["trades"] * self.tc
+        df.loc[df.index[0], STRATEGY_RETURNS] = 0
+
+        return df
+
     def _calculate_cumulative_returns(self, data):
         data[BUY_AND_HOLD] = data[self.returns_col].cumsum().apply(np.exp).fillna(1)
         data[CUM_SUM_STRATEGY_TC] = data[STRATEGY_RETURNS_TC].cumsum().apply(np.exp).fillna(1)

stratestic/backtesting/helpers/evaluation/_constants.py

@@ -16,10 +16,10 @@
     "accumulated_strategy_returns_tc": "Strategy returns (with trading costs)"
 }
 results_mapping = {
-    'initial_equity': lambda unit: f"Traded Initial Amount [{unit}]",
-    'exposed_capital': lambda unit: f"Exposed Capital [{unit}]",
-    'equity_final': lambda unit: f"Equity Final [{unit}]",
-    'equity_peak': lambda unit: f"Equity Peak [{unit}]",
+    'traded_amount': lambda unit: f"Traded Amount [{unit}]",
+    'equity_initial': lambda unit: f"Equity - Initial [{unit}]",
+    'equity_final': lambda unit: f"Equity - Final [{unit}]",
+    'equity_peak': lambda unit: f"Equity - Peak [{unit}]",
     'trading_costs': "Trading Costs [%]",
     'leverage': "Leverage [x]",
     'buy_and_hold_return': "Buy & Hold Return [%]",
@@ -57,9 +57,9 @@
 }
 
 results_sections = {
-    'Overview': ['total_duration', 'start_date', 'end_date', 'trading_costs',
-                 'leverage', 'initial_equity', 'exposed_capital', 'exposure_time'],
-    'Returns': ['return_pct', 'equity_final', 'equity_peak', 'return_pct_annualized',
+    'Overview': ['total_duration', 'start_date', 'end_date', 'trading_costs', 'exposure_time',
+                 'leverage', 'equity_initial', 'equity_final', 'equity_peak'],
+    'Returns': ['return_pct', 'return_pct_annualized',
                 'volatility_pct_annualized', 'buy_and_hold_return', ],
     'Drawdowns': ['max_drawdown', 'avg_drawdown', 'max_drawdown_duration', 'avg_drawdown_duration'],
     'Trades': ['nr_trades', 'win_rate', 'best_trade', 'worst_trade', 'avg_trade',

stratestic/backtesting/helpers/evaluation/_results.py

@@ -42,8 +42,8 @@ def get_overview_results(results, data, leverage, trading_costs, amount):
     if trading_costs is not None:
         results["trading_costs"] = trading_costs * 100
 
-    results["initial_equity"] = amount * leverage
-    results["exposed_capital"] = results["initial_equity"]
+    results["equity_initial"] = amount
+    results["traded_amount"] = amount * leverage
 
     if SIDE in data:
         results["exposure_time"] = exposure_time(data[SIDE])
@@ -106,7 +106,7 @@ def get_ratios_results(results, data, trades, trading_days):
 
 def log_results(results, backtesting=True):
 
-    length = 55
+    length = 60
 
     logging.info("")
 
@@ -142,7 +142,7 @@ def log_results(results, backtesting=True):
                 except TypeError:
                     value = str(value)
 
-            logging.info(f'{printed_title:<30}{value.rjust(25)}')
+            logging.info(f'{printed_title:<30}{value.rjust(30)}')
         logging.info('-' * length)
         logging.info('')
     logging.info('*' * length)

stratestic/backtesting/helpers/evaluation/metrics.py

@@ -21,10 +21,6 @@ def get_end_date(close_date: pd.Series) -> datetime:
     return close_date[-1]
 
 
-def calculate_multiple(cum_returns: pd.Series, leverage: int = 1) -> float:
-    return cum_returns[-1] / cum_returns[0] * leverage
-
-
 def exposure_time(positions: np.ndarray) -> float:
     """Calculate the percentage of time the strategy was exposed to the market."""
     return np.count_nonzero(positions) / len(positions) * 100
@@ -223,10 +219,6 @@ def win_rate_pct(trades: List[Trade]) -> float:
     return winning_trades / nr_trades * 100 if nr_trades > 0 else 0
 
 
-def calculate_pnl(trade):
-    return np.exp(np.log(trade.exit_price / trade.entry_price) * trade.side) - 1
-
-
 def best_trade_pct(trades: List[Trade]) -> float:
     """Calculate the percentage of the best trade."""
 

stratestic/backtesting/helpers/margin/_margin.py

@@ -26,18 +26,12 @@ def get_maintenance_margin_binance():
         Internal function for calculating maintenance margin for Binance exchange.
         """
 
-        # values = pd.concat([pd.DataFrame(notional_value).T] * len(notional_value)).T
-        #
-        # triangular_matrix = pd.DataFrame(np.tril(values))
-
         values = pd.Series(notional_value)
 
         comparison = pd.DataFrame()
         for i, notional_floor in enumerate(symbol_brackets['notionalFloor']):
             comparison[i] = values >= notional_floor
 
-        # greater_than = triangular_matrix.gt(symbol_brackets["notionalFloor"], axis=1)
-
         indexes = comparison.idxmin(axis=1) - 1
 
         brackets = symbol_brackets.iloc[indexes, :]

stratestic/backtesting/helpers/plotting/_plotting.py

@@ -258,6 +258,8 @@ def size_trade_markers(notional_value, min_marker_size=10, max_marker_size=35):
 
     normalized = (notional_value - min_value) / (max_value - min_value)
 
+    normalized = pd.Series(np.where(normalized, np.isnan(normalized), 0.5))
+
     marker_size = min_marker_size + normalized * (max_marker_size - min_marker_size)
 
     return marker_size

stratestic/backtesting/helpers/trade.py

@@ -32,17 +32,19 @@ class Trade:
     exit_price: float
     units: float
     side: int
-    equity: float
+    equity: float = None
     amount: float = None
     profit: float = None
     pnl: float = None
     liquidation_price: float = None
+    maintenance_rate: float = None
+    maintenance_amount: float = None
 
-    def calculate_profit(self, prev_amount):
+    def calculate_profit(self, prev_equity):
         if self.amount is None:
             return
 
-        self.profit = self.amount - prev_amount
+        self.profit = self.equity - prev_equity
 
     def calculate_pnl_pct(self, leverage):
         self.pnl = (np.exp(np.log(self.exit_price / self.entry_price) * self.side) - 1) * leverage

stratestic/backtesting/iterative/_iterative.py

@@ -104,20 +104,11 @@ def _reset_object(self, symbol):
         self.margin_ratios = []
         self.nr_trades = 0
         self.trades = []
+        self.initial_balance = self.amount * self.leverage
         self.current_balance = self.initial_balance
+        self.current_equity = self.amount
         self.units = 0
 
-    def _get_trades(self, _):
-        """
-        Gets the number of trades executed.
-
-        Returns
-        -------
-        int
-            The number of trades executed.
-        """
-        return self.nr_trades
-
     def _get_price(self, _, row):
         """
         Gets the price for the given row.
@@ -238,22 +229,24 @@ def _perform_iteration(self, data, print_results):
             trades = np.abs(signal - previous_position)
 
             if self.include_margin:
-                new_trade = trades >= 1 and previous_position != 0
+                new_trade = trades >= 1
                 self._calculate_margin_ratio(row, new_trade)
 
-            self.strategy_returns[self.symbol].append(row[self.returns_col] * previous_position)
-            self.strategy_returns_tc[self.symbol].append(self.strategy_returns[self.symbol][-1] - trades * self.tc)
+            strategy_return = row[self.returns_col] * previous_position - trades * self.tc
 
-            pnl = (np.exp(self.strategy_returns_tc[self.symbol][-1]) - 1) * amount
+            simple_return = np.exp(strategy_return) - 1
+
+            pnl = simple_return * amount
 
             equity = equity + pnl
+
+            self.equity.append(equity)
+
             if trades >= 1 and previous_position != 0:
                 amount = equity * self.leverage
             else:
                 amount = amount + pnl
 
-            self.equity.append(equity)
-
     def _iterative_backtest(self, data, print_results=True):
         """
         Iterate through the data, trade accordingly, and calculate the strategy's performance.
@@ -282,16 +275,17 @@ def _iterative_backtest(self, data, print_results=True):
             data = self._sanitize_margin_ratio(data)
 
         data = self._sanitize_equity(data, self.trades)
+        data = self._add_trades_rows(data)
         data = self._calculate_strategy_returns(data)
         data = self._calculate_cumulative_returns(data)
         trades = self._sanitize_trades(data, self.trades)
 
         return data, trades
 
     @staticmethod
-    def _calculate_strategy_returns(data):
-        data[STRATEGY_RETURNS_TC] = np.log(data['equity'] / data['equity'].shift(1)).fillna(0)
-        data[STRATEGY_RETURNS] = data[STRATEGY_RETURNS_TC].copy()
+    def _add_trades_rows(data):
+        data["trades"] = np.abs(data[SIDE].diff())
+        data.loc[data.index[0], "trades"] = np.abs(data.loc[data.index[0], SIDE] - 0)
 
         return data
 
@@ -310,8 +304,8 @@ def _calculate_margin_ratio(self, row, new_trade):
             trade.side,
             trade.entry_price,
             mark_price,
-            self.maintenance_rate,
-            self.maintenance_amount,
+            trade.maintenance_rate,
+            trade.maintenance_amount,
             exchange=self.exchange
         )
 
@@ -332,25 +326,6 @@ def _evaluate_backtest(self, data, trades):
 
         return results, self.nr_trades, perf, outperf
 
-    def _get_net_value(self, row):
-        """
-        Calculate the current net value of the strategy.
-
-        Parameters
-        ----------
-        row : pandas.Series
-            The current row of the data being processed.
-
-        Returns
-        -------
-        float
-            The current net value of the strategy.
-
-        """
-        price = self._get_price("", row)
-
-        return self.current_balance + self.units * price
-
     def buy_instrument(
         self,
         symbol,
@@ -541,7 +516,8 @@ def close_pos(self, symbol, date=None, row=None, header='', **kwargs):
 
     def _handle_trade(self, trades, open_trade, date, price, units, equity, amount, side):
         if open_trade:
-            liquidation_price = None
+
+            trades.append(Trade(date, None, price, None, units, side))
 
             if self.include_margin:
 
@@ -551,9 +527,6 @@ def _handle_trade(self, trades, open_trade, date, price, units, equity, amount,
                     self._symbol_bracket, [notional_value], exchange=self.exchange
                 )
 
-                self.maintenance_rate = maintenance_rate[0]
-                self.maintenance_amount = maintenance_amount[0]
-
                 liquidation_price = calculate_liquidation_price(
                     units,
                     price,
@@ -564,14 +537,16 @@ def _handle_trade(self, trades, open_trade, date, price, units, equity, amount,
                     exchange=self.exchange
                 )[0]
 
-            trades.append(Trade(date, None, price, None, units, side, None, None, None, None, liquidation_price))
+                trades[-1].liquidation_price = liquidation_price
+                trades[-1].maintenance_rate = maintenance_rate[0]
+                trades[-1].maintenance_amount = maintenance_amount[0]
         else:
             trades[-1].exit_date = date
             trades[-1].exit_price = price
             trades[-1].equity = equity
             trades[-1].amount = amount
 
-            trades[-1].calculate_profit(trades[-2].amount if len(trades) >= 2 else self.amount * self.leverage)
+            trades[-1].calculate_profit(trades[-2].equity if len(trades) >= 2 else self.amount)
             trades[-1].calculate_pnl_pct(self.leverage)
 
             self.nr_trades += 1

stratestic/backtesting/vectorized/_vectorized.py

@@ -146,6 +146,10 @@ def process_leveraged_returns(self, processed_data, trades_df):
 
         df.loc[df.index[0], self.returns_col] = 0
 
+        if len(trades_df) == 0:
+            df["equity"] = 0
+            return df
+
         df_filter = (df.trades != 0) & (df.side != 0)
 
         df.loc[df[df_filter].index, 'notional_value'] = trades_df["equity"].shift(1).values
@@ -161,9 +165,6 @@ def process_leveraged_returns(self, processed_data, trades_df):
         amount = self.amount * self.leverage
         for index, row in df.iterrows():
 
-            if index == Timestamp("2023-02-13 00:00:00"):
-                a = 1
-
             pnl = (np.exp(row[STRATEGY_RETURNS_TC]) - 1) * amount
 
             equity = equity + pnl
@@ -181,13 +182,6 @@ def process_leveraged_returns(self, processed_data, trades_df):
 
         return df
 
-    def _calculate_strategy_returns(self, df):
-        df[STRATEGY_RETURNS_TC] = np.log(df['equity'] / df['equity'].shift(1)).fillna(0)
-        df[STRATEGY_RETURNS] = df[STRATEGY_RETURNS_TC] + df["trades"] * self.tc
-        df.loc[df.index[0], STRATEGY_RETURNS] = 0
-
-        return df
-
     def _retrieve_trades(self, processed_data, trading_costs=0):
         """
         Computes the trades made based on the input processed data and returns a list of Trade objects.
@@ -255,7 +249,7 @@ def _retrieve_trades(self, processed_data, trading_costs=0):
 
             equity = self.amount
             for index, trade in trades.iterrows():
-                pnl = (np.exp(trade["log_return"]) - 1) * self.leverage # - self.tc * 2 * trade[SIDE]
+                pnl = (np.exp(trade["log_return"]) - 1) * self.leverage
 
                 equity = equity * (1 + pnl)
                 amount = equity * self.leverage
@@ -292,8 +286,6 @@ def _retrieve_trades(self, processed_data, trading_costs=0):
                 exchange=self.exchange
             )
 
-            columns_to_delete.extend(['maintenance_rate', 'maintenance_amount'])
-
         self._trades_df = trades.copy()
 
         trades.drop(columns_to_delete, axis=1, inplace=True)