Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _init_metrics(self):
if not backtesting_enabled(self.octobot.get_config()):
self.octobot.metrics_handler = MetricsManager(self.octobot)
# Advanced
AdvancedManager.init_advanced_classes_if_necessary(self.config)
# Debug tools
self.performance_analyser = None
if CONFIG_DEBUG_OPTION_PERF in self.config and self.config[CONFIG_DEBUG_OPTION_PERF]:
self.performance_analyser = PerformanceAnalyser()
# Init time frames using enabled strategies
EvaluatorCreator.init_time_frames_from_strategies(self.config)
self.time_frames = copy.copy(TimeFrameManager.get_config_time_frame(self.config))
# Init display time frame
config_time_frames = TimeFrameManager.get_config_time_frame(self.config)
if TimeFrames.ONE_HOUR not in config_time_frames and not backtesting_enabled(self.config):
config_time_frames.append(TimeFrames.ONE_HOUR)
TimeFrameManager.sort_config_time_frames(self.config)
# Init relevant evaluator names list using enabled strategies
self.relevant_evaluators = EvaluatorCreator.get_relevant_evaluators_from_strategies(self.config)
# Backtesting
self.backtesting_enabled = backtesting_enabled(self.config)
# Notifier
self.config[CONFIG_NOTIFICATION_INSTANCE] = Notification(self.config)
self.symbol_tasks_manager = {}
self.exchange_traders = {}
self.exchange_trader_simulators = {}
self.trading_mode = None
def _create_symbol_threads_managers(self, exchange, symbol_evaluator, global_price_updater):
if backtesting_enabled(self.config):
real_time_ta_eval_list = []
else:
# Create real time TA evaluators
real_time_ta_eval_list = EvaluatorCreator.create_real_time_ta_evals(self.config,
exchange,
symbol_evaluator.get_symbol(),
self.relevant_evaluators,
self.dispatchers_list)
self.real_time_eval_tasks = self.real_time_eval_tasks + real_time_ta_eval_list
if self.time_frames:
for time_frame in self.time_frames:
if exchange.get_exchange_manager().time_frame_exists(time_frame.value, symbol_evaluator.get_symbol()):
self.symbol_tasks_manager[time_frame] = \
EvaluatorTaskManager(self.config,
time_frame,
def should_be_ready(cls, config):
on_backtesting = backtesting_enabled(config)
return not on_backtesting or (on_backtesting and cls.BACKTESTING_ENABLED)
real_trader_activated=has_real_and_or_simulated_traders()[0],
ccxt_tested_exchanges=get_tested_exchange_list(),
ccxt_simulated_tested_exchanges=get_simulated_exchange_list(),
ccxt_other_exchanges=sorted(get_other_exchange_list()),
services_list=service_list,
service_name_list=service_name_list,
symbol_list=sorted(get_symbol_list([exchange
for exchange in display_config[CONFIG_EXCHANGES]])),
full_symbol_list=get_all_symbol_list(),
strategy_config=get_strategy_config(),
evaluator_startup_config=get_evaluator_startup_config(),
trading_startup_config=get_trading_startup_config(),
is_trading_persistence_activated=is_trading_persistence_activated(),
in_backtesting=backtesting_enabled(display_config)
)
async def initialize_impl(self):
if self.enable:
if self.previous_state_manager is not None:
self.load_previous_state_if_any()
try:
await self.portfolio.initialize()
await self.trades_manager.initialize()
except Exception as e:
self.enable = False
self.logger.error(f"Error when initializing portfolio: {e}. "
f"{self.exchange.get_name()} trader disabled.")
self.logger.exception(e)
if backtesting_enabled(self.config):
raise e
def __init__(self, config, crypto_currency, dispatchers_list, relevant_evaluators=None):
if relevant_evaluators is None:
relevant_evaluators = CONFIG_EVALUATORS_WILDCARD
self.config = config
self.crypto_currency = crypto_currency
self.dispatchers_list = dispatchers_list
self.symbol_evaluator_list = {}
if backtesting_enabled(self.config):
self.social_eval_list = []
self.social_not_tasked_list = []
else:
self.social_eval_list = EvaluatorCreator.create_social_eval(self.config,
self.crypto_currency,
self.dispatchers_list,
relevant_evaluators)
self.social_not_tasked_list = EvaluatorCreator.create_social_not_tasked_list(self.social_eval_list)
self.social_evaluator_refresh_task = SocialEvaluatorNotTaskedUpdateTask(self.social_not_tasked_list)
def run_in_main_asyncio_loop(self, coroutine):
# restart a new loop if necessary (for backtesting analysis)
if backtesting_enabled(self.config) and self.async_loop.is_closed():
self.logger.debug("Main loop is closed, starting a new main loop.")
self._create_new_asyncio_main_loop()
return run_coroutine_in_asyncio_loop(coroutine, self.async_loop)
def _fetch_recent_trades(self, symbol, timeframe, index):
# Use index + 1 to use prices from the current candle. Currently evaluated candle (previous candle) is closed.
current_candle = self.get_ohlcv(symbol)[timeframe.value][index + 1]
start_timestamp = current_candle[PriceIndexes.IND_PRICE_TIME.value] \
if backtesting_enabled(self.config) else time.time()
if not self.handles_trades_history(symbol) or not backtesting_enabled(self.config):
return self._generate_trades(current_candle, start_timestamp)
else:
end_timestamp = self.get_ohlcv(symbol)[timeframe.value][index+1][PriceIndexes.IND_PRICE_TIME.value] \
if len(self.get_ohlcv(symbol)[timeframe.value]) > index+1 else -1
return self.select_trades(start_timestamp, end_timestamp, symbol)
# remove service instances
for service in config[CONFIG_CATEGORY_SERVICES]:
config[CONFIG_CATEGORY_SERVICES][service].pop(CONFIG_SERVICE_INSTANCE, None)
# remove non config keys
config.pop(CONFIG_EVALUATOR, None)
config.pop(CONFIG_TRADING_TENTACLES, None)
config.pop(CONFIG_INTERFACES, None)
config.pop(CONFIG_ADVANCED_CLASSES, None)
config.pop(CONFIG_TIME_FRAME, None)
config.pop(CONFIG_NOTIFICATION_INSTANCE, None)
config.pop(CONFIG_ADVANCED_INSTANCES, None)
# remove backtesting specific differences
if backtesting_enabled(config):
if CONFIG_BACKTESTING in config:
config[CONFIG_BACKTESTING].pop(CONFIG_ENABLED_OPTION, None)
config[CONFIG_BACKTESTING].pop(CONFIG_ANALYSIS_ENABLED_OPTION, None)