From 30a841ee101aed8df2ccc2b8eda4efdbb6b65cf3 Mon Sep 17 00:00:00 2001 From: Chavithra PARANA Date: Sun, 23 May 2021 10:36:32 +0200 Subject: [PATCH] degiro integration : update mvc --- .../portfolio/degiro_controller.py | 244 ++++------------ gamestonk_terminal/portfolio/degiro_view.py | 262 +++++++++++++++++- 2 files changed, 295 insertions(+), 211 deletions(-) diff --git a/gamestonk_terminal/portfolio/degiro_controller.py b/gamestonk_terminal/portfolio/degiro_controller.py index 13730f70a30..d2b565feb69 100644 --- a/gamestonk_terminal/portfolio/degiro_controller.py +++ b/gamestonk_terminal/portfolio/degiro_controller.py @@ -3,23 +3,18 @@ import argparse # IMPORTATION THIRDPARTY from prompt_toolkit.completion import NestedCompleter -from trading.pb.trading_pb2 import ( - Credentials, - Order, -) # IMPORTATION INTERNAL +import gamestonk_terminal.config_terminal as config + from gamestonk_terminal import feature_flags as gtff from gamestonk_terminal.helper_funcs import ( get_flair, parse_known_args_and_warn, ) from gamestonk_terminal.menu import session -from gamestonk_terminal.portfolio.degiro_model import DegiroModel from gamestonk_terminal.portfolio.degiro_view import DegiroView -# pylint: disable=no-member - class DegiroController: CHOICES = [ @@ -38,26 +33,8 @@ class DegiroController: "update", ] - ORDER_ACTION = { - "buy": Order.Action.BUY, - "sell": Order.Action.SELL, - } - - ORDER_DURATION = { - "gtd": Order.TimeType.GOOD_TILL_DAY, - "gtc": Order.TimeType.GOOD_TILL_CANCELED, - } - - ORDER_TYPE = { - "limit": Order.OrderType.LIMIT, - "market": Order.OrderType.MARKET, - "stop-limit": Order.OrderType.STOP_LIMIT, - "stop-loss": Order.OrderType.STOP_LOSS, - } - def __init__(self): - self.__degiro_model = DegiroModel() - + self.__degiro_view = DegiroView() self.__degiro_parser = argparse.ArgumentParser( add_help=False, prog="degiro", @@ -79,17 +56,7 @@ class DegiroController: ) ns_parser = parse_known_args_and_warn(parser, l_args) - # GET ATTRIBUTES - degiro_model = self.__degiro_model - - # CANCEL ORDER - order_id = ns_parser.id - - # DISPLAY DATA - if degiro_model.cancel(order_id=order_id): - DegiroView.cancel_display_success(order_id=order_id) - else: - DegiroView.cancel_display_fail(order_id=order_id) + self.__degiro_view.cancel(ns_parser=ns_parser) def companynews(self, l_args): """Display news related to a company using its ISIN.""" @@ -106,26 +73,11 @@ class DegiroController: ) ns_parser = parse_known_args_and_warn(parser, l_args) - # GET ATTRIBUTES - degiro_model = self.__degiro_model - - # FETCH DATA - news_by_company = degiro_model.companynews(isin=ns_parser.isin) - - # DISPLAY DATA - DegiroView.companynews_display(news_by_company=news_by_company) + self.__degiro_view.companynews(ns_parser=ns_parser) def create(self, l_args): """Create an order.""" - # GET CONSTANTS - ORDER_ACTION = self.ORDER_ACTION - ORDER_DURATION = self.ORDER_DURATION - ORDER_TYPE = self.ORDER_TYPE - - # GET ATTRIBUTES - degiro_model = self.__degiro_model - # PARSE ARGS parser = argparse.ArgumentParser( add_help=False, @@ -134,7 +86,7 @@ class DegiroController: parser.add_argument( "-a", "--action", - choices=ORDER_ACTION.keys(), + choices=DegiroView.ORDER_ACTION.keys(), default="buy", help="Action wanted.", required=False, @@ -183,7 +135,7 @@ class DegiroController: "-d", "--duration", default="gtd", - choices=ORDER_DURATION.keys(), + choices=DegiroView.ORDER_DURATION.keys(), help="Duration of the Order.", required=False, type=str, @@ -191,7 +143,7 @@ class DegiroController: parser.add_argument( "-t", "--type", - choices=ORDER_TYPE.keys(), + choices=DegiroView.ORDER_TYPE.keys(), default="limit", help="Type of the Order.", required=False, @@ -199,77 +151,28 @@ class DegiroController: ) ns_parser = parse_known_args_and_warn(parser, l_args) - # SETUP ORDER - action = self.ORDER_ACTION[ns_parser.action] - order_type = self.ORDER_TYPE[ns_parser.type] - price = ns_parser.price - product_id = degiro_model.create_calculate_product_id( - product=ns_parser.product, - symbol=ns_parser.symbol, - ) - size = degiro_model.create_calculate_size( - price=ns_parser.price, - size=ns_parser.size, - up_to=ns_parser.up_to, - ) - time_type = self.ORDER_DURATION[ns_parser.duration] - order = Order( - action=action, - order_type=order_type, - price=price, - product_id=product_id, - size=size, - time_type=time_type, - ) - - # CHECK ORDER - checking_response = degiro_model.create_check(order=order) - DegiroView.create_display_check( - order=order, - checking_response=checking_response, - ) - - # USER INPUT - message_ask = DegiroView.create_message_ask_confirmation() - confirmation = input(message_ask) - - # EXECUTE ORDER - if confirmation in ["y", "yes"]: - confirmation_id = checking_response.confirmation_id - confirmation_response = degiro_model.create_confirm( - confirmation_id=confirmation_id, - order=order, - ) - order.id = confirmation_response.orderId - DegiroView.create_display_created_order(order=order) - else: - DegiroView.create_display_canceled() + self.__degiro_view.create(ns_parser=ns_parser) def help(self): """Show the help menu.""" DegiroView.help_display() - def hold(self, _): + def hold(self, l_args): """Display held products.""" - # GET ATTRIBUTES - degiro_model = self.__degiro_model + # PARSE ARGS + parser = argparse.ArgumentParser( + add_help=False, + prog="hold", + ) + ns_parser = parse_known_args_and_warn(parser, l_args) - # FETCH HELD PRODUCTS - positions = degiro_model.hold_positions() - - if len(positions) == 0: - DegiroView.hold_display_no_position() - else: - DegiroView.hold_display_positions(positions=positions) + self.__degiro_view.hold(ns_parser=ns_parser) def lastnews(self, l_args): """Display latest news.""" - # GET ATTRIBUTES - degiro_model = self.__degiro_model - # PARSE ARGS parser = argparse.ArgumentParser( add_help=False, @@ -284,19 +187,13 @@ class DegiroController: required=False, ) ns_parser = parse_known_args_and_warn(parser, l_args) - latest_news = degiro_model.lastnews(limit=ns_parser.limit) - # DISPLAY DATA - DegiroView.lastnews_display(latest_news=latest_news) + self.__degiro_view.lastnews(ns_parser=ns_parser) def login(self, l_args): """Connect to Degiro's API.""" - # GET ATTRIBUTES - degiro_model = self.__degiro_model - # PARSE ARGS - default_credentials = degiro_model.login_default_credentials() parser = argparse.ArgumentParser( add_help=False, prog="login", @@ -305,14 +202,14 @@ class DegiroController: "-u", "--username", type=str, - default=default_credentials.username, + default=config.DG_USERNAME, help="Username in Degiro's account.", ) parser.add_argument( "-p", "--password", type=str, - default=default_credentials.password, + default=config.DG_PASSWORD, help="Password in Degiro's account.", ) parser.add_argument( @@ -331,39 +228,23 @@ class DegiroController: ) ns_parser = parse_known_args_and_warn(parser, l_args) - if ns_parser: - credentials = Credentials() - credentials.CopyFrom(default_credentials) - credentials.username = ns_parser.username - credentials.password = ns_parser.password + self.__degiro_view.login(ns_parser=ns_parser) - if ns_parser.otp is not None: - credentials.one_time_password = ns_parser.otp - if ns_parser.topt_secret is not None: - credentials.totp_secret_key = ns_parser.topt_secret - - degiro_model.login(credentials=credentials) - - DegiroView.login_display_success() - - def logout(self, _): + def logout(self, l_args): """Log out from Degiro's API.""" - # GET ATTRIBUTES - degiro_model = self.__degiro_model + # PARSE ARGS + parser = argparse.ArgumentParser( + add_help=False, + prog="logout", + ) + ns_parser = parse_known_args_and_warn(parser, l_args) - # CALL API - if degiro_model.logout(): - DegiroView.logout_display_success() - else: - DegiroView.logout_display_fail() + self.__degiro_view.logout(ns_parser=ns_parser) def lookup(self, l_args): """Search for products by their name.""" - # GET ATTRIBUTES - degiro_model = self.__degiro_model - # PARSING ARGS parser = argparse.ArgumentParser( add_help=False, @@ -390,33 +271,19 @@ class DegiroController: ) ns_parser = parse_known_args_and_warn(parser, l_args) - # FECTH DATA - product_search = degiro_model.lookup( - limit=ns_parser.limit, - offset=ns_parser.offset, - search_text=ns_parser.search_text, - ) + self.__degiro_view.lookup(ns_parser=ns_parser) - # DISPLAY DATA - if len(product_search.products) > 0: - DegiroView.lookup_display(product_search=product_search) - else: - DegiroView.lookup_display_no_result() - - def pending(self, _): + def pending(self, l_args): """Display pending orders.""" - # GET ATTRIBUTES - degiro_model = self.__degiro_model + # PARSING ARGS + parser = argparse.ArgumentParser( + add_help=False, + prog="pending", + ) + ns_parser = parse_known_args_and_warn(parser, l_args) - # SETUP REQUEST - orders = degiro_model.pending() - - # DISPLAY DATA - if len(orders.values) > 0: - DegiroView.pending_display(orders=orders) - else: - DegiroView.pending_display_no_result() + self.__degiro_view.pending(ns_parser=ns_parser) def q(self, _): """Process Q command - quit the menu.""" @@ -432,24 +299,21 @@ class DegiroController: return True - def topnews(self, _): + def topnews(self, l_args): """Display top news.""" - # GET ATTRIBUTES - degiro_model = self.__degiro_model + # PARSING ARGS + parser = argparse.ArgumentParser( + add_help=False, + prog="topnews", + ) + ns_parser = parse_known_args_and_warn(parser, l_args) - # FETCH DATA - top_news = degiro_model.topnews() - - # DISPLAY DATA - DegiroView.topnews_display(top_news=top_news) + self.__degiro_view.topnews(ns_parser=ns_parser) def update(self, l_args): """Update an order.""" - # GET ATTRIBUTES - degiro_model = self.__degiro_model - # PARSING ARGS parser = argparse.ArgumentParser( add_help=False, @@ -469,21 +333,7 @@ class DegiroController: ) ns_parser = parse_known_args_and_warn(parser, l_args) - # FETCH DATA - order = degiro_model.update_pending_order( - order_id=ns_parser.id, - ) - - if order is None: - DegiroView.update_display_not_found(order_id=ns_parser.id) - else: - # SETUP ORDER - order.price = ns_parser.price - - if degiro_model.update(order=order): - DegiroView.update_display_success() - else: - DegiroView.update_display_fail() + self.__degiro_view.update(ns_parser=ns_parser) def switch(self, an_input: str): """Process and dispatch input diff --git a/gamestonk_terminal/portfolio/degiro_view.py b/gamestonk_terminal/portfolio/degiro_view.py index f0f5e8201e4..7119bb396d9 100644 --- a/gamestonk_terminal/portfolio/degiro_view.py +++ b/gamestonk_terminal/portfolio/degiro_view.py @@ -1,8 +1,10 @@ # IMPORTATION THIRDPARTY +from argparse import Namespace import pandas as pd import trading.helpers.payload_handler as payload_handler from trading.pb.trading_pb2 import ( + Credentials, LatestNews, NewsByCompany, Order, @@ -11,8 +13,33 @@ from trading.pb.trading_pb2 import ( Update, ) +# IMPORTATION INTERNAL +from gamestonk_terminal.portfolio.degiro_model import DegiroModel + +# pylint: disable=no-member + class DegiroView: + ORDER_ACTION = { + "buy": Order.Action.BUY, + "sell": Order.Action.SELL, + } + + ORDER_DURATION = { + "gtd": Order.TimeType.GOOD_TILL_DAY, + "gtc": Order.TimeType.GOOD_TILL_CANCELED, + } + + ORDER_TYPE = { + "limit": Order.OrderType.LIMIT, + "market": Order.OrderType.MARKET, + "stop-limit": Order.OrderType.STOP_LIMIT, + "stop-loss": Order.OrderType.STOP_LOSS, + } + + def __init__(self): + self.__degiro_model = DegiroModel() + @staticmethod def help_display(): print( @@ -32,6 +59,19 @@ class DegiroView: " update view top news preview\n" ) + def cancel(self, ns_parser: Namespace): + # GET ATTRIBUTES + degiro_model = self.__degiro_model + + # CANCEL ORDER + order_id = ns_parser.id + + # DISPLAY DATA + if degiro_model.cancel(order_id=order_id): + DegiroView.cancel_display_success(order_id=order_id) + else: + DegiroView.cancel_display_fail(order_id=order_id) + @staticmethod def cancel_display_success(order_id: str): print(f"Following `Order` was canceled : {order_id}") @@ -40,6 +80,83 @@ class DegiroView: def cancel_display_fail(order_id: str): print(f"Following `Order` cancellation failed : {order_id}") + def companynews(self, ns_parser: Namespace): + # GET ATTRIBUTES + degiro_model = self.__degiro_model + + # FETCH DATA + news_by_company = degiro_model.companynews(isin=ns_parser.isin) + + # DISPLAY DATA + DegiroView.companynews_display(news_by_company=news_by_company) + + @staticmethod + def companynews_display(news_by_company: NewsByCompany): + news_dict = payload_handler.message_to_dict( + message=news_by_company, + ) + for article in news_dict["items"]: + print("date", article["date"]) + print("title", article["title"]) + print("content", article["content"]) + print("isins", article["isins"]) + print("---") + + def create(self, ns_parser: Namespace): + # GET CONSTANTS + ORDER_ACTION = self.ORDER_ACTION + ORDER_DURATION = self.ORDER_DURATION + ORDER_TYPE = self.ORDER_TYPE + + # GET ATTRIBUTES + degiro_model = self.__degiro_model + + # SETUP ORDER + action = ORDER_ACTION[ns_parser.action] + order_type = ORDER_TYPE[ns_parser.type] + price = ns_parser.price + product_id = degiro_model.create_calculate_product_id( + product=ns_parser.product, + symbol=ns_parser.symbol, + ) + size = degiro_model.create_calculate_size( + price=ns_parser.price, + size=ns_parser.size, + up_to=ns_parser.up_to, + ) + time_type = ORDER_DURATION[ns_parser.duration] + order = Order( + action=action, + order_type=order_type, + price=price, + product_id=product_id, + size=size, + time_type=time_type, + ) + + # CHECK ORDER + checking_response = degiro_model.create_check(order=order) + DegiroView.create_display_check( + order=order, + checking_response=checking_response, + ) + + # USER INPUT + message_ask = DegiroView.create_message_ask_confirmation() + confirmation = input(message_ask) + + # EXECUTE ORDER + if confirmation in ["y", "yes"]: + confirmation_id = checking_response.confirmation_id + confirmation_response = degiro_model.create_confirm( + confirmation_id=confirmation_id, + order=order, + ) + order.id = confirmation_response.orderId + DegiroView.create_display_created_order(order=order) + else: + DegiroView.create_display_canceled() + @staticmethod def create_display_canceled(): print("`Order` creation canceled.\n") @@ -73,18 +190,6 @@ class DegiroView: fees, ) - @staticmethod - def companynews_display(news_by_company: NewsByCompany): - news_dict = payload_handler.message_to_dict( - message=news_by_company, - ) - for article in news_dict["items"]: - print("date", article["date"]) - print("title", article["title"]) - print("content", article["content"]) - print("isins", article["isins"]) - print("---") - @staticmethod def create_display_created_order(order: Order): order_dict = payload_handler.message_to_dict(message=order) @@ -108,6 +213,20 @@ class DegiroView: def create_message_ask_confirmation(): return "\nDo you confirm this `Order`?\n" + def hold(self, ns_parser: Namespace): + _ = ns_parser + + # GET ATTRIBUTES + degiro_model = self.__degiro_model + + # FETCH HELD PRODUCTS + positions = degiro_model.hold_positions() + + if len(positions) == 0: + DegiroView.hold_display_no_position() + else: + DegiroView.hold_display_positions(positions=positions) + @staticmethod def hold_display_no_position(): print("0 position found.") @@ -138,6 +257,14 @@ class DegiroView: print(fmt_positions) + def lastnews(self, ns_parser: Namespace): + # GET ATTRIBUTES + degiro_model = self.__degiro_model + latest_news = degiro_model.lastnews(limit=ns_parser.limit) + + # DISPLAY DATA + DegiroView.lastnews_display(latest_news=latest_news) + @staticmethod def lastnews_display(latest_news: LatestNews): news_dict = payload_handler.message_to_dict( @@ -149,18 +276,66 @@ class DegiroView: print("content", article["content"]) print("---") + def login(self, ns_parser: Namespace): + # GET ATTRIBUTES + degiro_model = self.__degiro_model + default_credentials = degiro_model.login_default_credentials() + + credentials = Credentials() + credentials.CopyFrom(default_credentials) + credentials.username = ns_parser.username + credentials.password = ns_parser.password + + if ns_parser.otp is not None: + credentials.one_time_password = ns_parser.otp + if ns_parser.topt_secret is not None: + credentials.totp_secret_key = ns_parser.topt_secret + + degiro_model.login(credentials=credentials) + + DegiroView.login_display_success() + @staticmethod def login_display_success(): print("You are now logged in !") + def logout(self, ns_parser: Namespace): + _ = ns_parser + + # GET ATTRIBUTES + degiro_model = self.__degiro_model + + # CALL API + if degiro_model.logout(): + DegiroView.logout_display_success() + else: + DegiroView.logout_display_fail() + @staticmethod def logout_display_fail(): - print("You are now logged out !") + print("Logged out failed.") @staticmethod def logout_display_success(): print("You are now logged out !") + def lookup(self, ns_parser: Namespace): + # GET ATTRIBUTES + degiro_model = self.__degiro_model + + # FECTH DATA + product_search = degiro_model.lookup( + limit=ns_parser.limit, + offset=ns_parser.offset, + search_text=ns_parser.search_text, + ) + + # DISPLAY DATA + if len(product_search.products) > 0: + DegiroView.lookup_display(product_search=product_search) + else: + DegiroView.lookup_display_no_result() + @staticmethod def lookup_display(product_search: ProductSearch): products_dict = payload_handler.message_to_dict( @@ -185,11 +360,38 @@ class DegiroView: def lookup_display_no_result(): print("0 result found.") + def pending(self, ns_parser: Namespace): + _ = ns_parser + + # GET ATTRIBUTES + degiro_model = self.__degiro_model + + # SETUP REQUEST + orders = degiro_model.pending() + + # DISPLAY DATA + if len(orders.values) > 0: + DegiroView.pending_display(orders=orders) + else: + DegiroView.pending_display_no_result() + @staticmethod def pending_display(orders: Update.Orders): orders_dict = payload_handler.message_to_dict(message=orders) orders_df = pd.DataFrame(orders_dict["values"]) - print(orders_df) + fields = [ + "action", + "currency", + "id", + "quantity", + "order_type", + "price", + "product", + "product_id", + "stop_price", + "total_order_value", + ] + print(orders_df[orders_df.columns.intersection(fields)]) @staticmethod def pending_display_no_result(): @@ -203,6 +405,18 @@ class DegiroView: def quit_display(): print("Quit the app.") + def topnews(self, ns_parser: Namespace): + _ = ns_parser + + # GET ATTRIBUTES + degiro_model = self.__degiro_model + + # FETCH DATA + top_news = degiro_model.topnews() + + # DISPLAY DATA + DegiroView.topnews_display(top_news=top_news) + @staticmethod def topnews_display(top_news: TopNewsPreview): news_dict = payload_handler.message_to_dict( @@ -216,6 +430,26 @@ class DegiroView: print("brief", article["brief"]) print("---") + def update(self, ns_parser: Namespace): + # GET ATTRIBUTES + degiro_model = self.__degiro_model + + # FETCH DATA + order = degiro_model.update_pending_order( + order_id=ns_parser.id, + ) + + if order is None: + DegiroView.update_display_not_found(order_id=ns_parser.id) + else: + # SETUP ORDER + order.price = ns_parser.price + + if degiro_model.update(order=order): + DegiroView.update_display_success() + else: + DegiroView.update_display_fail() + @staticmethod def update_display_fail(): print("`Order` update failed.")