Fix flake8 warnings

Author: MDUYN

investing_algorithm_framework/app/analysis/backtest_utils.py

@@ -14,6 +14,7 @@ def save_backtests_to_directory(
     backtests: List[Backtest],
     directory_path: Union[str, Path],
     dir_name_generation_function: Callable[[Backtest], str] = None,
+    number_of_backtests_to_save: int = None,
     filter_function: Callable[[Backtest], bool] = None
 ) -> None:
     """
@@ -28,6 +29,8 @@ def save_backtests_to_directory(
             a string to be used as the directory name for that backtest.
             If not provided, the backtest's metadata 'id' will be used.
             Defaults to None.
+        number_of_backtests_to_save (int, optional): Maximum number of
+            backtests to save. If None, all backtests will be saved.
         filter_function (Callable[[Backtest], bool], optional): A function
             that takes a Backtest object as input and returns True if the
             backtest should be saved. Defaults to None.
@@ -41,6 +44,12 @@ def save_backtests_to_directory(
 
     for backtest in backtests:
 
+        # Check if we have reached the limit of backtests to save
+        if number_of_backtests_to_save is not None:
+            if number_of_backtests_to_save <= 0:
+                break
+            number_of_backtests_to_save -= 1
+
         if filter_function is not None:
             if not filter_function(backtest):
                 continue
@@ -63,7 +72,8 @@ def save_backtests_to_directory(
 
 def load_backtests_from_directory(
     directory_path: Union[str, Path],
-    filter_function: Callable[[Backtest], bool] = None
+    filter_function: Callable[[Backtest], bool] = None,
+    number_of_backtests_to_load: int = None
 ) -> List[Backtest]:
     """
     Loads Backtest objects from the specified directory.
@@ -74,6 +84,8 @@ def load_backtests_from_directory(
         filter_function (Callable[[Backtest], bool], optional): A function
             that takes a Backtest object as input and returns True if the
             backtest should be included in the result. Defaults to None.
+        number_of_backtests_to_load (int, optional): Maximum number of
+            backtests to load. If None, all backtests will be loaded.
 
     Returns:
         List[Backtest]: List of loaded Backtest objects.
@@ -89,6 +101,13 @@ def load_backtests_from_directory(
         return backtests
 
     for file_name in os.listdir(directory_path):
+
+        # Check if we have reached the limit of backtests to load
+        if number_of_backtests_to_load is not None:
+            if number_of_backtests_to_load <= 0:
+                break
+            number_of_backtests_to_load -= 1
+
         file_path = os.path.join(directory_path, file_name)
 
         try:

investing_algorithm_framework/services/__init__.py

@@ -13,11 +13,11 @@
 from .repository_service import RepositoryService
 from .trade_service import TradeService, TradeStopLossService, \
     TradeTakeProfitService
-from .metrics import get_annual_volatility, \
+from .metrics import get_annual_volatility, get_mean_daily_return, \
     get_sortino_ratio, get_drawdown_series, get_max_drawdown, \
     get_equity_curve, get_price_efficiency_ratio, get_sharpe_ratio, \
     get_profit_factor, get_cumulative_profit_factor_series, \
-    get_rolling_profit_factor_series, \
+    get_rolling_profit_factor_series, get_daily_returns_std, \
     get_cagr, get_standard_deviation_returns, \
     get_standard_deviation_downside_returns, \
     get_total_return, get_cumulative_exposure, get_exposure_ratio, \
@@ -41,6 +41,8 @@
     get_current_average_trade_gain, create_backtest_metrics_for_backtest
 
 __all__ = [
+    "get_mean_daily_return",
+    "get_daily_returns_std",
     "OrderService",
     "RepositoryService",
     "PortfolioService",

investing_algorithm_framework/services/metrics/__init__.py

@@ -38,8 +38,12 @@
     get_current_average_trade_duration, get_current_average_trade_gain, \
     get_current_average_trade_return, get_number_of_open_trades, \
     get_average_trade_duration
+from .mean_daily_return import get_mean_daily_return
+from .standard_deviation import get_daily_returns_std
 
 __all__ = [
+    "get_mean_daily_return",
+    "get_daily_returns_std",
     "get_annual_volatility",
     "get_sortino_ratio",
     "get_drawdown_series",

investing_algorithm_framework/services/metrics/mean_daily_return.py

@@ -38,10 +38,13 @@ def get_mean_daily_return(snapshots):
 
     # Check if the period is less than a year
     if (end_date - start_date).days < 365:
+        print("Less than a year of data, using CAGR to calculate mean daily return.")
         # Use CAGR to calculate mean daily return
         cagr = get_cagr(snapshots)
         if cagr == 0.0:
             return 0.0
+
+        print(f"CAGR: {cagr}")
         return (1 + cagr) ** (1 / 365) - 1
 
     # Resample to daily frequency using last value of the day

investing_algorithm_framework/services/metrics/sharpe_ratio.py

@@ -44,12 +44,12 @@
 
 """
 
-from typing import Optional, List, Tuple
-
 import math
-import pandas as pd
-import numpy as np
 from datetime import datetime
+from typing import List, Tuple
+
+import numpy as np
+import pandas as pd
 
 from investing_algorithm_framework.domain import PortfolioSnapshot
 from .mean_daily_return import get_mean_daily_return
@@ -76,7 +76,10 @@ def get_sharpe_ratio(
     """
     snapshots = sorted(snapshots, key=lambda s: s.created_at)
     mean_daily_return = get_mean_daily_return(snapshots)
+
+    print(f"mean daily return {mean_daily_return}")
     std_daily_return = get_daily_returns_std(snapshots)
+    print(f"std daily return {std_daily_return}")
 
     if std_daily_return == 0:
         return float('nan')  # Avoid division by zero

investing_algorithm_framework/services/metrics/standard_deviation.py

@@ -108,9 +108,8 @@ def get_daily_returns_std(snapshots):
     df["created_at"] = pd.to_datetime(df["created_at"])
     df = df.drop_duplicates("created_at").set_index("created_at")
     df = df.sort_index()
-
     # Resample to daily frequency (end of day)
-    daily_df = df.resample("D").last().dropna()
+    daily_df = df.resample("D").last().ffill().dropna()
 
     # Calculate daily returns
     daily_df["return"] = daily_df["total_value"].pct_change().dropna()