mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-10-28 04:52:07 +00:00
Refactoring Due Diligence controller
This commit is contained in:
Artem Veremey
219
gamestonk_terminal/due_diligence/dd_controller.py
Normal file
219
gamestonk_terminal/due_diligence/dd_controller.py
Normal file
@@ -0,0 +1,219 @@
|
||||
""" Due Diligence Controller """
|
||||
__docformat__ = "numpy"
|
||||
|
||||
import argparse
|
||||
from typing import List
|
||||
from pandas.core.frame import DataFrame
|
||||
from prompt_toolkit.completion import NestedCompleter
|
||||
|
||||
from gamestonk_terminal.due_diligence import business_insider_api as bi_api
|
||||
from gamestonk_terminal.due_diligence import financial_modeling_prep_api as fmp_api
|
||||
from gamestonk_terminal.due_diligence import finviz_api as fvz_api
|
||||
from gamestonk_terminal.due_diligence import market_watch_api as mw_api
|
||||
from gamestonk_terminal.due_diligence import quandl_api as q_api
|
||||
from gamestonk_terminal.due_diligence import reddit_api as r_api
|
||||
from gamestonk_terminal.due_diligence import news_api
|
||||
from gamestonk_terminal import feature_flags as gtff
|
||||
from gamestonk_terminal.helper_funcs import get_flair
|
||||
from gamestonk_terminal.menu import session
|
||||
|
||||
|
||||
class DueDiligenceController:
|
||||
""" Due Diligence Controller """
|
||||
|
||||
# Command choices
|
||||
CHOICES = [
|
||||
"help",
|
||||
"q",
|
||||
"quit",
|
||||
"red",
|
||||
"short",
|
||||
"rating",
|
||||
"pt",
|
||||
"est",
|
||||
"ins",
|
||||
"insider",
|
||||
"news",
|
||||
"analyst",
|
||||
"warnings",
|
||||
"sec",
|
||||
]
|
||||
|
||||
def __init__(self, stock: DataFrame, ticker: str, start: str, interval: str):
|
||||
"""Constructor
|
||||
|
||||
Parameters
|
||||
----------
|
||||
stock : DataFrame
|
||||
Due diligence stock dataframe
|
||||
ticker : str
|
||||
Due diligence ticker symbol
|
||||
start : str
|
||||
Start date of the stock data
|
||||
interval : str
|
||||
Stock data interval
|
||||
"""
|
||||
self.stock = stock
|
||||
self.ticker = ticker
|
||||
self.start = start
|
||||
self.interval = interval
|
||||
self.dd_parser = argparse.ArgumentParser(add_help=False, prog="dd")
|
||||
self.dd_parser.add_argument(
|
||||
"cmd",
|
||||
choices=self.CHOICES,
|
||||
)
|
||||
|
||||
def print_help(self):
|
||||
""" Print help """
|
||||
|
||||
intraday = (f"Intraday {self.interval}", "Daily")[self.interval == "1440min"]
|
||||
|
||||
if self.start:
|
||||
print(
|
||||
f"\n{intraday} Stock: {self.ticker} (from {self.start.strftime('%Y-%m-%d')})"
|
||||
)
|
||||
else:
|
||||
print(f"\n{intraday} Stock: {self.ticker}")
|
||||
|
||||
print("\nDue Diligence:")
|
||||
print(" help show this fundamental analysis menu again")
|
||||
print(" q quit this menu, and shows back to main menu")
|
||||
print(" quit quit to abandon program")
|
||||
print("")
|
||||
print(" news latest news of the company [News API]")
|
||||
print(" red gets due diligence from another user's post [Reddit]")
|
||||
print(" analyst analyst prices and ratings of the company [Finviz]")
|
||||
print(
|
||||
" rating rating of the company from strong sell to strong buy [FMP]"
|
||||
)
|
||||
print(" pt price targets over time [Business Insider]")
|
||||
print(
|
||||
" est quarter and year analysts earnings estimates [Business Insider]"
|
||||
)
|
||||
print(" ins insider activity over time [Business Insider]")
|
||||
print(" insider insider trading of the company [Finviz]")
|
||||
print(" sec SEC filings [Market Watch]")
|
||||
print(" short short interest [Quandl]")
|
||||
print(
|
||||
" warnings company warnings according to Sean Seah book [Market Watch]"
|
||||
)
|
||||
print("")
|
||||
|
||||
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.dd_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()
|
||||
|
||||
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_red(self, other_args: List[str]):
|
||||
""" Process red command """
|
||||
r_api.due_diligence(other_args, self.ticker)
|
||||
|
||||
def call_insider(self, other_args: List[str]):
|
||||
""" Process insider command """
|
||||
fvz_api.insider(other_args, self.ticker)
|
||||
|
||||
def call_news(self, other_args: List[str]):
|
||||
""" Process news command """
|
||||
news_api.news(other_args, self.ticker)
|
||||
|
||||
def call_analyst(self, other_args: List[str]):
|
||||
""" Process analyst command """
|
||||
fvz_api.analyst(other_args, self.ticker)
|
||||
|
||||
def call_pt(self, other_args: List[str]):
|
||||
""" Process pt command """
|
||||
bi_api.price_target_from_analysts(
|
||||
other_args, self.stock, self.ticker, self.start, self.interval
|
||||
)
|
||||
|
||||
def call_est(self, other_args: List[str]):
|
||||
""" Process est command """
|
||||
bi_api.estimates(other_args, self.ticker)
|
||||
|
||||
def call_ins(self, other_args: List[str]):
|
||||
""" Process ins command """
|
||||
bi_api.insider_activity(
|
||||
other_args, self.stock, self.ticker, self.start, self.interval
|
||||
)
|
||||
|
||||
def call_rating(self, other_args: List[str]):
|
||||
""" Process rating command """
|
||||
fmp_api.rating(other_args, self.ticker)
|
||||
|
||||
def call_warnings(self, other_args: List[str]):
|
||||
""" Process rating command """
|
||||
mw_api.sean_seah_warnings(other_args, self.ticker)
|
||||
|
||||
def call_sec(self, other_args: List[str]):
|
||||
""" Process sec command """
|
||||
mw_api.sec_fillings(other_args, self.ticker)
|
||||
|
||||
def call_short(self, other_args: List[str]):
|
||||
""" Process short command """
|
||||
q_api.short_interest(other_args, self.ticker, self.start)
|
||||
|
||||
|
||||
def menu(stock: DataFrame, ticker: str, start: str, interval: str):
|
||||
"""Due Diligence Menu
|
||||
|
||||
Parameters
|
||||
----------
|
||||
stock : DataFrame
|
||||
Due diligence stock dataframe
|
||||
ticker : str
|
||||
Due diligence ticker symbol
|
||||
start : str
|
||||
Start date of the stock data
|
||||
interval : str
|
||||
Stock data interval
|
||||
"""
|
||||
|
||||
dd_controller = DueDiligenceController(stock, ticker, start, interval)
|
||||
dd_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 dd_controller.CHOICES}
|
||||
)
|
||||
|
||||
an_input = session.prompt(
|
||||
f"{get_flair()} (dd)> ",
|
||||
completer=completer,
|
||||
)
|
||||
else:
|
||||
an_input = input(f"{get_flair()} (dd)> ")
|
||||
|
||||
try:
|
||||
process_input = dd_controller.switch(an_input)
|
||||
|
||||
if process_input is not None:
|
||||
return process_input
|
||||
|
||||
except SystemExit:
|
||||
print("The command selected doesn't exist\n")
|
||||
continue
|
||||
@@ -1,148 +0,0 @@
|
||||
import argparse
|
||||
|
||||
from gamestonk_terminal.due_diligence import business_insider_api as bi_api
|
||||
from gamestonk_terminal.due_diligence import financial_modeling_prep_api as fmp_api
|
||||
from gamestonk_terminal.due_diligence import finviz_api as fvz_api
|
||||
from gamestonk_terminal.due_diligence import market_watch_api as mw_api
|
||||
from gamestonk_terminal.due_diligence import quandl_api as q_api
|
||||
from gamestonk_terminal.due_diligence import reddit_api as r_api
|
||||
from gamestonk_terminal.due_diligence import news_api
|
||||
from gamestonk_terminal import feature_flags as gtff
|
||||
from gamestonk_terminal.helper_funcs import get_flair
|
||||
from gamestonk_terminal.menu import session
|
||||
from prompt_toolkit.completion import NestedCompleter
|
||||
|
||||
|
||||
def print_due_diligence(s_ticker, s_start, s_interval):
|
||||
""" Print help """
|
||||
|
||||
s_intraday = (f"Intraday {s_interval}", "Daily")[s_interval == "1440min"]
|
||||
|
||||
if s_start:
|
||||
print(f"\n{s_intraday} Stock: {s_ticker} (from {s_start.strftime('%Y-%m-%d')})")
|
||||
else:
|
||||
print(f"\n{s_intraday} Stock: {s_ticker}")
|
||||
|
||||
print("\nDue Diligence:")
|
||||
print(" help show this fundamental analysis menu again")
|
||||
print(" q quit this menu, and shows back to main menu")
|
||||
print(" quit quit to abandon program")
|
||||
print("")
|
||||
print(" news latest news of the company [News API]")
|
||||
print(" red gets due diligence from another user's post [Reddit]")
|
||||
print(" analyst analyst prices and ratings of the company [Finviz]")
|
||||
print(" rating rating of the company from strong sell to strong buy [FMP]")
|
||||
print(" pt price targets over time [Business Insider]")
|
||||
print(
|
||||
" est quarter and year analysts earnings estimates [Business Insider]"
|
||||
)
|
||||
print(" ins insider activity over time [Business Insider]")
|
||||
print(" insider insider trading of the company [Finviz]")
|
||||
print(" sec SEC filings [Market Watch]")
|
||||
print(" short short interest [Quandl]")
|
||||
print(
|
||||
" warnings company warnings according to Sean Seah book [Market Watch]"
|
||||
)
|
||||
print("")
|
||||
return
|
||||
|
||||
|
||||
def dd_menu(df_stock, s_ticker, s_start, s_interval):
|
||||
|
||||
# Add list of arguments that the due diligence parser accepts
|
||||
dd_parser = argparse.ArgumentParser(prog="dd", add_help=False)
|
||||
choices = [
|
||||
"info",
|
||||
"help",
|
||||
"q",
|
||||
"quit",
|
||||
"red",
|
||||
"short",
|
||||
"rating",
|
||||
"pt",
|
||||
"est",
|
||||
"ins",
|
||||
"insider",
|
||||
"news",
|
||||
"analyst",
|
||||
"warnings",
|
||||
"sec",
|
||||
]
|
||||
dd_parser.add_argument("cmd", choices=choices)
|
||||
completer = NestedCompleter.from_nested_dict({c: None for c in choices})
|
||||
|
||||
print_due_diligence(s_ticker, s_start, s_interval)
|
||||
|
||||
# Loop forever and ever
|
||||
while True:
|
||||
# Get input command from user
|
||||
if session and gtff.USE_PROMPT_TOOLKIT:
|
||||
as_input = session.prompt(
|
||||
f"{get_flair()} (dd)> ",
|
||||
completer=completer,
|
||||
)
|
||||
else:
|
||||
as_input = input(f"{get_flair()} (dd)> ")
|
||||
|
||||
# Parse due diligence command of the list of possible commands
|
||||
try:
|
||||
(ns_known_args, l_args) = dd_parser.parse_known_args(as_input.split())
|
||||
|
||||
except SystemExit:
|
||||
print("The command selected doesn't exist\n")
|
||||
continue
|
||||
|
||||
if ns_known_args.cmd == "help":
|
||||
print_due_diligence(s_ticker, s_start, s_interval)
|
||||
|
||||
elif ns_known_args.cmd == "q":
|
||||
# Just leave the DD menu
|
||||
return False
|
||||
|
||||
elif ns_known_args.cmd == "quit":
|
||||
# Abandon the program
|
||||
return True
|
||||
|
||||
# REDDIT API
|
||||
elif ns_known_args.cmd == "red":
|
||||
r_api.due_diligence(l_args, s_ticker)
|
||||
|
||||
# FINVIZ API
|
||||
elif ns_known_args.cmd == "insider":
|
||||
fvz_api.insider(l_args, s_ticker)
|
||||
|
||||
elif ns_known_args.cmd == "news":
|
||||
news_api.news(l_args, s_ticker)
|
||||
|
||||
elif ns_known_args.cmd == "analyst":
|
||||
fvz_api.analyst(l_args, s_ticker)
|
||||
|
||||
# BUSINESS INSIDER API
|
||||
elif ns_known_args.cmd == "pt":
|
||||
bi_api.price_target_from_analysts(
|
||||
l_args, df_stock, s_ticker, s_start, s_interval
|
||||
)
|
||||
|
||||
elif ns_known_args.cmd == "est":
|
||||
bi_api.estimates(l_args, s_ticker)
|
||||
|
||||
elif ns_known_args.cmd == "ins":
|
||||
bi_api.insider_activity(l_args, df_stock, s_ticker, s_start, s_interval)
|
||||
|
||||
# FINANCIAL MODELING PREP API
|
||||
elif ns_known_args.cmd == "rating":
|
||||
fmp_api.rating(l_args, s_ticker)
|
||||
|
||||
# MARKET WATCH API
|
||||
elif ns_known_args.cmd == "sec":
|
||||
mw_api.sec_fillings(l_args, s_ticker)
|
||||
|
||||
elif ns_known_args.cmd == "warnings":
|
||||
mw_api.sean_seah_warnings(l_args, s_ticker)
|
||||
|
||||
# QUANDL API
|
||||
elif ns_known_args.cmd == "short":
|
||||
q_api.short_interest(l_args, s_ticker, s_start)
|
||||
|
||||
else:
|
||||
print("Command not recognized!")
|
||||
@@ -15,7 +15,7 @@ from gamestonk_terminal import feature_flags as gtff
|
||||
from gamestonk_terminal import thought_of_the_day as thought
|
||||
from gamestonk_terminal import res_menu as rm
|
||||
from gamestonk_terminal.discovery import disc_controller
|
||||
from gamestonk_terminal.due_diligence import dd_menu as ddm
|
||||
from gamestonk_terminal.due_diligence import dd_controller
|
||||
from gamestonk_terminal.fundamental_analysis import fa_menu as fam
|
||||
from gamestonk_terminal.helper_funcs import b_is_stock_market_open, get_flair
|
||||
from gamestonk_terminal.main_helper import clear, export, load, print_help, view, candle
|
||||
@@ -208,7 +208,7 @@ def main():
|
||||
b_quit = ta_controller.menu(df_stock, s_ticker, s_start, s_interval)
|
||||
|
||||
elif ns_known_args.opt == "dd":
|
||||
b_quit = ddm.dd_menu(df_stock, s_ticker, s_start, s_interval)
|
||||
b_quit = dd_controller.menu(df_stock, s_ticker, s_start, s_interval)
|
||||
|
||||
elif ns_known_args.opt == "eda":
|
||||
if s_interval == "1440min":
|
||||
|
||||
Reference in New Issue
Block a user