From 2d6131e7afc95316398e8e3e5ac5b7dfa2d1ef84 Mon Sep 17 00:00:00 2001 From: Axel Nilsson Date: Sun, 4 Feb 2024 20:40:26 +0100 Subject: [PATCH] Changed layout of legacy code. --- .../abacus/legacy/leg_garch.py | 0 .../abacus/legacy/leg_generalized_pareto.py | 0 .../legacy/leg_gjr_garch_normal_poisson.py | 0 .../abacus/legacy/leg_gjr_grach.py | 0 .../ma.py => src/abacus/legacy/leg_ma.py | 0 .../abacus/legacy/leg_model.py | 0 .../abacus/legacy/leg_model_factory.py | 0 .../nnar.py => src/abacus/legacy/leg_nnar.py | 0 .../abacus/legacy/leg_norm_poisson_mixture.py | 0 .../legacy/leg_student_poisson_mixture.py | 0 .../abacus/utils/header.txt | 0 src_legacy/abacus/__init__.py | 6 - src_legacy/abacus/instruments.py | 87 --------- src_legacy/abacus/simulator/__init__.py | 0 src_legacy/abacus/simulator/ar.py | 181 ------------------ src_legacy/abacus/simulator/forecaster.py | 42 ---- src_legacy/abacus/simulator/hmm.py | 8 - src_legacy/abacus/simulator/simulator.py | 167 ---------------- src_legacy/abacus/utilities/__init__.py | 0 src_legacy/abacus/utilities/dataloader.py | 148 -------------- src_legacy/abacus/utilities/email_service.py | 102 ---------- .../abacus/utilities/instrument_enum.py | 7 - src_legacy/abacus/utilities/model_enum.py | 7 - src_legacy/abacus/utilities/risk_assessor.py | 154 --------------- 24 files changed, 909 deletions(-) rename src_legacy/abacus/simulator/garch.py => src/abacus/legacy/leg_garch.py (100%) rename src_legacy/abacus/utilities/generalized_pareto.py => src/abacus/legacy/leg_generalized_pareto.py (100%) rename src_legacy/abacus/simulator/gjr_garch_normal_poisson.py => src/abacus/legacy/leg_gjr_garch_normal_poisson.py (100%) rename src_legacy/abacus/simulator/gjr_grach.py => src/abacus/legacy/leg_gjr_grach.py (100%) rename src_legacy/abacus/simulator/ma.py => src/abacus/legacy/leg_ma.py (100%) rename src_legacy/abacus/simulator/model.py => src/abacus/legacy/leg_model.py (100%) rename src_legacy/abacus/simulator/model_factory.py => src/abacus/legacy/leg_model_factory.py (100%) rename src_legacy/abacus/simulator/nnar.py => src/abacus/legacy/leg_nnar.py (100%) rename src_legacy/abacus/utilities/norm_poisson_mixture.py => src/abacus/legacy/leg_norm_poisson_mixture.py (100%) rename src_legacy/abacus/utilities/student_poisson_mixture.py => src/abacus/legacy/leg_student_poisson_mixture.py (100%) rename src_legacy/abacus/utilities/email_header.txt => src/abacus/utils/header.txt (100%) delete mode 100644 src_legacy/abacus/__init__.py delete mode 100644 src_legacy/abacus/instruments.py delete mode 100644 src_legacy/abacus/simulator/__init__.py delete mode 100644 src_legacy/abacus/simulator/ar.py delete mode 100644 src_legacy/abacus/simulator/forecaster.py delete mode 100644 src_legacy/abacus/simulator/hmm.py delete mode 100644 src_legacy/abacus/simulator/simulator.py delete mode 100644 src_legacy/abacus/utilities/__init__.py delete mode 100644 src_legacy/abacus/utilities/dataloader.py delete mode 100644 src_legacy/abacus/utilities/email_service.py delete mode 100644 src_legacy/abacus/utilities/instrument_enum.py delete mode 100644 src_legacy/abacus/utilities/model_enum.py delete mode 100644 src_legacy/abacus/utilities/risk_assessor.py diff --git a/src_legacy/abacus/simulator/garch.py b/src/abacus/legacy/leg_garch.py similarity index 100% rename from src_legacy/abacus/simulator/garch.py rename to src/abacus/legacy/leg_garch.py diff --git a/src_legacy/abacus/utilities/generalized_pareto.py b/src/abacus/legacy/leg_generalized_pareto.py similarity index 100% rename from src_legacy/abacus/utilities/generalized_pareto.py rename to src/abacus/legacy/leg_generalized_pareto.py diff --git a/src_legacy/abacus/simulator/gjr_garch_normal_poisson.py b/src/abacus/legacy/leg_gjr_garch_normal_poisson.py similarity index 100% rename from src_legacy/abacus/simulator/gjr_garch_normal_poisson.py rename to src/abacus/legacy/leg_gjr_garch_normal_poisson.py diff --git a/src_legacy/abacus/simulator/gjr_grach.py b/src/abacus/legacy/leg_gjr_grach.py similarity index 100% rename from src_legacy/abacus/simulator/gjr_grach.py rename to src/abacus/legacy/leg_gjr_grach.py diff --git a/src_legacy/abacus/simulator/ma.py b/src/abacus/legacy/leg_ma.py similarity index 100% rename from src_legacy/abacus/simulator/ma.py rename to src/abacus/legacy/leg_ma.py diff --git a/src_legacy/abacus/simulator/model.py b/src/abacus/legacy/leg_model.py similarity index 100% rename from src_legacy/abacus/simulator/model.py rename to src/abacus/legacy/leg_model.py diff --git a/src_legacy/abacus/simulator/model_factory.py b/src/abacus/legacy/leg_model_factory.py similarity index 100% rename from src_legacy/abacus/simulator/model_factory.py rename to src/abacus/legacy/leg_model_factory.py diff --git a/src_legacy/abacus/simulator/nnar.py b/src/abacus/legacy/leg_nnar.py similarity index 100% rename from src_legacy/abacus/simulator/nnar.py rename to src/abacus/legacy/leg_nnar.py diff --git a/src_legacy/abacus/utilities/norm_poisson_mixture.py b/src/abacus/legacy/leg_norm_poisson_mixture.py similarity index 100% rename from src_legacy/abacus/utilities/norm_poisson_mixture.py rename to src/abacus/legacy/leg_norm_poisson_mixture.py diff --git a/src_legacy/abacus/utilities/student_poisson_mixture.py b/src/abacus/legacy/leg_student_poisson_mixture.py similarity index 100% rename from src_legacy/abacus/utilities/student_poisson_mixture.py rename to src/abacus/legacy/leg_student_poisson_mixture.py diff --git a/src_legacy/abacus/utilities/email_header.txt b/src/abacus/utils/header.txt similarity index 100% rename from src_legacy/abacus/utilities/email_header.txt rename to src/abacus/utils/header.txt diff --git a/src_legacy/abacus/__init__.py b/src_legacy/abacus/__init__.py deleted file mode 100644 index d914c9f..0000000 --- a/src_legacy/abacus/__init__.py +++ /dev/null @@ -1,6 +0,0 @@ -# -*- coding: utf-8 -*- -import logging -import logging.config - -logging.config.fileConfig("src/abacus/logging.cfg") -logger = logging.getLogger(__name__) diff --git a/src_legacy/abacus/instruments.py b/src_legacy/abacus/instruments.py deleted file mode 100644 index bfc20e7..0000000 --- a/src_legacy/abacus/instruments.py +++ /dev/null @@ -1,87 +0,0 @@ -# -*- coding: utf-8 -*- - -import numpy as np -import pandas as pd - -from datetime import datetime -from abc import ABC -from abacus.simulator.model import Model -from abacus.utilities.currency_enum import Currency - - -class Instrument(ABC): - def __init__( - self, - ric: str, - currency: Currency, - start_date: datetime, - end_date: datetime, - interval: str, - ): - self.ric = ric - self.start_date = start_date - self.end_date = end_date - self.interval = interval - self.local_currency = currency - - self.price_history = None - self.art_return_history = None - self.log_return_history = None - self.model = None - self.has_model = False - - def set_model(self, model: Model): - self.model = model - self.has_model = True - - def set_price_history(self, price_history: pd.DataFrame): - self.price_history = price_history - self.art_return_history = self._art_return_history() - self.log_return_history = self._log_return_history() - - def _art_return_history(self) -> pd.DataFrame: - if self._has_prices: - return self.price_history / self.price_history.shift(1)[1:] - else: - raise ValueError("No price history exists.") - - def _log_return_history(self) -> pd.DataFrame: - if self._has_prices: - return np.log(self.price_history / self.price_history.shift(1))[1:] - else: - raise ValueError("No price history exists.") - - def _has_prices(self) -> bool: - """ - Checks if instrument has price history dataframe. - - Returns: - bool: status of price history. - """ - if self.price_hisotry is None: - return False - return True - - -class Equity(Instrument): - def __init__( - self, - ric: str, - currency: Currency, - start_date: datetime, - end_date: datetime, - interval: str, - ): - super().__init__(ric, currency, start_date, end_date, interval) - - -class FX(Instrument): - def __init__( - self, - ric: str, - currency: Currency, - start_date: datetime, - end_date: datetime, - interval: str, - ): - super().__init__(ric, currency, start_date, end_date, interval) diff --git a/src_legacy/abacus/simulator/__init__.py b/src_legacy/abacus/simulator/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src_legacy/abacus/simulator/ar.py b/src_legacy/abacus/simulator/ar.py deleted file mode 100644 index 2882ea6..0000000 --- a/src_legacy/abacus/simulator/ar.py +++ /dev/null @@ -1,181 +0,0 @@ -# -*- coding: utf-8 -*- -import numpy as np - -from numpy.linalg import inv -from scipy.stats import norm -from abacus.simulator.model import Model, StationarityError - - -class AR(Model): - def __init__(self, data, p): - super().__init__(data) - self.p = p - - @property - def initial_solution(self) -> np.array: - """ - Not required for AR(p) model. - """ - pass - - @property - def mse(self) -> float: - number_of_observations = len(self.data) - self.p - return ( - np.sum(self._generate_residuals(self.solution) ** 2) - / number_of_observations - ) - - def fit_model(self) -> np.array: - """ - Fits parameters of the AR(p) model by solving the normal equations. - - Returns: - np.array: optimal parameters. - """ - lag = self.p - number_of_observations = len(self.data) - - y = self.data[lag:] - x = [] - for i in range(lag): - x.append(self.data[lag - i - 1 : number_of_observations - i - 1]) - x = np.stack(x).T - phi = inv(x.T @ x) @ x.T @ y - - parameters = np.zeros(self.p + 2) - parameters[0] = np.mean(self.data) - parameters[1] = np.std(self.data) - parameters[2:] = phi - - self.solution = parameters - self._check_unit_roots() - - return parameters - - def _cost_function(self) -> float: - """ - Not required for AR(p) model. - """ - pass - - def run_simulation(self, number_of_steps: int) -> np.array: - """ - Runs univariate simulation of process. - - Args: - number_of_steps (int): number of simulation steps into the future. - - Returns: - np.array: simulated process. - """ - simulated_process = np.zeros(number_of_steps) - current_regression_values = self.data[-self.p :] - mu = self.solution[0] - sigma = self.solution[1] - phi = self.solution[2:] - - for i in range(number_of_steps): - residual = np.random.normal() - simulated_process[i] = ( - mu + phi.T @ current_regression_values + sigma * residual - ) - current_regression_values = np.insert( - current_regression_values[:-1], 0, simulated_process[i] - ) - - return simulated_process - - def transform_to_true(self, uniform_sample: np.array) -> np.array: - """ - Transforms a predicted uniform sample to true values of the process. Very similar to the - univarite simulation case, the difference is only that uniform samples are obtained from - elsewhere. - - Args: - uniform_sample (np.array): sample of uniform variables U(0,1). - - Returns: - np.array: simulated process. - """ - number_of_observations = len(uniform_sample) - simulated_process = np.zeros(number_of_observations) - current_regression_values = self.data[-self.p :] - mu = self.solution[0] - sigma = self.solution[1] - phi = self.solution[2:] - - for i in range(number_of_observations): - residual = norm.ppf(uniform_sample[i]) - simulated_process[i] = ( - mu + phi.T @ current_regression_values + sigma * residual - ) - current_regression_values = np.insert( - current_regression_values[:-1], 0, simulated_process[i] - ) - - return simulated_process - - def transform_to_uniform(self) -> np.array: - """ - Transformes the normalized time series to uniform variables, assuming Gaussian White Noise. - Uses a standard normalization approach without using regression for the first p values to avoid - shrinking the dataset. - - Returns: - np.array: sample of uniform variables U(0,1). - """ - number_of_observations = len(self.data) - uniform_sample = np.zeros(number_of_observations) - residuals = self._generate_residuals(self.solution) - mu = self.solution[0] - sigma = self.solution[1] - - for i in range(number_of_observations): - if i <= self.p - 1: - uniform_sample[i] = norm.cdf((self.data[i] - mu) / sigma) - else: - uniform_sample[i] = norm.cdf( - (self.data[i] - residuals[i - self.p]) / sigma - ) - - return uniform_sample - - def _generate_residuals(self, params: np.array) -> np.array: - """ - Helper method to recursivley generate residuals based on some set of values for params. - - Args: - params (np.array): parameters of the model. - - Returns: - np.array: residuals calculated based of the guessed parameters. - """ - number_of_observations = len(self.data) - residuals = np.zeros(number_of_observations - self.p) - current_regression_values = self.data[: self.p] - mu = params[0] - phi = params[2:] - - for i in range(number_of_observations - self.p): - residuals[i] = self.data[i] - mu - phi.T @ current_regression_values - current_regression_values = np.insert( - current_regression_values[:-1], 0, self.data[i] - ) - - return residuals - - def _check_unit_roots(self) -> None: - """ - Checks for unit roots outside the unit circle. - - Raises: - StationarityError: raised if unit root is found. - """ - coefficients = np.ones(self.p + 1) - coefficients[1:] = self.solution[2:] - roots = np.roots(coefficients) - - for root in roots: - if np.abs(root) >= 1: - raise StationarityError("non-stationarity encountered.") diff --git a/src_legacy/abacus/simulator/forecaster.py b/src_legacy/abacus/simulator/forecaster.py deleted file mode 100644 index 04487a0..0000000 --- a/src_legacy/abacus/simulator/forecaster.py +++ /dev/null @@ -1,42 +0,0 @@ -# -*- coding: utf-8 -*- -import numpy as np - -from abacus.simulator import simulator - - -class Forecaster: - def __init__(self, instruments, number_of_steps): - self.instruments = instruments - self.number_of_steps = number_of_steps - self.simulator = simulator.Simulator(instruments=instruments) - self.init_prices = self.simulator._last_prices() - - def forecast_returns(self): - dim = (len(self.instruments), self.number_of_steps) - dependency = True - number_of_simulations = 1000 - result = np.zeros(dim) - - for _ in range(number_of_simulations): - simulated_matrix = self.simulator.run_simultion_assets( - number_of_steps=self.number_of_steps, dependency=True - ) - simulated_matrix = np.vstack(simulated_matrix) - result += simulated_matrix - - result = 1 / number_of_simulations * result - - # for _ in range(len(self.instruments)): - # prices_XOM = [] - # prices_XOM.append(self.init_prices[_]) - # returns_XOM = result[_, :] - # for i in range(self.number_of_steps): - # prev_price = prices_XOM[i] - # return_ = np.prod(np.exp(returns_XOM[:i])) - # prices_XOM.append(prev_price*return_) - # print(prices_XOM) - - return result - - def forecast_prices(self): - raise NotImplemented diff --git a/src_legacy/abacus/simulator/hmm.py b/src_legacy/abacus/simulator/hmm.py deleted file mode 100644 index 044a112..0000000 --- a/src_legacy/abacus/simulator/hmm.py +++ /dev/null @@ -1,8 +0,0 @@ -# -*- coding: utf-8 -*- -import numpy as np - -from abacus.simulator.model import Model - - -class HMM(Model): - pass diff --git a/src_legacy/abacus/simulator/simulator.py b/src_legacy/abacus/simulator/simulator.py deleted file mode 100644 index aebfdaa..0000000 --- a/src_legacy/abacus/simulator/simulator.py +++ /dev/null @@ -1,167 +0,0 @@ -# -*- coding: utf-8 -*- -import numpy as np -import copulae as cop -import pyvinecopulib as pv - -from abacus.simulator.model_factory import ModelFactory -from abacus.instruments import Instrument -from abacus.config import DEFALUT_STEPS, VINE_COPULA_FAMILIES, DEFALUT_SIMULATIONS - - -class Simulator: - def __init__(self, instruments: list[Instrument]): - self.instruments = instruments - self.model_factory = ModelFactory(instruments=instruments) - - try: - self.number_of_instruments = len(instruments) - except ValueError: - self.number_of_instruments = -1 - - try: - last_prices = self._last_prices() - except ValueError: - self.value = -1 - - self.find_models() - self.fit_portfolio() - - def find_models(self) -> None: - self.model_factory.build_all() - - def fit_portfolio(self) -> None: - # Check if more than 1 asset exists. - if self.number_of_instruments == 1: - raise ValueError("To few instruments to run dependency.") - - # Check if more than 1 asset exists. - if self.number_of_instruments == 1: - raise ValueError("To few instruments to run dependency.") - - # Creating uniform data. - # TODO: Remove list to make this faster! - # TODO: Assert lenght -> pick smallest if error. - # TODO: Create dict for insturments in portfolio in order to never mix up returns! - uniforms = [] - for instrument in self.instruments: - uniforms.append(instrument.model.transform_to_uniform()) - - smallest_sample_size = len(min(uniforms, key=len)) - uniforms = [uniform[:smallest_sample_size] for uniform in uniforms] - uniforms = np.stack(uniforms).T - - if self.number_of_instruments == 2: - # TODO: Function which picks optimal vanilla copula from a family of copulae. - copula = cop.StudentCopula() - copula.fit(uniforms) - - if self.number_of_instruments > 2: - # Function which picks optimal vine copula. - controls = pv.FitControlsVinecop(family_set=VINE_COPULA_FAMILIES) - copula = pv.Vinecop(uniforms, controls=controls) - - self.copula = copula - - def run_simultion_assets( - self, number_of_steps: int = DEFALUT_STEPS, dependency: bool = True - ) -> np.array: - # Check if portfolio has instruments. - if not self._has_instruments(): - raise ValueError("Portfolio has no instruments.") - - # Check if all instruments has a model. - if not self._has_models(): - raise ValueError("One instrument has no model.") - - # Check if all models are fitted. - if not self._has_solution(): - raise ValueError("One model has no solution.") - - if dependency: - return self._generate_multivariate_simulation( - number_of_steps=number_of_steps - ) - else: - return self._generate_univariate_simulation(number_of_steps=number_of_steps) - - def run_simulation_portfolio( - self, - number_of_steps: int = DEFALUT_STEPS, - number_of_simulations: int = DEFALUT_SIMULATIONS, - dependency: bool = True, - ) -> list[np.array]: - - init_prices = self._last_prices() - for simulation in range(number_of_simulations): - # Portfolio constituance simulation. - simultion_matrix = self.run_simultion_assets( - number_of_steps=number_of_steps, dependency=dependency - ) - # Portfolio prices. - temp_prices = init_prices * np.prod(np.exp(simultion_matrix), axis=1) - print(simultion_matrix) - print(simulation) - - return temp_prices - - def _generate_univariate_simulation(self, number_of_steps: int) -> np.array: - # TODO: Remove list to make this faster! - result = [] - for instrument in self.instruments: - result.append( - instrument.model.run_simulation(number_of_steps=number_of_steps) - ) - return np.vstack(result) - - def _generate_multivariate_simulation(self, number_of_steps: int) -> np.array: - # Check if copula has been fitted. - if not self._has_copula(): - raise ValueError("Portfolio has no multivarite model/copula.") - - if self.number_of_instruments == 2: - simulated_uniforms = self.copula.random(number_of_steps) - - if self.number_of_instruments > 2: - simulated_uniforms = self.copula.simulate(number_of_steps) - - # TODO: Remove list to make this faster if need be. - result = [] - for i in range(self.number_of_instruments): - current_instrument = self.instruments[i] - current_uniform_sample = simulated_uniforms[:, i] - result.append( - current_instrument.model.transform_to_true(current_uniform_sample) - ) - return result - - def _last_prices(self) -> np.array: - try: - # TODO: Assert that all dates are the same! - last_prices = np.zeros(self.number_of_instruments) - for i in range(self.number_of_instruments): - last_prices[i] = self.instruments[i].price_history[-1] - return last_prices - except ValueError: - self._last_prices = np.zeros(self.number_of_instruments) - - def _has_models(self) -> bool: - for instrument in self.instruments: - if instrument.has_model == False: - return False - return True - - def _has_instruments(self) -> bool: - if self.number_of_instruments == 0: - return False - return True - - def _has_copula(self) -> bool: - if self.copula is None: - return False - return True - - def _has_solution(self) -> bool: - for instrument in self.instruments: - if instrument.model.solution is None: - return False - return True diff --git a/src_legacy/abacus/utilities/__init__.py b/src_legacy/abacus/utilities/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/src_legacy/abacus/utilities/dataloader.py b/src_legacy/abacus/utilities/dataloader.py deleted file mode 100644 index 6a5a0d1..0000000 --- a/src_legacy/abacus/utilities/dataloader.py +++ /dev/null @@ -1,148 +0,0 @@ -# -*- coding: utf-8 -*- -import os -import pymysql - -from pandas_datareader import data as pdr -from abacus.instruments import Instrument, Equity, FX -from abacus.utilities.instrument_enum import InstrumentType -from abacus.utilities.currency_enum import Currency - -import logging - -logger = logging.getLogger(__name__) - - -class DataLoader: - """ - Data loading class to fetch and market and portfolio data from different providers. - """ - - def __init__(self, start, end, interval): - self.has_connection = False - self.start = start - self.end = end - self.interval = interval - - def load_yahoo_data(self, instrument_specification: dict) -> list[Instrument]: - """ - Returns a list of instruments with historical price data. Data fetched - from Yahoo Finance. - - Args: - instrument_specification (dict): specifying code, type and currency for instruments. - - Returns: - list[Instrument]: A list of insturments with historical data. - """ - instruments = [] - for key, value in instrument_specification.items(): - current_code = key - current_currency = value["Currency"] - current_type = value["Type"] - current_instrument = self._yahoo_instrument_builder( - type=current_type, - code=current_code, - currency=current_currency, - start=self.start, - end=self.end, - interval=self.interval, - ) - - try: - price_history = pdr.get_data_yahoo( - current_code, start=self.start, end=self.end, interval=self.interval - )["Adj Close"] - current_instrument.set_price_history(price_history=price_history) - except: - logger.error("Cannot fetch yahoo data.") - - instruments.append(current_instrument) - - return instruments - - def _yahoo_instrument_builder( - self, type: str, code: str, currency: str, start: str, end: str, interval: str - ) -> Instrument: - """ - Creates a instrument instance depending on type specified. - - Args: - type (str): Type of instrument. - code (str): Code for yfiance identification. - currency (str): Local currency of instrument. - start (str): Start date. - end (str): End date. - interval (str): Time interval. - - Raises: - ValueError: If instrument type is unrecognized. - - Returns: - Instrument: Built instrument. - """ - - if InstrumentType[type].name == "Equity": - return Equity(code, Currency[currency].value, start, end, interval) - elif InstrumentType[type].name == "FX": - return FX(code, Currency[currency].value, start, end, interval) - else: - raise ValueError("Instrument Type not recognized.") - - # Using AWS / DB connection with appropriate schema. - def create_connection(self) -> pymysql.Connection: - """ - Creates connection to database using environment variables. - .env-mock gives direction of configuration. - - Returns: - pymysql.Connection: - """ - try: - connection = pymysql.connect( - host=os.getenv("AWS_DB_HOST"), - port=int(os.getenv("AWS_DB_PORT")), - user=os.getenv("AWS_DB_USER"), - passwd=os.getenv("AWS_DB_PASW"), - db=os.getenv("AWS_DB_NAME"), - ) - cursor = connection.cursor() - sql = "DESCRIBE Assets" - print(cursor.execute(sql)) - print("[+] RDS Connection Successful") - connection.close() - except Exception as e: - print(f"[-] RDS Connection Failed: {e}") - - return None - - def load_instrument_codes(): - pass - - def update_instrument_data(): - pass - - def insert_instrument_data(): - pass - - def load_instrument_data(): - pass - - def insert_portfolio(): - pass - - def load_portfolio(): - pass - - def load_current_portfolio(): - pass - - def _has_connection(self) -> bool: - """ - Checks if connection to db is established. - - Returns: - bool: connection status. - """ - if self.has_connection: - return True - return False diff --git a/src_legacy/abacus/utilities/email_service.py b/src_legacy/abacus/utilities/email_service.py deleted file mode 100644 index 66476b4..0000000 --- a/src_legacy/abacus/utilities/email_service.py +++ /dev/null @@ -1,102 +0,0 @@ -# -*- coding: utf-8 -*- -import os -import smtplib -import datetime as dt - -from email.mime.text import MIMEText -from email.mime.multipart import MIMEMultipart - - -class EmailService: - """ - EmailService to send formatted email as a status update once program is finished. - """ - - def __init__(self, msg, status): - self.msg = msg - self.status = status - - def send_email(self): - """ - Connects to email service and sends a message. - """ - host = os.getenv("EMAIL_HOST") - port = int(os.getenv("EMAIL_PORT")) - adrs = os.getenv("EMAIL_ADRS") - pasw = os.getenv("EMAIL_PASW") - - recipient = adrs - sender = adrs - - message = self._build_full_msg() - - with smtplib.SMTP(host=host, port=port) as server: - server.starttls() - server.login(user=adrs, password=pasw) - server.sendmail(sender, recipient, message) - server.close() - - def _build_full_msg(self) -> str: - """ - Creates full email message with HTML formatting. - - Returns: - str: HTML formatted message. - """ - header = self._build_header() - log = self._build_log() - msg = f"{header}\n\n{self.msg}\n\n{log}" - msg = msg.replace("\n", "
") - msg = "
" + msg + "
" - - message = MIMEMultipart() - message["Subject"] = f"Investment Report {self.status}" - html = MIMEText(msg, "html") - message.attach(html) - - return message.as_string() - - def _build_header(self) -> str: - """ - Creates the email header including a status code and date. - - Raises: - ValueError: Invalid status code. - - Returns: - str: Formatted header for email. - """ - file = open("src/abacus/utilities/email_header.txt", "r") - logo_offset = " " * 11 - logo = "" - for line in file: - logo += logo_offset + line.rstrip("\n") + "\n" - file.close() - date = str(dt.date.today()) - date_offset = "-" * 32 - if self.status == "OK": - stat_offset = "-" * 32 - elif self.status == "CRITICAL": - stat_offset = "-" * 29 - else: - raise ValueError("Invalid status for email header.") - - header = f"{logo}\n\n{date_offset} {date} {date_offset}\n{stat_offset} STATUS: {self.status} {stat_offset}" - - return header - - @staticmethod - def _build_log() -> str: - """ - Reads the .log file and creates a formatted string. - - Returns: - str: Formatted log for email. - """ - file = open(".log", "r") - log = file.read() - file.close() - offset = "-" * 35 - log = f"{offset} .LOG {offset}\n{log}" - - return log diff --git a/src_legacy/abacus/utilities/instrument_enum.py b/src_legacy/abacus/utilities/instrument_enum.py deleted file mode 100644 index cf2c4d0..0000000 --- a/src_legacy/abacus/utilities/instrument_enum.py +++ /dev/null @@ -1,7 +0,0 @@ -# -*- coding: utf-8 -*- -from enum import Enum - - -class InstrumentType(Enum): - Equity = (0,) - FX = 1 diff --git a/src_legacy/abacus/utilities/model_enum.py b/src_legacy/abacus/utilities/model_enum.py deleted file mode 100644 index e1f03b8..0000000 --- a/src_legacy/abacus/utilities/model_enum.py +++ /dev/null @@ -1,7 +0,0 @@ -# -*- coding: utf-8 -*- -from enum import Enum - - -class ModelType(Enum): - GARCH = (0,) - GJRGARCH = 1 diff --git a/src_legacy/abacus/utilities/risk_assessor.py b/src_legacy/abacus/utilities/risk_assessor.py deleted file mode 100644 index 5119ce1..0000000 --- a/src_legacy/abacus/utilities/risk_assessor.py +++ /dev/null @@ -1,154 +0,0 @@ -# -*- coding: utf-8 -*- -import numpy as np - -from scipy.optimize import minimize -from abacus.config import EPSILON - - -class RiskAssessor: - """ - Used for the risk assessment of portfolio returns given an empirical return distribution. Risk assessment is found - through standard measurements of Value at Risk and Expected Shortfall with user-specified confidence levels. Such - measurements are calculated with Extreme Value Theory (EVT). - - Portfolio returns are translated into portfolio losses in the class. - """ - - def __init__(self, portfolio_returns: list): - self._portfolio_losses = -portfolio_returns - - self._evt_threshold = np.quantile(self._portfolio_losses, 0.95) - self._excess_losses = [ - loss - self._evt_threshold - for loss in self._portfolio_losses - if loss > self._evt_threshold - ] - self._evt_params = np.zeros(2) - - def value_at_risk_non_parametric(self, quantile: float) -> float: - """ - Calculates the Value at Risk of portfolio returns (either in percentage terms or in absolute value) as specified - in the instantiation of the RiskAssessor. - - Args: - quantile: the quantile of the Value at Risk measurement. Note, portfolio losses are specified! - - Returns: Value at Risk for a given confidence level. - """ - return np.quantile(self._portfolio_losses, quantile) - - def expected_shortfall_non_parametric(self, quantile: float) -> float: - """ - Calculates the Expected Shortfall of portfolio returns (either in percentage terms or in absolute value) as - specified in the instantiation of the RiskAssessor. - - Args: - quantile: the quantile of the Value at Risk measurement. Note, portfolio losses are specified! - - Returns: Expected Shortfall for a given confidence level. - """ - var = np.quantile(self._portfolio_losses, quantile) - return np.mean([x for x in self._portfolio_losses if x >= var]) - - def value_at_risk_evt(self, quantile: float) -> float: - """ - Calculates the Value at Risk of portfolio returns (either in percentage terms or in absolute value) as specified - in the instantiation of the RiskAssessor, based on EVT. - - Args: - quantile: the quantile of the Value at Risk measurement. Note, portfolio losses are specified! - - Returns: EVT based Value at Risk for a given confidence level. - """ - if not self._evt_params.any(): - self._evt_params = self._evt_parameter_generator() - - total_n_observations = len(self._portfolio_losses) - excess_n_observations = len(self._excess_losses) - threshold = self._evt_threshold - xi, beta = self._evt_params[0], self._evt_params[1] - - return threshold + beta / xi * ( - (total_n_observations / excess_n_observations * (1 - quantile)) ** (-xi) - 1 - ) - - def expected_shortfall_evt(self, quantile: float) -> float: - """ - Calculates the Expected Shortfall of portfolio returns (either in percentage terms or in absolute value) as - specified in the instantiation of the RiskAssessor, based on EVT. - - Args: - quantile: the quantile of the Value at Risk measurement. Note, portfolio losses are specified! - - Returns: EVT based Expected Shortfall for a given confidence level. - """ - if not self._evt_params.any(): - self._evt_params = self._evt_parameter_generator() - - var = self.value_at_risk_evt(quantile) - threshold = self._evt_threshold - xi, beta = self._evt_params[0], self._evt_params[1] - - return var / (1 - xi) + (beta - threshold * xi) / (1 - xi) - - def risk_summary(self): - """ - Prints a summary of risk measurements. - """ - var_np_95 = self.value_at_risk_non_parametric(0.95) - var_np_99 = self.value_at_risk_non_parametric(0.99) - es_np_95 = self.expected_shortfall_non_parametric(0.95) - es_np_99 = self.expected_shortfall_non_parametric(0.99) - var_evt_99 = self.value_at_risk_evt(0.99) - es_evt_99 = self.expected_shortfall_evt(0.99) - - print("======== RISK ASSESSMENT ========") - print(f"VaR95 {var_np_95}, ES95 {es_np_95}") - print(f"VaR99 {var_np_99}, ES99 {es_np_99}") - print(f"EVT VaR99 {var_evt_99}, EVT ES99 {es_evt_99}") - - def _evt_parameter_generator(self) -> list: - """ - Estimates generalized pareto parameters used for Value at Risk and Expected Shortfall calculations based on EVT. - - Returns: list of parameters, xi and beta, for a generalized pareto distribution. - """ - cons = [] - for obs in self._excess_losses: - cons.append({"type": "ineq", "fun": lambda x: 1 + x[0] / x[1] * obs}) - - x0 = [0.15, 0.01] - sol = minimize( - self._evt_ml_objective_function, - x0, - constraints=cons, - args=self._excess_losses, - ) - - print(f"{sol.x} {sol.success}") - return sol.x - - @staticmethod - def _evt_ml_objective_function(params, data) -> float: - """ - Internal log-likelihood function. Note, this returns the negative of the log-likelihood. - - Args: - params: parameters for the generalized pareto distribution - data: observations above threshold determined in EVT. - - Returns: the negative log-likelihood value of generalized pareto distribution. - """ - - n_observations = len(data) - log_likelihood = 0 - - for obs in data: - log_likelihood = log_likelihood + np.log( - 1 + params[0] / params[1] * obs + EPSILON - ) - - return ( - n_observations * np.log(params[1]) - + (1 + 1 / params[0] + EPSILON) * log_likelihood - )