9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

95.89

/freqtrade/rpc/rpc_manager.py

"""
This module contains class to manage RPC communications (Telegram, API, ...)
"""
import logging
from collections import deque
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__)


class RPCManager:
    """
    Class to manage RPC objects (Telegram, API, ...)
    """

    def __init__(self, freqtrade) -> None:
        """ Initializes all enabled rpc modules """
        self.registered_modules: List[RPCHandler] = []
        self._rpc = RPC(freqtrade)
        config = freqtrade.config
        # Enable telegram
        if config.get('telegram', {}).get('enabled', False):
            logger.info('Enabling rpc.telegram ...')
            from freqtrade.rpc.telegram import Telegram
            self.registered_modules.append(Telegram(self._rpc, config))

        # Enable discord
        if config.get('discord', {}).get('enabled', False):
            logger.info('Enabling rpc.discord ...')
            from freqtrade.rpc.discord import Discord
            self.registered_modules.append(Discord(self._rpc, config))

        # Enable Webhook
        if config.get('webhook', {}).get('enabled', False):
            logger.info('Enabling rpc.webhook ...')
            from freqtrade.rpc.webhook import Webhook
            self.registered_modules.append(Webhook(self._rpc, config))

        # Enable local rest api server for cmd line control
        if config.get('api_server', {}).get('enabled', False):
            logger.info('Enabling rpc.api_server')
            from freqtrade.rpc.api_server import ApiServer
            apiserver = ApiServer(config)
            apiserver.add_rpc_handler(self._rpc)
            self.registered_modules.append(apiserver)

    def cleanup(self) -> None:
        """ Stops all enabled rpc modules """
        logger.info('Cleaning up rpc modules ...')
        while self.registered_modules:
            mod = self.registered_modules.pop()
            logger.info('Cleaning up rpc.%s ...', mod.name)
            mod.cleanup()
            del mod

    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.
        e.g.:
        {
            'status': 'stopping bot'
        }
        """
        if msg.get('type') not in NO_ECHO_MESSAGES:
            logger.info('Sending rpc message: %s', msg)
        for mod in self.registered_modules:
            logger.debug('Forwarding message to rpc.%s', mod.name)
            try:
                mod.send_msg(msg)
            except NotImplementedError:
                logger.error(f"Message type '{msg['type']}' not implemented by handler {mod.name}.")
            except Exception:
                logger.exception('Exception occurred within RPC module %s', mod.name)

    def process_msg_queue(self, queue: deque) -> None:
        """
        Process all messages in the queue.
        """
        while queue:
            msg = queue.popleft()
            logger.info('Sending rpc strategy_msg: %s', msg)
            for mod in self.registered_modules:
                if mod._config.get(mod.name, {}).get('allow_custom_messages', False):
                    mod.send_msg({
                        'type': RPCMessageType.STRATEGY_MSG,
                        'msg': msg,
                    })

    def startup_messages(self, config: Config, pairlist, protections) -> None:
        if config['dry_run']:
            self.send_msg({
                'type': RPCMessageType.WARNING,
                'status': 'Dry run is enabled. All trades are simulated.'
            })
        stake_currency = config['stake_currency']
        stake_amount = config['stake_amount']
        minimal_roi = config['minimal_roi']
        stoploss = config['stoploss']
        trailing_stop = config['trailing_stop']
        timeframe = config['timeframe']
        exchange_name = config['exchange']['name']
        strategy_name = config.get('strategy', '')
        pos_adjust_enabled = 'On' if config['position_adjustment_enable'] else 'Off'
        self.send_msg({
            'type': RPCMessageType.STARTUP,
            'status': f'*Exchange:* `{exchange_name}`\n'
                      f'*Stake per trade:* `{stake_amount} {stake_currency}`\n'
                      f'*Minimum ROI:* `{minimal_roi}`\n'
                      f'*{"Trailing " if trailing_stop else ""}Stoploss:* `{stoploss}`\n'
                      f'*Position adjustment:* `{pos_adjust_enabled}`\n'
                      f'*Timeframe:* `{timeframe}`\n'
                      f'*Strategy:* `{strategy_name}`'
        })
        self.send_msg({
            'type': RPCMessageType.STARTUP,
            'status': f'Searching for {stake_currency} pairs to buy and sell '
                      f'based on {pairlist.short_desc()}'
        })
        if len(protections.name_list) > 0:
            prots = '\n'.join([p for prot in protections.short_desc() for k, p in prot.items()])
            self.send_msg({
                'type': RPCMessageType.STARTUP,
                'status': f'Using Protections: \n{prots}'
            })

1	"""
2	This module contains class to manage RPC communications (Telegram, API, ...)
3	"""
4	import logging	1✔
5	from collections import deque	1✔
6	from typing import List	1✔
7
8	from freqtrade.constants import Config	1✔
9	from freqtrade.enums import NO_ECHO_MESSAGES, RPCMessageType	1✔
10	from freqtrade.rpc import RPC, RPCHandler	1✔
11	from freqtrade.rpc.rpc_types import RPCSendMsg	1✔
12
13
14	logger = logging.getLogger(__name__)	1✔
15
16
17	class RPCManager:	1✔
18	"""
19	Class to manage RPC objects (Telegram, API, ...)
20	"""
21
22	def __init__(self, freqtrade) -> None:	1✔
23	""" Initializes all enabled rpc modules """
24	self.registered_modules: List[RPCHandler] = []	1✔
25	self._rpc = RPC(freqtrade)	1✔
26	config = freqtrade.config	1✔
27	# Enable telegram
28	if config.get('telegram', {}).get('enabled', False):	1✔
29	logger.info('Enabling rpc.telegram ...')	1✔
30	from freqtrade.rpc.telegram import Telegram	1✔
31	self.registered_modules.append(Telegram(self._rpc, config))	1✔
32
33	# Enable discord
34	if config.get('discord', {}).get('enabled', False):	1✔
35	logger.info('Enabling rpc.discord ...')	×
36	from freqtrade.rpc.discord import Discord	×
37	self.registered_modules.append(Discord(self._rpc, config))	×
38
39	# Enable Webhook
40	if config.get('webhook', {}).get('enabled', False):	1✔
41	logger.info('Enabling rpc.webhook ...')	1✔
42	from freqtrade.rpc.webhook import Webhook	1✔
43	self.registered_modules.append(Webhook(self._rpc, config))	1✔
44
45	# Enable local rest api server for cmd line control
46	if config.get('api_server', {}).get('enabled', False):	1✔
47	logger.info('Enabling rpc.api_server')	1✔
48	from freqtrade.rpc.api_server import ApiServer	1✔
49	apiserver = ApiServer(config)	1✔
50	apiserver.add_rpc_handler(self._rpc)	1✔
51	self.registered_modules.append(apiserver)	1✔
52
53	def cleanup(self) -> None:	1✔
54	""" Stops all enabled rpc modules """
55	logger.info('Cleaning up rpc modules ...')	1✔
56	while self.registered_modules:	1✔
57	mod = self.registered_modules.pop()	1✔
58	logger.info('Cleaning up rpc.%s ...', mod.name)	1✔
59	mod.cleanup()	1✔
60	del mod	1✔
61
62	def send_msg(self, msg: RPCSendMsg) -> None:	1✔
63	"""
64	Send given message to all registered rpc modules.
65	A message consists of one or more key value pairs of strings.
66	e.g.:
67	{
68	'status': 'stopping bot'
69	}
70	"""
71	if msg.get('type') not in NO_ECHO_MESSAGES:	1✔
72	logger.info('Sending rpc message: %s', msg)	1✔
73	for mod in self.registered_modules:	1✔
74	logger.debug('Forwarding message to rpc.%s', mod.name)	1✔
75	try:	1✔
76	mod.send_msg(msg)	1✔
77	except NotImplementedError:	1✔
78	logger.error(f"Message type '{msg['type']}' not implemented by handler {mod.name}.")	1✔
79	except Exception:	1✔
80	logger.exception('Exception occurred within RPC module %s', mod.name)	1✔
81
82	def process_msg_queue(self, queue: deque) -> None:	1✔
83	"""
84	Process all messages in the queue.
85	"""
86	while queue:	1✔
87	msg = queue.popleft()	1✔
88	logger.info('Sending rpc strategy_msg: %s', msg)	1✔
89	for mod in self.registered_modules:	1✔
90	if mod._config.get(mod.name, {}).get('allow_custom_messages', False):	1✔
91	mod.send_msg({	1✔
92	'type': RPCMessageType.STRATEGY_MSG,
93	'msg': msg,
94	})
95
96	def startup_messages(self, config: Config, pairlist, protections) -> None:	1✔
97	if config['dry_run']:	1✔
98	self.send_msg({	1✔
99	'type': RPCMessageType.WARNING,
100	'status': 'Dry run is enabled. All trades are simulated.'
101	})
102	stake_currency = config['stake_currency']	1✔
103	stake_amount = config['stake_amount']	1✔
104	minimal_roi = config['minimal_roi']	1✔
105	stoploss = config['stoploss']	1✔
106	trailing_stop = config['trailing_stop']	1✔
107	timeframe = config['timeframe']	1✔
108	exchange_name = config['exchange']['name']	1✔
109	strategy_name = config.get('strategy', '')	1✔
110	pos_adjust_enabled = 'On' if config['position_adjustment_enable'] else 'Off'	1✔
111	self.send_msg({	1✔
112	'type': RPCMessageType.STARTUP,
113	'status': f'Exchange: `{exchange_name}`\n'
114	f'Stake per trade: `{stake_amount} {stake_currency}`\n'
115	f'Minimum ROI: `{minimal_roi}`\n'
116	f'{"Trailing " if trailing_stop else ""}Stoploss: `{stoploss}`\n'
117	f'Position adjustment: `{pos_adjust_enabled}`\n'
118	f'Timeframe: `{timeframe}`\n'
119	f'Strategy: `{strategy_name}`'
120	})
121	self.send_msg({	1✔
122	'type': RPCMessageType.STARTUP,
123	'status': f'Searching for {stake_currency} pairs to buy and sell '
124	f'based on {pairlist.short_desc()}'
125	})
126	if len(protections.name_list) > 0:	1✔
127	prots = '\n'.join([p for prot in protections.short_desc() for k, p in prot.items()])	1✔
128	self.send_msg({	1✔
129	'type': RPCMessageType.STARTUP,
130	'status': f'Using Protections: \n{prots}'
131	})

freqtrade / freqtrade / 9394559170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous