Merge pull request #414 from coding-kitties/release/dev-to-main

Merge dev to main (v7.30.0)

Author: MDUYN

investing_algorithm_framework/domain/models/time_frame.py

@@ -10,12 +10,16 @@ class TimeFrame(Enum):
     FIVE_MINUTE = "5m"
     TEN_MINUTE = "10m"
     FIFTEEN_MINUTE = "15m"
+    TWENTY_MINUTE = "20m"
     THIRTY_MINUTE = "30m"
     ONE_HOUR = "1h"
     TWO_HOUR = "2h"
     FOUR_HOUR = "4h"
+    SIX_HOUR = "6h"
+    EIGHT_HOUR = "8h"
     TWELVE_HOUR = "12h"
     ONE_DAY = "1d"
+    THREE_DAY = "3d"
     ONE_WEEK = "1W"
     ONE_MONTH = "1M"
     ONE_YEAR = "1Y"
@@ -36,7 +40,16 @@ def from_string(value: str):
                     if value == entry.value.replace("H", "h"):
                         return entry
 
-                # For hour timeframes compare with and without H
+                # For hour timeframes compare with and without h
+                if "h" in entry.value:
+
+                    if value == entry.value:
+                        return entry
+
+                    if value == entry.value.replace("h", "H"):
+                        return entry
+
+                # For day timeframes compare with and without D
                 if "d" in entry.value:
 
                     if value == entry.value:
@@ -100,6 +113,9 @@ def amount_of_minutes(self):
         if self.equals(TimeFrame.FIFTEEN_MINUTE):
             return 15
 
+        if self.equals(TimeFrame.TWENTY_MINUTE):
+            return 20
+
         if self.equals(TimeFrame.THIRTY_MINUTE):
             return 30
 
@@ -112,12 +128,21 @@ def amount_of_minutes(self):
         if self.equals(TimeFrame.FOUR_HOUR):
             return 240
 
+        if self.equals(TimeFrame.SIX_HOUR):
+            return 360
+
+        if self.equals(TimeFrame.EIGHT_HOUR):
+            return 480
+
         if self.equals(TimeFrame.TWELVE_HOUR):
             return 720
 
         if self.equals(TimeFrame.ONE_DAY):
             return 1440
 
+        if self.equals(TimeFrame.THREE_DAY):
+            return 4320
+
         if self.equals(TimeFrame.ONE_WEEK):
             return 10080
 

investing_algorithm_framework/services/metrics/drawdown.py

@@ -109,16 +109,17 @@ def get_max_drawdown(snapshots: List[PortfolioSnapshot]) -> float:
 
 def get_max_daily_drawdown(snapshots: List[PortfolioSnapshot]) -> float:
     """
-    Calculate the maximum daily drawdown of the portfolio as a percentage from the peak.
+    Calculate the worst single-day decline of the portfolio as a percentage.
 
-    This is the largest drop in equity (in percentage) from a peak to a trough
-    during the backtest period, calculated on a daily basis.
+    This is the largest day-over-day percentage drop in equity,
+    NOT the peak-to-trough drawdown (use get_max_drawdown for that).
 
     Args:
         snapshots (List[PortfolioSnapshot]): List of portfolio snapshots
 
     Returns:
-        float: The maximum daily drawdown as a negative percentage (e.g., -5.0 for a 5% drawdown).
+        float: The maximum single-day drawdown as a positive percentage
+            (e.g., 0.05 for a 5% single-day decline).
     """
     # Create DataFrame from snapshots
     data = [(s.created_at, s.total_value) for s in snapshots]
@@ -136,54 +137,58 @@ def get_max_daily_drawdown(snapshots: List[PortfolioSnapshot]) -> float:
     # Filter out non-positive values
     positive_values = daily_df[daily_df['total_value'] > 0]['total_value']
 
-    if positive_values.empty:
+    if positive_values.empty or len(positive_values) < 2:
         return 0.0
 
-    peak = positive_values.iloc[0]
-    max_daily_drawdown_pct = 0.0
+    # Compute day-over-day returns; the worst single-day decline
+    # is the most negative return (ignore positive returns)
+    daily_returns = positive_values.pct_change().dropna()
+    negative_returns = daily_returns[daily_returns < 0]
 
-    for equity in positive_values:
-        if equity > peak:
-            peak = equity
-
-        # Avoid division by zero (shouldn't happen but extra safety)
-        if peak <= 0:
-            continue
-
-        drawdown_pct = (equity - peak) / peak
-        max_daily_drawdown_pct = min(max_daily_drawdown_pct, drawdown_pct)
+    if negative_returns.empty:
+        return 0.0
 
-    return abs(max_daily_drawdown_pct)  # Return as positive percentage
+    return abs(negative_returns.min())
 
 def get_max_drawdown_duration(snapshots: List[PortfolioSnapshot]) -> int:
     """
     Calculate the maximum duration of drawdown in days.
 
-    This is the longest period where the portfolio equity was below its peak.
+    This is the longest period (in calendar days) where the portfolio
+    equity was below its peak.
 
     Args:
         snapshots (List[PortfolioSnapshot]): List of portfolio snapshots
 
     Returns:
-        int: The maximum drawdown duration in days.
+        int: The maximum drawdown duration in calendar days.
     """
     equity_curve = get_equity_curve(snapshots)
     if not equity_curve:
         return 0
 
     peak = equity_curve[0][0]
     max_duration = 0
-    current_duration = 0
+    drawdown_start = None
 
-    for equity, _ in equity_curve:
+    for equity, timestamp in equity_curve:
         if equity < peak:
-            current_duration += 1
+            # Entering or continuing a drawdown
+            if drawdown_start is None:
+                drawdown_start = timestamp
         else:
-            max_duration = max(max_duration, current_duration)
-            current_duration = 0
-            peak = equity  # Reset peak to current equity
+            # Recovered to or above the peak
+            if drawdown_start is not None:
+                elapsed = (timestamp - drawdown_start).days
+                max_duration = max(max_duration, elapsed)
+                drawdown_start = None
+            peak = equity
 
-    max_duration = max(max_duration, current_duration)  # Final check
+    # If still in drawdown at the end of the series
+    if drawdown_start is not None and len(equity_curve) > 0:
+        last_timestamp = equity_curve[-1][1]
+        elapsed = (last_timestamp - drawdown_start).days
+        max_duration = max(max_duration, elapsed)
 
     return max_duration
 

investing_algorithm_framework/services/metrics/equity_curve.py

@@ -21,4 +21,7 @@ def get_equity_curve(
         total_size = snapshot.total_value
         series.append((total_size, timestamp))
 
+    # Sort by timestamp to ensure chronological order
+    series.sort(key=lambda x: x[1])
+
     return series

tests/domain/models/test_time_frame.py

@@ -0,0 +1,169 @@
+from unittest import TestCase
+
+from investing_algorithm_framework.domain.models.time_frame import TimeFrame
+
+
+class TestTimeFrameNewEnumValues(TestCase):
+    """Test that the new enum members have the correct string values."""
+
+    def test_twenty_minute_value(self):
+        self.assertEqual(TimeFrame.TWENTY_MINUTE.value, "20m")
+
+    def test_six_hour_value(self):
+        self.assertEqual(TimeFrame.SIX_HOUR.value, "6h")
+
+    def test_eight_hour_value(self):
+        self.assertEqual(TimeFrame.EIGHT_HOUR.value, "8h")
+
+    def test_three_day_value(self):
+        self.assertEqual(TimeFrame.THREE_DAY.value, "3d")
+
+
+class TestTimeFrameNewAmountOfMinutes(TestCase):
+    """Test that amount_of_minutes returns correct values for new members."""
+
+    def test_twenty_minute_minutes(self):
+        self.assertEqual(TimeFrame.TWENTY_MINUTE.amount_of_minutes, 20)
+
+    def test_six_hour_minutes(self):
+        self.assertEqual(TimeFrame.SIX_HOUR.amount_of_minutes, 360)
+
+    def test_eight_hour_minutes(self):
+        self.assertEqual(TimeFrame.EIGHT_HOUR.amount_of_minutes, 480)
+
+    def test_three_day_minutes(self):
+        self.assertEqual(TimeFrame.THREE_DAY.amount_of_minutes, 4320)
+
+
+class TestTimeFrameNewFromString(TestCase):
+    """Test from_string parsing for new members, including case variants."""
+
+    def test_twenty_minute_lowercase(self):
+        self.assertEqual(TimeFrame.from_string("20m"), TimeFrame.TWENTY_MINUTE)
+
+    def test_six_hour_lowercase(self):
+        self.assertEqual(TimeFrame.from_string("6h"), TimeFrame.SIX_HOUR)
+
+    def test_six_hour_uppercase(self):
+        self.assertEqual(TimeFrame.from_string("6H"), TimeFrame.SIX_HOUR)
+
+    def test_eight_hour_lowercase(self):
+        self.assertEqual(TimeFrame.from_string("8h"), TimeFrame.EIGHT_HOUR)
+
+    def test_eight_hour_uppercase(self):
+        self.assertEqual(TimeFrame.from_string("8H"), TimeFrame.EIGHT_HOUR)
+
+    def test_three_day_lowercase(self):
+        self.assertEqual(TimeFrame.from_string("3d"), TimeFrame.THREE_DAY)
+
+    def test_three_day_uppercase(self):
+        self.assertEqual(TimeFrame.from_string("3D"), TimeFrame.THREE_DAY)
+
+
+class TestTimeFrameNewFromValue(TestCase):
+    """Test from_value parsing for new members."""
+
+    def test_twenty_minute_from_value_string(self):
+        self.assertEqual(
+            TimeFrame.from_value("20m"), TimeFrame.TWENTY_MINUTE
+        )
+
+    def test_six_hour_from_value_string(self):
+        self.assertEqual(TimeFrame.from_value("6h"), TimeFrame.SIX_HOUR)
+
+    def test_eight_hour_from_value_string(self):
+        self.assertEqual(TimeFrame.from_value("8h"), TimeFrame.EIGHT_HOUR)
+
+    def test_three_day_from_value_string(self):
+        self.assertEqual(TimeFrame.from_value("3d"), TimeFrame.THREE_DAY)
+
+    def test_twenty_minute_from_value_enum(self):
+        self.assertEqual(
+            TimeFrame.from_value(TimeFrame.TWENTY_MINUTE),
+            TimeFrame.TWENTY_MINUTE,
+        )
+
+    def test_six_hour_from_value_enum(self):
+        self.assertEqual(
+            TimeFrame.from_value(TimeFrame.SIX_HOUR), TimeFrame.SIX_HOUR
+        )
+
+    def test_eight_hour_from_value_enum(self):
+        self.assertEqual(
+            TimeFrame.from_value(TimeFrame.EIGHT_HOUR), TimeFrame.EIGHT_HOUR
+        )
+
+    def test_three_day_from_value_enum(self):
+        self.assertEqual(
+            TimeFrame.from_value(TimeFrame.THREE_DAY), TimeFrame.THREE_DAY
+        )
+
+
+class TestTimeFrameNewEquals(TestCase):
+    """Test equals method for new members against string and enum values."""
+
+    def test_twenty_minute_equals_string(self):
+        self.assertTrue(TimeFrame.TWENTY_MINUTE.equals("20m"))
+
+    def test_twenty_minute_equals_enum(self):
+        self.assertTrue(TimeFrame.TWENTY_MINUTE.equals(TimeFrame.TWENTY_MINUTE))
+
+    def test_twenty_minute_not_equals_other(self):
+        self.assertFalse(TimeFrame.TWENTY_MINUTE.equals(TimeFrame.THIRTY_MINUTE))
+
+    def test_six_hour_equals_string(self):
+        self.assertTrue(TimeFrame.SIX_HOUR.equals("6h"))
+
+    def test_six_hour_equals_enum(self):
+        self.assertTrue(TimeFrame.SIX_HOUR.equals(TimeFrame.SIX_HOUR))
+
+    def test_eight_hour_equals_string(self):
+        self.assertTrue(TimeFrame.EIGHT_HOUR.equals("8h"))
+
+    def test_eight_hour_equals_enum(self):
+        self.assertTrue(TimeFrame.EIGHT_HOUR.equals(TimeFrame.EIGHT_HOUR))
+
+    def test_three_day_equals_string(self):
+        self.assertTrue(TimeFrame.THREE_DAY.equals("3d"))
+
+    def test_three_day_equals_enum(self):
+        self.assertTrue(TimeFrame.THREE_DAY.equals(TimeFrame.THREE_DAY))
+
+    def test_three_day_not_equals_other(self):
+        self.assertFalse(TimeFrame.THREE_DAY.equals(TimeFrame.ONE_DAY))
+
+
+class TestTimeFrameNewOrdering(TestCase):
+    """Test ordering of new members relative to their neighbors."""
+
+    # TWENTY_MINUTE sits between FIFTEEN_MINUTE and THIRTY_MINUTE
+    def test_fifteen_lt_twenty_minute(self):
+        self.assertLess(TimeFrame.FIFTEEN_MINUTE, TimeFrame.TWENTY_MINUTE)
+
+    def test_twenty_lt_thirty_minute(self):
+        self.assertLess(TimeFrame.TWENTY_MINUTE, TimeFrame.THIRTY_MINUTE)
+
+    # SIX_HOUR sits between FOUR_HOUR and EIGHT_HOUR
+    def test_four_hour_lt_six_hour(self):
+        self.assertLess(TimeFrame.FOUR_HOUR, TimeFrame.SIX_HOUR)
+
+    def test_six_hour_lt_eight_hour(self):
+        self.assertLess(TimeFrame.SIX_HOUR, TimeFrame.EIGHT_HOUR)
+
+    # EIGHT_HOUR sits between SIX_HOUR and TWELVE_HOUR
+    def test_eight_hour_lt_twelve_hour(self):
+        self.assertLess(TimeFrame.EIGHT_HOUR, TimeFrame.TWELVE_HOUR)
+
+    # THREE_DAY sits between ONE_DAY and ONE_WEEK
+    def test_one_day_lt_three_day(self):
+        self.assertLess(TimeFrame.ONE_DAY, TimeFrame.THREE_DAY)
+
+    def test_three_day_lt_one_week(self):
+        self.assertLess(TimeFrame.THREE_DAY, TimeFrame.ONE_WEEK)
+
+    # Verify >= and <= also work
+    def test_twenty_minute_le_thirty_minute(self):
+        self.assertLessEqual(TimeFrame.TWENTY_MINUTE, TimeFrame.THIRTY_MINUTE)
+
+    def test_eight_hour_ge_six_hour(self):
+        self.assertGreaterEqual(TimeFrame.EIGHT_HOUR, TimeFrame.SIX_HOUR)