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

96.43

/freqtrade/worker.py

"""
Main Freqtrade worker class.
"""
import logging
import time
import traceback
from os import getpid
from typing import Any, Callable, Dict, Optional

import sdnotify

from freqtrade import __version__
from freqtrade.configuration import Configuration
from freqtrade.constants import PROCESS_THROTTLE_SECS, RETRY_TIMEOUT, Config
from freqtrade.enums import RPCMessageType, State
from freqtrade.exceptions import OperationalException, TemporaryError
from freqtrade.exchange import timeframe_to_next_date
from freqtrade.freqtradebot import FreqtradeBot


logger = logging.getLogger(__name__)


class Worker:
    """
    Freqtradebot worker class
    """

    def __init__(self, args: Dict[str, Any], config: Optional[Config] = None) -> None:
        """
        Init all variables and objects the bot needs to work
        """
        logger.info(f"Starting worker {__version__}")

        self._args = args
        self._config = config
        self._init(False)

        self._heartbeat_msg: float = 0

        # Tell systemd that we completed initialization phase
        self._notify("READY=1")

    def _init(self, reconfig: bool) -> None:
        """
        Also called from the _reconfigure() method (with reconfig=True).
        """
        if reconfig or self._config is None:
            # Load configuration
            self._config = Configuration(self._args, None).get_config()

        # Init the instance of the bot
        self.freqtrade = FreqtradeBot(self._config)

        internals_config = self._config.get('internals', {})
        self._throttle_secs = internals_config.get('process_throttle_secs',
                                                   PROCESS_THROTTLE_SECS)
        self._heartbeat_interval = internals_config.get('heartbeat_interval', 60)

        self._sd_notify = sdnotify.SystemdNotifier() if \
            self._config.get('internals', {}).get('sd_notify', False) else None

    def _notify(self, message: str) -> None:
        """
        Removes the need to verify in all occurrences if sd_notify is enabled
        :param message: Message to send to systemd if it's enabled.
        """
        if self._sd_notify:
            logger.debug(f"sd_notify: {message}")
            self._sd_notify.notify(message)

    def run(self) -> None:
        state = None
        while True:
            state = self._worker(old_state=state)
            if state == State.RELOAD_CONFIG:
                self._reconfigure()

    def _worker(self, old_state: Optional[State]) -> State:
        """
        The main routine that runs each throttling iteration and handles the states.
        :param old_state: the previous service state from the previous call
        :return: current service state
        """
        state = self.freqtrade.state

        # Log state transition
        if state != old_state:

            if old_state != State.RELOAD_CONFIG:
                self.freqtrade.notify_status(f'{state.name.lower()}')

            logger.info(
                f"Changing state{f' from {old_state.name}' if old_state else ''} to: {state.name}")
            if state == State.RUNNING:
                self.freqtrade.startup()

            if state == State.STOPPED:
                self.freqtrade.check_for_open_trades()

            # Reset heartbeat timestamp to log the heartbeat message at
            # first throttling iteration when the state changes
            self._heartbeat_msg = 0

        if state == State.STOPPED:
            # Ping systemd watchdog before sleeping in the stopped state
            self._notify("WATCHDOG=1\nSTATUS=State: STOPPED.")

            self._throttle(func=self._process_stopped, throttle_secs=self._throttle_secs)

        elif state == State.RUNNING:
            # Ping systemd watchdog before throttling
            self._notify("WATCHDOG=1\nSTATUS=State: RUNNING.")

            # Use an offset of 1s to ensure a new candle has been issued
            self._throttle(func=self._process_running, throttle_secs=self._throttle_secs,
                           timeframe=self._config['timeframe'] if self._config else None,
                           timeframe_offset=1)

        if self._heartbeat_interval:
            now = time.time()
            if (now - self._heartbeat_msg) > self._heartbeat_interval:
                version = __version__
                strategy_version = self.freqtrade.strategy.version()
                if (strategy_version is not None):
                    version += ', strategy_version: ' + strategy_version
                logger.info(f"Bot heartbeat. PID={getpid()}, "
                            f"version='{version}', state='{state.name}'")
                self._heartbeat_msg = now

        return state

    def _throttle(self, func: Callable[..., Any], throttle_secs: float,
                  timeframe: Optional[str] = None, timeframe_offset: float = 1.0,
                  *args, **kwargs) -> Any:
        """
        Throttles the given callable that it
        takes at least `min_secs` to finish execution.
        :param func: Any callable
        :param throttle_secs: throttling iteration execution time limit in seconds
        :param timeframe: ensure iteration is executed at the beginning of the next candle.
        :param timeframe_offset: offset in seconds to apply to the next candle time.
        :return: Any (result of execution of func)
        """
        last_throttle_start_time = time.time()
        logger.debug("========================================")
        result = func(*args, **kwargs)
        time_passed = time.time() - last_throttle_start_time
        sleep_duration = throttle_secs - time_passed
        if timeframe:
            next_tf = timeframe_to_next_date(timeframe)
            # Maximum throttling should be until new candle arrives
            # Offset is added to ensure a new candle has been issued.
            next_tft = next_tf.timestamp() - time.time()
            next_tf_with_offset = next_tft + timeframe_offset
            if next_tft < sleep_duration and sleep_duration < next_tf_with_offset:
                # Avoid hitting a new loop between the new candle and the candle with offset
                sleep_duration = next_tf_with_offset
            sleep_duration = min(sleep_duration, next_tf_with_offset)
        sleep_duration = max(sleep_duration, 0.0)
        # next_iter = datetime.now(timezone.utc) + timedelta(seconds=sleep_duration)

        logger.debug(f"Throttling with '{func.__name__}()': sleep for {sleep_duration:.2f} s, "
                     f"last iteration took {time_passed:.2f} s."
                     #  f"next: {next_iter}"
                     )
        self._sleep(sleep_duration)
        return result

    @staticmethod
    def _sleep(sleep_duration: float) -> None:
        """Local sleep method - to improve testability"""
        time.sleep(sleep_duration)

    def _process_stopped(self) -> None:
        self.freqtrade.process_stopped()

    def _process_running(self) -> None:
        try:
            self.freqtrade.process()
        except TemporaryError as error:
            logger.warning(f"Error: {error}, retrying in {RETRY_TIMEOUT} seconds...")
            time.sleep(RETRY_TIMEOUT)
        except OperationalException:
            tb = traceback.format_exc()
            hint = 'Issue `/start` if you think it is safe to restart.'

            self.freqtrade.notify_status(
                f'*OperationalException:*\n```\n{tb}```\n {hint}',
                msg_type=RPCMessageType.EXCEPTION
            )

            logger.exception('OperationalException. Stopping trader ...')
            self.freqtrade.state = State.STOPPED

    def _reconfigure(self) -> None:
        """
        Cleans up current freqtradebot instance, reloads the configuration and
        replaces it with the new instance
        """
        # Tell systemd that we initiated reconfiguration
        self._notify("RELOADING=1")

        # Clean up current freqtrade modules
        self.freqtrade.cleanup()

        # Load and validate config and create new instance of the bot
        self._init(True)

        self.freqtrade.notify_status('config reloaded')

        # Tell systemd that we completed reconfiguration
        self._notify("READY=1")

    def exit(self) -> None:
        # Tell systemd that we are exiting now
        self._notify("STOPPING=1")

        if self.freqtrade:
            self.freqtrade.notify_status('process died')
            self.freqtrade.cleanup()

1	"""
2	Main Freqtrade worker class.
3	"""
4	import logging	1✔
5	import time	1✔
6	import traceback	1✔
7	from os import getpid	1✔
8	from typing import Any, Callable, Dict, Optional	1✔
9
10	import sdnotify	1✔
11
12	from freqtrade import __version__	1✔
13	from freqtrade.configuration import Configuration	1✔
14	from freqtrade.constants import PROCESS_THROTTLE_SECS, RETRY_TIMEOUT, Config	1✔
15	from freqtrade.enums import RPCMessageType, State	1✔
16	from freqtrade.exceptions import OperationalException, TemporaryError	1✔
17	from freqtrade.exchange import timeframe_to_next_date	1✔
18	from freqtrade.freqtradebot import FreqtradeBot	1✔
19
20
21	logger = logging.getLogger(__name__)	1✔
22
23
24	class Worker:	1✔
25	"""
26	Freqtradebot worker class
27	"""
28
29	def __init__(self, args: Dict[str, Any], config: Optional[Config] = None) -> None:	1✔
30	"""
31	Init all variables and objects the bot needs to work
32	"""
33	logger.info(f"Starting worker {__version__}")	1✔
34
35	self._args = args	1✔
36	self._config = config	1✔
37	self._init(False)	1✔
38
39	self._heartbeat_msg: float = 0	1✔
40
41	# Tell systemd that we completed initialization phase
42	self._notify("READY=1")	1✔
43
44	def _init(self, reconfig: bool) -> None:	1✔
45	"""
46	Also called from the _reconfigure() method (with reconfig=True).
47	"""
48	if reconfig or self._config is None:	1✔
49	# Load configuration
50	self._config = Configuration(self._args, None).get_config()	1✔
51
52	# Init the instance of the bot
53	self.freqtrade = FreqtradeBot(self._config)	1✔
54
55	internals_config = self._config.get('internals', {})	1✔
56	self._throttle_secs = internals_config.get('process_throttle_secs',	1✔
57	PROCESS_THROTTLE_SECS)
58	self._heartbeat_interval = internals_config.get('heartbeat_interval', 60)	1✔
59
60	self._sd_notify = sdnotify.SystemdNotifier() if \	1✔
61	self._config.get('internals', {}).get('sd_notify', False) else None
62
63	def _notify(self, message: str) -> None:	1✔
64	"""
65	Removes the need to verify in all occurrences if sd_notify is enabled
66	:param message: Message to send to systemd if it's enabled.
67	"""
68	if self._sd_notify:	1✔
69	logger.debug(f"sd_notify: {message}")	×
70	self._sd_notify.notify(message)	×
71
72	def run(self) -> None:	1✔
73	state = None	1✔
74	while True:	1✔
75	state = self._worker(old_state=state)	1✔
76	if state == State.RELOAD_CONFIG:	1✔
77	self._reconfigure()	1✔
78
79	def _worker(self, old_state: Optional[State]) -> State:	1✔
80	"""
81	The main routine that runs each throttling iteration and handles the states.
82	:param old_state: the previous service state from the previous call
83	:return: current service state
84	"""
85	state = self.freqtrade.state	1✔
86
87	# Log state transition
88	if state != old_state:	1✔
89
90	if old_state != State.RELOAD_CONFIG:	1✔
91	self.freqtrade.notify_status(f'{state.name.lower()}')	1✔
92
93	logger.info(	1✔
94	f"Changing state{f' from {old_state.name}' if old_state else ''} to: {state.name}")
95	if state == State.RUNNING:	1✔
96	self.freqtrade.startup()	1✔
97
98	if state == State.STOPPED:	1✔
99	self.freqtrade.check_for_open_trades()	1✔
100
101	# Reset heartbeat timestamp to log the heartbeat message at
102	# first throttling iteration when the state changes
103	self._heartbeat_msg = 0	1✔
104
105	if state == State.STOPPED:	1✔
106	# Ping systemd watchdog before sleeping in the stopped state
107	self._notify("WATCHDOG=1\nSTATUS=State: STOPPED.")	1✔
108
109	self._throttle(func=self._process_stopped, throttle_secs=self._throttle_secs)	1✔
110
111	elif state == State.RUNNING:	1✔
112	# Ping systemd watchdog before throttling
113	self._notify("WATCHDOG=1\nSTATUS=State: RUNNING.")	1✔
114
115	# Use an offset of 1s to ensure a new candle has been issued
116	self._throttle(func=self._process_running, throttle_secs=self._throttle_secs,	1✔
117	timeframe=self._config['timeframe'] if self._config else None,
118	timeframe_offset=1)
119
120	if self._heartbeat_interval:	1✔
121	now = time.time()	1✔
122	if (now - self._heartbeat_msg) > self._heartbeat_interval:	1✔
123	version = __version__	1✔
124	strategy_version = self.freqtrade.strategy.version()	1✔
125	if (strategy_version is not None):	1✔
126	version += ', strategy_version: ' + strategy_version	×
127	logger.info(f"Bot heartbeat. PID={getpid()}, "	1✔
128	f"version='{version}', state='{state.name}'")
129	self._heartbeat_msg = now	1✔
130
131	return state	1✔
132
133	def _throttle(self, func: Callable[..., Any], throttle_secs: float,	1✔
134	timeframe: Optional[str] = None, timeframe_offset: float = 1.0,
135	args, *kwargs) -> Any:
136	"""
137	Throttles the given callable that it
138	takes at least `min_secs` to finish execution.
139	:param func: Any callable
140	:param throttle_secs: throttling iteration execution time limit in seconds
141	:param timeframe: ensure iteration is executed at the beginning of the next candle.
142	:param timeframe_offset: offset in seconds to apply to the next candle time.
143	:return: Any (result of execution of func)
144	"""
145	last_throttle_start_time = time.time()	1✔
146	logger.debug("========================================")	1✔
147	result = func(args, *kwargs)	1✔
148	time_passed = time.time() - last_throttle_start_time	1✔
149	sleep_duration = throttle_secs - time_passed	1✔
150	if timeframe:	1✔
151	next_tf = timeframe_to_next_date(timeframe)	1✔
152	# Maximum throttling should be until new candle arrives
153	# Offset is added to ensure a new candle has been issued.
154	next_tft = next_tf.timestamp() - time.time()	1✔
155	next_tf_with_offset = next_tft + timeframe_offset	1✔
156	if next_tft < sleep_duration and sleep_duration < next_tf_with_offset:	1✔
157	# Avoid hitting a new loop between the new candle and the candle with offset
158	sleep_duration = next_tf_with_offset	1✔
159	sleep_duration = min(sleep_duration, next_tf_with_offset)	1✔
160	sleep_duration = max(sleep_duration, 0.0)	1✔
161	# next_iter = datetime.now(timezone.utc) + timedelta(seconds=sleep_duration)
162
163	logger.debug(f"Throttling with '{func.__name__}()': sleep for {sleep_duration:.2f} s, "	1✔
164	f"last iteration took {time_passed:.2f} s."
165	# f"next: {next_iter}"
166	)
167	self._sleep(sleep_duration)	1✔
168	return result	1✔
169
170	@staticmethod	1✔
171	def _sleep(sleep_duration: float) -> None:	1✔
172	"""Local sleep method - to improve testability"""
173	time.sleep(sleep_duration)	1✔
174
175	def _process_stopped(self) -> None:	1✔
176	self.freqtrade.process_stopped()	×
177
178	def _process_running(self) -> None:	1✔
179	try:	1✔
180	self.freqtrade.process()	1✔
181	except TemporaryError as error:	1✔
182	logger.warning(f"Error: {error}, retrying in {RETRY_TIMEOUT} seconds...")	1✔
183	time.sleep(RETRY_TIMEOUT)	1✔
184	except OperationalException:	1✔
185	tb = traceback.format_exc()	1✔
186	hint = 'Issue `/start` if you think it is safe to restart.'	1✔
187
188	self.freqtrade.notify_status(	1✔
189	f'OperationalException:\n```\n{tb}```\n {hint}',
190	msg_type=RPCMessageType.EXCEPTION
191	)
192
193	logger.exception('OperationalException. Stopping trader ...')	1✔
194	self.freqtrade.state = State.STOPPED	1✔
195
196	def _reconfigure(self) -> None:	1✔
197	"""
198	Cleans up current freqtradebot instance, reloads the configuration and
199	replaces it with the new instance
200	"""
201	# Tell systemd that we initiated reconfiguration
202	self._notify("RELOADING=1")	1✔
203
204	# Clean up current freqtrade modules
205	self.freqtrade.cleanup()	1✔
206
207	# Load and validate config and create new instance of the bot
208	self._init(True)	1✔
209
210	self.freqtrade.notify_status('config reloaded')	1✔
211
212	# Tell systemd that we completed reconfiguration
213	self._notify("READY=1")	1✔
214
215	def exit(self) -> None:	1✔
216	# Tell systemd that we are exiting now
217	self._notify("STOPPING=1")	1✔
218
219	if self.freqtrade:	1✔
220	self.freqtrade.notify_status('process died')	1✔
221	self.freqtrade.cleanup()	1✔

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