Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def start_backtesting_using_specific_files(files, source, reset_tentacle_config=False, run_on_common_part_only=True):
try:
tools = WebInterface.tools
previous_independant_backtesting = tools[BOT_TOOLS_BACKTESTING]
if tools[BOT_TOOLS_STRATEGY_OPTIMIZER] and is_optimizer_in_progress(tools[BOT_TOOLS_STRATEGY_OPTIMIZER]):
return False, "Optimizer already running"
elif previous_independant_backtesting and \
is_independent_backtesting_in_progress(previous_independant_backtesting):
return False, "A backtesting is already running"
else:
if previous_independant_backtesting:
run_in_bot_main_loop(stop_independent_backtesting(previous_independant_backtesting))
if reset_tentacle_config:
tentacles_setup_config = get_tentacles_setup_config()
else:
tentacles_setup_config = get_bot_api().get_edited_tentacles_config()
config = get_global_config()
independent_backtesting = create_independent_backtesting(config,
tentacles_setup_config,
files,
run_on_common_part_only=run_on_common_part_only)
run_in_bot_main_loop(initialize_and_run_independent_backtesting(independent_backtesting), blocking=False)
tools[BOT_TOOLS_BACKTESTING] = independent_backtesting
tools[BOT_TOOLS_BACKTESTING_SOURCE] = source
return True, "Backtesting started"
except Exception as e:
LOGGER.exception(e)
return False, f"Error when starting backtesting: {e}"
# init_logger()
backtesting = await _run_backtesting(config, tentacles_setup_config)
exchange_manager = get_exchange_manager_from_exchange_id(
get_independent_backtesting_exchange_manager_ids(backtesting)[0])
trades = get_trade_history(exchange_manager)
open_orders = get_open_orders(exchange_manager)
# ensure at least one order is either open or got filled
assert trades + open_orders
trades = open_orders = exchange_manager = None # prevent memory leak
await stop_independent_backtesting(backtesting, memory_check=True)
asyncio.get_event_loop().call_soon(check_independent_backtesting_remaining_objects, backtesting)
await asyncio.create_task(error_container.check())
except Exception as e:
if backtesting is not None:
# do not get stuck in running backtesting
await stop_independent_backtesting(backtesting, memory_check=False)
raise e
config[CONFIG_SIMULATOR][CONFIG_STARTING_PORTFOLIO]["ETH"] = 20
for _ in range(backtesting_count):
error_container = ErrorContainer()
asyncio.get_event_loop().set_exception_handler(error_container.exception_handler)
# enabling loggers is slowing down backtesting but can display useful debug info
# from octobot.logger import init_logger
# init_logger()
backtesting = await _run_backtesting(config, tentacles_setup_config)
exchange_manager = get_exchange_manager_from_exchange_id(
get_independent_backtesting_exchange_manager_ids(backtesting)[0])
trades = get_trade_history(exchange_manager)
open_orders = get_open_orders(exchange_manager)
# ensure at least one order is either open or got filled
assert trades + open_orders
trades = open_orders = exchange_manager = None # prevent memory leak
await stop_independent_backtesting(backtesting, memory_check=True)
asyncio.get_event_loop().call_soon(check_independent_backtesting_remaining_objects, backtesting)
await asyncio.create_task(error_container.check())
except Exception as e:
if backtesting is not None:
# do not get stuck in running backtesting
await stop_independent_backtesting(backtesting, memory_check=False)
raise e
async def initialize(self):
try:
await self.initializer.create()
self.independent_backtesting = create_independent_backtesting(
self.config,
self.tentacles_setup_config,
get_backtesting_data_files(self.config),
run_on_common_part_only=self.run_on_common_part_only)
await initialize_and_run_independent_backtesting(self.independent_backtesting, log_errors=False)
await join_independent_backtesting(self.independent_backtesting)
if self.log_report:
log_independent_backtesting_report(self.independent_backtesting)
await stop_independent_backtesting(self.independent_backtesting, memory_check=False)
except Exception as e:
self.logger.error(f"Error when starting backtesting: {e}")
finally:
self.task_manager.stop_tasks()