zipline.finance.risk.information_ratio() - Code Metrics - Inspection of "REF: Remove unused parameter." - quantopian/zipline - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( e4fe44...6106cb )

by Eddie

created 2015-12-14 19:26 UTC

zipline.finance.risk.information_ratio() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	2
dl	0
loc	25
rs	8.8571

#
# Copyright 2014 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import functools
import logbook
import math
import numpy as np

import zipline.utils.math_utils as zp_math

import pandas as pd
from pandas.tseries.tools import normalize_date

from six import iteritems

from . risk import (
    alpha,
    check_entry,
    choose_treasury,
    downside_risk,
    sharpe_ratio,
    sortino_ratio,
)

from zipline.utils.serialization_utils import (
    VERSION_LABEL
)

log = logbook.Logger('Risk Cumulative')


choose_treasury = functools.partial(choose_treasury, lambda *args: '10year',
                                    compound=False)


def information_ratio(algo_volatility, algorithm_return, benchmark_return):
    """
    http://en.wikipedia.org/wiki/Information_ratio

    Args:
        algorithm_returns (np.array-like):
            All returns during algorithm lifetime.
        benchmark_returns (np.array-like):
            All benchmark returns during algo lifetime.

    Returns:
        float. Information ratio.
    """
    if zp_math.tolerant_equals(algo_volatility, 0):
        return np.nan

    # The square of the annualization factor is in the volatility,
    # because the volatility is also annualized,
    # i.e. the sqrt(annual factor) is in the volatility's numerator.
    # So to have the the correct annualization factor for the
    # Sharpe value's numerator, which should be the sqrt(annual factor).
    # The square of the sqrt of the annual factor, i.e. the annual factor
    # itself, is needed in the numerator to factor out the division by
    # its square root.
    return (algorithm_return - benchmark_return) / algo_volatility


class RiskMetricsCumulative(object):
    """
    :Usage:
        Instantiate RiskMetricsCumulative once.
        Call update() method on each dt to update the metrics.
    """

    METRIC_NAMES = (
        'alpha',
        'beta',
        'sharpe',
        'algorithm_volatility',
        'benchmark_volatility',
        'downside_risk',
        'sortino',
        'information',
    )

    def __init__(self, sim_params, env,
                 create_first_day_stats=False,
                 account=None):
        self.treasury_curves = env.treasury_curves
        self.start_date = sim_params.period_start.replace(
            hour=0, minute=0, second=0, microsecond=0
        )
        self.end_date = sim_params.period_end.replace(
            hour=0, minute=0, second=0, microsecond=0
        )

        self.trading_days = env.days_in_range(self.start_date, self.end_date)

        # Hold on to the trading day before the start,
        # used for index of the zero return value when forcing returns
        # on the first day.
        self.day_before_start = self.start_date - env.trading_days.freq

        last_day = normalize_date(sim_params.period_end)
        if last_day not in self.trading_days:
            last_day = pd.tseries.index.DatetimeIndex(
                [last_day]
            )
            self.trading_days = self.trading_days.append(last_day)

        self.sim_params = sim_params
        self.env = env

        self.create_first_day_stats = create_first_day_stats

        cont_index = self.trading_days

        self.cont_index = cont_index
        self.cont_len = len(self.cont_index)

        empty_cont = np.full(self.cont_len, np.nan)

        self.algorithm_returns_cont = empty_cont.copy()
        self.benchmark_returns_cont = empty_cont.copy()
        self.algorithm_cumulative_leverages_cont = empty_cont.copy()
        self.mean_returns_cont = empty_cont.copy()
        self.annualized_mean_returns_cont = empty_cont.copy()
        self.mean_benchmark_returns_cont = empty_cont.copy()
        self.annualized_mean_benchmark_returns_cont = empty_cont.copy()

        # The returns at a given time are read and reset from the respective
        # returns container.
        self.algorithm_returns = None
        self.benchmark_returns = None
        self.mean_returns = None
        self.annualized_mean_returns = None
        self.mean_benchmark_returns = None
        self.annualized_mean_benchmark_returns = None

        self.algorithm_cumulative_returns = empty_cont.copy()
        self.benchmark_cumulative_returns = empty_cont.copy()
        self.algorithm_cumulative_leverages = empty_cont.copy()
        self.excess_returns = empty_cont.copy()

        self.latest_dt_loc = 0
        self.latest_dt = cont_index[0]

        self.benchmark_volatility = empty_cont.copy()
        self.algorithm_volatility = empty_cont.copy()
        self.beta = empty_cont.copy()
        self.alpha = empty_cont.copy()
        self.sharpe = empty_cont.copy()
        self.downside_risk = empty_cont.copy()
        self.sortino = empty_cont.copy()
        self.information = empty_cont.copy()

        self.drawdowns = empty_cont.copy()
        self.max_drawdowns = empty_cont.copy()
        self.max_drawdown = 0
        self.max_leverages = empty_cont.copy()
        self.max_leverage = 0
        self.current_max = -np.inf
        self.daily_treasury = pd.Series(index=self.trading_days)
        self.treasury_period_return = np.nan

        self.num_trading_days = 0

    def update(self, dt, algorithm_returns, benchmark_returns, account):
        # Keep track of latest dt for use in to_dict and other methods
        # that report current state.
        self.latest_dt = dt
        dt_loc = self.cont_index.get_loc(dt)
        self.latest_dt_loc = dt_loc

        self.algorithm_returns_cont[dt_loc] = algorithm_returns
        self.algorithm_returns = self.algorithm_returns_cont[:dt_loc + 1]

        self.num_trading_days = len(self.algorithm_returns)

        if self.create_first_day_stats:
            if len(self.algorithm_returns) == 1:
                self.algorithm_returns = np.append(0.0, self.algorithm_returns)

        self.algorithm_cumulative_returns[dt_loc] = \
            self.calculate_cumulative_returns(self.algorithm_returns)

        algo_cumulative_returns_to_date = \
            self.algorithm_cumulative_returns[:dt_loc + 1]

        self.mean_returns_cont[dt_loc] = \
            algo_cumulative_returns_to_date[dt_loc] / self.num_trading_days

        self.mean_returns = self.mean_returns_cont[:dt_loc + 1]

        self.annualized_mean_returns_cont[dt_loc] = \
            self.mean_returns_cont[dt_loc] * 252

        self.annualized_mean_returns = \
            self.annualized_mean_returns_cont[:dt_loc + 1]

        if self.create_first_day_stats:
            if len(self.mean_returns) == 1:
                self.mean_returns = np.append(0.0, self.mean_returns)
                self.annualized_mean_returns = np.append(
                    0.0, self.annualized_mean_returns)

        self.benchmark_returns_cont[dt_loc] = benchmark_returns
        self.benchmark_returns = self.benchmark_returns_cont[:dt_loc + 1]

        if self.create_first_day_stats:
            if len(self.benchmark_returns) == 1:
                self.benchmark_returns = np.append(0.0, self.benchmark_returns)

        self.benchmark_cumulative_returns[dt_loc] = \
            self.calculate_cumulative_returns(self.benchmark_returns)

        benchmark_cumulative_returns_to_date = \
            self.benchmark_cumulative_returns[:dt_loc + 1]

        self.mean_benchmark_returns_cont[dt_loc] = \
            benchmark_cumulative_returns_to_date[dt_loc] / \
            self.num_trading_days

        self.mean_benchmark_returns = self.mean_benchmark_returns_cont[:dt_loc]

        self.annualized_mean_benchmark_returns_cont[dt_loc] = \
            self.mean_benchmark_returns_cont[dt_loc] * 252

        self.annualized_mean_benchmark_returns = \
            self.annualized_mean_benchmark_returns_cont[:dt_loc + 1]

        self.algorithm_cumulative_leverages_cont[dt_loc] = account['leverage']
        self.algorithm_cumulative_leverages = \
            self.algorithm_cumulative_leverages_cont[:dt_loc + 1]

        if self.create_first_day_stats:
            if len(self.algorithm_cumulative_leverages) == 1:
                self.algorithm_cumulative_leverages = np.append(
                    0.0,
                    self.algorithm_cumulative_leverages)

        if not len(self.algorithm_returns) and len(self.benchmark_returns):
            message = "Mismatch between benchmark_returns ({bm_count}) and \
algorithm_returns ({algo_count}) in range {start} : {end} on {dt}"
            message = message.format(
                bm_count=len(self.benchmark_returns),
                algo_count=len(self.algorithm_returns),
                start=self.start_date,
                end=self.end_date,
                dt=dt
            )
            raise Exception(message)

        self.update_current_max()
        self.benchmark_volatility[dt_loc] = \
            self.calculate_volatility(self.benchmark_returns)
        self.algorithm_volatility[dt_loc] = \
            self.calculate_volatility(self.algorithm_returns)

        # caching the treasury rates for the minutely case is a
        # big speedup, because it avoids searching the treasury
        # curves on every minute.
        # In both minutely and daily, the daily curve is always used.
        treasury_end = dt.replace(hour=0, minute=0)
        if np.isnan(self.daily_treasury[treasury_end]):
            treasury_period_return = choose_treasury(
                self.treasury_curves,
                self.start_date,
                treasury_end,
                self.env,
            )
            self.daily_treasury[treasury_end] = treasury_period_return
        self.treasury_period_return = self.daily_treasury[treasury_end]
        self.excess_returns[dt_loc] = (
            self.algorithm_cumulative_returns[dt_loc] -
            self.treasury_period_return)
        self.beta[dt_loc] = self.calculate_beta()
        self.alpha[dt_loc] = self.calculate_alpha()
        self.sharpe[dt_loc] = self.calculate_sharpe()
        self.downside_risk[dt_loc] = \
            self.calculate_downside_risk()
        self.sortino[dt_loc] = self.calculate_sortino()
        self.information[dt_loc] = self.calculate_information()
        self.max_drawdown = self.calculate_max_drawdown()
        self.max_drawdowns[dt_loc] = self.max_drawdown
        self.max_leverage = self.calculate_max_leverage()
        self.max_leverages[dt_loc] = self.max_leverage

    def to_dict(self):
        """
        Creates a dictionary representing the state of the risk report.
        Returns a dict object of the form:
        """
        dt = self.latest_dt
        dt_loc = self.latest_dt_loc
        period_label = dt.strftime("%Y-%m")
        rval = {
            'trading_days': self.num_trading_days,
            'benchmark_volatility':
            self.benchmark_volatility[dt_loc],
            'algo_volatility':
            self.algorithm_volatility[dt_loc],
            'treasury_period_return': self.treasury_period_return,
            # Though the two following keys say period return,
            # they would be more accurately called the cumulative return.
            # However, the keys need to stay the same, for now, for backwards
            # compatibility with existing consumers.
            'algorithm_period_return':
            self.algorithm_cumulative_returns[dt_loc],
            'benchmark_period_return':
            self.benchmark_cumulative_returns[dt_loc],
            'beta': self.beta[dt_loc],
            'alpha': self.alpha[dt_loc],
            'sharpe': self.sharpe[dt_loc],
            'sortino': self.sortino[dt_loc],
            'information': self.information[dt_loc],
            'excess_return': self.excess_returns[dt_loc],
            'max_drawdown': self.max_drawdown,
            'max_leverage': self.max_leverage,
            'period_label': period_label
        }

        return {k: (None if check_entry(k, v) else v)
                for k, v in iteritems(rval)}

    def __repr__(self):
        statements = []
        for metric in self.METRIC_NAMES:
            value = getattr(self, metric)[-1]
            if isinstance(value, list):
                if len(value) == 0:
                    value = np.nan
                else:
                    value = value[-1]
            statements.append("{m}:{v}".format(m=metric, v=value))

        return '\n'.join(statements)

    def calculate_cumulative_returns(self, returns):
        return (1. + returns).prod() - 1

    def update_current_max(self):
        if len(self.algorithm_cumulative_returns) == 0:
            return
        current_cumulative_return = \
            self.algorithm_cumulative_returns[self.latest_dt_loc]
        if self.current_max < current_cumulative_return:
            self.current_max = current_cumulative_return

    def calculate_max_drawdown(self):
        if len(self.algorithm_cumulative_returns) == 0:
            return self.max_drawdown

        # The drawdown is defined as: (high - low) / high
        # The above factors out to: 1.0 - (low / high)
        #
        # Instead of explicitly always using the low, use the current total
        # return value, and test that against the max drawdown, which will
        # exceed the previous max_drawdown iff the current return is lower than
        # the previous low in the current drawdown window.
        cur_drawdown = 1.0 - (
            (1.0 + self.algorithm_cumulative_returns[self.latest_dt_loc])
            /
            (1.0 + self.current_max))

        self.drawdowns[self.latest_dt_loc] = cur_drawdown

        if self.max_drawdown < cur_drawdown:
            return cur_drawdown
        else:
            return self.max_drawdown

    def calculate_max_leverage(self):
        # The leverage is defined as: the gross_exposure/net_liquidation
        # gross_exposure = long_exposure + abs(short_exposure)
        # net_liquidation = ending_cash + long_exposure + short_exposure
        cur_leverage = self.algorithm_cumulative_leverages_cont[
            self.latest_dt_loc]

        return max(cur_leverage, self.max_leverage)

    def calculate_sharpe(self):
        """
        http://en.wikipedia.org/wiki/Sharpe_ratio
        """
        return sharpe_ratio(
            self.algorithm_volatility[self.latest_dt_loc],
            self.annualized_mean_returns_cont[self.latest_dt_loc],
            self.daily_treasury[self.latest_dt.date()])

    def calculate_sortino(self):
        """
        http://en.wikipedia.org/wiki/Sortino_ratio
        """
        return sortino_ratio(
            self.annualized_mean_returns_cont[self.latest_dt_loc],
            self.daily_treasury[self.latest_dt.date()],
            self.downside_risk[self.latest_dt_loc])

    def calculate_information(self):
        """
        http://en.wikipedia.org/wiki/Information_ratio
        """
        return information_ratio(
            self.algorithm_volatility[self.latest_dt_loc],
            self.annualized_mean_returns_cont[self.latest_dt_loc],
            self.annualized_mean_benchmark_returns_cont[self.latest_dt_loc])

    def calculate_alpha(self):
        """
        http://en.wikipedia.org/wiki/Alpha_(investment)
        """
        return alpha(
            self.annualized_mean_returns_cont[self.latest_dt_loc],
            self.treasury_period_return,
            self.annualized_mean_benchmark_returns_cont[self.latest_dt_loc],
            self.beta[self.latest_dt_loc])

    def calculate_volatility(self, daily_returns):
        if len(daily_returns) <= 1:
            return 0.0
        return np.std(daily_returns, ddof=1) * math.sqrt(252)

    def calculate_downside_risk(self):
        return downside_risk(self.algorithm_returns,
                             self.mean_returns,
                             252)

    def calculate_beta(self):
        """

        .. math::

            \\beta_a = \\frac{\mathrm{Cov}(r_a,r_p)}{\mathrm{Var}(r_p)}

        http://en.wikipedia.org/wiki/Beta_(finance)
        """
        # it doesn't make much sense to calculate beta for less than two
        # values, so return none.
        if len(self.algorithm_returns) < 2:
            return 0.0

        returns_matrix = np.vstack([self.algorithm_returns,
                                    self.benchmark_returns])
        C = np.cov(returns_matrix, ddof=1)
        algorithm_covariance = C[0][1]
        benchmark_variance = C[1][1]
        beta = algorithm_covariance / benchmark_variance

        return beta

    def __getstate__(self):
        state_dict = {k: v for k, v in iteritems(self.__dict__)
                      if not k.startswith('_')}

        STATE_VERSION = 3
        state_dict[VERSION_LABEL] = STATE_VERSION

        return state_dict

    def __setstate__(self, state):

        OLDEST_SUPPORTED_STATE = 3
        version = state.pop(VERSION_LABEL)

        if version < OLDEST_SUPPORTED_STATE:
            raise BaseException("RiskMetricsCumulative \
                    saved state is too old.")

        self.__dict__.update(state)


1			#
2			# Copyright 2014 Quantopian, Inc.
3			#
4			# Licensed under the Apache License, Version 2.0 (the "License");
5			# you may not use this file except in compliance with the License.
6			# You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import functools
17			import logbook
18			import math
19			import numpy as np
20
21			import zipline.utils.math_utils as zp_math
22
23			import pandas as pd
24			from pandas.tseries.tools import normalize_date
25
26			from six import iteritems
27
28			from . risk import (
29			alpha,
30			check_entry,
31			choose_treasury,
32			downside_risk,
33			sharpe_ratio,
34			sortino_ratio,
35			)
36
37			from zipline.utils.serialization_utils import (
38			VERSION_LABEL
39			)
40
41			log = logbook.Logger('Risk Cumulative')
42
43
44			choose_treasury = functools.partial(choose_treasury, lambda *args: '10year',
45			compound=False)
46
47
48			def information_ratio(algo_volatility, algorithm_return, benchmark_return):
49			"""
50			http://en.wikipedia.org/wiki/Information_ratio
51
52			Args:
53			algorithm_returns (np.array-like):
54			All returns during algorithm lifetime.
55			benchmark_returns (np.array-like):
56			All benchmark returns during algo lifetime.
57
58			Returns:
59			float. Information ratio.
60			"""
61			if zp_math.tolerant_equals(algo_volatility, 0):
62			return np.nan
63
64			# The square of the annualization factor is in the volatility,
65			# because the volatility is also annualized,
66			# i.e. the sqrt(annual factor) is in the volatility's numerator.
67			# So to have the the correct annualization factor for the
68			# Sharpe value's numerator, which should be the sqrt(annual factor).
69			# The square of the sqrt of the annual factor, i.e. the annual factor
70			# itself, is needed in the numerator to factor out the division by
71			# its square root.
72			return (algorithm_return - benchmark_return) / algo_volatility
73
74
75			class RiskMetricsCumulative(object):
76			"""
77			:Usage:
78			Instantiate RiskMetricsCumulative once.
79			Call update() method on each dt to update the metrics.
80			"""
81
82			METRIC_NAMES = (
83			'alpha',
84			'beta',
85			'sharpe',
86			'algorithm_volatility',
87			'benchmark_volatility',
88			'downside_risk',
89			'sortino',
90			'information',
91			)
92
93			def __init__(self, sim_params, env,
94			create_first_day_stats=False,
95			account=None):
96			self.treasury_curves = env.treasury_curves
97			self.start_date = sim_params.period_start.replace(
98			hour=0, minute=0, second=0, microsecond=0
99			)
100			self.end_date = sim_params.period_end.replace(
101			hour=0, minute=0, second=0, microsecond=0
102			)
103
104			self.trading_days = env.days_in_range(self.start_date, self.end_date)
105
106			# Hold on to the trading day before the start,
107			# used for index of the zero return value when forcing returns
108			# on the first day.
109			self.day_before_start = self.start_date - env.trading_days.freq
110
111			last_day = normalize_date(sim_params.period_end)
112			if last_day not in self.trading_days:
113			last_day = pd.tseries.index.DatetimeIndex(
114			[last_day]
115			)
116			self.trading_days = self.trading_days.append(last_day)
117
118			self.sim_params = sim_params
119			self.env = env
120
121			self.create_first_day_stats = create_first_day_stats
122
123			cont_index = self.trading_days
124
125			self.cont_index = cont_index
126			self.cont_len = len(self.cont_index)
127
128			empty_cont = np.full(self.cont_len, np.nan)
129
130			self.algorithm_returns_cont = empty_cont.copy()
131			self.benchmark_returns_cont = empty_cont.copy()
132			self.algorithm_cumulative_leverages_cont = empty_cont.copy()
133			self.mean_returns_cont = empty_cont.copy()
134			self.annualized_mean_returns_cont = empty_cont.copy()
135			self.mean_benchmark_returns_cont = empty_cont.copy()
136			self.annualized_mean_benchmark_returns_cont = empty_cont.copy()
137
138			# The returns at a given time are read and reset from the respective
139			# returns container.
140			self.algorithm_returns = None
141			self.benchmark_returns = None
142			self.mean_returns = None
143			self.annualized_mean_returns = None
144			self.mean_benchmark_returns = None
145			self.annualized_mean_benchmark_returns = None
146
147			self.algorithm_cumulative_returns = empty_cont.copy()
148			self.benchmark_cumulative_returns = empty_cont.copy()
149			self.algorithm_cumulative_leverages = empty_cont.copy()
150			self.excess_returns = empty_cont.copy()
151
152			self.latest_dt_loc = 0
153			self.latest_dt = cont_index[0]
154
155			self.benchmark_volatility = empty_cont.copy()
156			self.algorithm_volatility = empty_cont.copy()
157			self.beta = empty_cont.copy()
158			self.alpha = empty_cont.copy()
159			self.sharpe = empty_cont.copy()
160			self.downside_risk = empty_cont.copy()
161			self.sortino = empty_cont.copy()
162			self.information = empty_cont.copy()
163
164			self.drawdowns = empty_cont.copy()
165			self.max_drawdowns = empty_cont.copy()
166			self.max_drawdown = 0
167			self.max_leverages = empty_cont.copy()
168			self.max_leverage = 0
169			self.current_max = -np.inf
170			self.daily_treasury = pd.Series(index=self.trading_days)
171			self.treasury_period_return = np.nan
172
173			self.num_trading_days = 0
174
175			def update(self, dt, algorithm_returns, benchmark_returns, account):
176			# Keep track of latest dt for use in to_dict and other methods
177			# that report current state.
178			self.latest_dt = dt
179			dt_loc = self.cont_index.get_loc(dt)
180			self.latest_dt_loc = dt_loc
181
182			self.algorithm_returns_cont[dt_loc] = algorithm_returns
183			self.algorithm_returns = self.algorithm_returns_cont[:dt_loc + 1]
184
185			self.num_trading_days = len(self.algorithm_returns)
186
187			if self.create_first_day_stats:
188			if len(self.algorithm_returns) == 1:
189			self.algorithm_returns = np.append(0.0, self.algorithm_returns)
190
191			self.algorithm_cumulative_returns[dt_loc] = \
192			self.calculate_cumulative_returns(self.algorithm_returns)
193
194			algo_cumulative_returns_to_date = \
195			self.algorithm_cumulative_returns[:dt_loc + 1]
196
197			self.mean_returns_cont[dt_loc] = \
198			algo_cumulative_returns_to_date[dt_loc] / self.num_trading_days
199
200			self.mean_returns = self.mean_returns_cont[:dt_loc + 1]
201
202			self.annualized_mean_returns_cont[dt_loc] = \
203			self.mean_returns_cont[dt_loc] * 252
204
205			self.annualized_mean_returns = \
206			self.annualized_mean_returns_cont[:dt_loc + 1]
207
208			if self.create_first_day_stats:
209			if len(self.mean_returns) == 1:
210			self.mean_returns = np.append(0.0, self.mean_returns)
211			self.annualized_mean_returns = np.append(
212			0.0, self.annualized_mean_returns)
213
214			self.benchmark_returns_cont[dt_loc] = benchmark_returns
215			self.benchmark_returns = self.benchmark_returns_cont[:dt_loc + 1]
216
217			if self.create_first_day_stats:
218			if len(self.benchmark_returns) == 1:
219			self.benchmark_returns = np.append(0.0, self.benchmark_returns)
220
221			self.benchmark_cumulative_returns[dt_loc] = \
222			self.calculate_cumulative_returns(self.benchmark_returns)
223
224			benchmark_cumulative_returns_to_date = \
225			self.benchmark_cumulative_returns[:dt_loc + 1]
226
227			self.mean_benchmark_returns_cont[dt_loc] = \
228			benchmark_cumulative_returns_to_date[dt_loc] / \
229			self.num_trading_days
230
231			self.mean_benchmark_returns = self.mean_benchmark_returns_cont[:dt_loc]
232
233			self.annualized_mean_benchmark_returns_cont[dt_loc] = \
234			self.mean_benchmark_returns_cont[dt_loc] * 252
235
236			self.annualized_mean_benchmark_returns = \
237			self.annualized_mean_benchmark_returns_cont[:dt_loc + 1]
238
239			self.algorithm_cumulative_leverages_cont[dt_loc] = account['leverage']
240			self.algorithm_cumulative_leverages = \
241			self.algorithm_cumulative_leverages_cont[:dt_loc + 1]
242
243			if self.create_first_day_stats:
244			if len(self.algorithm_cumulative_leverages) == 1:
245			self.algorithm_cumulative_leverages = np.append(
246			0.0,
247			self.algorithm_cumulative_leverages)
248
249			if not len(self.algorithm_returns) and len(self.benchmark_returns):
250			message = "Mismatch between benchmark_returns ({bm_count}) and \
251			algorithm_returns ({algo_count}) in range {start} : {end} on {dt}"
252			message = message.format(
253			bm_count=len(self.benchmark_returns),
254			algo_count=len(self.algorithm_returns),
255			start=self.start_date,
256			end=self.end_date,
257			dt=dt
258			)
259			raise Exception(message)
260
261			self.update_current_max()
262			self.benchmark_volatility[dt_loc] = \
263			self.calculate_volatility(self.benchmark_returns)
264			self.algorithm_volatility[dt_loc] = \
265			self.calculate_volatility(self.algorithm_returns)
266
267			# caching the treasury rates for the minutely case is a
268			# big speedup, because it avoids searching the treasury
269			# curves on every minute.
270			# In both minutely and daily, the daily curve is always used.
271			treasury_end = dt.replace(hour=0, minute=0)
272			if np.isnan(self.daily_treasury[treasury_end]):
273			treasury_period_return = choose_treasury(
274			self.treasury_curves,
275			self.start_date,
276			treasury_end,
277			self.env,
278			)
279			self.daily_treasury[treasury_end] = treasury_period_return
280			self.treasury_period_return = self.daily_treasury[treasury_end]
281			self.excess_returns[dt_loc] = (
282			self.algorithm_cumulative_returns[dt_loc] -
283			self.treasury_period_return)
284			self.beta[dt_loc] = self.calculate_beta()
285			self.alpha[dt_loc] = self.calculate_alpha()
286			self.sharpe[dt_loc] = self.calculate_sharpe()
287			self.downside_risk[dt_loc] = \
288			self.calculate_downside_risk()
289			self.sortino[dt_loc] = self.calculate_sortino()
290			self.information[dt_loc] = self.calculate_information()
291			self.max_drawdown = self.calculate_max_drawdown()
292			self.max_drawdowns[dt_loc] = self.max_drawdown
293			self.max_leverage = self.calculate_max_leverage()
294			self.max_leverages[dt_loc] = self.max_leverage
295
296			def to_dict(self):
297			"""
298			Creates a dictionary representing the state of the risk report.
299			Returns a dict object of the form:
300			"""
301			dt = self.latest_dt
302			dt_loc = self.latest_dt_loc
303			period_label = dt.strftime("%Y-%m")
304			rval = {
305			'trading_days': self.num_trading_days,
306			'benchmark_volatility':
307			self.benchmark_volatility[dt_loc],
308			'algo_volatility':
309			self.algorithm_volatility[dt_loc],
310			'treasury_period_return': self.treasury_period_return,
311			# Though the two following keys say period return,
312			# they would be more accurately called the cumulative return.
313			# However, the keys need to stay the same, for now, for backwards
314			# compatibility with existing consumers.
315			'algorithm_period_return':
316			self.algorithm_cumulative_returns[dt_loc],
317			'benchmark_period_return':
318			self.benchmark_cumulative_returns[dt_loc],
319			'beta': self.beta[dt_loc],
320			'alpha': self.alpha[dt_loc],
321			'sharpe': self.sharpe[dt_loc],
322			'sortino': self.sortino[dt_loc],
323			'information': self.information[dt_loc],
324			'excess_return': self.excess_returns[dt_loc],
325			'max_drawdown': self.max_drawdown,
326			'max_leverage': self.max_leverage,
327			'period_label': period_label
328			}
329
330			return {k: (None if check_entry(k, v) else v)
331			for k, v in iteritems(rval)}
332
333			def __repr__(self):
334			statements = []
335			for metric in self.METRIC_NAMES:
336			value = getattr(self, metric)[-1]
337			if isinstance(value, list):
338			if len(value) == 0:
339			value = np.nan
340			else:
341			value = value[-1]
342			statements.append("{m}:{v}".format(m=metric, v=value))
343
344			return '\n'.join(statements)
345
346			def calculate_cumulative_returns(self, returns):
347			return (1. + returns).prod() - 1
348
349			def update_current_max(self):
350			if len(self.algorithm_cumulative_returns) == 0:
351			return
352			current_cumulative_return = \
353			self.algorithm_cumulative_returns[self.latest_dt_loc]
354			if self.current_max < current_cumulative_return:
355			self.current_max = current_cumulative_return
356
357			def calculate_max_drawdown(self):
358			if len(self.algorithm_cumulative_returns) == 0:
359			return self.max_drawdown
360
361			# The drawdown is defined as: (high - low) / high
362			# The above factors out to: 1.0 - (low / high)
363			#
364			# Instead of explicitly always using the low, use the current total
365			# return value, and test that against the max drawdown, which will
366			# exceed the previous max_drawdown iff the current return is lower than
367			# the previous low in the current drawdown window.
368			cur_drawdown = 1.0 - (
369			(1.0 + self.algorithm_cumulative_returns[self.latest_dt_loc])
370			/
371			(1.0 + self.current_max))
372
373			self.drawdowns[self.latest_dt_loc] = cur_drawdown
374
375			if self.max_drawdown < cur_drawdown:
376			return cur_drawdown
377			else:
378			return self.max_drawdown
379
380			def calculate_max_leverage(self):
381			# The leverage is defined as: the gross_exposure/net_liquidation
382			# gross_exposure = long_exposure + abs(short_exposure)
383			# net_liquidation = ending_cash + long_exposure + short_exposure
384			cur_leverage = self.algorithm_cumulative_leverages_cont[
385			self.latest_dt_loc]
386
387			return max(cur_leverage, self.max_leverage)
388
389			def calculate_sharpe(self):
390			"""
391			http://en.wikipedia.org/wiki/Sharpe_ratio
392			"""
393			return sharpe_ratio(
394			self.algorithm_volatility[self.latest_dt_loc],
395			self.annualized_mean_returns_cont[self.latest_dt_loc],
396			self.daily_treasury[self.latest_dt.date()])
397
398			def calculate_sortino(self):
399			"""
400			http://en.wikipedia.org/wiki/Sortino_ratio
401			"""
402			return sortino_ratio(
403			self.annualized_mean_returns_cont[self.latest_dt_loc],
404			self.daily_treasury[self.latest_dt.date()],
405			self.downside_risk[self.latest_dt_loc])
406
407			def calculate_information(self):
408			"""
409			http://en.wikipedia.org/wiki/Information_ratio
410			"""
411			return information_ratio(
412			self.algorithm_volatility[self.latest_dt_loc],
413			self.annualized_mean_returns_cont[self.latest_dt_loc],
414			self.annualized_mean_benchmark_returns_cont[self.latest_dt_loc])
415
416			def calculate_alpha(self):
417			"""
418			http://en.wikipedia.org/wiki/Alpha_(investment)
419			"""
420			return alpha(
421			self.annualized_mean_returns_cont[self.latest_dt_loc],
422			self.treasury_period_return,
423			self.annualized_mean_benchmark_returns_cont[self.latest_dt_loc],
424			self.beta[self.latest_dt_loc])
425
426			def calculate_volatility(self, daily_returns):
427			if len(daily_returns) <= 1:
428			return 0.0
429			return np.std(daily_returns, ddof=1) * math.sqrt(252)
430
431			def calculate_downside_risk(self):
432			return downside_risk(self.algorithm_returns,
433			self.mean_returns,
434			252)
435
436			def calculate_beta(self):
437			"""
438
439			.. math::
440
441			\\beta_a = \\frac{\mathrm{Cov}(r_a,r_p)}{\mathrm{Var}(r_p)}
442
443			http://en.wikipedia.org/wiki/Beta_(finance)
444			"""
445			# it doesn't make much sense to calculate beta for less than two
446			# values, so return none.
447			if len(self.algorithm_returns) < 2:
448			return 0.0
449
450			returns_matrix = np.vstack([self.algorithm_returns,
451			self.benchmark_returns])
452			C = np.cov(returns_matrix, ddof=1)
453			algorithm_covariance = C[0][1]
454			benchmark_variance = C[1][1]
455			beta = algorithm_covariance / benchmark_variance
456
457			return beta
458
459			def __getstate__(self):
460			state_dict = {k: v for k, v in iteritems(self.__dict__)
461			if not k.startswith('_')}
462
463			STATE_VERSION = 3
464			state_dict[VERSION_LABEL] = STATE_VERSION
465
466			return state_dict
467
468			def __setstate__(self, state):
469
470			OLDEST_SUPPORTED_STATE = 3
471			version = state.pop(VERSION_LABEL)
472
473			if version < OLDEST_SUPPORTED_STATE:
474			raise BaseException("RiskMetricsCumulative \
475			saved state is too old.")
476
477			self.__dict__.update(state)
478

quantopian / zipline

Push — master ( e4fe44...6106cb )

zipline.finance.risk.information_ratio() B

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like