diff --git a/gamestonk_terminal/cryptocurrency/crypto_controller.py b/gamestonk_terminal/cryptocurrency/crypto_controller.py index 0d137e25f60..a0b1501d45f 100644 --- a/gamestonk_terminal/cryptocurrency/crypto_controller.py +++ b/gamestonk_terminal/cryptocurrency/crypto_controller.py @@ -63,7 +63,7 @@ class CryptoController(CryptoBaseController): "prt", "resources", ] - CHOICES_MENUS = ["ta", "dd", "ov", "disc", "onchain", "defi", "nft", "pred"] + CHOICES_MENUS = ["ta", "dd", "ov", "disc", "onchain", "defi", "nft", "pred", "qa"] DD_VIEWS_MAPPING = { "cg": pycoingecko_view, @@ -109,7 +109,8 @@ class CryptoController(CryptoBaseController): > nft non-fungible tokens, e.g.: today drops{has_ticker_start} > dd due-diligence for loaded coin, e.g.: coin information, social media, market stats > ta technical analysis for loaded coin, e.g.: ema, macd, rsi, adx, bbands, obv -> pred prediction techniques e.g.: regression, arima, rnn, lstm, conv1d, monte carlo[/menu] +> pred prediction techniques e.g.: regression, arima, rnn, lstm, conv1d, monte carlo +> qa quantitative analysis, \t e.g.: decompose, cusum, residuals analysis[/menu] {has_ticker_end} """ console.print(text=help_text, menu="Cryptocurrency") @@ -507,6 +508,24 @@ class CryptoController(CryptoBaseController): else: console.print("No coin selected. Use 'load' to load a coin.\n") + @log_start_end(log=logger) + def call_qa(self, _): + """Process pred command""" + if self.coin: + from gamestonk_terminal.cryptocurrency.quantitative_analysis import ( + qa_controller, + ) + + if self.current_interval != "1day": + console.print("Only interval `1day` is possible for now.\n") + else: + self.queue = self.load_class( + qa_controller.QaController, + self.coin, + self.current_df, + self.queue, + ) + @log_start_end(log=logger) def call_pred(self, _): """Process pred command""" diff --git a/gamestonk_terminal/cryptocurrency/quantitative_analysis/__init__.py b/gamestonk_terminal/cryptocurrency/quantitative_analysis/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/gamestonk_terminal/cryptocurrency/quantitative_analysis/qa_controller.py b/gamestonk_terminal/cryptocurrency/quantitative_analysis/qa_controller.py new file mode 100644 index 00000000000..448527b7fe5 --- /dev/null +++ b/gamestonk_terminal/cryptocurrency/quantitative_analysis/qa_controller.py @@ -0,0 +1,699 @@ +"""Quantitative Analysis Controller Module""" +__docformat__ = "numpy" + +import argparse +import logging +from typing import List + +import numpy as np +import pandas as pd +from prompt_toolkit.completion import NestedCompleter + +from gamestonk_terminal import feature_flags as gtff +from gamestonk_terminal.common.quantitative_analysis import qa_view, rolling_view +from gamestonk_terminal.decorators import log_start_end +from gamestonk_terminal.cryptocurrency import cryptocurrency_helpers as c_help +from gamestonk_terminal.helper_funcs import ( + EXPORT_ONLY_FIGURES_ALLOWED, + EXPORT_ONLY_RAW_DATA_ALLOWED, + check_positive, + check_proportion_range, + parse_known_args_and_warn, + check_list_dates, +) +from gamestonk_terminal.menu import session +from gamestonk_terminal.parent_classes import CryptoBaseController +from gamestonk_terminal.rich_config import console + +logger = logging.getLogger(__name__) + + +class QaController(CryptoBaseController): + """Quantitative Analysis Controller class""" + + CHOICES_COMMANDS = [ + "load", + "pick", + "raw", + "summary", + "line", + "hist", + "cdf", + "bw", + "rolling", + "decompose", + "cusum", + "acf", + "spread", + "quantile", + "skew", + "kurtosis", + "normality", + "qqplot", + "unitroot", + "goodness", + "unitroot", + ] + + PATH = "/crypto/qa/" + + def __init__( + self, + coin: str, + data: pd.DataFrame, + queue: List[str] = None, + ): + """Constructor""" + super().__init__(queue) + + data["Returns"] = data["Close"].pct_change() + data["LogRet"] = np.log(data["Close"]) - np.log(data["Close"].shift(1)) + data = data.dropna() + + self.data = data + self.coin = coin + self.target = "Close" + + if session and gtff.USE_PROMPT_TOOLKIT: + choices: dict = {c: {} for c in self.controller_choices} + choices["pick"] = {c: None for c in list(data.columns)} + choices["load"]["-r"] = {c: {} for c in c_help.INTERVALS} + self.completer = NestedCompleter.from_nested_dict(choices) + + def print_help(self): + """Print help""" + help_text = f"""[cmds] + load load new coin + pick pick target column for analysis[/cmds] + +[param]Coin: [/param]{self.coin} +[param]Target Column: [/param]{self.target} +[cmds] +[info]Statistics:[/info] + summary brief summary statistics of loaded coin. + normality normality statistics and tests + unitroot unit root test for stationarity (ADF, KPSS) +[info]Plots:[/info] + line line plot of selected target + hist histogram with density plot + cdf cumulative distribution function + bw box and whisker plot + acf (partial) auto-correlation function differentials of prices + qqplot residuals against standard normal curve +[info]Rolling Metrics:[/info] + rolling rolling mean and std deviation of prices + spread rolling variance and std deviation of prices + quantile rolling median and quantile of prices + skew rolling skewness of distribution of prices + kurtosis rolling kurtosis of distribution of prices +[info]Risk:[/info] + var display value at risk + es display expected shortfall +[info]Other:[/info] + raw print raw data + decompose decomposition in cyclic-trend, season, and residuals of prices + cusum detects abrupt changes using cumulative sum algorithm of prices[/cmds] + """ + console.print(text=help_text, menu="Crypto - Quantitative Analysis") + + def custom_reset(self): + """Class specific component of reset command""" + if self.coin: + return ["crypto", f"load {self.coin}", "qa"] + return [] + + @log_start_end(log=logger) + def call_pick(self, other_args: List[str]): + """Process pick command""" + parser = argparse.ArgumentParser( + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + add_help=False, + prog="pick", + description=""" + Change target variable + """, + ) + parser.add_argument( + "-t", + "--target", + dest="target", + choices=list(self.data.columns), + help="Select variable to analyze", + ) + if other_args and "-t" not in other_args and "-h" not in other_args: + other_args.insert(0, "-t") + + ns_parser = parse_known_args_and_warn(parser, other_args) + if ns_parser: + self.target = ns_parser.target + console.print("") + + @log_start_end(log=logger) + def call_raw(self, other_args: List[str]): + parser = argparse.ArgumentParser( + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + add_help=False, + prog="raw", + description=""" + Print raw data to console + """, + ) + parser.add_argument( + "-l", + "--limit", + help="Number to show", + type=check_positive, + default=20, + dest="limit", + ) + parser.add_argument( + "-d", + "--descend", + action="store_true", + default=False, + dest="descend", + help="Sort in descending order", + ) + + ns_parser = parse_known_args_and_warn( + parser, other_args, export_allowed=EXPORT_ONLY_RAW_DATA_ALLOWED + ) + if ns_parser: + qa_view.display_raw( + self.data[self.target], + num=ns_parser.limit, + sort="", + des=ns_parser.descend, + export=ns_parser.export, + ) + + @log_start_end(log=logger) + def call_summary(self, other_args: List[str]): + """Process summary command""" + parser = argparse.ArgumentParser( + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + add_help=False, + prog="summary", + description=""" + Summary statistics + """, + ) + ns_parser = parse_known_args_and_warn( + parser, other_args, export_allowed=EXPORT_ONLY_RAW_DATA_ALLOWED + ) + if ns_parser: + qa_view.display_summary(df=self.data, export=ns_parser.export) + + @log_start_end(log=logger) + def call_line(self, other_args: List[str]): + """Process line command""" + parser = argparse.ArgumentParser( + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + add_help=False, + prog="line", + description="Show line plot of selected data and allow to draw lines or highlight specific datetimes.", + ) + parser.add_argument( + "--log", + help="Plot with y on log scale", + dest="log", + action="store_true", + default=False, + ) + parser.add_argument( + "-d", + "--draw", + help="Draw lines and annotate on the plot", + dest="draw", + action="store_true", + default=False, + ) + parser.add_argument( + "--ml", + help="Draw vertical line markers to highlight certain events", + dest="ml", + type=check_list_dates, + default="", + ) + parser.add_argument( + "--ms", + help="Draw scatter markers to highlight certain events", + dest="ms", + type=check_list_dates, + default="", + ) + + ns_parser = parse_known_args_and_warn( + parser, other_args, export_allowed=EXPORT_ONLY_FIGURES_ALLOWED + ) + if ns_parser: + qa_view.display_line( + self.data[self.target], + title=f"{self.coin} {self.target}", + log_y=ns_parser.log, + draw=ns_parser.draw, + markers_lines=ns_parser.ml, + markers_scatter=ns_parser.ms, + ) + + @log_start_end(log=logger) + def call_hist(self, other_args: List[str]): + """Process hist command""" + parser = argparse.ArgumentParser( + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + add_help=False, + prog="hist", + description=""" + Histogram with density and rug + """, + ) + parser.add_argument( + "-b", "--bins", type=check_positive, default=15, dest="n_bins" + ) + ns_parser = parse_known_args_and_warn(parser, other_args) + if ns_parser: + qa_view.display_hist( + name=self.coin, + df=self.data, + target=self.target, + bins=ns_parser.n_bins, + ) + + @log_start_end(log=logger) + def call_cdf(self, other_args: List[str]): + """Process cdf command""" + parser = argparse.ArgumentParser( + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + add_help=False, + prog="cdf", + description=""" + Cumulative distribution function + """, + ) + ns_parser = parse_known_args_and_warn( + parser, other_args, export_allowed=EXPORT_ONLY_RAW_DATA_ALLOWED + ) + if ns_parser: + qa_view.display_cdf( + name=self.coin, + df=self.data, + target=self.target, + export=ns_parser.export, + ) + + @log_start_end(log=logger) + def call_bw(self, other_args: List[str]): + """Process bwy command""" + parser = argparse.ArgumentParser( + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + add_help=False, + prog="bw", + description=""" + Box and Whisker plot + """, + ) + parser.add_argument( + "-y", + "--yearly", + action="store_true", + default=False, + dest="year", + help="Flag to show yearly bw plot", + ) + ns_parser = parse_known_args_and_warn(parser, other_args) + if ns_parser: + qa_view.display_bw( + name=self.coin, + df=self.data, + target=self.target, + yearly=ns_parser.year, + ) + + @log_start_end(log=logger) + def call_decompose(self, other_args: List[str]): + """Process decompose command""" + parser = argparse.ArgumentParser( + add_help=False, + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + prog="decompose", + description=""" + Decompose time series as: + - Additive Time Series = Level + CyclicTrend + Residual + Seasonality + - Multiplicative Time Series = Level * CyclicTrend * Residual * Seasonality + """, + ) + parser.add_argument( + "-m", + "--multiplicative", + action="store_true", + default=False, + dest="multiplicative", + help="decompose using multiplicative model instead of additive", + ) + ns_parser = parse_known_args_and_warn( + parser, other_args, export_allowed=EXPORT_ONLY_RAW_DATA_ALLOWED + ) + if ns_parser: + qa_view.display_seasonal( + name=self.coin, + df=self.data, + target=self.target, + multiplicative=ns_parser.multiplicative, + export=ns_parser.export, + ) + + @log_start_end(log=logger) + def call_cusum(self, other_args: List[str]): + """Process cusum command""" + parser = argparse.ArgumentParser( + add_help=False, + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + prog="cusum", + description=""" + Cumulative sum algorithm (CUSUM) to detect abrupt changes in data + """, + ) + parser.add_argument( + "-t", + "--threshold", + dest="threshold", + type=float, + default=( + max(self.data[self.target].values) - min(self.data[self.target].values) + ) + / 40, + help="threshold", + ) + parser.add_argument( + "-d", + "--drift", + dest="drift", + type=float, + default=( + max(self.data[self.target].values) - min(self.data[self.target].values) + ) + / 80, + help="drift", + ) + ns_parser = parse_known_args_and_warn(parser, other_args) + if ns_parser: + qa_view.display_cusum( + df=self.data, + target=self.target, + threshold=ns_parser.threshold, + drift=ns_parser.drift, + ) + + @log_start_end(log=logger) + def call_acf(self, other_args: List[str]): + """Process acf command""" + parser = argparse.ArgumentParser( + add_help=False, + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + prog="acf", + description=""" + Auto-Correlation and Partial Auto-Correlation Functions for diff and diff diff crypto data + """, + ) + parser.add_argument( + "-l", + "--lags", + dest="lags", + type=check_positive, + default=15, + help="maximum lags to display in plots", + ) + ns_parser = parse_known_args_and_warn(parser, other_args) + if ns_parser: + if self.target != "Close": + console.print( + "Target not Close. For best results, use `pick Close` first." + ) + + qa_view.display_acf( + name=self.coin, + df=self.data, + target=self.target, + lags=ns_parser.lags, + ) + + @log_start_end(log=logger) + def call_rolling(self, other_args: List[str]): + """Process rolling command""" + parser = argparse.ArgumentParser( + add_help=False, + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + prog="rolling", + description=""" + Rolling mean and std deviation + """, + ) + parser.add_argument( + "-w", + "--window", + action="store", + dest="n_window", + type=check_positive, + default=14, + help="Window length", + ) + ns_parser = parse_known_args_and_warn( + parser, other_args, export_allowed=EXPORT_ONLY_RAW_DATA_ALLOWED + ) + if ns_parser: + rolling_view.display_mean_std( + name=self.coin, + df=self.data, + target=self.target, + window=ns_parser.n_window, + export=ns_parser.export, + ) + + @log_start_end(log=logger) + def call_spread(self, other_args: List[str]): + """Process spread command""" + parser = argparse.ArgumentParser( + add_help=False, + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + prog="spread", + description="""Shows rolling spread measurement + """, + ) + parser.add_argument( + "-w", + "--window", + action="store", + dest="n_window", + type=check_positive, + default=14, + help="Window length", + ) + ns_parser = parse_known_args_and_warn( + parser, other_args, export_allowed=EXPORT_ONLY_RAW_DATA_ALLOWED + ) + if ns_parser: + rolling_view.display_spread( + name=self.coin, + df=self.data, + target=self.target, + window=ns_parser.n_window, + export=ns_parser.export, + ) + + @log_start_end(log=logger) + def call_quantile(self, other_args: List[str]): + """Process quantile command""" + parser = argparse.ArgumentParser( + add_help=False, + formatter_class=argparse.ArgumentDefaultsHelpFormatter, + prog="quantile", + description=""" + The quantiles are values which divide the distribution such that + there is a given proportion of observations below the quantile. + For example, the median is a quantile. The median is the central + value of the distribution, such that half the points are less than + or equal to it and half are greater than or equal to it. + + By default, q is set at 0.5, which effectively is median. Change q to + get the desired quantile (0 dd due-diligence for loaded coin, e.g.: coin information, social media, market stats > ta technical analysis for loaded coin, e.g.: ema, macd, rsi, adx, bbands, obv > pred prediction techniques e.g.: regression, arima, rnn, lstm, conv1d, monte carlo +> qa quantitative analysis, e.g.: decompose, cusum, residuals analysis [/unvl]