degiro integration : update mvc

2026-10-27 20:42:26 +00:00 · 2021-05-23 10:36:32 +02:00
parent f53d818508
commit 30a841ee10
2 changed files with 295 additions and 211 deletions
--- 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
--- 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.")