Source code for backtrader.analyzers.annualreturn

#!/usr/bin/env python
"""Annual Return Analyzer Module - Annual return calculation.

This module provides the AnnualReturn analyzer for calculating
year-by-year returns of a strategy.

Classes:
    AnnualReturn: Analyzer that calculates annual returns.

Example:
    >>> cerebro = bt.Cerebro()
    >>> cerebro.addanalyzer(bt.analyzers.AnnualReturn, _name='annret')
    >>> results = cerebro.run()
    >>> print(results[0].analyzers.annret.get_analysis())
"""

import math
from collections import OrderedDict

from ..analyzer import Analyzer
from ..utils.date import num2date
from ..utils.log_message import get_logger
from ..utils.py3 import range

logger = get_logger(__name__)


# Calculate annual returns. The algorithm implementation is somewhat complex, so a pandas-based version MyAnnualReturn was written later with much simpler logic


class AnnualReturn(Analyzer):
    """
    This analyzer calculates the AnnualReturns by looking at the beginning
    and end of the year

    Params:

      - (None)

    Member Attributes:

      - ``rets``: list of calculated annual returns

      - ``ret``: dictionary (key: year) of annual returns

    **get_analysis**:

      - Returns a dictionary of annual returns (key: year)
    """



    def __init__(self):
        """Initialize the AnnualReturn analyzer.

        Initializes cache lists for storing dates and values during backtesting.
        """
        super().__init__()
        # Cache data
        self._dt_cache = []
        self._value_cache = []




    def next(self):
        """Cache current date and account value on each bar.

        Stores the current datetime and portfolio value for later
        annual return calculation.
        """
        # Cache current date and account value each time next is called
        dt_val = self.data.datetime[0]
        value_val = self.strategy.broker.getvalue()
        self._dt_cache.append(dt_val)
        self._value_cache.append(value_val)




    def stop(self):
        """Calculate annual returns from cached data.

        Iterates through cached date-value pairs to calculate returns
        for each calendar year. Stores results in self.rets (list) and
        self.ret (dictionary keyed by year).
        """
        # Must have stats.broker
        # Current year
        cur_year = -1
        # Start value
        value_start = 0.0
        # End value
        value_end = 0.0
        # Save return data
        setattr(self, "rets", [])
        setattr(self, "ret", OrderedDict())

        # Calculate using cached data
        for i in range(len(self._dt_cache)):
            dt_val = self._dt_cache[i]
            value_cur = self._value_cache[i]

            # Convert date
            try:
                dt = num2date(dt_val)
            except (ValueError, TypeError, OverflowError) as e:
                logger.debug("Failed to convert date value %s: %s", dt_val, e)
                continue

            # If the year at index i is greater than current year, if current year > 0, calculate return and save to self.ret, and start value equals end value
            # When years are not equal, it indicates current i is a new year
            if dt.year > cur_year:
                if cur_year >= 0:
                    annual_ret = self._safe_annual_return(value_start, value_end)
                    self.rets.append(annual_ret)
                    self.ret[cur_year] = annual_ret

                    # changing between real years, use last value as new start
                    value_start = value_end
                else:
                    # No value set whatsoever, use the currently loaded value
                    value_start = value_cur

                cur_year = dt.year

            # No matter what, the last value is always the last loaded value
            value_end = value_cur
        # If current year hasn't ended and return hasn't been calculated, calculate at the end even if less than a full year
        if cur_year >= 0 and cur_year not in self.ret:
            # finish calculating pending data
            annual_ret = self._safe_annual_return(value_start, value_end)
            self.rets.append(annual_ret)
            self.ret[cur_year] = annual_ret


    @staticmethod
    def _safe_annual_return(value_start, value_end):
        """Compute (value_end/value_start - 1), returning 0.0 for any invalid
        (zero/NaN/inf/complex) inputs or result. Shared by the year-boundary
        and final-pending paths in stop()."""
        try:
            valid_values = (
                value_start != 0 and math.isfinite(value_start) and math.isfinite(value_end)
            )
        except TypeError:
            valid_values = False
        if not valid_values:
            return 0.0
        try:
            annual_ret = (value_end / value_start) - 1.0
            if isinstance(annual_ret, complex) or not math.isfinite(annual_ret):
                return 0.0
            return annual_ret
        except (ZeroDivisionError, TypeError, ValueError):
            return 0.0



    def get_analysis(self):
        """Return the annual return analysis results.

        Returns:
            OrderedDict: Dictionary mapping years to their annual returns.
        """
        return self.ret






class MyAnnualReturn(Analyzer):
    """
    This analyzer calculates the AnnualReturns by looking at the beginning
    and end of the year

    Params:

      - (None)

    Member Attributes:

      - ``rets``: list of calculated annual returns

      - ``ret``: dictionary (key: year) of annual returns

    **get_analysis**:

      - Returns a dictionary of annual returns (key: year)
    """



    def stop(self):
        """Calculate annual returns using pandas.

        Uses pandas DataFrame operations to group data by year and
        calculate annual returns based on beginning and ending values
        for each year.

        Note:
            This method requires pandas to be installed.
        """
        # Container for saving data - dictionary
        if not hasattr(self, "ret"):
            setattr(self, "ret", OrderedDict())
        # Get data time and convert to date
        dt_list = self.data.datetime.get(0, size=len(self.data))
        dt_list = [num2date(i) for i in dt_list]
        # Get account assets
        value_list = self.strategy.stats.broker.value.get(0, size=len(self.data))
        # Convert to pandas format
        import pandas as pd

        df = pd.DataFrame([dt_list, value_list]).T
        df.columns = ["datetime", "value"]
        df["pre_value"] = df["value"].shift(1)
        # Calculate simple returns for each year
        df["year"] = [i.year for i in df["datetime"]]
        for year, data in df.groupby("year"):
            begin_value = list(data["pre_value"])[0]
            end_value = list(data["value"])[-1]
            try:
                valid_values = math.isfinite(begin_value) and math.isfinite(end_value)
            except TypeError:
                valid_values = False
            if not valid_values or begin_value == 0:
                annual_return = 0.0
            else:
                try:
                    annual_return = (end_value / begin_value) - 1
                    if isinstance(annual_return, complex) or not math.isfinite(annual_return):
                        annual_return = 0.0
                except (ZeroDivisionError, TypeError, ValueError):
                    annual_return = 0.0
            self.ret[year] = annual_return




    def get_analysis(self):
        """Return the annual return analysis results.

        Returns:
            OrderedDict: Dictionary mapping years to their annual returns.
        """
        return self.ret