Module serve

Classes

class Job (strategy: Strategy, data: DataPipe, lookback: datetime.timedelta, *, result_path: Optional[str] = None, params: Optional[Dict[str, Any]] = None, index_pipe: Optional[DataPipe] = None, working_directory: pathlib.PurePath = PurePosixPath('prod_data'), indicators: Union[IndicatorSet, List[IndicatorSet], Dict[int, IndicatorSet]] = IndicatorSet(n_indicators=0), trigger_cb: Optional[Callable[[StateSignals, pathlib.PurePath], None]] = None, renderer: Union[RendererList, Renderer] = None, cash_controller: CashControllerBase = <backtest.engine.cashController.CashControllerBase object>, time_res_extender: Optional[TimeResExtender] = None)

This class is designed to run a strategy in inference mode. It is derived from the backtest class to keep as much similarity as possible. The main difference is that the Job class will not run the strategy in a loop, but will run it once and return the signals. It is designed to be used in a server environment where the strategy will be run on a schedule. (Cron schedule or with the schedule module)

Overall mechanism: The Job class receives as parameter a datapipe. When the job class is run, it starts by fetching the data from the datapipe. It then prepares the data as it is prepared in the backtest class. It then runs the strategy on the data and records the signals emitted by the strategy. It builds a StateSignals object with the signals and the state of the system (i.e. the account, the broker, the cash controller, etc.). It then calls the renderer. The renderer converts the StateSignals object into a file (or a set of files) that can be used to visualize the signals or be imported into an application. Finally, the Job class calls a trigger callback that can be used to trigger other actions such as sending a notification, buying or selling assets based on an api, etc.

How to use: The Job class is designed to be run once a backtest has been completed (To see the performance expected by the strategy). Also, there are two ways to run the Job class. The first way is to call run the run_job method once in a while (Could be scheduled with a Cron job). The second way, is to pass a schedule.Job object to the run_job method and the script will run the job at the specified interval. In that case, the script never finishes.

Example

The first example is designed to be run by a Cron job. The script will brun and exit.

>>> # imports ...
>>> data_pipe = ...  # A data pipe that fetches the data
>>> strategy = ...  # A strategy object
>>> # In case a backtest has been run prior to this, we pass the path to the backtest results.  This way, the
>>> # Job class will load the initial parameters from the backtest results.
>>> job = Job(strategy, data_pipe, timedelta(days=1), result_path="path/to/backtest/result", renderer=...,
...          trigger_cb=...)
>>> # In case a backtest hasn't been run, we pass the parameters directly to the Job class.
>>> # The parameters are the parameters passed to the backtest class.  Almost every parameters must be passed,
>>> # because it doesn't have defaults.
>>> job = Job(strategy, data_pipe, timedelta(days=1), params={"initial_cash": 10000, "commission": 0.01, ...},
...          renderer=..., trigger_cb=...)
>>> job.run_job()

The second example is designed to be run in a loop. The script will run the job every day at 5pm.

>>> # imports ...
>>> data_pipe = ...  # A data pipe that fetches the data
>>> strategy = ...  # A strategy object
>>> # In case a backtest has been run prior to this, we pass the path to
>>> # the backtest results.  This way, the Job class will load the initial parameters from the backtest results.
>>> job = Job(strategy, data_pipe, timedelta(days=1), result_path="path/to/backtest/result", renderer=...,
...          trigger_cb=...)
>>> # In case a backtest hasn't been run, we pass the parameters directly to the Job class.
>>> # The parameters are the parameters passed to the backtest class.  Almost every parameters must be passed,
>>> # because it doesn't have defaults.
>>> job = Job(strategy, data_pipe, timedelta(days=1), params={"initial_cash": 10000, "commission": 0.01, ...},
...          renderer=..., trigger_cb=...)
>>> job.run_job(schedule.every().day.at("17:00"))

Args:

strategy: The strategy to use and extract signals from.
data: The data pipeline used to fetch the data for the strategy. It must return a list of dict of TSDatawhere position in the list are different time resolutions (like in the backtest) and the keys in the dict arethe tickers.
lookback: The lookback period to fetch the data. For example, let's say the lookback is 1y, the datafetched will be from (now - 1y) to now at every run.
result_path: The path to the backtest results. If this is provided, the Job class will load theparameters from the backtest results. If this is not provided, the Job class will load the parameters from theparams dictionary.
params: The parameters to use for the backtest. If the result_path is provided, this parameter isignored. The parameters are the same as the parameters passed to the backtest class. Almost every parametersmust be passed, because it doesn't have defaults. The parameters required are: - main_timestep: The main timestep to use for the backtest. - initial_cash: The initial cash to start the backtest. - commission: The commission to apply to the trades. - relative_commission: The relative commission to apply to the trades. - margin_interest: The margin interest to apply to the trades. - min_initial_margin: The minimum initial margin to apply to the trades. - min_maintenance_margin: The minimum maintenance margin to apply to the trades. - liquidation_delay: The liquidation delay to apply to the trades. - min_initial_margin_short: The minimum initial margin for short trades. - min_maintenance_margin_short: The minimum maintenance margin for short trades. - window: The window to use for the backtest. - default_marginable: The default marginable status used in the backtest. - default_shortable: The default shortable status used in the backtest. - default_short_rate: The default short rate used in the backtest. - risk_free_rate: The risk-free rate used in the backtest. - verbose: The verbosity level of the backtest.
index_pipe: A datapipe designed to fetch the index data. This is useful to compare the strategyperformance with the index performance. It must return a list of TSData objects. Although it is recommended touse only one reference index because the renderers are designed to render only one index, it is possible toreturn more than one. It is required that the time resolution of the index is the same as the main timeresolution.
working_directory: The working directory to use to export files. This will be considered as the root bythe renderers. For example, the prebuilt renderers will export their reports in the reports directory of theworking directory.
indicators: The same indicators used for the backtest.
trigger_cb: The trigger callback to call after the renderer has rendered the StateSignals.This callback is designed to trigger other actions like sending a notification, buying or selling assets basedon an api, etc.
renderer: A renderer object designed to convert the StateSignals object to an output file. The renderercan be a single renderer or a list of renderers (RendererList). The renderers are run in the same order as theyare in the list.
cash_controller: The cash controller used during the backtest.
time_res_extender: The time resolution extender used during the backtest.

Parent

Methods

def pipeline(self, now_override: Optional[datetime.datetime] = None)

This method will run the whole pipeline. The steps of the pipeline are: 1. Fetch the data 2. Setup the Job. (Initialize the different variables) 3. Prepare the data 4. Run the cash controller 5. Run the broker (It will always be run on the previous timestep, i.e. lagging by one timestep) 6. Run the strategy 7. Save the state of the job for later warm restarts. 8. Package the signals and the state in a StateSignals object 9. Render the signals using the renderer 10. Call the trigger callback

Args:

now_override: A datetime object to override the current time. (Useful for testing in simulated environments)

Returns:

None
def run_job(self, every: Optional[ForwardRef('schedule.Job')] = None)

This method will run the job. If the every parameter is None, the job will run once and render the signals. If the every parameter is not None, the job will run at the specified interval and never return (never exit).

Args:

every: The interval to run the job.

Returns:

None
def setup(self) ‑> Optional[datetime.datetime]

This method will setup the backtest. It will load from cache, if it exists, the backtest state. Loading from cache the backtest state simulate a warm restart, (like the script has never stopped). If the script is runned in a cron job, so it exits after each run, and restarts at the next run, it wouldn't be able to remember the state of the job and would think it is always the first run. This is why the cache is used. It is used to remember the state of the job between runs. It is useful to remember the state to calulate the current statistics of the strategy. It will later be possible to compare the current performance with the backtest performances.