Fix report formatting

Author: MDUYN

investing_algorithm_framework/app/analysis/__init__.py

@@ -0,0 +1,5 @@
+from .backtest_data_ranges import select_backtest_date_ranges
+
+__all__ = [
+    "select_backtest_date_ranges"
+]

investing_algorithm_framework/app/analysis/backtest_data_ranges.py

@@ -0,0 +1,96 @@
+import pandas as pd
+from investing_algorithm_framework.domain import BacktestDateRange, \
+    OperationalException
+from typing import List, Union
+
+
+def select_backtest_date_ranges(
+    df: pd.DataFrame, window: Union[str, int] = '365D'
+) -> List[BacktestDateRange]:
+    """
+    Identifies the best upturn, worst downturn, and sideways periods
+    for the given window duration. This allows you to quickly select
+    interesting periods for backtesting.
+    """
+    df = df.copy()
+    df = df.sort_index()
+
+    if isinstance(window, int):
+        window = pd.Timedelta(days=window)
+    elif isinstance(window, str):
+        window = pd.to_timedelta(window)
+    else:
+        raise OperationalException("window must be a string or integer")
+
+    if len(df) < 2 or df.index[-1] - df.index[0] < window:
+        raise OperationalException(
+            "DataFrame must contain at least two rows and span "
+            "the full window duration"
+        )
+
+    best_upturn = {
+        "name": "UpTurn", "return": float('-inf'), "start": None, "end": None
+    }
+    worst_downturn = {
+        "name": "DownTurn", "return": float('inf'), "start": None, "end": None
+    }
+    most_sideways = {
+        "name": "SideWays",
+        "volatility": float('inf'),
+        "return": None,
+        "start": None,
+        "end": None
+    }
+
+    for i in range(len(df)):
+        start_time = df.index[i]
+        end_time = start_time + window
+        window_df = df[(df.index >= start_time) & (df.index <= end_time)]
+
+        if len(window_df) < 2 or (window_df.index[-1] - start_time) < window:
+            continue
+
+        start_price = window_df['Close'].iloc[0]
+        end_price = window_df['Close'].iloc[-1]
+        ret = (end_price / start_price) - 1  # relative return
+        volatility = window_df['Close'].std()
+
+        # Ensure datetime for BacktestDateRange
+        start_time = pd.Timestamp(start_time).to_pydatetime()
+        end_time = pd.Timestamp(window_df.index[-1]).to_pydatetime()
+
+        if ret > best_upturn["return"]:
+            best_upturn.update(
+                {"return": ret, "start": start_time, "end": end_time}
+            )
+
+        if ret < worst_downturn["return"]:
+            worst_downturn.update(
+                {"return": ret, "start": start_time, "end": end_time}
+            )
+
+        if volatility < most_sideways["volatility"]:
+            most_sideways.update({
+                "return": ret,
+                "volatility": volatility,
+                "start": start_time,
+                "end": end_time
+            })
+
+    return [
+        BacktestDateRange(
+            start_date=best_upturn['start'],
+            end_date=best_upturn['end'],
+            name=best_upturn['name']
+        ),
+        BacktestDateRange(
+            start_date=worst_downturn['start'],
+            end_date=worst_downturn['end'],
+            name=worst_downturn['name']
+        ),
+        BacktestDateRange(
+            start_date=most_sideways['start'],
+            end_date=most_sideways['end'],
+            name=most_sideways['name']
+        )
+    ]

investing_algorithm_framework/app/reporting/charts/ohlcv_data_completeness.py

@@ -0,0 +1,51 @@
+import plotly.graph_objects as go
+import pandas as pd
+import plotly.io as pio
+
+def get_ohlcv_data_completeness_chart(
+    df,
+    timeframe='1min',
+    windowsize=100,
+    title="OHLCV Data completenes"
+):
+    df = df.copy()
+    df['Datetime'] = pd.to_datetime(df['Datetime'])
+    df = df.sort_values('Datetime').tail(windowsize)
+    start = df['Datetime'].iloc[0]
+    end = df['Datetime'].iloc[-1]
+    freq = pd.to_timedelta(timeframe)
+    expected = pd.date_range(start, end, freq=freq)
+    actual = df['Datetime']
+    missing = expected.difference(actual)
+
+    # Calculte the percentage completeness
+    completeness = len(actual) / len(expected) * 100
+    title += f" ({completeness:.2f}% complete)"
+    fig = go.Figure()
+    fig.add_trace(
+        go.Scatter(
+            x=actual,
+            y=[1]*len(actual),
+            mode='markers',
+            name='Present',
+            marker=dict(color='green', size=6)
+        )
+    )
+    fig.add_trace(
+        go.Scatter(
+            x=missing,
+            y=[1]*len(missing),
+            mode='markers',
+            name='Missing',
+            marker=dict(color='red', size=6, symbol='x')
+        )
+    )
+    fig.update_layout(
+        title=title,
+        xaxis_title='Datetime',
+        yaxis=dict(showticklabels=False),
+        height=300,
+        showlegend=True
+    )
+
+    return pio.to_html(fig, full_html=False, include_plotlyjs='cdn')

tests/app/reporting/metrics/test_cagr.py

@@ -0,0 +1,87 @@
+import unittest
+from datetime import datetime, timedelta
+from unittest.mock import MagicMock
+
+from investing_algorithm_framework import get_cagr
+
+
+# Mock Snapshot class
+class Snapshot:
+    def __init__(self, total_value, created_at):
+        self.total_value = total_value
+        self.created_at = created_at
+
+class TestGetCagr(unittest.TestCase):
+
+    def create_report(self, prices, start_date):
+        """ Helper to create a mocked BacktestReport from a price list """
+        snapshots = [
+            Snapshot(net_size, start_date + timedelta(weeks=i))
+            for i, net_size in enumerate(prices)
+        ]
+        report = MagicMock()
+        report.get_snapshots.return_value = snapshots
+        return report
+
+    def test_cagr_for_return_less_then_a_year(self):
+        """
+        Test a scenario where the CAGR is calculated 
+        for a period less than a year (3 months). Given that the 
+        portfolio return 12% annually, 
+        the CAGR should be approximately 3% for 3 months.
+        """
+        # Convert annualized return to approximate weekly returns
+        weekly_return = (1 + 0.12) ** (1 / 52) - 1  # ≈ 0.0022
+        weeks = 13  # 3 months ≈ 13 weeks
+
+        # Generate weekly prices (cumulative compounding)
+        start_price = 100
+        prices = [start_price * ((1 + weekly_return) ** i) for i in range(weeks + 1)]
+
+        start_date = datetime(2022, 1, 1)
+        report_x = self.create_report(prices, start_date)
+
+        cagr = get_cagr(report_x)
+        self.assertAlmostEqual(cagr, 0.12034875793587707, delta=1)
+
+    def test_cagr_for_return_exactly_a_year(self):
+        """
+        Test a scenario where the CAGR is calculated
+        for a period less than a year (3 months). Given that the
+        portfolio return 12% annually,
+        the CAGR should be approximately 3% for 3 months.
+        """
+        # Convert annualized return to approximate weekly returns
+        weekly_return = (1 + 0.12) ** (1 / 52) - 1  # ≈ 0.0022
+        weeks = 52
+
+        # Generate weekly prices (cumulative compounding)
+        start_price = 100
+        prices = [start_price * ((1 + weekly_return) ** i) for i in range(weeks + 1)]
+
+        start_date = datetime(2022, 1, 1)
+        report_x = self.create_report(prices, start_date)
+
+        cagr = get_cagr(report_x)
+        self.assertAlmostEqual(cagr, 0.12034875793587663, delta=1)
+
+    def test_cagr_for_return_more_then_a_year(self):
+        """
+        Test a scenario where the CAGR is calculated
+        for a period less than a year (3 months). Given that the
+        portfolio return 12% annually,
+        the CAGR should be approximately 3% for 3 months.
+        """
+        # Convert annualized return to approximate weekly returns
+        weekly_return = (1 + 0.12) ** (1 / 52) - 1  # ≈ 0.0022
+        weeks = 73
+
+        # Generate weekly prices (cumulative compounding)
+        start_price = 100
+        prices = [start_price * ((1 + weekly_return) ** i) for i in range(weeks + 1)]
+
+        start_date = datetime(2022, 1, 1)
+        report_x = self.create_report(prices, start_date)
+
+        cagr = get_cagr(report_x)
+        self.assertAlmostEqual(cagr, 0.1203487579358764, delta=1)

tests/app/reporting/metrics/test_calmer_ratio.py

@@ -0,0 +1,68 @@
+import unittest
+from datetime import datetime
+from unittest.mock import patch, MagicMock
+from investing_algorithm_framework.domain import BacktestReport
+
+from investing_algorithm_framework import get_calmar_ratio
+
+
+class TestGetCalmarRatio(unittest.TestCase):
+
+    def setUp(self):
+        # Generate mocked equity curve: net_size over time
+        self.timestamps = [
+            datetime(2024, 1, 1),
+            datetime(2024, 1, 2),
+            datetime(2024, 1, 3),
+            datetime(2024, 1, 4),
+            datetime(2024, 1, 5),
+        ]
+
+        self.net_sizes = [1000, 1200, 900, 1100, 1300]  # Simulates rise, fall, recovery, new high
+
+        # Create mock snapshot objects
+        self.snapshots = []
+        for ts, net_size in zip(self.timestamps, self.net_sizes):
+            snapshot = MagicMock()
+            snapshot.created_at = ts
+            snapshot.total_value = net_size
+            self.snapshots.append(snapshot)
+
+        # Create a mocked BacktestReport
+        self.backtest_report = MagicMock()
+        self.backtest_report.get_snapshots.return_value = self.snapshots
+
+    def _create_report(self, total_size_series, timestamps):
+        report = MagicMock(spec=BacktestReport)
+        report.get_snapshots.return_value = [
+            MagicMock(created_at=ts, total_value=size)
+            for ts, size in zip(timestamps, total_size_series)
+        ]
+        return report
+
+    def test_typical_case(self):
+        # Create a report with total sizes for a whole year, with intervals of 31 days.
+        report = self._create_report(
+            [1000, 1200, 900, 1100, 1300],
+            [datetime(2024, i, 1) for i in range(1, 6)]
+        )
+        ratio = get_calmar_ratio(report)
+        self.assertEqual(ratio, 4.8261927891975365)  # Expected ratio based on the mock data
+
+    def test_calmar_ratio_zero_drawdown(self):
+        # Create a report with total sizes for a whole year, with intervals of 31 days, and no drawdowns.
+        report = self._create_report(
+            [1000, 1200, 1300, 1400, 1500, 1600, 1700, 1800, 1900, 2000],
+            [datetime(2024, 1, i) for i in range(1, 11)]
+        )
+        ratio = get_calmar_ratio(report)
+        self.assertEqual(ratio, 0.0)
+
+    def test_calmar_ratio_with_only_drawdown(self):
+        # Create a report with total sizes for a whole year, with intervals of 31 days, and no drawdowns.
+        report = self._create_report(
+            [1000, 900, 800, 700, 600, 500, 400, 300, 200, 100],
+            [datetime(2024, 1, i) for i in range(1, 11)]
+        )
+        ratio = get_calmar_ratio(report)
+        self.assertEqual(ratio, -1.1111111111111112)

tests/app/reporting/metrics/test_drawdowns.py

@@ -0,0 +1,57 @@
+import unittest
+from datetime import datetime, timedelta
+from unittest.mock import MagicMock
+from investing_algorithm_framework import get_drawdown_series, \
+    get_max_drawdown, get_max_drawdown_absolute
+
+
+class TestDrawdownFunctions(unittest.TestCase):
+
+    def setUp(self):
+        # Generate mocked equity curve: net_size over time
+        self.timestamps = [
+            datetime(2024, 1, 1),
+            datetime(2024, 1, 2),
+            datetime(2024, 1, 3),
+            datetime(2024, 1, 4),
+            datetime(2024, 1, 5),
+        ]
+
+        self.net_sizes = [1000, 1200, 900, 1100, 1300]  # Simulates rise, fall, recovery, new high
+
+        # Create mock snapshot objects
+        self.snapshots = []
+        for ts, net_size in zip(self.timestamps, self.net_sizes):
+            snapshot = MagicMock()
+            snapshot.created_at = ts
+            snapshot.total_value = net_size
+            self.snapshots.append(snapshot)
+
+        # Create a mocked BacktestReport
+        self.backtest_report = MagicMock()
+        self.backtest_report.get_snapshots.return_value = self.snapshots
+
+    def test_drawdown_series(self):
+        drawdown_series = get_drawdown_series(self.backtest_report)
+
+        expected_drawdowns = [
+            0.0,                                # baseline
+            0.0,                                # new high
+            (900 - 1200) / 1200,                # drop from peak
+            (1100 - 1200) / 1200,               # partial recovery
+            0.0                                 # new peak again
+        ]
+
+        self.assertEqual(len(drawdown_series), len(expected_drawdowns))
+
+        for i, (_, drawdown) in enumerate(drawdown_series):
+            self.assertAlmostEqual(drawdown, expected_drawdowns[i], places=6)
+
+    def test_max_drawdown(self):
+        max_drawdown = get_max_drawdown(self.backtest_report)
+        expected_max = abs((900 - 1200) / 1200) # 0.25
+        self.assertAlmostEqual(max_drawdown, expected_max, places=6)
+
+    def test_max_drawdown_absolute(self):
+        max_drawdown = get_max_drawdown_absolute(self.backtest_report)
+        self.assertEqual(max_drawdown, 300)  # 1200 - 900 = 300