add exploratory data analysis menu

This commit is contained in:
didier
2021-03-27 17:22:39 +00:00
parent 137ea2147d
commit bb9ab3a9d2
4 changed files with 817 additions and 0 deletions

View File

@@ -0,0 +1,619 @@
""" Exploratory Data Analysis API """
__docformat__ = "numpy"
import argparse
from typing import List
from detecta import detect_cusum
from matplotlib import pyplot as plt
import pandas as pd
from pandas.plotting import register_matplotlib_converters
import numpy as np
import seaborn as sns
import statsmodels.api as sm
import matplotlib.gridspec as gridspec
from statsmodels.tsa.seasonal import seasonal_decompose
from gamestonk_terminal.helper_funcs import (
parse_known_args_and_warn,
plot_autoscale,
check_positive,
)
from gamestonk_terminal.config_plot import PLOT_DPI
from gamestonk_terminal import feature_flags as gtff
register_matplotlib_converters()
def summary(other_args: List[str], stock: pd.DataFrame):
"""Print summary statistics
Parameters
----------
other_args : str
Command line arguments to be processed with argparse
ticker : str
Ticker of the stock
stock : pd.DataFrame
Stock data
"""
parser = argparse.ArgumentParser(
add_help=False,
prog="summary",
description="""
Summary statistics
""",
)
try:
ns_parser = parse_known_args_and_warn(parser, other_args)
if not ns_parser:
return
df_stats = stock.describe(percentiles=[0.1, 0.25, 0.5, 0.75, 0.9])
df_stats.loc["var"] = df_stats.loc["std"] ** 2
print(df_stats.round(2).to_string())
print("")
except Exception as e:
print(e, "\n")
return
def hist(other_args: List[str], ticker: str, stock: pd.DataFrame):
"""Plot histogram and density
Parameters
----------
other_args : str
Command line arguments to be processed with argparse
ticker : str
Ticker of the stock
stock : pd.DataFrame
Stock data
"""
parser = argparse.ArgumentParser(
add_help=False,
prog="hist",
description="""
Histogram with depicted density and rug
""",
)
try:
ns_parser = parse_known_args_and_warn(parser, other_args)
if not ns_parser:
return
plt.figure(figsize=plot_autoscale(), dpi=PLOT_DPI)
stock = stock["5. adjusted close"]
sns.distplot(
stock,
bins=35,
color="blue",
hist=True,
hist_kws={"edgecolor": "black"},
kde=True,
kde_kws={"color": "black", "lw": 3, "label": "KDE"},
rug=True,
rug_kws={"edgecolor": "orange"},
)
plt.title(ticker + " (Histogram with Density)")
plt.ylabel("Density")
plt.xlabel("Share Price")
plt.grid(True)
if gtff.USE_ION:
plt.ion()
plt.show()
print("")
except Exception as e:
print(e, "\n")
return
def cdf(other_args: List[str], ticker: str, stock: pd.DataFrame):
"""Plot cumulative distribution function
Parameters
----------
other_args : str
Command line arguments to be processed with argparse
ticker : str
Ticker of the stock
stock : pd.DataFrame
Stock data
"""
parser = argparse.ArgumentParser(
add_help=False,
prog="cdf",
description="""
Cumulative distribution function
""",
)
try:
ns_parser = parse_known_args_and_warn(parser, other_args)
if not ns_parser:
return
plt.figure(figsize=plot_autoscale(), dpi=PLOT_DPI)
stock = stock["5. adjusted close"]
cdf = stock.value_counts().sort_index().div(len(stock)).cumsum()
cdf.plot(lw=2)
plt.title(ticker + " (Cumulative Distribution Function)")
plt.ylabel("Probability")
plt.xlabel("Share Price")
minVal = stock.values.min()
q25 = np.quantile(stock.values, 0.25)
medianVal = np.quantile(stock.values, 0.5)
q75 = np.quantile(stock.values, 0.75)
data = [
(minVal, q25),
(0.25, 0.25),
"r",
(q25, q25),
(0, 0.25),
"r",
(minVal, medianVal),
(0.5, 0.5),
"r",
(medianVal, medianVal),
(0, 0.5),
"r",
(minVal, q75),
(0.75, 0.75),
"r",
(q75, q75),
(0, 0.75),
"r",
]
plt.plot(*data, ls="--")
plt.text(minVal + (q25 - minVal) / 2, 0.27, "Q1", color="r", fontweight="bold")
plt.text(
minVal + (medianVal - minVal) / 2,
0.52,
"Median",
color="r",
fontweight="bold",
)
plt.text(minVal + (q75 - minVal) / 2, 0.77, "Q3", color="r", fontweight="bold")
plt.xlim(cdf.index[0], cdf.index[-1])
plt.grid(True)
if gtff.USE_ION:
plt.ion()
plt.show()
print("")
except Exception as e:
print(e, "\n")
return
def bwy(other_args: List[str], ticker: str, stock: pd.DataFrame):
"""Box and Whisker plot yearly
Parameters
----------
other_args : str
Command line arguments to be processed with argparse
ticker : str
Ticker of the stock
stock : pd.DataFrame
Stock data
"""
parser = argparse.ArgumentParser(
add_help=False,
prog="bwy",
description="""
Box and Whisker plot yearly
""",
)
try:
ns_parser = parse_known_args_and_warn(parser, other_args)
if not ns_parser:
return
plt.figure(figsize=plot_autoscale(), dpi=PLOT_DPI)
stock = stock["5. adjusted close"]
sns.set(style="whitegrid")
box_plot = sns.boxplot(x=stock.index.year, y=stock)
box_plot.set(
xlabel="Year", ylabel="Share Price", title=ticker + " (Box-plot per Year)"
)
if gtff.USE_ION:
plt.ion()
plt.show()
print("")
except Exception as e:
print(e, "\n")
return
def bwm(other_args: List[str], ticker: str, stock: pd.DataFrame):
"""Box and Whisker plot monthly
Parameters
----------
other_args : str
Command line arguments to be processed with argparse
ticker : str
Ticker of the stock
stock : pd.DataFrame
Stock data
"""
parser = argparse.ArgumentParser(
add_help=False,
prog="bwm",
description="""
Box and Whisker plot monthly
""",
)
try:
ns_parser = parse_known_args_and_warn(parser, other_args)
if not ns_parser:
return
plt.figure(figsize=plot_autoscale(), dpi=PLOT_DPI)
stock = stock["5. adjusted close"]
sns.set(style="whitegrid")
box_plot = sns.boxplot(x=stock.index.month, y=stock)
box_plot.set(
xlabel="Month",
ylabel="Share Price",
title=ticker + " (Box-plot per Month)",
)
l_months = [
"Jan",
"Feb",
"Mar",
"Apr",
"May",
"Jun",
"Jul",
"Aug",
"Sep",
"Oct",
"Nov",
"Dec",
]
l_ticks = list()
for val in box_plot.get_xticklabels():
l_ticks.append(l_months[int(val.get_text()) - 1])
box_plot.set_xticklabels(l_ticks)
if gtff.USE_ION:
plt.ion()
plt.show()
print("")
except Exception as e:
print(e, "\n")
return
def rolling(other_args: List[str], ticker: str, stock: pd.DataFrame):
"""Rolling mean and std deviation
Parameters
----------
other_args : str
Command line arguments to be processed with argparse
ticker : str
Ticker of the stock
stock : pd.DataFrame
Stock data
"""
parser = argparse.ArgumentParser(
add_help=False,
prog="rolling",
description="""
Rolling mean and std deviation
""",
)
parser.add_argument(
"-w",
"--window",
dest="rolling_window",
type=check_positive,
default=100,
help="rolling window",
)
try:
ns_parser = parse_known_args_and_warn(parser, other_args)
if not ns_parser:
return
stock = stock["5. adjusted close"]
rolling_mean = stock.rolling(
ns_parser.rolling_window, center=True, min_periods=1
).mean()
rolling_std = stock.rolling(
ns_parser.rolling_window, center=True, min_periods=1
).std()
fig, axMean = plt.subplots(figsize=plot_autoscale(), dpi=PLOT_DPI)
axMean.plot(stock.index, stock.values, label=ticker, linewidth=2, color="black")
axMean.plot(rolling_mean, linestyle="--", linewidth=3, color="blue")
axMean.set_xlabel("Time")
axMean.set_ylabel("Share Price", color="blue")
axMean.legend(["Real values", "Rolling Mean"], loc=2)
axMean.tick_params(axis="y", labelcolor="blue")
axStd = axMean.twinx()
axStd.plot(
rolling_std, label="Rolling std", linestyle="--", color="green", linewidth=3
)
axStd.set_ylabel("Std Deviation")
axStd.legend(["Rolling std"], loc=1)
axStd.set_ylabel("Share Price standard deviation", color="green")
axStd.tick_params(axis="y", labelcolor="green")
axMean.set_title(
"Rolling mean and std with window "
+ str(ns_parser.rolling_window)
+ " applied to "
+ ticker
)
plt.xlim([stock.index[0], stock.index[-1]])
plt.grid(b=True, which="major", color="#666666", linestyle="-")
plt.minorticks_on()
plt.grid(b=True, which="minor", color="#999999", linestyle="-", alpha=0.2)
if gtff.USE_ION:
plt.ion()
plt.show()
print("")
except Exception as e:
print(e, "\n")
return
def decompose(other_args: List[str], ticker: str, stock: pd.DataFrame):
"""Decompose time series as:
- Additive Time Series = Level + CyclicTrend + Residual + Seasonality
- Multiplicative Time Series = Level * CyclicTrend * Residual * Seasonality
Parameters
----------
other_args : str
Command line arguments to be processed with argparse
ticker : str
Ticker of the stock
stock : pd.DataFrame
Stock data
"""
parser = argparse.ArgumentParser(
add_help=False,
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",
)
try:
ns_parser = parse_known_args_and_warn(parser, other_args)
if not ns_parser:
return
stock = stock["5. adjusted close"]
seasonal_periods = 5
# Hodrick-Prescott filter
# See Ravn and Uhlig: http://home.uchicago.edu/~huhlig/papers/uhlig.ravn.res.2002.pdf
lamb = 107360000000
fig = plt.figure(
figsize=plot_autoscale(), dpi=PLOT_DPI, constrained_layout=True
)
spec = gridspec.GridSpec(ncols=4, nrows=5, figure=fig)
fig.add_subplot(spec[0, :])
plt.plot(stock)
plt.title(ticker + " (Time-Series)")
if ns_parser.multiplicative:
resultMul = seasonal_decompose(
stock, model="multiplicative", period=seasonal_periods
)
cycleMul, trendMul = sm.tsa.filters.hpfilter(
resultMul.trend[resultMul.trend.notna().values], lamb=lamb
)
# Multiplicative model
fig.add_subplot(spec[1, :4])
plt.plot(resultMul.trend, lw=2, c="purple")
plt.xlim([stock.index[0], stock.index[-1]])
plt.title("Multiplicative Cyclic-Trend")
fig.add_subplot(spec[2, 0:2])
plt.plot(trendMul, lw=2, c="tab:blue")
plt.xlim([stock.index[0], stock.index[-1]])
plt.title("Multiplicative Trend component")
fig.add_subplot(spec[2, 2:])
plt.plot(cycleMul, lw=2, c="green")
plt.xlim([stock.index[0], stock.index[-1]])
plt.title("Multiplicative Cycle component")
fig.add_subplot(spec[3, :])
plt.plot(resultMul.seasonal, lw=2, c="orange")
plt.xlim([stock.index[0], stock.index[-1]])
plt.title("Multiplicative Seasonal effect")
fig.add_subplot(spec[4, :])
plt.plot(resultMul.resid, lw=2, c="red")
plt.xlim([stock.index[0], stock.index[-1]])
plt.title("Multiplicative Residuals")
else:
resultAdd = seasonal_decompose(
stock, model="additive", period=seasonal_periods
)
cycleAdd, trendAdd = sm.tsa.filters.hpfilter(
resultAdd.trend[resultAdd.trend.notna().values], lamb=lamb
)
# Additive model
fig.add_subplot(spec[1, :4])
plt.plot(resultAdd.trend, lw=2, c="purple")
plt.xlim([stock.index[0], stock.index[-1]])
plt.title("Additive Cyclic-Trend")
fig.add_subplot(spec[2, 0:2])
plt.plot(trendAdd, lw=2, c="tab:blue")
plt.xlim([stock.index[0], stock.index[-1]])
plt.title("Additive Trend component")
fig.add_subplot(spec[2, 2:])
plt.plot(cycleAdd, lw=2, c="green")
plt.xlim([stock.index[0], stock.index[-1]])
plt.title("Additive Cycle component")
fig.add_subplot(spec[3, :])
plt.plot(resultAdd.seasonal, lw=2, c="orange")
plt.xlim([stock.index[0], stock.index[-1]])
plt.title("Additive Seasonal effect")
fig.add_subplot(spec[4, :])
plt.plot(resultAdd.resid, lw=2, c="red")
plt.xlim([stock.index[0], stock.index[-1]])
plt.title("Additive Residuals")
if gtff.USE_ION:
plt.ion()
plt.show()
print("")
# From # https://otexts.com/fpp2/seasonal-strength.html
print("Time-Series Level is " + str(round(stock.mean(), 2)))
if ns_parser.multiplicative:
FtMul = max(0, 1 - np.var(resultMul.resid)) / np.var(
resultMul.trend + resultMul.resid
)
print("Strenght of Trend: %.4f" % FtMul)
FsMul = max(
0,
1
- np.var(resultMul.resid)
/ np.var(resultMul.seasonal + resultMul.resid),
)
print("Strenght of Seasonality: %.4f" % FsMul)
else:
FtAdd = max(
0,
1 - np.var(resultAdd.resid) / np.var(resultAdd.trend + resultAdd.resid),
)
print("Strenght of Trend: %.4f" % FtAdd)
FsAdd = max(
0,
1
- np.var(resultAdd.resid)
/ np.var(resultAdd.seasonal + resultAdd.resid),
)
print("Strenght of Seasonality: %.4f" % FsAdd)
print("")
except Exception as e:
print(e, "\n")
return
def cusum(other_args: List[str], ticker: str, stock: pd.DataFrame):
"""Cumulative sum algorithm (CUSUM) to detect abrupt changes in data
Parameters
----------
other_args : str
Command line arguments to be processed with argparse
ticker : str
Ticker of the stock
stock : pd.DataFrame
Stock data
"""
parser = argparse.ArgumentParser(
add_help=False,
prog="cusum",
description="""
Cumulative sum algorithm (CUSUM) to detect abrupt changes in data
""",
)
parser.add_argument(
"-t",
"--threshold",
dest="threshold",
type=float,
default=(
max(stock["5. adjusted close"].values)
- min(stock["5. adjusted close"].values)
)
/ 40,
help="threshold",
)
parser.add_argument(
"-d",
"--drift",
dest="drift",
type=float,
default=(
max(stock["5. adjusted close"].values)
- min(stock["5. adjusted close"].values)
)
/ 80,
help="drift",
)
try:
ns_parser = parse_known_args_and_warn(parser, other_args)
if not ns_parser:
return
stock = stock["5. adjusted close"]
detect_cusum(stock.values, ns_parser.threshold, ns_parser.drift, True, True)
if gtff.USE_ION:
plt.ion()
plt.show()
print("")
except Exception as e:
print(e, "\n")
return

View File

@@ -0,0 +1,172 @@
"""Exploratory Data Analysis Controller Module"""
__docformat__ = "numpy"
import argparse
from typing import List
from datetime import datetime
import pandas as pd
from matplotlib import pyplot as plt
from gamestonk_terminal.exploratory_data_analysis import eda_api
from gamestonk_terminal import feature_flags as gtff
from gamestonk_terminal import config_terminal as cfg
from gamestonk_terminal.helper_funcs import get_flair, parse_known_args_and_warn
from gamestonk_terminal.menu import session
from prompt_toolkit.completion import NestedCompleter
class EdaController:
"""EDA Controller class"""
# Command choices
CHOICES = [
"help",
"q",
"quit",
"summary",
"hist",
"cdf",
"bwy",
"bwm",
"rolling",
"decompose",
"cusum",
]
def __init__(
self,
stock: pd.DataFrame,
ticker: str,
start: datetime,
interval: str,
):
"""Constructor"""
self.stock = stock
self.ticker = ticker
self.start = start
self.interval = interval
self.stats_parser = argparse.ArgumentParser(add_help=False, prog="stats")
self.stats_parser.add_argument(
"cmd",
choices=self.CHOICES,
)
@staticmethod
def print_help(self):
"""Print help"""
s_intraday = (f"Intraday {self.interval}", "Daily")[self.interval == "1440min"]
if self.start:
print(
f"\n{s_intraday} Stock: {self.ticker} (from {self.start.strftime('%Y-%m-%d')})"
)
else:
print(f"\n{s_intraday} Stock: {self.ticker}")
print("\nExploratory Data Analysis:")
print(" help show this comparison analysis menu again")
print(" q quit this menu, and shows back to main menu")
print(" quit quit to abandon program")
print("")
print(" summary brief summary statistics")
print(" hist histogram with density plot")
print(" cdf cumulative distribution function")
print(" bwy box and whisker yearly plot")
print(" bwm box and whisker monthly plot")
print(" rolling rolling mean and std deviation")
print(" decompose decomposition in cyclic-trend, season, and residuals")
print(" cusum detects abrupt changes using cumulative sum algorithm")
print("")
return
def switch(self, an_input: str):
"""Process and dispatch input
Returns
-------
True, False or None
False - quit the menu
True - quit the program
None - continue in the menu
"""
(known_args, other_args) = self.stats_parser.parse_known_args(an_input.split())
return getattr(
self, "call_" + known_args.cmd, lambda: "Command not recognized!"
)(other_args)
def call_help(self, _):
"""Process Help command"""
self.print_help(self)
def call_q(self, _):
"""Process Q command - quit the menu"""
return False
def call_quit(self, _):
"""Process Quit command - quit the program"""
return True
def call_summary(self, other_args: List[str]):
"""Process summary command"""
eda_api.summary(other_args, self.stock)
def call_hist(self, other_args: List[str]):
"""Process hist command"""
eda_api.hist(other_args, self.ticker, self.stock)
def call_cdf(self, other_args: List[str]):
"""Process cdf command"""
eda_api.cdf(other_args, self.ticker, self.stock)
def call_bwy(self, other_args: List[str]):
"""Process bwy command"""
eda_api.bwy(other_args, self.ticker, self.stock)
def call_bwm(self, other_args: List[str]):
"""Process bwm command"""
eda_api.bwm(other_args, self.ticker, self.stock)
def call_rolling(self, other_args: List[str]):
"""Process rolling command"""
eda_api.rolling(other_args, self.ticker, self.stock)
def call_decompose(self, other_args: List[str]):
"""Process decompose command"""
eda_api.decompose(other_args, self.ticker, self.stock)
def call_cusum(self, other_args: List[str]):
"""Process cusum command"""
eda_api.cusum(other_args, self.ticker, self.stock)
def menu(stock: pd.DataFrame, ticker: str, start: datetime, interval: str):
"""Statistics Menu"""
eda_controller = EdaController(stock, ticker, start, interval)
eda_controller.call_help(None)
while True:
# Get input command from user
if session and gtff.USE_PROMPT_TOOLKIT:
completer = NestedCompleter.from_nested_dict(
{c: None for c in eda_controller.CHOICES}
)
an_input = session.prompt(
f"{get_flair()} (eda)> ",
completer=completer,
)
else:
an_input = input(f"{get_flair()} (eda)> ")
try:
plt.close("all")
process_input = eda_controller.switch(an_input)
if process_input is not None:
return process_input
except SystemExit:
print("The command selected doesn't exist\n")
continue

View File

@@ -75,6 +75,9 @@ def print_help(s_ticker, s_start, s_interval, b_is_market_open):
print(
" dd in-depth due-diligence, \t e.g.: news, analyst, shorts, insider, sec"
)
print(
" eda exploratory data analysis,\t e.g.: decompose, cusum, residuals analysis"
)
print(
" pred prediction techniques, \t e.g.: regression, arima, rnn, lstm, prophet"
)

View File

@@ -23,6 +23,7 @@ from gamestonk_terminal.papermill import papermill_controller as mill
from gamestonk_terminal.behavioural_analysis import ba_controller
from gamestonk_terminal.technical_analysis import ta_menu as tam
from gamestonk_terminal.comparison_analysis import ca_controller
from gamestonk_terminal.exploratory_data_analysis import eda_controller
from gamestonk_terminal.options import op_controller
from gamestonk_terminal.fred import fred_menu as fm
from gamestonk_terminal.portfolio import port_controller
@@ -75,6 +76,7 @@ def main():
"fa",
"ta",
"dd",
"eda",
"pred",
"ca",
"op",
@@ -202,6 +204,27 @@ def main():
elif ns_known_args.opt == "dd":
b_quit = ddm.dd_menu(df_stock, s_ticker, s_start, s_interval)
elif ns_known_args.opt == "eda":
if s_interval == "1440min":
b_quit = eda_controller.menu(df_stock, s_ticker, s_start, s_interval)
else:
df_stock = yf.download(
ns_parser.s_ticker, start=ns_parser.s_start_date, progress=False
)
df_stock = df_stock.rename(
columns={
"Open": "1. open",
"High": "2. high",
"Low": "3. low",
"Close": "4. close",
"Adj Close": "5. adjusted close",
"Volume": "6. volume",
}
)
df_stock.index.name = "date"
b_quit = eda_controller.menu(df_stock, s_ticker, s_start, s_interval)
elif ns_known_args.opt == "op":
b_quit = op_controller.menu(s_ticker)