Add typedDict for RPC messages

Currently not fully functional.

freqtrade/freqtradebot.py

@@ -30,6 +30,7 @@ from freqtrade.plugins.protectionmanager import ProtectionManager
 from freqtrade.resolvers import ExchangeResolver, StrategyResolver
 from freqtrade.rpc import RPCManager
 from freqtrade.rpc.external_message_consumer import ExternalMessageConsumer
+from freqtrade.rpc.rpc_types import RPCBuyMsg, RPCCancelMsg, RPCSellCancelMsg, RPCSellMsg
 from freqtrade.strategy.interface import IStrategy
 from freqtrade.strategy.strategy_wrapper import strategy_safe_wrapper
 from freqtrade.util import FtPrecise
@@ -957,7 +958,7 @@ class FreqtradeBot(LoggingMixin):
             current_rate = self.exchange.get_rate(
                 trade.pair, side='entry', is_short=trade.is_short, refresh=False)
 
-        msg = {
+        msg: RPCBuyMsg = {
             'trade_id': trade.id,
             'type': msg_type,
             'buy_tag': trade.enter_tag,
@@ -989,7 +990,7 @@ class FreqtradeBot(LoggingMixin):
         current_rate = self.exchange.get_rate(
             trade.pair, side='entry', is_short=trade.is_short, refresh=False)
 
-        msg = {
+        msg: RPCCancelMsg = {
             'trade_id': trade.id,
             'type': RPCMessageType.ENTRY_CANCEL,
             'buy_tag': trade.enter_tag,
@@ -1001,6 +1002,7 @@ class FreqtradeBot(LoggingMixin):
             'limit': trade.open_rate,
             'order_type': order_type,
             'stake_amount': trade.stake_amount,
+            'open_rate': trade.open_rate,
             'stake_currency': self.config['stake_currency'],
             'fiat_currency': self.config.get('fiat_display_currency', None),
             'amount': trade.amount,
@@ -1666,7 +1668,7 @@ class FreqtradeBot(LoggingMixin):
             amount = trade.amount
         gain = "profit" if profit_ratio > 0 else "loss"
 
-        msg = {
+        msg: RPCSellMsg = {
             'type': (RPCMessageType.EXIT_FILL if fill
                      else RPCMessageType.EXIT),
             'trade_id': trade.id,
@@ -1722,7 +1724,7 @@ class FreqtradeBot(LoggingMixin):
         profit_ratio = trade.calc_profit_ratio(profit_rate)
         gain = "profit" if profit_ratio > 0 else "loss"
 
-        msg = {
+        msg: RPCSellCancelMsg = {
             'type': RPCMessageType.EXIT_CANCEL,
             'trade_id': trade.id,
             'exchange': trade.exchange.capitalize(),

freqtrade/rpc/api_server/webserver.py

@@ -1,6 +1,6 @@
 import logging
 from ipaddress import IPv4Address
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 import orjson
 import uvicorn
@@ -13,6 +13,7 @@ from freqtrade.exceptions import OperationalException
 from freqtrade.rpc.api_server.uvicorn_threaded import UvicornServer
 from freqtrade.rpc.api_server.ws.message_stream import MessageStream
 from freqtrade.rpc.rpc import RPC, RPCException, RPCHandler
+from freqtrade.rpc.rpc_types import RPCSendMsg
 
 
 logger = logging.getLogger(__name__)
@@ -108,7 +109,7 @@ class ApiServer(RPCHandler):
         cls._has_rpc = False
         cls._rpc = None
 
-    def send_msg(self, msg: Dict[str, Any]) -> None:
+    def send_msg(self, msg: RPCSendMsg) -> None:
         """
         Publish the message to the message stream
         """

freqtrade/rpc/rpc.py

@@ -30,6 +30,7 @@ from freqtrade.persistence import Order, PairLocks, Trade
 from freqtrade.persistence.models import PairLock
 from freqtrade.plugins.pairlist.pairlist_helpers import expand_pairlist
 from freqtrade.rpc.fiat_convert import CryptoToFiatConverter
+from freqtrade.rpc.rpc_types import RPCSendMsg
 from freqtrade.wallets import PositionWallet, Wallet
 
 
@@ -79,7 +80,7 @@ class RPCHandler:
         """ Cleanup pending module resources """
 
     @abstractmethod
-    def send_msg(self, msg: Dict[str, str]) -> None:
+    def send_msg(self, msg: RPCSendMsg) -> None:
         """ Sends a message to all registered rpc modules """
 
 

freqtrade/rpc/rpc_manager.py

@@ -3,11 +3,12 @@ This module contains class to manage RPC communications (Telegram, API, ...)
 """
 import logging
 from collections import deque
-from typing import Any, Dict, List
+from typing import List
 
 from freqtrade.constants import Config
 from freqtrade.enums import NO_ECHO_MESSAGES, RPCMessageType
 from freqtrade.rpc import RPC, RPCHandler
+from freqtrade.rpc.rpc_types import RPCSendMsg
 
 
 logger = logging.getLogger(__name__)
@@ -58,7 +59,7 @@ class RPCManager:
             mod.cleanup()
             del mod
 
-    def send_msg(self, msg: Dict[str, Any]) -> None:
+    def send_msg(self, msg: RPCSendMsg) -> None:
         """
         Send given message to all registered rpc modules.
         A message consists of one or more key value pairs of strings.

freqtrade/rpc/rpc_types.py

@@ -0,0 +1,82 @@
+from datetime import datetime
+from typing import Optional, TypedDict, Union
+
+from freqtrade.enums import RPCMessageType
+
+
+class RPCSendMsgBase(TypedDict):
+    type: RPCMessageType
+
+
+class RPCStatusMsg(RPCSendMsgBase):
+    """Used for Status, Startup and Warning messages"""
+    status: str
+
+
+class RPCProtectionMsg(RPCSendMsgBase):
+    id: int
+    pair: str
+    base_currency: Optional[str]
+    lock_time: str
+    lock_timestamp: int
+    lock_end_time: str
+    lock_end_timestamp: int
+    reason: str
+    side: str
+    active: bool
+
+
+class RPCBuyMsg(RPCSendMsgBase):
+    trade_id: str
+    buy_tag: str
+    enter_tag: str
+    exchange: str
+    pair: str
+    leverage: float
+    direction: str
+    limit: float
+    open_rate: float
+    order_type: Optional[str]  # TODO: why optional??
+    stake_amount: float
+    stake_currency: str
+    fiat_currency: Optional[str]
+    amount: float
+    open_date: datetime
+    current_rate: float
+    sub_trade: bool
+
+
+class RPCCancelMsg(RPCBuyMsg):
+    reason: str
+
+
+class RPCSellMsg(RPCBuyMsg):
+    cumulative_profit: float
+    gain: str  # Literal["profit", "loss"]
+    close_rate: float
+    profit_amount: float
+    profit_ratio: float
+    sell_reason: str
+    exit_reason: str
+    close_date: datetime
+    current_rate: Optional[float]
+
+
+class RPCSellCancelMsg(RPCBuyMsg):
+    reason: str
+    gain: str  # Literal["profit", "loss"]
+    profit_amount: float
+    profit_ratio: float
+    sell_reason: str
+    exit_reason: str
+    close_date: datetime
+
+
+RPCSendMsg = Union[
+    RPCStatusMsg,
+    RPCProtectionMsg,
+    RPCBuyMsg,
+    RPCCancelMsg,
+    RPCSellMsg,
+    RPCSellCancelMsg
+    ]

freqtrade/rpc/telegram.py

@@ -30,6 +30,7 @@ from freqtrade.exceptions import OperationalException
 from freqtrade.misc import chunks, plural, round_coin_value
 from freqtrade.persistence import Trade
 from freqtrade.rpc import RPC, RPCException, RPCHandler
+from freqtrade.rpc.rpc_types import RPCSendMsg
 
 
 logger = logging.getLogger(__name__)
@@ -429,7 +430,7 @@ class Telegram(RPCHandler):
             return None
         return message
 
-    def send_msg(self, msg: Dict[str, Any]) -> None:
+    def send_msg(self, msg: RPCSendMsg) -> None:
         """ Send a message to telegram channel """
 
         default_noti = 'on'

freqtrade/rpc/webhook.py

@@ -10,6 +10,7 @@ from requests import RequestException, post
 from freqtrade.constants import Config
 from freqtrade.enums import RPCMessageType
 from freqtrade.rpc import RPC, RPCHandler
+from freqtrade.rpc.rpc_types import RPCSendMsg
 
 
 logger = logging.getLogger(__name__)
@@ -41,7 +42,7 @@ class Webhook(RPCHandler):
         """
         pass
 
-    def _get_value_dict(self, msg: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    def _get_value_dict(self, msg: RPCSendMsg) -> Optional[Dict[str, Any]]:
         whconfig = self._config['webhook']
         # Deprecated 2022.10 - only keep generic method.
         if msg['type'] in [RPCMessageType.ENTRY]:
@@ -75,7 +76,7 @@ class Webhook(RPCHandler):
             return None
         return valuedict
 
-    def send_msg(self, msg: Dict[str, Any]) -> None:
+    def send_msg(self, msg: RPCSendMsg) -> None:
         """ Send a message to telegram channel """
         try: