zipline.finance.risk.RiskMetricsPeriod.calculate_max_leverage() - Code Metrics - quantopian/zipline - Measure and Improve Code Quality continuously with Scrutinizer

calculate_max_leverage() A
last analyzed 2016-01-08 22:03 UTC

↳ Parent: zipline.finance.risk.RiskMetricsPeriod

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	2
dl	0
loc	5
rs	9.4286

#
# Copyright 2013 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 numpy.linalg as la

from six import iteritems

import pandas as pd

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

from zipline.utils.serialization_utils import (
    VERSION_LABEL
)

log = logbook.Logger('Risk Period')

choose_treasury = functools.partial(risk.choose_treasury,
                                    risk.select_treasury_duration)


class RiskMetricsPeriod(object):
    def __init__(self, start_date, end_date, returns, env,
                 benchmark_returns=None, algorithm_leverages=None):

        self.env = env
        treasury_curves = env.treasury_curves
        if treasury_curves.index[-1] >= start_date:
            mask = ((treasury_curves.index >= start_date) &
                    (treasury_curves.index <= end_date))

            self.treasury_curves = treasury_curves[mask]
        else:
            # our test is beyond the treasury curve history
            # so we'll use the last available treasury curve
            self.treasury_curves = treasury_curves[-1:]

        self.start_date = start_date
        self.end_date = end_date

        if benchmark_returns is None:
            br = env.benchmark_returns
            benchmark_returns = br[(br.index >= returns.index[0]) &
                                   (br.index <= returns.index[-1])]

        self.algorithm_returns = self.mask_returns_to_period(returns,
                                                             env)
        self.benchmark_returns = self.mask_returns_to_period(benchmark_returns,
                                                             env)
        self.algorithm_leverages = algorithm_leverages

        self.calculate_metrics()

    def calculate_metrics(self):

        self.benchmark_period_returns = \
            self.calculate_period_returns(self.benchmark_returns)

        self.algorithm_period_returns = \
            self.calculate_period_returns(self.algorithm_returns)

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

        self.num_trading_days = len(self.benchmark_returns)
        self.trading_day_counts = pd.stats.moments.rolling_count(
            self.algorithm_returns, self.num_trading_days)

        self.mean_algorithm_returns = \
            self.algorithm_returns.cumsum() / self.trading_day_counts

        self.benchmark_volatility = self.calculate_volatility(
            self.benchmark_returns)
        self.algorithm_volatility = self.calculate_volatility(
            self.algorithm_returns)
        self.treasury_period_return = choose_treasury(
            self.treasury_curves,
            self.start_date,
            self.end_date,
            self.env,
        )
        self.sharpe = self.calculate_sharpe()
        # The consumer currently expects a 0.0 value for sharpe in period,
        # this differs from cumulative which was np.nan.
        # When factoring out the sharpe_ratio, the different return types
        # were collapsed into `np.nan`.
        # TODO: Either fix consumer to accept `np.nan` or make the
        # `sharpe_ratio` return type configurable.
        # In the meantime, convert nan values to 0.0
        if pd.isnull(self.sharpe):
            self.sharpe = 0.0
        self.sortino = self.calculate_sortino()
        self.information = self.calculate_information()
        self.beta, self.algorithm_covariance, self.benchmark_variance, \
            self.condition_number, self.eigen_values = self.calculate_beta()
        self.alpha = self.calculate_alpha()
        self.excess_return = self.algorithm_period_returns - \
            self.treasury_period_return
        self.max_drawdown = self.calculate_max_drawdown()
        self.max_leverage = self.calculate_max_leverage()

    def to_dict(self):
        """
        Creates a dictionary representing the state of the risk report.
        Returns a dict object of the form:
        """
        period_label = self.end_date.strftime("%Y-%m")
        rval = {
            'trading_days': self.num_trading_days,
            'benchmark_volatility': self.benchmark_volatility,
            'algo_volatility': self.algorithm_volatility,
            'treasury_period_return': self.treasury_period_return,
            'algorithm_period_return': self.algorithm_period_returns,
            'benchmark_period_return': self.benchmark_period_returns,
            'sharpe': self.sharpe,
            'sortino': self.sortino,
            'information': self.information,
            'beta': self.beta,
            'alpha': self.alpha,
            'excess_return': self.excess_return,
            '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 = []
        metrics = [
            "algorithm_period_returns",
            "benchmark_period_returns",
            "excess_return",
            "num_trading_days",
            "benchmark_volatility",
            "algorithm_volatility",
            "sharpe",
            "sortino",
            "information",
            "algorithm_covariance",
            "benchmark_variance",
            "beta",
            "alpha",
            "max_drawdown",
            "max_leverage",
            "algorithm_returns",
            "benchmark_returns",
            "condition_number",
            "eigen_values"
        ]

        for metric in metrics:
            value = getattr(self, metric)
            statements.append("{m}:{v}".format(m=metric, v=value))

        return '\n'.join(statements)

    def mask_returns_to_period(self, daily_returns, env):
        if isinstance(daily_returns, list):
            returns = pd.Series([x.returns for x in daily_returns],
                                index=[x.date for x in daily_returns])
        else:  # otherwise we're receiving an index already
            returns = daily_returns

        trade_days = env.trading_days
        trade_day_mask = returns.index.normalize().isin(trade_days)

        mask = ((returns.index >= self.start_date) &
                (returns.index <= self.end_date) & trade_day_mask)

        returns = returns[mask]
        return returns

    def calculate_period_returns(self, returns):
        period_returns = (1. + returns).prod() - 1
        return period_returns

    def calculate_volatility(self, daily_returns):
        return np.std(daily_returns, ddof=1) * math.sqrt(self.num_trading_days)

    def calculate_sharpe(self):
        """
        http://en.wikipedia.org/wiki/Sharpe_ratio
        """
        return sharpe_ratio(self.algorithm_volatility,
                            self.algorithm_period_returns,
                            self.treasury_period_return)

    def calculate_sortino(self):
        """
        http://en.wikipedia.org/wiki/Sortino_ratio
        """
        mar = downside_risk(self.algorithm_returns,
                            self.mean_algorithm_returns,
                            self.num_trading_days)
        # Hold on to downside risk for debugging purposes.
        self.downside_risk = mar
        return sortino_ratio(self.algorithm_period_returns,
                             self.treasury_period_return,
                             mar)

    def calculate_information(self):
        """
        http://en.wikipedia.org/wiki/Information_ratio
        """
        return information_ratio(self.algorithm_returns,
                                 self.benchmark_returns)

    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 days,
        # so return nan.
        if len(self.algorithm_returns) < 2:
            return np.nan, np.nan, np.nan, np.nan, []

        returns_matrix = np.vstack([self.algorithm_returns,
                                    self.benchmark_returns])
        C = np.cov(returns_matrix, ddof=1)

        # If there are missing benchmark values, then we can't calculate the
        # beta.
        if not np.isfinite(C).all():
            return np.nan, np.nan, np.nan, np.nan, []

        eigen_values = la.eigvals(C)
        condition_number = max(eigen_values) / min(eigen_values)
        algorithm_covariance = C[0][1]
        benchmark_variance = C[1][1]
        beta = algorithm_covariance / benchmark_variance

        return (
            beta,
            algorithm_covariance,
            benchmark_variance,
            condition_number,
            eigen_values
        )

    def calculate_alpha(self):
        """
        http://en.wikipedia.org/wiki/Alpha_(investment)
        """
        return alpha(self.algorithm_period_returns,
                     self.treasury_period_return,
                     self.benchmark_period_returns,
                     self.beta)

    def calculate_max_drawdown(self):
        compounded_returns = []
        cur_return = 0.0
        for r in self.algorithm_returns:
            try:
                cur_return += math.log(1.0 + r)
            # this is a guard for a single day returning -100%, if returns are
            # greater than -1.0 it will throw an error because you cannot take
            # the log of a negative number
            except ValueError:
                log.debug("{cur} return, zeroing the returns".format(
                    cur=cur_return))
                cur_return = 0.0
            compounded_returns.append(cur_return)

        cur_max = None
        max_drawdown = None
        for cur in compounded_returns:
            if cur_max is None or cur > cur_max:
                cur_max = cur

            drawdown = (cur - cur_max)
            if max_drawdown is None or drawdown < max_drawdown:
                max_drawdown = drawdown

        if max_drawdown is None:
            return 0.0

        return 1.0 - math.exp(max_drawdown)

    def calculate_max_leverage(self):
        if self.algorithm_leverages is None:
            return 0.0
        else:
            return max(self.algorithm_leverages)

    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("RiskMetricsPeriod saved state \
                    is too old.")

        self.__dict__.update(state)


1			#
2			# Copyright 2013 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
18			import logbook
19			import math
20			import numpy as np
21			import numpy.linalg as la
22
23			from six import iteritems
24
25			import pandas as pd
26
27			from . import risk
28			from . risk import (
29			alpha,
30			check_entry,
31			downside_risk,
32			information_ratio,
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 Period')
42
43			choose_treasury = functools.partial(risk.choose_treasury,
44			risk.select_treasury_duration)
45
46
47			class RiskMetricsPeriod(object):
48			def __init__(self, start_date, end_date, returns, env,
49			benchmark_returns=None, algorithm_leverages=None):
50
51			self.env = env
52			treasury_curves = env.treasury_curves
53			if treasury_curves.index[-1] >= start_date:
54			mask = ((treasury_curves.index >= start_date) &
55			(treasury_curves.index <= end_date))
56
57			self.treasury_curves = treasury_curves[mask]
58			else:
59			# our test is beyond the treasury curve history
60			# so we'll use the last available treasury curve
61			self.treasury_curves = treasury_curves[-1:]
62
63			self.start_date = start_date
64			self.end_date = end_date
65
66			if benchmark_returns is None:
67			br = env.benchmark_returns
68			benchmark_returns = br[(br.index >= returns.index[0]) &
69			(br.index <= returns.index[-1])]
70
71			self.algorithm_returns = self.mask_returns_to_period(returns,
72			env)
73			self.benchmark_returns = self.mask_returns_to_period(benchmark_returns,
74			env)
75			self.algorithm_leverages = algorithm_leverages
76
77			self.calculate_metrics()
78
79			def calculate_metrics(self):
80
81			self.benchmark_period_returns = \
82			self.calculate_period_returns(self.benchmark_returns)
83
84			self.algorithm_period_returns = \
85			self.calculate_period_returns(self.algorithm_returns)
86
87			if not self.algorithm_returns.index.equals(
88			self.benchmark_returns.index
89			):
90			message = "Mismatch between benchmark_returns ({bm_count}) and \
91			algorithm_returns ({algo_count}) in range {start} : {end}"
92			message = message.format(
93			bm_count=len(self.benchmark_returns),
94			algo_count=len(self.algorithm_returns),
95			start=self.start_date,
96			end=self.end_date
97			)
98			raise Exception(message)
99
100			self.num_trading_days = len(self.benchmark_returns)
101			self.trading_day_counts = pd.stats.moments.rolling_count(
102			self.algorithm_returns, self.num_trading_days)
103
104			self.mean_algorithm_returns = \
105			self.algorithm_returns.cumsum() / self.trading_day_counts
106
107			self.benchmark_volatility = self.calculate_volatility(
108			self.benchmark_returns)
109			self.algorithm_volatility = self.calculate_volatility(
110			self.algorithm_returns)
111			self.treasury_period_return = choose_treasury(
112			self.treasury_curves,
113			self.start_date,
114			self.end_date,
115			self.env,
116			)
117			self.sharpe = self.calculate_sharpe()
118			# The consumer currently expects a 0.0 value for sharpe in period,
119			# this differs from cumulative which was np.nan.
120			# When factoring out the sharpe_ratio, the different return types
121			# were collapsed into `np.nan`.
122			# TODO: Either fix consumer to accept `np.nan` or make the
123			# `sharpe_ratio` return type configurable.
124			# In the meantime, convert nan values to 0.0
125			if pd.isnull(self.sharpe):
126			self.sharpe = 0.0
127			self.sortino = self.calculate_sortino()
128			self.information = self.calculate_information()
129			self.beta, self.algorithm_covariance, self.benchmark_variance, \
130			self.condition_number, self.eigen_values = self.calculate_beta()
131			self.alpha = self.calculate_alpha()
132			self.excess_return = self.algorithm_period_returns - \
133			self.treasury_period_return
134			self.max_drawdown = self.calculate_max_drawdown()
135			self.max_leverage = self.calculate_max_leverage()
136
137			def to_dict(self):
138			"""
139			Creates a dictionary representing the state of the risk report.
140			Returns a dict object of the form:
141			"""
142			period_label = self.end_date.strftime("%Y-%m")
143			rval = {
144			'trading_days': self.num_trading_days,
145			'benchmark_volatility': self.benchmark_volatility,
146			'algo_volatility': self.algorithm_volatility,
147			'treasury_period_return': self.treasury_period_return,
148			'algorithm_period_return': self.algorithm_period_returns,
149			'benchmark_period_return': self.benchmark_period_returns,
150			'sharpe': self.sharpe,
151			'sortino': self.sortino,
152			'information': self.information,
153			'beta': self.beta,
154			'alpha': self.alpha,
155			'excess_return': self.excess_return,
156			'max_drawdown': self.max_drawdown,
157			'max_leverage': self.max_leverage,
158			'period_label': period_label
159			}
160
161			return {k: None if check_entry(k, v) else v
162			for k, v in iteritems(rval)}
163
164			def __repr__(self):
165			statements = []
166			metrics = [
167			"algorithm_period_returns",
168			"benchmark_period_returns",
169			"excess_return",
170			"num_trading_days",
171			"benchmark_volatility",
172			"algorithm_volatility",
173			"sharpe",
174			"sortino",
175			"information",
176			"algorithm_covariance",
177			"benchmark_variance",
178			"beta",
179			"alpha",
180			"max_drawdown",
181			"max_leverage",
182			"algorithm_returns",
183			"benchmark_returns",
184			"condition_number",
185			"eigen_values"
186			]
187
188			for metric in metrics:
189			value = getattr(self, metric)
190			statements.append("{m}:{v}".format(m=metric, v=value))
191
192			return '\n'.join(statements)
193
194			def mask_returns_to_period(self, daily_returns, env):
195			if isinstance(daily_returns, list):
196			returns = pd.Series([x.returns for x in daily_returns],
197			index=[x.date for x in daily_returns])
198			else: # otherwise we're receiving an index already
199			returns = daily_returns
200
201			trade_days = env.trading_days
202			trade_day_mask = returns.index.normalize().isin(trade_days)
203
204			mask = ((returns.index >= self.start_date) &
205			(returns.index <= self.end_date) & trade_day_mask)
206
207			returns = returns[mask]
208			return returns
209
210			def calculate_period_returns(self, returns):
211			period_returns = (1. + returns).prod() - 1
212			return period_returns
213
214			def calculate_volatility(self, daily_returns):
215			return np.std(daily_returns, ddof=1) * math.sqrt(self.num_trading_days)
216
217			def calculate_sharpe(self):
218			"""
219			http://en.wikipedia.org/wiki/Sharpe_ratio
220			"""
221			return sharpe_ratio(self.algorithm_volatility,
222			self.algorithm_period_returns,
223			self.treasury_period_return)
224
225			def calculate_sortino(self):
226			"""
227			http://en.wikipedia.org/wiki/Sortino_ratio
228			"""
229			mar = downside_risk(self.algorithm_returns,
230			self.mean_algorithm_returns,
231			self.num_trading_days)
232			# Hold on to downside risk for debugging purposes.
233			self.downside_risk = mar
234			return sortino_ratio(self.algorithm_period_returns,
235			self.treasury_period_return,
236			mar)
237
238			def calculate_information(self):
239			"""
240			http://en.wikipedia.org/wiki/Information_ratio
241			"""
242			return information_ratio(self.algorithm_returns,
243			self.benchmark_returns)
244
245			def calculate_beta(self):
246			"""
247
248			.. math::
249
250			\\beta_a = \\frac{\mathrm{Cov}(r_a,r_p)}{\mathrm{Var}(r_p)}
251
252			http://en.wikipedia.org/wiki/Beta_(finance)
253			"""
254			# it doesn't make much sense to calculate beta for less than two days,
255			# so return nan.
256			if len(self.algorithm_returns) < 2:
257			return np.nan, np.nan, np.nan, np.nan, []
258
259			returns_matrix = np.vstack([self.algorithm_returns,
260			self.benchmark_returns])
261			C = np.cov(returns_matrix, ddof=1)
262
263			# If there are missing benchmark values, then we can't calculate the
264			# beta.
265			if not np.isfinite(C).all():
266			return np.nan, np.nan, np.nan, np.nan, []
267
268			eigen_values = la.eigvals(C)
269			condition_number = max(eigen_values) / min(eigen_values)
270			algorithm_covariance = C[0][1]
271			benchmark_variance = C[1][1]
272			beta = algorithm_covariance / benchmark_variance
273
274			return (
275			beta,
276			algorithm_covariance,
277			benchmark_variance,
278			condition_number,
279			eigen_values
280			)
281
282			def calculate_alpha(self):
283			"""
284			http://en.wikipedia.org/wiki/Alpha_(investment)
285			"""
286			return alpha(self.algorithm_period_returns,
287			self.treasury_period_return,
288			self.benchmark_period_returns,
289			self.beta)
290
291			def calculate_max_drawdown(self):
292			compounded_returns = []
293			cur_return = 0.0
294			for r in self.algorithm_returns:
295			try:
296			cur_return += math.log(1.0 + r)
297			# this is a guard for a single day returning -100%, if returns are
298			# greater than -1.0 it will throw an error because you cannot take
299			# the log of a negative number
300			except ValueError:
301			log.debug("{cur} return, zeroing the returns".format(
302			cur=cur_return))
303			cur_return = 0.0
304			compounded_returns.append(cur_return)
305
306			cur_max = None
307			max_drawdown = None
308			for cur in compounded_returns:
309			if cur_max is None or cur > cur_max:
310			cur_max = cur
311
312			drawdown = (cur - cur_max)
313			if max_drawdown is None or drawdown < max_drawdown:
314			max_drawdown = drawdown
315
316			if max_drawdown is None:
317			return 0.0
318
319			return 1.0 - math.exp(max_drawdown)
320
321			def calculate_max_leverage(self):
322			if self.algorithm_leverages is None:
323			return 0.0
324			else:
325			return max(self.algorithm_leverages)
326
327			def __getstate__(self):
328			state_dict = {k: v for k, v in iteritems(self.__dict__)
329			if not k.startswith('_')}
330
331			STATE_VERSION = 3
332			state_dict[VERSION_LABEL] = STATE_VERSION
333
334			return state_dict
335
336			def __setstate__(self, state):
337
338			OLDEST_SUPPORTED_STATE = 3
339			version = state.pop(VERSION_LABEL)
340
341			if version < OLDEST_SUPPORTED_STATE:
342			raise BaseException("RiskMetricsPeriod saved state \
343			is too old.")
344
345			self.__dict__.update(state)
346

quantopian / zipline

calculate_max_leverage() A last analyzed 2016-01-08 22:03 UTC

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like

calculate_max_leverage() A
last analyzed 2016-01-08 22:03 UTC