diff --git a/config.py b/config.py index b674863a..042848ba 100644 --- a/config.py +++ b/config.py @@ -210,6 +210,26 @@ class AppSettings(BaseSettings): ) +class BacktestingSettings(BaseSettings): + """Backtest result retention. + + A finished backtest is ~98% bulk arrays (processed_data, pnl_timeseries) and only a + few KB of metrics, so full payloads are archived to disk and only metrics stay + resident. Retention is therefore a count of results, not a memory budget. + """ + + max_results: int = Field( + default=100, + description="How many finished backtests to retain before the oldest are reaped" + ) + results_path: str = Field( + default="bots/data/backtests", + description="Directory holding archived backtest payloads (inside the bots volume, so it survives redeploys)" + ) + + model_config = SettingsConfigDict(env_prefix="BACKTESTING_", extra="ignore") + + class Settings(BaseSettings): """Combined application settings.""" @@ -221,6 +241,7 @@ class Settings(BaseSettings): gateway: GatewaySettings = Field(default_factory=GatewaySettings) cors: CORSSettings = Field(default_factory=CORSSettings) app: AppSettings = Field(default_factory=AppSettings) + backtesting: BacktestingSettings = Field(default_factory=BacktestingSettings) # Direct banned_tokens field to handle env parsing banned_tokens: List[str] = Field( @@ -235,4 +256,5 @@ class Settings(BaseSettings): extra="ignore" ) + settings = Settings() diff --git a/routers/backtesting.py b/routers/backtesting.py index f58f0f73..ff8f2b5a 100644 --- a/routers/backtesting.py +++ b/routers/backtesting.py @@ -43,10 +43,17 @@ async def get_backtest_task( service: BacktestingService = Depends(get_backtesting_service), ): """Get a backtest task by ID, including results if completed.""" - task = service.get_task(task_id) - if task is None: + payload = service.get_task_payload(task_id) + if payload is None: + # A reaped task is reported as 410 rather than 404: it did exist and its result is + # permanently gone, which a caller cannot infer from "not found". + if service.was_reaped(task_id): + raise HTTPException( + status_code=410, + detail=f"Task {task_id} completed but its result was reaped to honour the retention limit", + ) raise HTTPException(status_code=404, detail=f"Task {task_id} not found") - return task.to_dict(include_result=True) + return payload @router.delete("/tasks/{task_id}") diff --git a/services/backtesting_service.py b/services/backtesting_service.py index 897a0515..606da1bc 100644 --- a/services/backtesting_service.py +++ b/services/backtesting_service.py @@ -1,12 +1,28 @@ """ BacktestingService manages background backtesting tasks. -Stores task state and results in memory for polling. + +Results are archived to gzipped JSON on completion; only their metrics stay resident. + +A finished backtest is dominated by bulk arrays -- processed_data and pnl_timeseries are +~98% of a payload (7.6 MB for a 7-day run at 1m), while the metrics anyone actually reads +back are under 10 KB. Holding whole results in memory therefore rationed the wrong +resource: a flat 50-task cap was ~380 MB for 7-day runs but ~3.2 GB for 60-day ones, and +because reaping can only drop *terminal* tasks it did nothing at all during a submission +burst -- exactly when nothing has finished yet and memory is climbing fastest. + +With the bulk on disk a resident task costs a few KB, so retention is a count of results +rather than a memory budget, the full payload is rehydrated on read, and results survive +a restart instead of dying with the process. """ import asyncio +import gzip +import json import logging import uuid +from collections import OrderedDict from datetime import datetime, timezone from enum import Enum +from pathlib import Path from typing import Any, Dict, Optional from hummingbot.strategy_v2.backtesting.backtesting_engine_base import BacktestingEngineBase @@ -15,6 +31,10 @@ logger = logging.getLogger(__name__) +# How many reaped ids to remember, so a caller polling a result that was dropped gets a +# definite "it existed and is gone" instead of a bare 404 it cannot distinguish from a typo. +_REAPED_MEMORY = 1000 + class BacktestTaskStatus(str, Enum): PENDING = "pending" @@ -24,6 +44,23 @@ class BacktestTaskStatus(str, Enum): CANCELLED = "cancelled" +_TERMINAL = (BacktestTaskStatus.COMPLETED, BacktestTaskStatus.FAILED, BacktestTaskStatus.CANCELLED) + + +def _json_default(obj: Any) -> Any: + """Coerce values the stdlib encoder rejects (numpy scalars, timestamps). + + Float dict keys -- processed_data is keyed by epoch seconds -- are coerced to strings + by json.dump itself, which is exactly what FastAPI's encoder already did on the wire. + A disk round-trip therefore reproduces the same JSON the API returned before. + """ + if hasattr(obj, "item"): # numpy scalar + return obj.item() + if isinstance(obj, datetime): + return obj.isoformat() + return str(obj) + + class BacktestTask: def __init__(self, task_id: str, config: dict): self.task_id = task_id @@ -32,7 +69,9 @@ def __init__(self, task_id: str, config: dict): self.created_at = datetime.now(timezone.utc) self.started_at: Optional[datetime] = None self.completed_at: Optional[datetime] = None + # Once archived this holds metrics only; the full payload lives on disk. self.result: Optional[Dict[str, Any]] = None + self.archived = False self.error: Optional[str] = None self._asyncio_task: Optional[asyncio.Task] = None @@ -50,32 +89,61 @@ def to_dict(self, include_result: bool = True) -> dict: data["result"] = self.result return data + def metadata(self) -> dict: + """Light record for the on-disk index -- everything but the bulk arrays.""" + return {**self.to_dict(include_result=False), "result": self.result, "archived": self.archived} + class BacktestingService: - def __init__(self, max_tasks: int = 50): - self._tasks: Dict[str, BacktestTask] = {} + def __init__(self, max_results: Optional[int] = None, results_path: Optional[str] = None): + self._tasks: "OrderedDict[str, BacktestTask]" = OrderedDict() self._engine = BacktestingEngineBase() - self._max_tasks = max_tasks + self._max_results = max_results if max_results is not None else settings.backtesting.max_results + self._results_dir = Path(results_path if results_path is not None else settings.backtesting.results_path) + self._reaped: "OrderedDict[str, str]" = OrderedDict() + self._results_dir.mkdir(parents=True, exist_ok=True) + self._restore() + # Honour a limit that was lowered since the last run. + self._reap() @property def tasks(self) -> Dict[str, BacktestTask]: return self._tasks + # -- submission and lifecycle -- + def submit_task(self, config: dict) -> BacktestTask: """Submit a new backtesting task to run in the background.""" - self._cleanup_old_tasks() task_id = str(uuid.uuid4())[:8] task = BacktestTask(task_id=task_id, config=config) self._tasks[task_id] = task task._asyncio_task = asyncio.create_task(self._run_task(task)) + self._reap() logger.info(f"Backtesting task {task_id} submitted") return task def get_task(self, task_id: str) -> Optional[BacktestTask]: return self._tasks.get(task_id) + def get_task_payload(self, task_id: str) -> Optional[dict]: + """Task dict with the complete result, rehydrated from disk when archived.""" + task = self._tasks.get(task_id) + if task is None: + return None + data = task.to_dict(include_result=False) + result = self._read_archive(task_id) if task.archived else None + if result is None: + result = task.result # never archived, or the archive went missing + if result is not None: + data["result"] = result + return data + + def was_reaped(self, task_id: str) -> bool: + """True if the task completed and was later dropped to honour the retention limit.""" + return task_id in self._reaped + def cancel_task(self, task_id: str) -> bool: - """Cancel a running task or remove a completed one.""" + """Cancel a running task or remove a completed one, discarding its archive.""" task = self._tasks.get(task_id) if task is None: return False @@ -83,15 +151,15 @@ def cancel_task(self, task_id: str) -> bool: task._asyncio_task.cancel() task.status = BacktestTaskStatus.CANCELLED task.completed_at = datetime.now(timezone.utc) - del self._tasks[task_id] + self._drop(task_id, reaped=False) return True def list_tasks(self) -> list: - """List all tasks (without full results for brevity).""" + """List all tasks (without results for brevity).""" return [t.to_dict(include_result=False) for t in self._tasks.values()] async def run_backtest_sync(self, config: dict) -> dict: - """Run a backtest synchronously (returns full result directly).""" + """Run a backtest synchronously (returns full result directly, stores nothing).""" return await self._execute_backtest(config) async def _run_task(self, task: BacktestTask): @@ -99,18 +167,25 @@ async def _run_task(self, task: BacktestTask): task.status = BacktestTaskStatus.RUNNING task.started_at = datetime.now(timezone.utc) try: - task.result = await self._execute_backtest(task.config) + result = await self._execute_backtest(task.config) task.status = BacktestTaskStatus.COMPLETED + self._archive(task, result) logger.info(f"Backtesting task {task.task_id} completed") except asyncio.CancelledError: task.status = BacktestTaskStatus.CANCELLED logger.info(f"Backtesting task {task.task_id} cancelled") + raise except Exception as e: task.status = BacktestTaskStatus.FAILED task.error = str(e) logger.error(f"Backtesting task {task.task_id} failed: {e}", exc_info=True) finally: task.completed_at = datetime.now(timezone.utc) + self._persist_index() + # Reap here too: a burst that is submitted all at once and only then starts + # finishing would otherwise never be trimmed, since nothing is terminal at + # submit time and no further submissions arrive to trigger it. + self._reap() async def _execute_backtest(self, config: dict) -> dict: """Core backtest execution logic shared by sync and async modes.""" @@ -163,15 +238,103 @@ async def _execute_backtest(self, config: dict) -> dict: "pnl_timeseries": backtesting_results.get("pnl_timeseries", []), } - def _cleanup_old_tasks(self): - """Remove oldest completed/failed tasks if we exceed max_tasks.""" - if len(self._tasks) < self._max_tasks: + # -- archive -- + + def _archive_path(self, task_id: str) -> Path: + return self._results_dir / f"{task_id}.json.gz" + + @property + def _index_path(self) -> Path: + return self._results_dir / "_index.json" + + def _archive(self, task: BacktestTask, result: dict) -> None: + """Write the full payload to disk and keep only its metrics resident.""" + try: + with gzip.open(self._archive_path(task.task_id), "wt", encoding="utf-8") as fh: + json.dump(result, fh, default=_json_default) + except (OSError, TypeError, ValueError) as e: + # Losing the result would be worse than holding it: keep it in memory and let + # the retention limit reclaim it later. + logger.error(f"Could not archive backtest {task.task_id}, keeping it in memory: {e}") + task.result = result + return + task.result = {"results": result.get("results", {})} + task.archived = True + + def _read_archive(self, task_id: str) -> Optional[dict]: + path = self._archive_path(task_id) + if not path.exists(): + return None + try: + with gzip.open(path, "rt", encoding="utf-8") as fh: + return json.load(fh) + except (OSError, ValueError) as e: + logger.error(f"Could not read archived backtest {task_id}: {e}") + return None + + def _persist_index(self) -> None: + try: + index = {tid: t.metadata() for tid, t in self._tasks.items() if t.status in _TERMINAL} + with open(self._index_path, "w", encoding="utf-8") as fh: + json.dump(index, fh, default=_json_default) + except (OSError, TypeError, ValueError) as e: + logger.error(f"Could not persist backtest index: {e}") + + def _restore(self) -> None: + """Rebuild finished tasks from the index so results outlive a restart.""" + if not self._index_path.exists(): + return + try: + with open(self._index_path, encoding="utf-8") as fh: + index = json.load(fh) + except (OSError, ValueError) as e: + logger.error(f"Could not read backtest index, starting empty: {e}") return - completed = [ - (tid, t) for tid, t in self._tasks.items() - if t.status in (BacktestTaskStatus.COMPLETED, BacktestTaskStatus.FAILED, BacktestTaskStatus.CANCELLED) - ] - completed.sort(key=lambda x: x[1].completed_at or x[1].created_at) - while len(self._tasks) >= self._max_tasks and completed: - tid, _ = completed.pop(0) - del self._tasks[tid] + + for task_id, meta in index.items(): + try: + task = BacktestTask(task_id=task_id, config=meta.get("config") or {}) + task.status = BacktestTaskStatus(meta.get("status", "completed")) + task.created_at = datetime.fromisoformat(meta["created_at"]) + for field in ("started_at", "completed_at"): + if meta.get(field): + setattr(task, field, datetime.fromisoformat(meta[field])) + task.error = meta.get("error") + task.result = meta.get("result") + task.archived = bool(meta.get("archived")) and self._archive_path(task_id).exists() + self._tasks[task_id] = task + except (KeyError, TypeError, ValueError) as e: + logger.warning(f"Skipping unreadable backtest index entry '{task_id}': {e}") + + if self._tasks: + logger.info(f"Restored {len(self._tasks)} backtest result(s) from {self._results_dir}") + + # -- retention -- + + def _reap(self) -> None: + """Drop the oldest finished results beyond the retention limit. + + Running tasks are never dropped -- there is nothing to reclaim from one anyway, + since a result exists only once it completes. A large burst can therefore briefly + exceed the limit, which is safe now that a resident task is a few KB. + """ + if len(self._tasks) <= self._max_results: + return + terminal = [(tid, t) for tid, t in self._tasks.items() if t.status in _TERMINAL] + terminal.sort(key=lambda kv: kv[1].completed_at or kv[1].created_at) + while len(self._tasks) > self._max_results and terminal: + task_id, _ = terminal.pop(0) + self._drop(task_id, reaped=True) + + def _drop(self, task_id: str, reaped: bool) -> None: + self._tasks.pop(task_id, None) + path = self._archive_path(task_id) + try: + path.unlink(missing_ok=True) + except OSError as e: + logger.error(f"Could not delete archived backtest {task_id}: {e}") + if reaped: + self._reaped[task_id] = datetime.now(timezone.utc).isoformat() + while len(self._reaped) > _REAPED_MEMORY: + self._reaped.popitem(last=False) + self._persist_index() diff --git a/test/test_backtest_storage.py b/test/test_backtest_storage.py new file mode 100644 index 00000000..ba12b62d --- /dev/null +++ b/test/test_backtest_storage.py @@ -0,0 +1,195 @@ +""" +Tests for backtest result archiving, retention and restart-survival. + +The behaviour being pinned down: a finished backtest is ~98% bulk arrays, so the payload +is written to a gzipped file and only its metrics stay resident. Reads rehydrate from +disk, the retention limit counts results rather than bytes, and a reaped result is +reported as 410 rather than an ambiguous 404. + +The repo has no async test setup, so coroutines are driven with asyncio.run(). + +Run with: pytest test/test_backtest_storage.py -v +""" +import asyncio +import gzip +import json + +import numpy as np +import pytest + +from services.backtesting_service import BacktestingService, BacktestTaskStatus + +# Mirrors the real payload shape, including the two things that trip a naive json.dump: +# numpy scalars, and processed_data being keyed by float epoch seconds. +PAYLOAD = { + "executors": [{"id": "e1", "net_pnl_quote": 1.5}], + "processed_data": {"close_bt": {1785429000.0: np.float64(3.5), 1785429060.0: np.float64(3.6)}}, + "results": {"net_pnl": -0.0179, "total_executors": 74, "sharpe_ratio": np.float64(-0.9375)}, + "position_holds": [], + "position_held_timeseries": [], + "pnl_timeseries": [{"timestamp": 1785429000.0, "total_pnl": np.int64(0)}], +} + + +class StubService(BacktestingService): + """BacktestingService with the engine replaced, so no market data is needed.""" + + def __init__(self, tmp_path, max_results=3, fail=False): + super().__init__(max_results=max_results, results_path=str(tmp_path)) + self._fail = fail + + async def _execute_backtest(self, config): + await asyncio.sleep(0) + if self._fail: + raise RuntimeError("no candles for ETH-USDT") + return json.loads(json.dumps(PAYLOAD, default=lambda o: o.item())) + + +def _config(tag="c1"): + return {"start_time": 1, "end_time": 2, "config": {"id": tag, "controller_name": "ema_trend_v1"}} + + +async def _submit_and_wait(service, tag="c1"): + task = service.submit_task(_config(tag)) + await task._asyncio_task + return task + + +def test_result_is_archived_and_only_metrics_stay_resident(tmp_path): + async def scenario(): + service = StubService(tmp_path) + task = await _submit_and_wait(service) + + assert task.status == BacktestTaskStatus.COMPLETED + assert task.archived + # The bulk is gone from memory; metrics remain. + assert set(task.result) == {"results"} + assert task.result["results"]["net_pnl"] == -0.0179 + + archive = tmp_path / f"{task.task_id}.json.gz" + assert archive.exists() + with gzip.open(archive, "rt", encoding="utf-8") as fh: + stored = json.load(fh) + assert "processed_data" in stored and "pnl_timeseries" in stored + return service, task + + service, task = asyncio.run(scenario()) + + # A read rehydrates the whole payload, so the HTTP contract is unchanged. + payload = service.get_task_payload(task.task_id) + assert set(payload["result"]) == set(PAYLOAD) + # json coerces float keys to strings exactly as FastAPI already did on the wire. + assert payload["result"]["processed_data"]["close_bt"]["1785429000.0"] == 3.5 + assert payload["result"]["pnl_timeseries"][0]["total_pnl"] == 0 + + +def test_archive_is_much_smaller_than_the_payload(tmp_path): + async def scenario(): + service = StubService(tmp_path) + return await _submit_and_wait(service) + + task = asyncio.run(scenario()) + raw = len(json.dumps(PAYLOAD, default=lambda o: o.item())) + assert (tmp_path / f"{task.task_id}.json.gz").stat().st_size < raw + + +def test_oldest_results_are_reaped_beyond_the_limit(tmp_path): + async def scenario(): + service = StubService(tmp_path, max_results=3) + tasks = [await _submit_and_wait(service, f"c{i}") for i in range(5)] + return service, tasks + + service, tasks = asyncio.run(scenario()) + + assert len(service.tasks) == 3, "retention limit not enforced" + dropped, kept = tasks[:2], tasks[2:] + + for task in dropped: + assert task.task_id not in service.tasks + assert not (tmp_path / f"{task.task_id}.json.gz").exists(), "archive left behind" + assert service.was_reaped(task.task_id), "reaped id must stay distinguishable from a typo" + assert service.get_task_payload(task.task_id) is None + for task in kept: + assert service.get_task_payload(task.task_id)["result"]["results"]["net_pnl"] == -0.0179 + + +def test_results_survive_a_restart(tmp_path): + async def scenario(): + service = StubService(tmp_path) + return await _submit_and_wait(service) + + task = asyncio.run(scenario()) + + # A fresh service over the same directory: what a container restart looks like. + restarted = BacktestingService(max_results=3, results_path=str(tmp_path)) + + assert task.task_id in restarted.tasks + restored = restarted.tasks[task.task_id] + assert restored.status == BacktestTaskStatus.COMPLETED + assert restored.config["config"]["id"] == "c1" + + payload = restarted.get_task_payload(task.task_id) + assert payload["result"]["results"]["total_executors"] == 74 + assert "processed_data" in payload["result"], "bulk payload should still be readable" + + +def test_failed_task_keeps_its_error_and_writes_no_archive(tmp_path): + async def scenario(): + service = StubService(tmp_path, fail=True) + return service, await _submit_and_wait(service) + + service, task = asyncio.run(scenario()) + + assert task.status == BacktestTaskStatus.FAILED + assert "no candles" in task.error + assert not (tmp_path / f"{task.task_id}.json.gz").exists() + assert service.get_task_payload(task.task_id)["error"] == "no candles for ETH-USDT" + + +def test_result_is_kept_in_memory_when_archiving_fails(tmp_path): + """Losing a result is worse than holding it, so a write failure must not discard it.""" + + class Unwritable(StubService): + def _archive_path(self, task_id): + return tmp_path / "missing-dir" / f"{task_id}.json.gz" + + async def scenario(): + service = Unwritable(tmp_path) + return service, await _submit_and_wait(service) + + service, task = asyncio.run(scenario()) + + assert not task.archived + assert set(task.result) == set(PAYLOAD), "full result should have been retained in memory" + assert "processed_data" in service.get_task_payload(task.task_id)["result"] + + +def test_deleting_a_task_removes_its_archive(tmp_path): + async def scenario(): + service = StubService(tmp_path) + return service, await _submit_and_wait(service) + + service, task = asyncio.run(scenario()) + + assert service.cancel_task(task.task_id) is True + assert not (tmp_path / f"{task.task_id}.json.gz").exists() + # Deleted on request, not reaped -- the caller knows why it is gone. + assert not service.was_reaped(task.task_id) + assert service.cancel_task(task.task_id) is False + + +def test_sync_run_stores_nothing(tmp_path): + async def scenario(): + service = StubService(tmp_path) + result = await service.run_backtest_sync(_config()) + return service, result + + service, result = asyncio.run(scenario()) + + assert "processed_data" in result + assert service.tasks == {} + assert list(tmp_path.glob("*.json.gz")) == [] + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/test/test_controller_config_class_loading.py b/test/test_controller_config_class_loading.py new file mode 100644 index 00000000..16c368c0 --- /dev/null +++ b/test/test_controller_config_class_loading.py @@ -0,0 +1,86 @@ +""" +Tests for resolving a controller's concrete config class. + +Regression: a controller module imports its own base class, so the base is a member of the +module namespace too -- and every base except ControllerConfigBase is itself a strict +subclass of ControllerConfigBase. The old check ("subclass of a base, but not that same +base") therefore accepted the *sibling* bases, and because inspect.getmembers() returns +members sorted by name, whichever name sorted first won. + +That silently resolved every controller whose config class sorts after its base: +ema_trend_v1 and supertrend_v1 and macd_bb_v1 -> DirectionalTradingControllerConfigBase, +pmm_simple and pmm_dynamic -> MarketMakingControllerConfigBase. Since the bases set +extra="forbid", /config/validate then rejected every controller-specific field +(ema_fast, ema_slow, adx_period, ...) as "Extra inputs are not permitted", and +/config/template advertised only base fields. Controllers whose names happen to sort +first (bollinger_v1, dman_v3) worked, which is what made this look intermittent. + +Run with: pytest test/test_controller_config_class_loading.py -v +""" +import pytest +from hummingbot.strategy_v2.controllers.controller_base import ControllerConfigBase +from hummingbot.strategy_v2.controllers.directional_trading_controller_base import DirectionalTradingControllerConfigBase +from hummingbot.strategy_v2.controllers.market_making_controller_base import MarketMakingControllerConfigBase + +from utils.file_system import fs_util + +BASE_CLASSES = { + ControllerConfigBase.__name__, + DirectionalTradingControllerConfigBase.__name__, + MarketMakingControllerConfigBase.__name__, +} + +# Controllers whose config class name sorts AFTER its own base class name -- the exact +# set that the name-ordering bug used to resolve to a base class. +SORTS_AFTER_ITS_BASE = [ + ("directional_trading", "ema_trend_v1", "EmaTrendV1Config"), + ("directional_trading", "macd_bb_v1", "MACDBBV1ControllerConfig"), + ("directional_trading", "supertrend_v1", "SuperTrendConfig"), + ("market_making", "pmm_simple", "PMMSimpleConfig"), + ("market_making", "pmm_dynamic", "PMMDynamicControllerConfig"), +] + +# Controllers that sorted before their base and so worked even with the bug -- kept here +# so a future "fix" cannot regress them. +SORTS_BEFORE_ITS_BASE = [ + ("directional_trading", "bollinger_v1", "BollingerV1ControllerConfig"), + ("directional_trading", "dman_v3", "DManV3ControllerConfig"), +] + + +@pytest.mark.parametrize( + "controller_type,controller_name,expected", + SORTS_AFTER_ITS_BASE + SORTS_BEFORE_ITS_BASE, +) +def test_resolves_the_concrete_config_class(controller_type, controller_name, expected): + config_class = fs_util.load_controller_config_class(controller_type, controller_name) + + assert config_class is not None, f"{controller_name} did not resolve to any config class" + assert config_class.__name__ not in BASE_CLASSES, ( + f"{controller_name} resolved to the base class {config_class.__name__}; " + "controller-specific fields would be rejected as 'Extra inputs are not permitted'" + ) + assert config_class.__name__ == expected + + +def test_controller_specific_fields_are_accepted_by_the_resolved_class(): + """The end-to-end symptom: /config/validate instantiates the resolved class.""" + config_class = fs_util.load_controller_config_class("directional_trading", "ema_trend_v1") + + config = config_class( + id="ema_eth_5_55", + controller_name="ema_trend_v1", + controller_type="directional_trading", + connector_name="binance_perpetual", + trading_pair="ETH-USDT", + interval="15m", + ema_fast=5, + ema_slow=55, + ) + + assert config.ema_fast == 5 + assert config.ema_slow == 55 + + +def test_unknown_controller_still_returns_none(): + assert fs_util.load_controller_config_class("directional_trading", "no_such_controller") is None