feat: add backtesting mode w/ silverback test command

.github/workflows/test.yaml

@@ -79,7 +79,10 @@ jobs:
             pip install .[test]
 
         - name: Run Tests
-          run: pytest -m "not fuzzing" -n 0 -s --cov
+          run: pytest -p "no:silverback_test" -m "not fuzzing" -n 0 -s --cov
+
+        - name: Run Backtests
+          run: silverback test tests/backtest -n 0 -s --cov
 
 # NOTE: uncomment this block after you've marked tests with @pytest.mark.fuzzing
 #    fuzzing:

docs/commands/run.rst

@@ -6,6 +6,10 @@ CLI commands for local development of running Silverback bots and task workers.
   :prog: silverback run
   :nested: none
 
+.. click:: silverback._cli:test
+  :prog: silverback test
+  :nested: none
+
 .. click:: silverback._cli:worker
   :prog: silverback worker
   :nested: none

docs/userguides/development.md

@@ -306,7 +306,31 @@ The client will send tasks to the 2 worker subprocesses, and all task queue and
 
 ## Testing your Bot
 
-TODO: Add backtesting mode w/ `silverback test`
+You can test your bot across various ranges of history using the [`silverback test`](../commands/test) command to perform "backtesting".
+Backtesting is a super useful technique to measure the performance of your bot by checking it against different periods of history to see how it would react.
+Silverback has provisional support for special backtest cases that must have a name like `backtest_*.yaml` under your `tests/` folder in your project.
+
+Here is an example test case:
+
+```yaml
+# `bots` key is optional, defaults to `bot:bot` if omitted
+bots: "<bot name>"
+# can also use array of bot names, or "*" for all bots under `bots/`
+
+# required, recommended to be a public network
+network: "ethereum:mainnet"
+
+# recommended to pick a small range of blocks for your case
+start_block: 15_338_009
+stop_block: 15_338_018
+
+# NOTE: Ignores extra data
+something_else: blah
+```
+
+In a complex project (where your bot(s) are not the only component), it is recommended to place your backtests under a separate folder such as `tests/backtest/`.
+This will allow you to run backtests separately from your other tests as they take a long time to execute.
+Further, you can use any pytest arguments to help select your tests, for example using `-k <bot-name>` in a [test matrix](https://docs.github.com/en/actions/writing-workflows/choosing-what-your-workflow-does/running-variations-of-jobs-in-a-workflow#using-a-matrix-strategy), if you have multiple bots to test.
 
 ## Deploying your Bot
 

setup.py

@@ -76,6 +76,7 @@
     ],
     entry_points={
         "console_scripts": ["silverback=silverback._cli:cli"],
+        "pytest11": ["silverback_test=silverback.pytest"],
     },
     python_requires=">=3.10,<4",
     extras_require=extras_require,

silverback/__init__.py

@@ -22,6 +22,7 @@ def __getattr__(name: str):
 
 __all__ = [
     "StateSnapshot",
+    "BacktestRunner",
     "CircuitBreaker",
     "SilverbackBot",
     "SilverbackException",

silverback/_cli.py

@@ -1,11 +1,13 @@
 import asyncio
 import os
+import sys
 from collections import defaultdict
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import TYPE_CHECKING, Optional
 
 import click
+import pytest
 import yaml  # type: ignore[import-untyped]
 from ape import Contract, convert
 from ape.cli import (
@@ -15,6 +17,7 @@
     account_option,
     ape_cli_context,
     network_option,
+    verbosity_option,
 )
 from ape.exceptions import Abort, ApeException, ConversionError
 from ape.logging import LogLevel
@@ -182,6 +185,21 @@ def worker(cli_ctx, account, workers, max_exceptions, shutdown_timeout, debug, b
     )
 
 
+@cli.command(
+    section="Local Commands",
+    add_help_option=False,  # NOTE: This allows pass-through to pytest's help
+    short_help="Run bot backtests (`tests/**/backtest_*.yaml`)",
+    context_settings=dict(ignore_unknown_options=True),
+)
+@ape_cli_context()
+@verbosity_option()
+@click.argument("pytest_args", nargs=-1, type=click.UNPROCESSED)
+def test(cli_ctx, pytest_args):
+    if return_code := pytest.main([*pytest_args], ["silverback_test"]):
+        # only exit with non-zero status to make testing easier
+        sys.exit(return_code)
+
+
 @cli.command(section="Cloud Commands (https://silverback.apeworx.io)")
 @auth_required
 def login(auth: "FiefAuth"):

silverback/pytest.py

@@ -0,0 +1,147 @@
+import asyncio
+import os
+import sys
+from functools import cached_property
+from pathlib import Path
+
+import pytest
+import yaml  # type: ignore[import]
+from ape import networks
+
+from silverback._importer import import_from_string
+from silverback.exceptions import SilverbackException
+from silverback.recorder import TaskResult
+from silverback.runner import BacktestRunner
+
+
+class AssertionViolation(SilverbackException):
+    pass
+
+
+def pytest_collect_file(parent, file_path):
+    if file_path.suffix == ".yaml" and file_path.name.startswith("backtest"):
+        return BacktestFile.from_parent(parent, path=file_path)
+
+
+class BacktestFile(pytest.File):
+    def collect(self):
+        raw = yaml.safe_load(self.path.open())
+        if not (network_triple := raw.get("network")):
+            raise ValueError(f"{self.path} is missing key 'network'.")
+
+        start_block = raw.get("start_block", 0)
+        stop_block = raw.get("stop_block", -1)
+        assertion_checks = raw.get("assertions", {})
+
+        # NOTE: Use default assumption rest of CLI uses
+        raw_bot_paths = raw.get("bots", "bot:bot")
+        if raw_bot_paths == "*":
+            raw_bot_paths = list(f.stem for f in (Path.cwd() / "bots").glob("*.py"))
+
+        if isinstance(raw_bot_paths, list):
+            for bot_path in raw_bot_paths:
+                if ":" in bot_path:
+                    bot_path, bot_name = bot_path.split(":")
+                else:
+                    bot_name = "bot"
+
+                yield BacktestItem.from_parent(
+                    self,
+                    name=f"{self.path.stem}[{bot_path}]",
+                    file_path=self.path,
+                    bot_path=bot_path,
+                    bot_name=bot_name,
+                    network_triple=network_triple,
+                    start_block=start_block,
+                    stop_block=stop_block,
+                    assertion_checks=assertion_checks,
+                )
+
+        else:
+            if ":" in raw_bot_paths:
+                bot_path, bot_name = raw_bot_paths.split(":")
+            else:
+                bot_path = raw_bot_paths
+                bot_name = "bot"
+
+            yield BacktestItem.from_parent(
+                self,
+                name=self.path.stem,
+                file_path=self.path,
+                bot_path=bot_path,
+                bot_name=bot_name,
+                network_triple=network_triple,
+                start_block=start_block,
+                stop_block=stop_block,
+                assertion_checks=assertion_checks,
+            )
+
+
+class BacktestItem(pytest.Item):
+    def __init__(
+        self,
+        *,
+        file_path,
+        bot_path,
+        bot_name,
+        network_triple,
+        start_block,
+        stop_block,
+        assertion_checks,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.file_path = file_path
+        self.bot_path = bot_path
+        self.bot_name = bot_name
+        self.network_triple = network_triple
+        self.start_block = start_block
+        self.stop_block = stop_block
+        self.assertion_checks = assertion_checks
+
+        self.assertion_failures = 0
+        self.overruns = 0
+        self.total_tasks = 0
+        self.task_failures = 0
+
+    def check_assertions(self, result: TaskResult):
+        self.total_tasks += 1
+
+        if result.execution_time > float(self.assertion_checks.get("execution_time", "inf")):
+            self.overruns += 1
+
+        if result.error:
+            self.task_failures += 1
+
+    @cached_property
+    def runner(self) -> BacktestRunner:
+        # NOTE: Make sure it can find the path
+        if (bot_folder := (Path.cwd() / "bots")).exists():
+            sys.path.insert(0, str(bot_folder))
+
+        # NOTE: Set parameters for loading settings properly
+        os.environ["SILVERBACK_BOT_NAME"] = self.bot_path
+        with networks.parse_network_choice(self.network_triple):
+            # NOTE: Loading bot requires a network connection
+            bot = import_from_string(f"{self.bot_path}:{self.bot_name}")
+
+        return BacktestRunner(
+            bot,
+            start_block=self.start_block,
+            stop_block=self.stop_block,
+            network_triple=self.network_triple,
+            check_assertions=self.check_assertions,
+        )
+
+    def runtest(self):
+        asyncio.run(self.runner.run())
+
+        # Test is over, check test status (if it completed)
+        self.raise_run_status()
+
+    def raise_run_status(self):
+        if self.overruns > 0 or self.assertion_failures > 0:
+            raise AssertionViolation()
+
+    def reportinfo(self):
+        return self.path, None, self.name