mirror of
https://github.com/OpenBB-finance/OpenBB.git
synced 2026-11-07 18:01:10 +00:00
refactored portfolio to be class
This commit is contained in:
Jamie
153
gamestonk_terminal/portfolio/port_controller.py
Normal file
153
gamestonk_terminal/portfolio/port_controller.py
Normal file
@@ -0,0 +1,153 @@
|
||||
__docformat__ = "numpy"
|
||||
|
||||
import argparse
|
||||
from typing import List
|
||||
from matplotlib import pyplot as plt
|
||||
from prompt_toolkit.completion import NestedCompleter
|
||||
from gamestonk_terminal import feature_flags as gtff
|
||||
from gamestonk_terminal.helper_funcs import get_flair
|
||||
from gamestonk_terminal.menu import session
|
||||
from gamestonk_terminal.portfolio import rh_api
|
||||
|
||||
class PortfolioController:
|
||||
|
||||
CHOICES = [
|
||||
"help",
|
||||
"q",
|
||||
"quit",
|
||||
"login",
|
||||
"hold",
|
||||
"rhhist",
|
||||
]
|
||||
|
||||
BROKERS = [
|
||||
"rh",
|
||||
]
|
||||
|
||||
def __init__(self):
|
||||
self.port_parser = argparse.ArgumentParser(add_help=False, prog="port")
|
||||
self.port_parser.add_argument("cmd",
|
||||
choices = self.CHOICES)
|
||||
self.brokers_list = set()
|
||||
|
||||
|
||||
@staticmethod
|
||||
def print_help(broker_list):
|
||||
if not broker_list:
|
||||
print_broke = " None"
|
||||
if broker_list:
|
||||
print_broke=" "
|
||||
for broker in broker_list:
|
||||
print_broke += broker+" "
|
||||
|
||||
""" Print help """
|
||||
print("\nPortfolio:")
|
||||
print(" help show this menu again")
|
||||
print(" q quit this menu, and shows back to main menu")
|
||||
print(" quit quit to abandon program")
|
||||
print(" login login to your brokers")
|
||||
print("\nCurrent Brokers :" + print_broke)
|
||||
print("")
|
||||
print("\nCurrently Supported :")
|
||||
print(" rh Robinhood - fuck these guys")
|
||||
print("")
|
||||
|
||||
|
||||
def print_portfolio_menu(self):
|
||||
|
||||
print_broke = " "
|
||||
for broker in self.brokers_list:
|
||||
print_broke += broker + " "
|
||||
print("\nCurrent Brokers :" + print_broke)
|
||||
print("")
|
||||
print(" hold check holdings")
|
||||
print(" rhhist plot historical RH portfolio")
|
||||
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.port_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.brokers_list)
|
||||
|
||||
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_login(self, other_args):
|
||||
|
||||
if not other_args:
|
||||
print("Please enter brokers you wish to login to")
|
||||
print("")
|
||||
return
|
||||
for broker in other_args:
|
||||
api = broker + "_api"
|
||||
self.brokers_list.add(broker)
|
||||
try:
|
||||
eval(api+".login()")
|
||||
except Exception as e:
|
||||
print("")
|
||||
print(f"Error at broker : {broker}")
|
||||
print(e)
|
||||
print("Make sure credentials are defined in config_terminal.py ")
|
||||
print("")
|
||||
|
||||
self.print_portfolio_menu()
|
||||
|
||||
def call_rhhist(self, other_args: List[str]):
|
||||
rh_api.plot_historical(other_args)
|
||||
|
||||
def call_hold(self,_):
|
||||
try:
|
||||
rh_api.show_holdings()
|
||||
except Exception as e:
|
||||
print(e)
|
||||
print("")
|
||||
|
||||
def menu():
|
||||
"""Portfolio Analysis Menu"""
|
||||
|
||||
port_controller = PortfolioController()
|
||||
port_controller.print_help(port_controller.brokers_list)
|
||||
|
||||
while True:
|
||||
# Get input command from user
|
||||
if session and gtff.USE_PROMPT_TOOLKIT:
|
||||
completer = NestedCompleter.from_nested_dict(
|
||||
{c: None for c in port_controller.CHOICES}
|
||||
)
|
||||
an_input = session.prompt(
|
||||
f"{get_flair()} (port)> ",
|
||||
completer=completer,
|
||||
)
|
||||
else:
|
||||
an_input = input(f"{get_flair()} (port)> ")
|
||||
|
||||
try:
|
||||
process_input = port_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,100 +0,0 @@
|
||||
import argparse
|
||||
from matplotlib import pyplot as plt
|
||||
from prompt_toolkit.completion import NestedCompleter
|
||||
from gamestonk_terminal import feature_flags as gtff
|
||||
from gamestonk_terminal.helper_funcs import get_flair
|
||||
from gamestonk_terminal.menu import session
|
||||
from gamestonk_terminal.portfolio import rh_api
|
||||
|
||||
|
||||
def print_port(show_login):
|
||||
""" Print help """
|
||||
print("\nPortfolio:")
|
||||
print(" help show this menu again")
|
||||
print(" q quit this menu, and shows back to main menu")
|
||||
print(" quit quit to abandon program")
|
||||
print(" ")
|
||||
print("Robinhood:")
|
||||
print("")
|
||||
if show_login:
|
||||
print(" login login")
|
||||
print(" hold look at current holdings")
|
||||
print(" rhhist look at historical portfolio")
|
||||
print(" ")
|
||||
|
||||
|
||||
def port_menu():
|
||||
plt.close("all")
|
||||
port_parser = argparse.ArgumentParser(prog="port", add_help=False)
|
||||
choices = ["help", "q", "quit", "hold", "rhhist", "login"]
|
||||
port_parser.add_argument("cmd", choices=choices)
|
||||
completer = NestedCompleter.from_nested_dict({c: None for c in choices})
|
||||
should_print_help = True
|
||||
print_login = True
|
||||
while True:
|
||||
if should_print_help:
|
||||
print_port(print_login)
|
||||
should_print_help = False
|
||||
|
||||
# Get input command from user
|
||||
if session and gtff.USE_PROMPT_TOOLKIT:
|
||||
as_input = session.prompt(
|
||||
f"{get_flair()} (port)> ",
|
||||
completer=completer,
|
||||
)
|
||||
else:
|
||||
as_input = input(f"{get_flair()} (port)> ")
|
||||
try:
|
||||
(ns_known_args, l_args) = port_parser.parse_known_args(as_input.split())
|
||||
|
||||
except SystemExit:
|
||||
print("The command selected doesn't exist\n")
|
||||
continue
|
||||
|
||||
if ns_known_args.cmd == "help":
|
||||
should_print_help = True
|
||||
|
||||
elif ns_known_args.cmd == "q":
|
||||
# Leave the port menu + logout of robinhood
|
||||
# logoff raises error if not logged in
|
||||
try:
|
||||
rh_api.logoff()
|
||||
except Exception:
|
||||
pass
|
||||
return False
|
||||
|
||||
elif ns_known_args.cmd == "quit":
|
||||
# Will raise exception if not logged in
|
||||
try:
|
||||
rh_api.logoff()
|
||||
except Exception:
|
||||
pass
|
||||
# Abandon the program
|
||||
return True
|
||||
|
||||
elif ns_known_args.cmd == "hold":
|
||||
try:
|
||||
rh_api.show_holdings()
|
||||
except Exception as e:
|
||||
print(e)
|
||||
print("")
|
||||
|
||||
elif ns_known_args.cmd == "rhhist":
|
||||
rh_api.plot_historical(l_args)
|
||||
|
||||
elif ns_known_args.cmd == "login":
|
||||
try:
|
||||
rh_api.login()
|
||||
should_print_help = True
|
||||
print_login = False
|
||||
except Exception as e:
|
||||
print("")
|
||||
print(e)
|
||||
print("Make sure credentials are defined in config_terminal.py ")
|
||||
print("")
|
||||
should_print_help = False
|
||||
print_login = True
|
||||
|
||||
else:
|
||||
print("Command not recognized")
|
||||
print("")
|
||||
@@ -25,8 +25,7 @@ from gamestonk_terminal.technical_analysis import ta_menu as tam
|
||||
from gamestonk_terminal.comparison_analysis import ca_menu as cam
|
||||
from gamestonk_terminal.options import op_menu as opm
|
||||
from gamestonk_terminal.fred import fred_menu as fm
|
||||
from gamestonk_terminal.portfolio import port_menu
|
||||
|
||||
from gamestonk_terminal.portfolio import port_controller
|
||||
# import warnings
|
||||
# warnings.simplefilter("always")
|
||||
|
||||
@@ -199,8 +198,8 @@ def main():
|
||||
b_quit = fm.fred_menu()
|
||||
|
||||
elif ns_known_args.opt == "port":
|
||||
b_quit = port_menu.port_menu()
|
||||
|
||||
#b_quit = port_menu.port_menu()
|
||||
b_quit = port_controller.menu()
|
||||
elif ns_known_args.opt == "pred":
|
||||
|
||||
if not gtff.ENABLE_PREDICT:
|
||||
|
||||
Reference in New Issue
Block a user