freqtrade_origin/freqtrade/worker.py

235 lines
8.3 KiB
Python
Raw Permalink Normal View History

2019-03-25 14:45:03 +00:00
"""
Main Freqtrade worker class.
"""
2024-05-12 14:32:47 +00:00
2019-03-25 14:45:03 +00:00
import logging
import time
import traceback
2020-02-21 02:07:31 +00:00
from os import getpid
from typing import Any, Callable, Dict, Optional
2019-03-25 14:45:03 +00:00
import sdnotify
2022-09-18 11:20:36 +00:00
from freqtrade import __version__
2019-03-25 14:45:03 +00:00
from freqtrade.configuration import Configuration
2022-09-18 11:20:36 +00:00
from freqtrade.constants import PROCESS_THROTTLE_SECS, RETRY_TIMEOUT, Config
from freqtrade.enums import RPCMessageType, State
from freqtrade.exceptions import OperationalException, TemporaryError
2022-10-23 12:56:51 +00:00
from freqtrade.exchange import timeframe_to_next_date
from freqtrade.freqtradebot import FreqtradeBot
2019-03-25 14:45:03 +00:00
2020-09-28 17:39:41 +00:00
2019-03-25 14:45:03 +00:00
logger = logging.getLogger(__name__)
2019-09-12 01:39:52 +00:00
class Worker:
2019-03-25 14:45:03 +00:00
"""
Freqtradebot worker class
"""
2023-01-21 14:01:56 +00:00
def __init__(self, args: Dict[str, Any], config: Optional[Config] = None) -> None:
2019-03-25 14:45:03 +00:00
"""
Init all variables and objects the bot needs to work
"""
2020-02-20 05:19:22 +00:00
logger.info(f"Starting worker {__version__}")
2019-03-25 14:45:03 +00:00
self._args = args
self._config = config
self._init(False)
2019-03-25 14:45:03 +00:00
2020-02-21 02:31:21 +00:00
self._heartbeat_msg: float = 0
2019-03-25 14:45:03 +00:00
# Tell systemd that we completed initialization phase
2020-05-18 05:02:57 +00:00
self._notify("READY=1")
2019-03-25 14:45:03 +00:00
2019-04-30 16:47:55 +00:00
def _init(self, reconfig: bool) -> None:
2019-03-25 14:45:03 +00:00
"""
Also called from the _reconfigure() method (with reconfig=True).
2019-03-25 14:45:03 +00:00
"""
if reconfig or self._config is None:
# Load configuration
self._config = Configuration(self._args, None).get_config()
2019-03-25 14:45:03 +00:00
# Init the instance of the bot
self.freqtrade = FreqtradeBot(self._config)
2019-03-25 14:45:03 +00:00
2024-05-12 14:32:47 +00:00
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)
2019-03-25 14:45:03 +00:00
2024-05-12 14:32:47 +00:00
self._sd_notify = (
sdnotify.SystemdNotifier()
if self._config.get("internals", {}).get("sd_notify", False)
else None
)
2019-03-25 14:45:03 +00:00
2020-05-18 05:02:57 +00:00
def _notify(self, message: str) -> None:
"""
2021-06-25 13:45:49 +00:00
Removes the need to verify in all occurrences if sd_notify is enabled
2020-05-18 05:02:57 +00:00
: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)
2019-04-30 16:47:55 +00:00
def run(self) -> None:
2019-03-25 14:45:03 +00:00
state = None
while True:
state = self._worker(old_state=state)
if state == State.RELOAD_CONFIG:
2019-04-30 07:29:49 +00:00
self._reconfigure()
2019-03-25 14:45:03 +00:00
2020-02-21 01:00:23 +00:00
def _worker(self, old_state: Optional[State]) -> State:
2019-03-25 14:45:03 +00:00
"""
2020-02-23 19:50:58 +00:00
The main routine that runs each throttling iteration and handles the states.
2019-03-25 14:45:03 +00:00
:param old_state: the previous service state from the previous call
:return: current service state
"""
state = self.freqtrade.state
2019-03-25 14:45:03 +00:00
# Log state transition
if state != old_state:
if old_state != State.RELOAD_CONFIG:
2024-05-12 14:32:47 +00:00
self.freqtrade.notify_status(f"{state.name.lower()}")
logger.info(
2024-05-12 14:32:47 +00:00
f"Changing state{f' from {old_state.name}' if old_state else ''} to: {state.name}"
)
2019-03-25 14:45:03 +00:00
if state == State.RUNNING:
2019-05-19 18:06:26 +00:00
self.freqtrade.startup()
2019-03-25 14:45:03 +00:00
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
2019-03-25 14:45:03 +00:00
if state == State.STOPPED:
# Ping systemd watchdog before sleeping in the stopped state
2020-05-18 05:02:57 +00:00
self._notify("WATCHDOG=1\nSTATUS=State: STOPPED.")
2019-03-25 14:45:03 +00:00
2020-02-21 01:00:23 +00:00
self._throttle(func=self._process_stopped, throttle_secs=self._throttle_secs)
2019-03-25 14:45:03 +00:00
elif state == State.RUNNING:
# Ping systemd watchdog before throttling
2020-05-18 05:02:57 +00:00
self._notify("WATCHDOG=1\nSTATUS=State: RUNNING.")
2019-03-25 14:45:03 +00:00
2022-10-23 12:56:51 +00:00
# Use an offset of 1s to ensure a new candle has been issued
2024-05-12 14:32:47 +00:00
self._throttle(
func=self._process_running,
throttle_secs=self._throttle_secs,
timeframe=self._config["timeframe"] if self._config else None,
timeframe_offset=1,
)
2019-03-25 14:45:03 +00:00
2020-02-21 02:31:21 +00:00
if self._heartbeat_interval:
now = time.time()
if (now - self._heartbeat_msg) > self._heartbeat_interval:
version = __version__
strategy_version = self.freqtrade.strategy.version()
2024-05-12 14:32:47 +00:00
if strategy_version is not None:
version += ", strategy_version: " + strategy_version
logger.info(
2024-05-12 15:51:21 +00:00
f"Bot heartbeat. PID={getpid()}, version='{version}', state='{state.name}'"
2024-05-12 14:32:47 +00:00
)
2020-02-21 02:31:21 +00:00
self._heartbeat_msg = now
2020-02-21 02:07:31 +00:00
2019-03-25 14:45:03 +00:00
return state
2024-05-12 14:32:47 +00:00
def _throttle(
self,
func: Callable[..., Any],
throttle_secs: float,
timeframe: Optional[str] = None,
timeframe_offset: float = 1.0,
*args,
**kwargs,
) -> Any:
2019-03-25 14:45:03 +00:00
"""
Throttles the given callable that it
takes at least `min_secs` to finish execution.
:param func: Any callable
2024-04-18 20:51:25 +00:00
:param throttle_secs: throttling iteration execution time limit in seconds
2022-10-23 12:56:51 +00:00
: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.
2020-02-21 01:00:23 +00:00
:return: Any (result of execution of func)
2019-03-25 14:45:03 +00:00
"""
2022-10-23 12:51:17 +00:00
last_throttle_start_time = time.time()
2020-02-20 05:17:24 +00:00
logger.debug("========================================")
2020-02-21 00:37:38 +00:00
result = func(*args, **kwargs)
2022-10-23 12:51:17 +00:00
time_passed = time.time() - last_throttle_start_time
2022-10-23 12:56:51 +00:00
sleep_duration = throttle_secs - time_passed
if timeframe:
next_tf = timeframe_to_next_date(timeframe)
# Maximum throttling should be until new candle arrives
2022-11-06 10:18:13 +00:00
# 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
2022-10-23 12:56:51 +00:00
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)
2024-05-12 14:32:47 +00:00
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}"
)
2022-10-23 12:17:01 +00:00
self._sleep(sleep_duration)
2019-03-25 14:45:03 +00:00
return result
2022-10-23 12:17:01 +00:00
@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:
2019-03-25 14:45:03 +00:00
try:
2019-08-13 07:36:52 +00:00
self.freqtrade.process()
2019-03-25 14:45:03 +00:00
except TemporaryError as error:
2022-09-18 11:20:36 +00:00
logger.warning(f"Error: {error}, retrying in {RETRY_TIMEOUT} seconds...")
time.sleep(RETRY_TIMEOUT)
2019-03-25 14:45:03 +00:00
except OperationalException:
tb = traceback.format_exc()
2024-05-12 14:32:47 +00:00
hint = "Issue `/start` if you think it is safe to restart."
2020-01-27 00:34:53 +00:00
self.freqtrade.notify_status(
2024-05-12 14:32:47 +00:00
f"*OperationalException:*\n```\n{tb}```\n {hint}", msg_type=RPCMessageType.EXCEPTION
)
2020-01-27 00:34:53 +00:00
2024-05-12 14:32:47 +00:00
logger.exception("OperationalException. Stopping trader ...")
self.freqtrade.state = State.STOPPED
2019-03-25 14:45:03 +00:00
2019-04-30 16:47:55 +00:00
def _reconfigure(self) -> None:
2019-03-25 14:45:03 +00:00
"""
Cleans up current freqtradebot instance, reloads the configuration and
2019-03-30 20:33:52 +00:00
replaces it with the new instance
2019-03-25 14:45:03 +00:00
"""
# Tell systemd that we initiated reconfiguration
2020-05-18 05:02:57 +00:00
self._notify("RELOADING=1")
2019-03-25 14:45:03 +00:00
# Clean up current freqtrade modules
self.freqtrade.cleanup()
# Load and validate config and create new instance of the bot
self._init(True)
2019-03-25 14:45:03 +00:00
2024-05-12 14:32:47 +00:00
self.freqtrade.notify_status("config reloaded")
2019-03-25 14:45:03 +00:00
# Tell systemd that we completed reconfiguration
2020-05-18 05:02:57 +00:00
self._notify("READY=1")
2019-03-25 14:45:03 +00:00
2019-04-30 16:47:55 +00:00
def exit(self) -> None:
2019-03-25 14:45:03 +00:00
# Tell systemd that we are exiting now
2020-05-18 05:02:57 +00:00
self._notify("STOPPING=1")
2019-03-25 14:45:03 +00:00
if self.freqtrade:
2024-05-12 14:32:47 +00:00
self.freqtrade.notify_status("process died")
2019-03-25 14:45:03 +00:00
self.freqtrade.cleanup()