Add metadata specification

Author: MDUYN

investing_algorithm_framework/domain/backtesting/backtest_run.py

@@ -293,14 +293,23 @@ def get_trades(self, target_symbol=None, trade_status=None) -> List[Trade]:
 
     def get_portfolio_snapshots(
         self,
-        created_at_lt: Optional[datetime] = None
+        created_at_lt: Optional[datetime] = None,
+        created_at_lte: Optional[datetime] = None,
+        created_at_gt: Optional[datetime] = None,
+        created_at_gte: Optional[datetime] = None
     ) -> List[PortfolioSnapshot]:
         """
         Get the portfolio snapshots of the backtest report
 
         Args:
             created_at_lt (datetime): The created_at date to filter
                 the snapshots
+            created_at_lte (datetime): The created_at date to filter
+                the snapshots
+            created_at_gt (datetime): The created_at date to filter
+                the snapshots
+            created_at_gte (datetime): The created_at date to filter
+                the snapshots
 
         Returns:
             list: The portfolio snapshots of the backtest report
@@ -313,6 +322,24 @@ def get_portfolio_snapshots(
                 if snapshot.created_at < created_at_lt
             ]
 
+        if created_at_lte is not None:
+            selection = [
+                snapshot for snapshot in selection
+                if snapshot.created_at <= created_at_lte
+            ]
+
+        if created_at_gt is not None:
+            selection = [
+                snapshot for snapshot in selection
+                if snapshot.created_at > created_at_gt
+            ]
+
+        if created_at_gte is not None:
+            selection = [
+                snapshot for snapshot in selection
+                if snapshot.created_at >= created_at_gte
+            ]
+
         return selection
 
     def get_orders(

investing_algorithm_framework/domain/models/portfolio/portfolio_snapshot.py

@@ -128,6 +128,7 @@ def __repr__(self):
             total_revenue=self.total_revenue,
             total_cost=self.total_cost,
             cash_flow=self.cash_flow,
+            metadata=self.metadata,
         )
 
     def to_dict(self, datetime_format=None):

investing_algorithm_framework/services/backtesting/backtest_service.py

@@ -259,7 +259,7 @@ def create_vector_backtest(
                 # If we are not in a position, and we get a buy signal
                 if current_signal == 1 and last_trade is None:
                     amount = float(capital_for_trade / current_price)
-                    order = Order(
+                    buy_order = Order(
                         id=uuid4(),
                         target_symbol=symbol,
                         trading_symbol=trading_symbol,
@@ -271,12 +271,10 @@ def create_vector_backtest(
                         updated_at=current_date,
                         order_side=OrderSide.BUY
                     )
-                    orders.append(order)
+                    orders.append(buy_order)
                     trade = Trade(
                         id=uuid4(),
-                        orders=[
-                            order
-                        ],
+                        orders=[buy_order],
                         target_symbol=symbol,
                         trading_symbol=trading_symbol,
                         available_amount=amount,
@@ -286,11 +284,9 @@ def create_vector_backtest(
                         opened_at=current_date,
                         closed_at=None,
                         amount=amount,
-                        net_gain=0,
                         status=TradeStatus.OPEN.value,
                         cost=capital_for_trade
                     )
-                    trade.updated_at = current_date
                     last_trade = trade
                     trades.append(trade)
 
@@ -299,11 +295,7 @@ def create_vector_backtest(
                     net_gain_val = (
                         current_price - last_trade.open_price
                     ) * last_trade.available_amount
-                    last_trade.closed_at = current_date
-                    last_trade.updated_at = current_date
-                    last_trade.net_gain = net_gain_val
-                    last_trade.status = TradeStatus.CLOSED.value
-                    order = Order(
+                    sell_order = Order(
                         id=uuid4(),
                         target_symbol=symbol,
                         trading_symbol=trading_symbol,
@@ -315,8 +307,18 @@ def create_vector_backtest(
                         updated_at=current_date,
                         order_side=OrderSide.SELL
                     )
-                    orders.append(order)
-                    last_trade.orders.append(order)
+                    orders.append(sell_order)
+                    trade_orders = last_trade.orders
+                    trade_orders.append(sell_order)
+                    last_trade.update(
+                        {
+                            "orders": trade_orders,
+                            "closed_at": current_date,
+                            "trade_status": TradeStatus.CLOSED,
+                            "updated_at": current_date,
+                            "net_gain": net_gain_val
+                        }
+                    )
                     last_trade = None
 
         # Create portfolio snapshots

investing_algorithm_framework/services/metrics/trades.py

@@ -201,7 +201,8 @@ def get_average_trade_return(trades: List[Trade]) -> Tuple[float, float]:
             "Trades list is empty, cannot compute average return."
         )
 
-    closed_trades = [t for t in trades if t.status == TradeStatus.CLOSED]
+    closed_trades = [t for t in trades if TradeStatus.CLOSED.equals(t.status)]
+
     if not closed_trades:
         return 0.0, 0.0
 
@@ -265,15 +266,22 @@ def get_average_trade_loss(trades: List[Trade]) -> Tuple[float, float]:
             "Trades list is empty or None, cannot calculate average loss."
         )
 
-    losses = [t.net_gain for t in trades if t.net_gain < 0]
-    cost = sum(t.cost for t in trades if t.net_gain < 0)
+    closed_trades = [t for t in trades if TradeStatus.CLOSED.equals(t.status)]
+    losing_trades = [t for t in closed_trades if t.net_gain < 0]
 
-    if not losses:
+    if not losing_trades or len(losing_trades) == 0:
         return 0.0, 0.0
 
+    losses = [t.net_gain for t in losing_trades]
     average_loss = sum(losses) / len(losses)
-    percentage = (average_loss / cost) if cost > 0 else 0.0
-    return average_loss, percentage
+    percentage_returns = [
+        (t.net_gain / t.cost) * 100.0 for t in losing_trades if t.cost > 0
+    ]
+    average_return_percentage = (
+        sum(percentage_returns) / len(percentage_returns)
+        if percentage_returns else 0.0
+    )
+    return average_loss, average_return_percentage
 
 
 def get_median_trade_return(trades: List[Trade]) -> Tuple[float, float]:
@@ -339,5 +347,3 @@ def get_worst_trade(trades: List[Trade]) -> Trade:
         return None
 
     return min(trades, key=lambda t: t.net_gain)
-
-