diff --git a/qf_lib/containers/__init__.py b/qf_lib/containers/__init__.py index ee8c886a..863b60f0 100644 --- a/qf_lib/containers/__init__.py +++ b/qf_lib/containers/__init__.py @@ -1,26 +1,26 @@ -# Copyright 2016-present CERN – European Organization for Nuclear Research -# -# 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. - -""" -This module offers a comprehensive set of containers for financial data analysis, including series, dataframes, data arrays, and futures-related objects. - -Each container type is designed with specialized classes to handle various aspects of financial data, such as price series, return calculations, and futures contract management. -These classes streamline the processing and analysis of complex financial datasets, providing a robust framework for quantitative finance research. - -Notes: - - When constructing a container object, the input data is typically expected to be in a wide format. For example: - - A `PricesSeries` object might be initialized with a DataFrame where each column represents the price history of a different asset. - - A `QFDataArray` can be created using multi-dimensional arrays to represent time, asset, and other axes of data. - -""" +# Copyright 2016-present CERN – European Organization for Nuclear Research +# +# 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. + +""" +This module offers a comprehensive set of containers for financial data analysis, including series, dataframes, data arrays, and futures-related objects. + +Each container type is designed with specialized classes to handle various aspects of financial data, such as price series, return calculations, and futures contract management. +These classes streamline the processing and analysis of complex financial datasets, providing a robust framework for quantitative finance research. + +Notes: + - When constructing a container object, the input data is typically expected to be in a wide format. For example: + - A `PricesSeries` object might be initialized with a DataFrame where each column represents the price history of a different asset. + - A `QFDataArray` can be created using multi-dimensional arrays to represent time, asset, and other axes of data. + +""" diff --git a/qf_lib/data_providers/abstract_price_data_provider.py b/qf_lib/data_providers/abstract_price_data_provider.py index 84777503..8cf5c66e 100644 --- a/qf_lib/data_providers/abstract_price_data_provider.py +++ b/qf_lib/data_providers/abstract_price_data_provider.py @@ -1,229 +1,229 @@ -# Copyright 2016-present CERN – European Organization for Nuclear Research -# -# 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. - -from abc import abstractmethod, ABCMeta -from datetime import datetime -from typing import Union, Sequence, Dict - -from qf_lib.common.enums.expiration_date_field import ExpirationDateField -from qf_lib.common.enums.frequency import Frequency -from qf_lib.common.enums.price_field import PriceField -from qf_lib.common.tickers.tickers import Ticker -from qf_lib.common.utils.dateutils.relative_delta import RelativeDelta -from qf_lib.common.utils.miscellaneous.to_list_conversion import convert_to_list -from qf_lib.containers.dataframe.prices_dataframe import PricesDataFrame -from qf_lib.containers.dataframe.qf_dataframe import QFDataFrame -from qf_lib.containers.futures.future_tickers.future_ticker import FutureTicker -from qf_lib.containers.qf_data_array import QFDataArray -from qf_lib.containers.series.prices_series import PricesSeries -from qf_lib.data_providers.data_provider import DataProvider -from qf_lib.data_providers.helpers import normalize_data_array - - -class AbstractPriceDataProvider(DataProvider, metaclass=ABCMeta): - """ - Interface for data providers that supply historical data for various asset classes, including stocks, indices, and futures. - - This base class is designed for simple data providers, which are linked to a single data source (e.g., Quandl, Bloomberg, Yahoo). - It defines the standard structure and methods that any specific data provider implementation must adhere to in order to access and retrieve historical market data. - - Notes: - - When implementing the get_history method (which drivers a large portion of backtesting capabilities) careful consideration must be taken - to ensure the data is returned in the expected format depending on the specific input tickers and fields. - For Example: - - isinstance(tickers, str) and isinstance(fields, str) it is expected to return a PriceSeries object - - isinstance(tickers, str) and isinstance(fields, list) it is expected to return a PricesDataframe object - - otherwise it is expected to return a QFDataArray object - - """ - - def get_price(self, tickers: Union[Ticker, Sequence[Ticker]], fields: Union[PriceField, Sequence[PriceField]], - start_date: datetime, end_date: datetime = None, frequency: Frequency = Frequency.DAILY, **kwargs) -> \ - Union[None, PricesSeries, PricesDataFrame, QFDataArray]: - - end_date = end_date or datetime.now() - end_date = end_date + RelativeDelta(second=0, microsecond=0) - start_date = self._adjust_start_date(start_date, frequency) - got_single_date = self._got_single_date(start_date, end_date, frequency) - - tickers, got_single_ticker = convert_to_list(tickers, Ticker) - fields, got_single_field = convert_to_list(fields, PriceField) - - fields_str = self._map_field_to_str(fields) - container = self.get_history(tickers, fields_str, start_date, end_date, frequency, **kwargs) - - str_to_field_dict = self.str_to_price_field_map() - - # Map the specific fields onto the fields given by the str_to_field_dict - if isinstance(container, QFDataArray): - container = container.assign_coords(fields=[str_to_field_dict[field_str] for field_str in container.fields.values]) - normalized_result = normalize_data_array( - container, tickers, fields, got_single_date, got_single_ticker, got_single_field, use_prices_types=True - ) - else: - normalized_result = PricesDataFrame(container).rename(columns=str_to_field_dict) - if got_single_field: - normalized_result = normalized_result.squeeze(axis=1) - if got_single_ticker: - normalized_result = normalized_result.squeeze(axis=0) - - return normalized_result - - @abstractmethod - def price_field_to_str_map(self) -> Dict[PriceField, str]: - """ - Method has to be implemented in each data provider in order to be able to use get_price. - Returns dictionary containing mapping between PriceField and corresponding string that has to be used by - get_history method to get appropriate type of price series. - - Returns - ------- - Dict[PriceField, str] - mapping between PriceFields and corresponding strings - """ - raise NotImplementedError() - - @abstractmethod - def expiration_date_field_str_map(self, ticker: Ticker = None) -> Dict[ExpirationDateField, str]: - """ - Method has to be implemented in each data provider in order to be able to use get_futures_chain_tickers. - Returns dictionary containing mapping between ExpirationDateField and corresponding string that has to be used - by get_futures_chain_tickers method. - - Parameters - ----------- - ticker: None, Ticker - ticker is optional and might be uses by particular data providers to create appropriate dictionary - - Returns - ------- - Dict[ExpirationDateField, str] - mapping between ExpirationDateField and corresponding strings - """ - pass - - def get_futures_chain_tickers(self, tickers: Union[FutureTicker, Sequence[FutureTicker]], - expiration_date_fields: Union[ExpirationDateField, Sequence[ExpirationDateField]]) \ - -> Dict[FutureTicker, QFDataFrame]: - - expiration_date_fields, got_single_expiration_date_field = convert_to_list(expiration_date_fields, - ExpirationDateField) - mapping_dict = self.expiration_date_field_str_map() - expiration_date_fields_str = [mapping_dict[field] for field in expiration_date_fields] - exp_dates_dict = self._get_futures_chain_dict(tickers, expiration_date_fields_str) - - for future_ticker, exp_dates in exp_dates_dict.items(): - exp_dates = exp_dates.rename(columns=self.str_to_expiration_date_field_map()) - for ticker in exp_dates.index: - ticker.security_type = future_ticker.security_type - ticker.point_value = future_ticker.point_value - ticker.set_name(future_ticker.name) - if got_single_expiration_date_field: - exp_dates = exp_dates.squeeze() - exp_dates_dict[future_ticker] = exp_dates - - return exp_dates_dict - - @abstractmethod - def _get_futures_chain_dict(self, tickers: Union[FutureTicker, Sequence[FutureTicker]], - expiration_date_fields: Union[str, Sequence[str]]) -> Dict[FutureTicker, QFDataFrame]: - """ - Returns a dictionary, which maps Tickers to QFSeries, consisting of the expiration dates of Future - Contracts: Dict[FutureTicker, Union[QFSeries, QFDataFrame]]]. - - Parameters - ---------- - tickers: Ticker, Sequence[Ticker] - tickers for securities which should be retrieved - expiration_date_fields: str, Sequence[str] - expiration date fields of securities which should be retrieved. Specific for each data provider, - the mapping between strings and corresponding ExpirationDateField enum values should be implemented as - str_to_expiration_date_field_map function. - """ - pass - - def str_to_expiration_date_field_map(self, ticker: Ticker = None) -> Dict[str, ExpirationDateField]: - """ - Inverse of str_to_expiration_date_field_map. - """ - field_str_dict = self.expiration_date_field_str_map(ticker) - inv_dict = {v: k for k, v in field_str_dict.items()} - return inv_dict - - def str_to_price_field_map(self) -> Dict[str, PriceField]: - """ - Inverse of price_field_to_str_map. - """ - field_str_dict = self.price_field_to_str_map() - inv_dict = {v: k for k, v in field_str_dict.items()} - return inv_dict - - def _map_field_to_str( - self, fields: Union[None, PriceField, Sequence[PriceField]]) \ - -> Union[None, str, Sequence[str]]: - """ - The method maps enum to sting that is recognised by the specific database. - - Parameters - ---------- - fields - fields of securities which should be retrieved - - Returns - ------- - str, Sequence[str] - String representation of the field or fields that corresponds the API provided by a specific data provider - Depending on the input it returns: - None -> None - PriceField -> str - Sequence[PriceField] -> List[str] - """ - if fields is None: - return None - - field_str_dict = self.price_field_to_str_map() - - if self._is_single_price_field(fields): - if fields in field_str_dict: - return field_str_dict[fields] - else: - raise LookupError("Field {} is not recognised by the data provider. Available Fields: {}" - .format(fields, list(field_str_dict.keys()))) - - result = [] - for field in fields: - if field in field_str_dict: - result.append(field_str_dict[field]) - else: - raise LookupError("Field {} is not recognised by the data provider. Available Fields: {}" - .format(fields, list(field_str_dict.keys()))) - return result - - @staticmethod - def _is_single_price_field(fields: Union[None, PriceField, Sequence[PriceField]]): - return fields is not None and isinstance(fields, PriceField) - - @staticmethod - def _is_single_ticker(value): - if isinstance(value, Ticker): - return True - - return False - - def _get_first_ticker(self, tickers): - if self._is_single_ticker(tickers): - ticker = tickers - else: - ticker = tickers[0] - return ticker +# Copyright 2016-present CERN – European Organization for Nuclear Research +# +# 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. + +from abc import abstractmethod, ABCMeta +from datetime import datetime +from typing import Union, Sequence, Dict + +from qf_lib.common.enums.expiration_date_field import ExpirationDateField +from qf_lib.common.enums.frequency import Frequency +from qf_lib.common.enums.price_field import PriceField +from qf_lib.common.tickers.tickers import Ticker +from qf_lib.common.utils.dateutils.relative_delta import RelativeDelta +from qf_lib.common.utils.miscellaneous.to_list_conversion import convert_to_list +from qf_lib.containers.dataframe.prices_dataframe import PricesDataFrame +from qf_lib.containers.dataframe.qf_dataframe import QFDataFrame +from qf_lib.containers.futures.future_tickers.future_ticker import FutureTicker +from qf_lib.containers.qf_data_array import QFDataArray +from qf_lib.containers.series.prices_series import PricesSeries +from qf_lib.data_providers.data_provider import DataProvider +from qf_lib.data_providers.helpers import normalize_data_array + + +class AbstractPriceDataProvider(DataProvider, metaclass=ABCMeta): + """ + Interface for data providers that supply historical data for various asset classes, including stocks, indices, and futures. + + This base class is designed for simple data providers, which are linked to a single data source (e.g., Quandl, Bloomberg, Yahoo). + It defines the standard structure and methods that any specific data provider implementation must adhere to in order to access and retrieve historical market data. + + Notes: + - When implementing the get_history method (which drivers a large portion of backtesting capabilities) careful consideration must be taken + to ensure the data is returned in the expected format depending on the specific input tickers and fields. + For Example: + - isinstance(tickers, str) and isinstance(fields, str) it is expected to return a PriceSeries object + - isinstance(tickers, str) and isinstance(fields, list) it is expected to return a PricesDataframe object + - otherwise it is expected to return a QFDataArray object + + """ + + def get_price(self, tickers: Union[Ticker, Sequence[Ticker]], fields: Union[PriceField, Sequence[PriceField]], + start_date: datetime, end_date: datetime = None, frequency: Frequency = Frequency.DAILY, **kwargs) -> \ + Union[None, PricesSeries, PricesDataFrame, QFDataArray]: + + end_date = end_date or datetime.now() + end_date = end_date + RelativeDelta(second=0, microsecond=0) + start_date = self._adjust_start_date(start_date, frequency) + got_single_date = self._got_single_date(start_date, end_date, frequency) + + tickers, got_single_ticker = convert_to_list(tickers, Ticker) + fields, got_single_field = convert_to_list(fields, PriceField) + + fields_str = self._map_field_to_str(fields) + container = self.get_history(tickers, fields_str, start_date, end_date, frequency, **kwargs) + + str_to_field_dict = self.str_to_price_field_map() + + # Map the specific fields onto the fields given by the str_to_field_dict + if isinstance(container, QFDataArray): + container = container.assign_coords(fields=[str_to_field_dict[field_str] for field_str in container.fields.values]) + normalized_result = normalize_data_array( + container, tickers, fields, got_single_date, got_single_ticker, got_single_field, use_prices_types=True + ) + else: + normalized_result = PricesDataFrame(container).rename(columns=str_to_field_dict) + if got_single_field: + normalized_result = normalized_result.squeeze(axis=1) + if got_single_ticker: + normalized_result = normalized_result.squeeze(axis=0) + + return normalized_result + + @abstractmethod + def price_field_to_str_map(self) -> Dict[PriceField, str]: + """ + Method has to be implemented in each data provider in order to be able to use get_price. + Returns dictionary containing mapping between PriceField and corresponding string that has to be used by + get_history method to get appropriate type of price series. + + Returns + ------- + Dict[PriceField, str] + mapping between PriceFields and corresponding strings + """ + raise NotImplementedError() + + @abstractmethod + def expiration_date_field_str_map(self, ticker: Ticker = None) -> Dict[ExpirationDateField, str]: + """ + Method has to be implemented in each data provider in order to be able to use get_futures_chain_tickers. + Returns dictionary containing mapping between ExpirationDateField and corresponding string that has to be used + by get_futures_chain_tickers method. + + Parameters + ----------- + ticker: None, Ticker + ticker is optional and might be uses by particular data providers to create appropriate dictionary + + Returns + ------- + Dict[ExpirationDateField, str] + mapping between ExpirationDateField and corresponding strings + """ + pass + + def get_futures_chain_tickers(self, tickers: Union[FutureTicker, Sequence[FutureTicker]], + expiration_date_fields: Union[ExpirationDateField, Sequence[ExpirationDateField]]) \ + -> Dict[FutureTicker, QFDataFrame]: + + expiration_date_fields, got_single_expiration_date_field = convert_to_list(expiration_date_fields, + ExpirationDateField) + mapping_dict = self.expiration_date_field_str_map() + expiration_date_fields_str = [mapping_dict[field] for field in expiration_date_fields] + exp_dates_dict = self._get_futures_chain_dict(tickers, expiration_date_fields_str) + + for future_ticker, exp_dates in exp_dates_dict.items(): + exp_dates = exp_dates.rename(columns=self.str_to_expiration_date_field_map()) + for ticker in exp_dates.index: + ticker.security_type = future_ticker.security_type + ticker.point_value = future_ticker.point_value + ticker.set_name(future_ticker.name) + if got_single_expiration_date_field: + exp_dates = exp_dates.squeeze() + exp_dates_dict[future_ticker] = exp_dates + + return exp_dates_dict + + @abstractmethod + def _get_futures_chain_dict(self, tickers: Union[FutureTicker, Sequence[FutureTicker]], + expiration_date_fields: Union[str, Sequence[str]]) -> Dict[FutureTicker, QFDataFrame]: + """ + Returns a dictionary, which maps Tickers to QFSeries, consisting of the expiration dates of Future + Contracts: Dict[FutureTicker, Union[QFSeries, QFDataFrame]]]. + + Parameters + ---------- + tickers: Ticker, Sequence[Ticker] + tickers for securities which should be retrieved + expiration_date_fields: str, Sequence[str] + expiration date fields of securities which should be retrieved. Specific for each data provider, + the mapping between strings and corresponding ExpirationDateField enum values should be implemented as + str_to_expiration_date_field_map function. + """ + pass + + def str_to_expiration_date_field_map(self, ticker: Ticker = None) -> Dict[str, ExpirationDateField]: + """ + Inverse of str_to_expiration_date_field_map. + """ + field_str_dict = self.expiration_date_field_str_map(ticker) + inv_dict = {v: k for k, v in field_str_dict.items()} + return inv_dict + + def str_to_price_field_map(self) -> Dict[str, PriceField]: + """ + Inverse of price_field_to_str_map. + """ + field_str_dict = self.price_field_to_str_map() + inv_dict = {v: k for k, v in field_str_dict.items()} + return inv_dict + + def _map_field_to_str( + self, fields: Union[None, PriceField, Sequence[PriceField]]) \ + -> Union[None, str, Sequence[str]]: + """ + The method maps enum to sting that is recognised by the specific database. + + Parameters + ---------- + fields + fields of securities which should be retrieved + + Returns + ------- + str, Sequence[str] + String representation of the field or fields that corresponds the API provided by a specific data provider + Depending on the input it returns: + None -> None + PriceField -> str + Sequence[PriceField] -> List[str] + """ + if fields is None: + return None + + field_str_dict = self.price_field_to_str_map() + + if self._is_single_price_field(fields): + if fields in field_str_dict: + return field_str_dict[fields] + else: + raise LookupError("Field {} is not recognised by the data provider. Available Fields: {}" + .format(fields, list(field_str_dict.keys()))) + + result = [] + for field in fields: + if field in field_str_dict: + result.append(field_str_dict[field]) + else: + raise LookupError("Field {} is not recognised by the data provider. Available Fields: {}" + .format(fields, list(field_str_dict.keys()))) + return result + + @staticmethod + def _is_single_price_field(fields: Union[None, PriceField, Sequence[PriceField]]): + return fields is not None and isinstance(fields, PriceField) + + @staticmethod + def _is_single_ticker(value): + if isinstance(value, Ticker): + return True + + return False + + def _get_first_ticker(self, tickers): + if self._is_single_ticker(tickers): + ticker = tickers + else: + ticker = tickers[0] + return ticker diff --git a/qf_lib/portfolio_construction/covariance_estimation/robust_covariance.py b/qf_lib/portfolio_construction/covariance_estimation/robust_covariance.py index 29e0c85c..0e497b4e 100644 --- a/qf_lib/portfolio_construction/covariance_estimation/robust_covariance.py +++ b/qf_lib/portfolio_construction/covariance_estimation/robust_covariance.py @@ -1,77 +1,77 @@ -# Copyright 2016-present CERN – European Organization for Nuclear Research -# -# 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 numpy as np -from arch.univariate.volatility import VolatilityProcess - -from qf_lib.common.utils.volatility.volatility_forecast import VolatilityForecast -from qf_lib.containers.dataframe.qf_dataframe import QFDataFrame - - -class RobustCovariance: - """ - Class for volatility forecasting and covariance matrix estimation. - - This class provides methods for estimating future volatility and the covariance matrix - of financial assets. Note that the forecasting methods implemented here may have a tendency - to reduce the estimated volatility. - - Parameters - ---------- - simple_returns: QFDataFrame - QFDataFrame of simple returns of the assets - """ - - def __init__(self, simple_returns: QFDataFrame): - self.returns = simple_returns - - def calculate_covariance(self, vol_process: VolatilityProcess, horizon: int, method: str = 'analytic') -> QFDataFrame: - """ - Calculates the covariance matrix using foretasted volatility for each asset and - rank correlation between the assets. Covariance matrix contains NOT annualised values. - They are expressed in the frequency of the input returns - - Parameters - ----------- - vol_process: VolatilityProcess - volatility proces used for forecats calculation. For example EGARCH(p=p, o=o, q=q) - horizon: int - horizon for the volatility forecast. It is expressed in the frequency of the returns provided - method (optional): str - method of the volatility forecast calculation. Possible: 'analytic', 'simulation' or 'bootstrap' - 'analytic' is the fastest but is not available for EGARCH and possibly some other models. - For details check arch documentation - - Returns - ---------- - QFDataFrame - cov_matrix of the assets. - """ - vol_forecast_array = self._calculate_expected_volatilities(vol_process, horizon, method) - # use spearman rank correlation instead of simple pearson correlation - corr_matrix = self.returns.corr(method='spearman') - cov_matrix = corr_matrix.copy() - - for i, _ in enumerate(vol_forecast_array): - for j, _ in enumerate(vol_forecast_array): - cov_matrix.iloc[i, j] = corr_matrix.iloc[i, j] * vol_forecast_array[i] * vol_forecast_array[j] - return cov_matrix - - def _calculate_expected_volatilities(self, vol_process, horizon, method): - vol_forecast_array = np.zeros(self.returns.shape[1]) # size = nr o assets - for i, asset in enumerate(self.returns): - asset_returns = self.returns[asset] - vol_forecast = VolatilityForecast(asset_returns, vol_process, method, horizon, annualise=False) - vol_forecast_array[i] = vol_forecast.calculate_single_forecast() - return vol_forecast_array +# Copyright 2016-present CERN – European Organization for Nuclear Research +# +# 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 numpy as np +from arch.univariate.volatility import VolatilityProcess + +from qf_lib.common.utils.volatility.volatility_forecast import VolatilityForecast +from qf_lib.containers.dataframe.qf_dataframe import QFDataFrame + + +class RobustCovariance: + """ + Class for volatility forecasting and covariance matrix estimation. + + This class provides methods for estimating future volatility and the covariance matrix + of financial assets. Note that the forecasting methods implemented here may have a tendency + to reduce the estimated volatility. + + Parameters + ---------- + simple_returns: QFDataFrame + QFDataFrame of simple returns of the assets + """ + + def __init__(self, simple_returns: QFDataFrame): + self.returns = simple_returns + + def calculate_covariance(self, vol_process: VolatilityProcess, horizon: int, method: str = 'analytic') -> QFDataFrame: + """ + Calculates the covariance matrix using foretasted volatility for each asset and + rank correlation between the assets. Covariance matrix contains NOT annualised values. + They are expressed in the frequency of the input returns + + Parameters + ----------- + vol_process: VolatilityProcess + volatility proces used for forecats calculation. For example EGARCH(p=p, o=o, q=q) + horizon: int + horizon for the volatility forecast. It is expressed in the frequency of the returns provided + method (optional): str + method of the volatility forecast calculation. Possible: 'analytic', 'simulation' or 'bootstrap' + 'analytic' is the fastest but is not available for EGARCH and possibly some other models. + For details check arch documentation + + Returns + ---------- + QFDataFrame + cov_matrix of the assets. + """ + vol_forecast_array = self._calculate_expected_volatilities(vol_process, horizon, method) + # use spearman rank correlation instead of simple pearson correlation + corr_matrix = self.returns.corr(method='spearman') + cov_matrix = corr_matrix.copy() + + for i, _ in enumerate(vol_forecast_array): + for j, _ in enumerate(vol_forecast_array): + cov_matrix.iloc[i, j] = corr_matrix.iloc[i, j] * vol_forecast_array[i] * vol_forecast_array[j] + return cov_matrix + + def _calculate_expected_volatilities(self, vol_process, horizon, method): + vol_forecast_array = np.zeros(self.returns.shape[1]) # size = nr o assets + for i, asset in enumerate(self.returns): + asset_returns = self.returns[asset] + vol_forecast = VolatilityForecast(asset_returns, vol_process, method, horizon, annualise=False) + vol_forecast_array[i] = vol_forecast.calculate_single_forecast() + return vol_forecast_array