222 lines
8.3 KiB
Python
222 lines
8.3 KiB
Python
"""
|
|
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 interation 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()
|