update to v1.0.1, change return value in subprocess to avoid large object copy between processes

Author: 32ethers

demeter/core/_typing.py

@@ -55,5 +55,3 @@ class BacktestConfig:
     print_actions:bool = False
     print_result: bool = False
     interval: str = "1min"
-    callback: Callable[[Actuator], None] | None = None
-    # save_result = False

demeter/core/actuator.py

@@ -476,7 +476,7 @@ def run(self, print_result: bool = True):
             self.print_result()
 
         self.__backtest_duration = time.time() - self.__start_time
-        self.logger.info(f"Backtesting finished, execute time {time.time() - self.__start_time}s")
+        self.logger.info(f"Backtest with process id: {os.getpid()} finished, execute time {(time.time() - self.__start_time):.3f}s")
 
     def _generate_account_status_df(self):
         self._account_status_df: pd.DataFrame = AccountStatus.to_dataframe(self._account_status_list)
@@ -596,6 +596,7 @@ def init_strategy(self):
         self._strategy.actions = self._action_list
         self._strategy.assets = self.broker.assets
         self._strategy.account_status_df = self.account_status_df
+        self._strategy.actuator = self
         self._strategy.comment_last_action = self.comment_last_action
         self._strategy.log = self._log
         for k, v in self.broker.markets.items():

demeter/core/backtest.py

@@ -1,6 +1,7 @@
 import logging
 import os
 import platform
+import time
 import traceback
 from multiprocessing import Pool, cpu_count, set_start_method
 from multiprocessing.pool import ApplyResult
@@ -26,8 +27,8 @@ def _start_with_param_data(config: StrategyConfig, data: BacktestData, strategy:
     return _start(config, data, strategy, bk_config)
 
 
-def _start(config: StrategyConfig, data: BacktestData, strategy: Strategy, bk_config: BacktestConfig) -> Actuator:
-    logger.info(f"Process id: {os.getpid()}, id of data object {id(data)}")
+def _start(config: StrategyConfig, data: BacktestData, strategy: Strategy, bk_config: BacktestConfig):
+    logger.info(f"Start with process id: {os.getpid()}, id of data object {id(data)}")
     actuator = Actuator()
     for market in config.markets:
         # add market to broker
@@ -42,9 +43,6 @@ def _start(config: StrategyConfig, data: BacktestData, strategy: Strategy, bk_co
     actuator.print_action = bk_config.print_actions
     actuator.interval = bk_config.interval
     actuator.run(bk_config.print_result)
-    if bk_config.callback is not None:
-        bk_config.callback(actuator)
-    return actuator
 
 
 def e_callback(e):
@@ -78,14 +76,14 @@ def run(self):
             raise RuntimeError("Config has not set")
         if self.data is None:
             raise RuntimeError("Data has not set")
-
+        start_time = time.time()  # 1681718968.267463
         if len(self.strategies) < 1:
             return
         elif len(self.strategies) == 1 or self.threads == 1:
             # start in single thread by default
             for strategy in self.strategies:
-                _start_with_param_data(self.config, self.data, strategy, self.backtest_config)
-
+                actuator = _start_with_param_data(self.config, self.data, strategy, self.backtest_config)
+                e_callback(actuator)
         else:
             if self.threads > cpu_count():
                 raise RuntimeError("Threads should lower than " + cpu_count())
@@ -105,7 +103,7 @@ def run(self):
                         tasks.append(result1)
                     [x.wait() for x in tasks]
             else:
-                set_start_method("fork") # ensure linux and macos have the same behavior
+                set_start_method("fork")  # ensure linux and macos have the same behavior
                 global global_data
                 global_data = self.data  # to keep there only one instance among processes
                 with Pool(processes=self.threads) as pool:
@@ -123,3 +121,4 @@ def run(self):
                         tasks.append(result1)
                     [x.wait() for x in tasks]
                 pass
+        logger.info(f"All backtest finished, total execute time {(time.time() - start_time):.3f}s")

demeter/strategy/strategy.py

@@ -5,6 +5,8 @@
 
 from .trigger import Trigger
 from .. import Broker, MarketDict, AccountStatus, AssetDict, Asset, Snapshot
+
+# from ..core import Actuator
 from .._typing import DemeterError
 from ..broker import MarketInfo, BaseAction, Market
 
@@ -25,6 +27,7 @@ def __init__(self):
         self.comment_last_action: Callable[[str], None] | None = None
         self.assets: AssetDict[Asset] = AssetDict()
         self.actions: List[BaseAction] = []
+        self.actuator = None
         self.log: Callable[[datetime, str, int], None] | None = None
 
     def initialize(self):

docs/source/concurrent.md

@@ -41,10 +41,10 @@ class DemoStrategy(Strategy):
         self.comment_last_action("Add liquidity because ...")  # add comment to last transaction
 
 
-def after_backtest(actuator: Actuator):
-    # save result and etc.
-    files = actuator.save_result(path="./result", file_name=f"strategy-{actuator.strategy.lp_range}")
-    pass
+    def finalize(self):
+        # save result and etc.
+        files = self.actuator.save_result(path="./result", file_name=f"strategy-{self.lp_range}")
+        pass
 
 
 """

docs/source/conf.py

@@ -15,7 +15,7 @@
 project = 'Demeter'
 copyright = '2024, zelos research'
 author = 'zelos research'
-release = '1.0.0'
+release = '1.0.1'
 
 # -- General configuration ---------------------------------------------------
 # https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration

release_note.md

@@ -1,3 +1,9 @@
+# Ver 1.0.1
+
+[Breaking change] Remove callback in BacktestManager, because if subprocess return actuator, 
+it will cause object copy between subprocess and main process, which will cost a lot of time. 
+You can do saving or calculating performance in Strategy.finialize().
+
 # Ver 1.0.0
 
 * Add `BacktestManager` who can start multiple backtest in one or multiple subprocesses.

samples/strategy-example/62_multiprocess.py

@@ -46,11 +46,10 @@ def add(self, snapshot: Snapshot):
         )  # add liquidity
         self.comment_last_action("Add liquidity because ...")  # add comment to last transaction
 
-
-def after_backtest(actuator: Actuator):
-    # save result and etc.
-    files = actuator.save_result(path="./result", file_name=f"strategy-{actuator.strategy.lp_range}")
-    pass
+    def finalize(self):
+        # save result and etc.
+        files = self.actuator.save_result(path="./result", file_name=f"strategy-{self.lp_range}")
+        pass
 
 
 """
@@ -85,9 +84,9 @@ def after_backtest(actuator: Actuator):
     backtest = BacktestManager(
         config=strategy_config,
         data=BacktestData({market_key: data_df}, price_data),
-        strategies=[DemoStrategy((1000, 3000)), DemoStrategy((1500, 2500))], # two strategies
-        backtest_config=BacktestConfig(callback=after_backtest),
-        threads=2, # concurrent number
+        strategies=[DemoStrategy((1000, 3000)), DemoStrategy((1500, 2500))],  # two strategies
+        backtest_config=BacktestConfig(),
+        threads=2,  # concurrent number
     )
     backtest.run()
     pass

setup.py

@@ -7,7 +7,7 @@
 
 setup(
     name="zelos-demeter",
-    version="1.0.0",
+    version="1.0.1",
     packages=find_packages(exclude=["tests", "tests.*", "samples", "samples.*"]),
     url="https://zelos-demeter.readthedocs.io",
     license="MIT",