Skip to content

Backtest

API endpoints for backtest execution and status.

Provides endpoints for starting backtests and querying their status.

router = APIRouter(prefix='/api/backtest', tags=['backtest']) module-attribute

api_backtest_run(request) async

Enqueue a backtest and return the run ID.

Source code in src/onesecondtrader/dashboard/routers/backtest.py 
25
26
27
28
29
30
@router.post("/run")
async def api_backtest_run(request: BacktestRequest) -> dict:
    """Enqueue a backtest and return the run ID."""
    run_id = str(uuid.uuid4())
    enqueue_backtest(request, run_id)
    return {"run_id": run_id, "status": "queued"}

api_backtest_running() async

Return all currently running backtests with metadata and progress.

Source code in src/onesecondtrader/dashboard/routers/backtest.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
@router.get("/running")
async def api_backtest_running() -> dict:
    """Return all currently running backtests with metadata and progress."""
    result = []
    with _jobs_lock:
        items = list(running_jobs.items())
        meta_snapshot = dict(_running_metadata)
        orch_snapshot = dict(_orchestrator_refs)
    for run_id, status in items:
        if status in ("running", "queued"):
            meta = meta_snapshot.get(run_id, {})
            orch = orch_snapshot.get(run_id)
            progress = int(orch.progress * 100) if orch is not None else 0
            result.append(
                {
                    "run_id": run_id,
                    "status": status,
                    "progress": progress,
                    **meta,
                }
            )
    return {"running": result}

api_backtest_status(run_id) async

Return the current status of a backtest job.

Source code in src/onesecondtrader/dashboard/routers/backtest.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
@router.get("/status/{run_id}")
async def api_backtest_status(run_id: str) -> dict:
    """Return the current status of a backtest job."""
    with _jobs_lock:
        status = running_jobs.get(run_id, "not found")
        orch = _orchestrator_refs.get(run_id)
    progress = 0
    if status == "running":
        if orch is not None:
            progress = int(orch.progress * 100)
    elif status == "completed":
        progress = 100
    return {"run_id": run_id, "status": status, "progress": progress}