You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
835 lines
27 KiB
Python
835 lines
27 KiB
Python
"""Backtest execution API endpoints (paper backtest)."""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import re
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
_ANSI_RE = re.compile(r'\x1b\[[0-9;]*[mGKHABCDFrsu]|\x1b[()][AB012]')
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
from apps.web.dependencies import get_configs_dir, get_project_root, get_runs_dir
|
|
|
|
router = APIRouter(prefix="/backtest", tags=["backtest"])
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task registry — in-memory + disk persistence
|
|
# ---------------------------------------------------------------------------
|
|
_tasks: dict[str, dict[str, Any]] = {}
|
|
_tasks_lock = threading.Lock()
|
|
_tasks_initialized = False
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Direct result helpers (in-process backtest)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _result_file(task_id: str) -> Path:
|
|
return _log_dir() / f"{task_id}.result.json"
|
|
|
|
|
|
def _json_default(obj: Any) -> Any:
|
|
"""JSON serializer for types not handled by default encoder."""
|
|
if isinstance(obj, (dt.date, dt.datetime)):
|
|
return obj.isoformat()
|
|
raise TypeError(f"Object of type {type(obj)} is not JSON serializable")
|
|
|
|
|
|
def _persist_direct_result(task_id: str, result: dict[str, Any]) -> None:
|
|
"""Write direct backtest result to disk."""
|
|
try:
|
|
_log_dir().mkdir(parents=True, exist_ok=True)
|
|
_result_file(task_id).write_text(
|
|
json.dumps(result, default=_json_default, indent=2)
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _load_direct_result(task_id: str) -> dict[str, Any] | None:
|
|
"""Load direct backtest result from disk."""
|
|
try:
|
|
f = _result_file(task_id)
|
|
if f.exists():
|
|
return json.loads(f.read_text())
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def _parse_dates(
|
|
start: str | None,
|
|
end: str | None,
|
|
year: str | None,
|
|
) -> tuple[dt.date, dt.date]:
|
|
"""Parse date parameters into (start_date, end_date)."""
|
|
if year:
|
|
y = int(year)
|
|
return dt.date(y, 1, 1), dt.date(y, 12, 31)
|
|
if not start:
|
|
raise ValueError("Either year or start date is required")
|
|
if len(start) == 4:
|
|
start_date = dt.date(int(start), 1, 1)
|
|
end_date = dt.date.fromisoformat(end) if end else dt.date.today()
|
|
else:
|
|
start_date = dt.date.fromisoformat(start)
|
|
end_date = dt.date.fromisoformat(end) if end else dt.date.today()
|
|
return start_date, end_date
|
|
|
|
|
|
def _log_dir() -> Path:
|
|
return get_runs_dir() / ".backtest_tasks"
|
|
|
|
|
|
def _task_file(task_id: str) -> Path:
|
|
return _log_dir() / f"{task_id}.task.json"
|
|
|
|
|
|
def _persist_task(task: dict[str, Any]) -> None:
|
|
"""Write task state to disk (includes pid so we can re-attach after restart)."""
|
|
try:
|
|
_log_dir().mkdir(parents=True, exist_ok=True)
|
|
_task_file(task["task_id"]).write_text(json.dumps(task, indent=2))
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _delete_task_file(task_id: str) -> None:
|
|
try:
|
|
_task_file(task_id).unlink(missing_ok=True)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _is_pid_alive(pid: int) -> bool:
|
|
"""Return True if the process with the given PID is still running."""
|
|
try:
|
|
os.kill(pid, 0)
|
|
return True
|
|
except ProcessLookupError:
|
|
return False
|
|
except PermissionError:
|
|
return True # Exists but owned by a different user (unlikely but safe)
|
|
|
|
|
|
def _reattach_watcher(task_id: str, pid: int, output_root: Path) -> None:
|
|
"""Watch an orphaned subprocess by PID after server restart."""
|
|
import time
|
|
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
return
|
|
started_at_str = task.get("started_at") or datetime.now(timezone.utc).isoformat()
|
|
experiment_name = task["experiment_name"]
|
|
|
|
try:
|
|
started_at = datetime.fromisoformat(started_at_str.replace("Z", "+00:00"))
|
|
except Exception:
|
|
started_at = datetime.now(timezone.utc)
|
|
|
|
# Poll until the orphaned process exits
|
|
while _is_pid_alive(pid):
|
|
time.sleep(2)
|
|
|
|
# Process finished — check for results
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
return
|
|
mode = task.get("mode", "cli")
|
|
|
|
if mode == "direct":
|
|
# Direct mode: result is in a JSON file, not a run directory
|
|
direct_log = output_root / ".backtest_tasks" / f"{task_id}.direct.log"
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
return
|
|
if task["status"] == "cancelled":
|
|
_persist_task(task)
|
|
return
|
|
task["pid"] = None
|
|
if not task.get("finished_at"):
|
|
task["finished_at"] = datetime.now(timezone.utc).isoformat()
|
|
if _result_file(task_id).exists():
|
|
task["status"] = "completed"
|
|
task["has_direct_result"] = True
|
|
else:
|
|
task["status"] = "failed"
|
|
task["error"] = _log_tail_error(direct_log)
|
|
_persist_task(task)
|
|
return
|
|
|
|
run_id = _detect_any_run_id(output_root, experiment_name, started_at)
|
|
log_file = output_root / ".backtest_tasks" / f"{task_id}.log"
|
|
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
return
|
|
if task["status"] == "cancelled":
|
|
_persist_task(task)
|
|
return
|
|
task["pid"] = None
|
|
if not task.get("finished_at"):
|
|
task["finished_at"] = datetime.now(timezone.utc).isoformat()
|
|
if run_id:
|
|
task["status"] = "completed"
|
|
task["run_id"] = run_id
|
|
elif _log_success(log_file):
|
|
# Multi-config or run saved to unexpected path — log says success
|
|
task["status"] = "completed"
|
|
task["run_id"] = None
|
|
else:
|
|
task["status"] = "failed"
|
|
task["error"] = _log_tail_error(log_file)
|
|
_persist_task(task)
|
|
|
|
|
|
def _load_tasks_from_disk() -> list[tuple[str, int]]:
|
|
"""Load persisted task files. Returns list of (task_id, pid) needing re-attachment."""
|
|
ld = _log_dir()
|
|
reattach: list[tuple[str, int]] = []
|
|
if not ld.exists():
|
|
return reattach
|
|
for f in sorted(ld.glob("*.task.json")):
|
|
try:
|
|
data = json.loads(f.read_text())
|
|
task_id = data.get("task_id")
|
|
if not task_id or task_id in _tasks:
|
|
continue
|
|
|
|
if data.get("status") in ("running", "queued"):
|
|
pid = data.get("pid")
|
|
if pid and _is_pid_alive(pid):
|
|
# Process still running — schedule re-attachment
|
|
_tasks[task_id] = data
|
|
reattach.append((task_id, pid))
|
|
continue
|
|
# Process dead — check if it completed successfully
|
|
started_at_str = data.get("started_at")
|
|
run_id = None
|
|
log_file = _log_dir() / f"{data.get('task_id', '')}.log"
|
|
if started_at_str:
|
|
try:
|
|
started_at = datetime.fromisoformat(started_at_str.replace("Z", "+00:00"))
|
|
run_id = _detect_any_run_id(get_runs_dir(), data.get("experiment_name", ""), started_at)
|
|
except Exception:
|
|
pass
|
|
data["pid"] = None
|
|
if not data.get("finished_at"):
|
|
data["finished_at"] = datetime.now(timezone.utc).isoformat()
|
|
# Direct mode: check result file
|
|
if data.get("mode") == "direct":
|
|
if _result_file(task_id).exists():
|
|
data["status"] = "completed"
|
|
data["has_direct_result"] = True
|
|
else:
|
|
data["status"] = "failed"
|
|
log_file = _log_dir() / f"{task_id}.direct.log"
|
|
data["error"] = _log_tail_error(log_file)
|
|
elif run_id:
|
|
data["status"] = "completed"
|
|
data["run_id"] = run_id
|
|
elif _log_success(log_file):
|
|
data["status"] = "completed"
|
|
data["run_id"] = None
|
|
else:
|
|
data["status"] = "failed"
|
|
data["error"] = "Server restarted while task was running"
|
|
f.write_text(json.dumps(data, indent=2))
|
|
|
|
_tasks[task_id] = data
|
|
except Exception:
|
|
continue
|
|
return reattach
|
|
|
|
|
|
def _ensure_tasks_loaded() -> None:
|
|
global _tasks_initialized
|
|
if _tasks_initialized:
|
|
return
|
|
with _tasks_lock:
|
|
if _tasks_initialized:
|
|
return
|
|
_tasks_initialized = True
|
|
reattach = _load_tasks_from_disk()
|
|
|
|
# Start re-attachment threads outside the lock
|
|
for task_id, pid in reattach:
|
|
threading.Thread(
|
|
target=_reattach_watcher,
|
|
args=(task_id, pid, get_runs_dir()),
|
|
daemon=True,
|
|
).start()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Request / Response schemas
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class BacktestRequest(BaseModel):
|
|
experiment_name: str
|
|
capital: float = 10_000.0
|
|
start: str | None = None # YYYY-MM-DD or YYYY
|
|
end: str | None = None # YYYY-MM-DD
|
|
year: str | None = None # shorthand for full year (YYYY)
|
|
no_trades: bool = False
|
|
mode: str = "cli" # "cli" (subprocess) or "direct" (in-process)
|
|
|
|
|
|
class BatchBacktestRequest(BaseModel):
|
|
experiment_names: list[str]
|
|
capital: float = 10_000.0
|
|
start: str | None = None
|
|
end: str | None = None
|
|
year: str | None = None
|
|
no_trades: bool = False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Task helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_task(
|
|
experiment_name: str,
|
|
capital: float,
|
|
start: str | None,
|
|
end: str | None,
|
|
year: str | None,
|
|
no_trades: bool = False,
|
|
mode: str = "cli",
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"task_id": str(uuid.uuid4())[:8],
|
|
"experiment_name": experiment_name,
|
|
"capital": capital,
|
|
"start_date": year if year else start,
|
|
"end_date": None if year else end,
|
|
"year": year,
|
|
"no_trades": no_trades,
|
|
"status": "queued",
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"started_at": None,
|
|
"finished_at": None,
|
|
"run_id": None,
|
|
"error": None,
|
|
"pid": None,
|
|
"mode": mode,
|
|
"has_direct_result": False,
|
|
}
|
|
|
|
|
|
def _detect_run_id(output_root: Path, experiment_name: str, started_before: datetime) -> str | None:
|
|
"""Scan runs dir for a new run directory created after started_before."""
|
|
if not output_root.exists():
|
|
return None
|
|
candidates = []
|
|
for d in output_root.iterdir():
|
|
if not d.is_dir():
|
|
continue
|
|
meta_file = d / "metadata.json"
|
|
if not meta_file.exists():
|
|
continue
|
|
try:
|
|
mtime = datetime.fromtimestamp(d.stat().st_mtime, tz=timezone.utc)
|
|
if mtime < started_before:
|
|
continue
|
|
meta = json.loads(meta_file.read_text())
|
|
if meta.get("experiment_name") == experiment_name:
|
|
candidates.append((mtime, d.name))
|
|
except Exception:
|
|
continue
|
|
if not candidates:
|
|
return None
|
|
candidates.sort(reverse=True)
|
|
return candidates[0][1]
|
|
|
|
|
|
def _detect_any_run_id(output_root: Path, experiment_name: str, started_before: datetime) -> str | None:
|
|
"""Detect run_id for single OR multi-config experiments (tries each name in joined list)."""
|
|
# First try the full name (handles single config)
|
|
run_id = _detect_run_id(output_root, experiment_name, started_before)
|
|
if run_id:
|
|
return run_id
|
|
# For multi-config joined names ("exp1, exp2, ..."), try each individual name
|
|
parts = [n.strip() for n in experiment_name.split(",") if n.strip()]
|
|
if len(parts) > 1:
|
|
for name in parts:
|
|
run_id = _detect_run_id(output_root, name, started_before)
|
|
if run_id:
|
|
return run_id
|
|
return None
|
|
|
|
|
|
def _log_success(log_file: Path) -> bool:
|
|
"""Return True if the log file ends with a success marker."""
|
|
if not log_file.exists():
|
|
return False
|
|
try:
|
|
tail = log_file.read_text(errors="replace")[-1500:]
|
|
return "results saved to" in tail.lower() or "results saved" in tail.lower()
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _log_tail_error(log_file: Path, returncode: int | None = None) -> str:
|
|
"""Return last ~10 lines of log with ANSI stripped, for use as error message."""
|
|
if log_file.exists():
|
|
try:
|
|
lines = log_file.read_text(errors="replace").strip().splitlines()
|
|
# Strip ANSI codes from each line
|
|
clean = [_ANSI_RE.sub("", l) for l in lines[-10:]]
|
|
return "\n".join(clean) if clean else f"Process exited with code {returncode}"
|
|
except Exception:
|
|
pass
|
|
return f"Process exited with code {returncode}" if returncode is not None else "Process exited unexpectedly"
|
|
|
|
|
|
def _watch_process(task_id: str, proc: subprocess.Popen[bytes], output_root: Path) -> None:
|
|
"""Background thread: wait for process to finish and update task state."""
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
return
|
|
task["status"] = "running"
|
|
task["started_at"] = datetime.now(timezone.utc).isoformat()
|
|
task["pid"] = proc.pid
|
|
started_at = datetime.now(timezone.utc)
|
|
experiment_name = task["experiment_name"]
|
|
_persist_task(task)
|
|
|
|
returncode = proc.wait()
|
|
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
return
|
|
task["finished_at"] = datetime.now(timezone.utc).isoformat()
|
|
task["pid"] = None
|
|
|
|
if task["status"] == "cancelled":
|
|
_persist_task(task)
|
|
return
|
|
|
|
if returncode == 0:
|
|
task["status"] = "completed"
|
|
run_id = _detect_run_id(output_root, experiment_name, started_at)
|
|
task["run_id"] = run_id
|
|
else:
|
|
task["status"] = "failed"
|
|
log_file = output_root / ".backtest_tasks" / f"{task_id}.log"
|
|
task["error"] = _log_tail_error(log_file, returncode)
|
|
_persist_task(task)
|
|
|
|
|
|
def _build_cmd(config_paths: list[Path], capital: float, start: str | None, end: str | None, year: str | None, no_trades: bool, runs_dir: Path) -> list[str]:
|
|
"""Build the paper backtest CLI command."""
|
|
cmd = [
|
|
sys.executable, "-m", "apps.paper_trader.cli", "backtest",
|
|
"--capital", str(capital),
|
|
"--output", str(runs_dir),
|
|
]
|
|
for cp in config_paths:
|
|
cmd += ["--config", str(cp)]
|
|
if year:
|
|
cmd += ["--year", year]
|
|
elif start:
|
|
cmd += ["--start", start]
|
|
if end:
|
|
cmd += ["--end", end]
|
|
if no_trades:
|
|
cmd.append("--no-trades")
|
|
return cmd
|
|
|
|
|
|
def _launch_backtest(req: BacktestRequest) -> dict[str, Any]:
|
|
"""Build CLI args, launch subprocess, register task."""
|
|
project_root = get_project_root()
|
|
configs_dir = get_configs_dir()
|
|
runs_dir = get_runs_dir()
|
|
|
|
config_path = configs_dir / f"{req.experiment_name}.json"
|
|
if not config_path.exists():
|
|
raise HTTPException(status_code=404, detail=f"Experiment config not found: {req.experiment_name}")
|
|
|
|
task = _make_task(req.experiment_name, req.capital, req.start, req.end, req.year, req.no_trades)
|
|
task_id = task["task_id"]
|
|
|
|
# Log directory
|
|
log_dir = runs_dir / ".backtest_tasks"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
log_file = log_dir / f"{task_id}.log"
|
|
|
|
# Save params for last-params lookup
|
|
params_file = log_dir / f"{task_id}.params.json"
|
|
params_file.write_text(json.dumps({
|
|
"experiment_name": req.experiment_name,
|
|
"capital": req.capital,
|
|
"start": req.start,
|
|
"end": req.end,
|
|
"year": req.year,
|
|
"no_trades": req.no_trades,
|
|
"created_at": task["created_at"],
|
|
}, indent=2))
|
|
|
|
cmd = _build_cmd([config_path], req.capital, req.start, req.end, req.year, req.no_trades, runs_dir)
|
|
|
|
with open(log_file, "wb") as log_fp:
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
stdout=log_fp,
|
|
stderr=subprocess.STDOUT,
|
|
cwd=str(project_root),
|
|
)
|
|
|
|
with _tasks_lock:
|
|
_tasks[task_id] = task
|
|
_persist_task(task)
|
|
|
|
thread = threading.Thread(
|
|
target=_watch_process,
|
|
args=(task_id, proc, runs_dir),
|
|
daemon=True,
|
|
)
|
|
thread.start()
|
|
|
|
return task
|
|
|
|
|
|
def _launch_backtest_multi(names: list[str], capital: float, start: str | None, end: str | None, year: str | None, no_trades: bool) -> dict[str, Any]:
|
|
"""Launch ONE subprocess with multiple --config flags for batch experiments."""
|
|
project_root = get_project_root()
|
|
configs_dir = get_configs_dir()
|
|
runs_dir = get_runs_dir()
|
|
|
|
config_paths: list[Path] = []
|
|
for name in names:
|
|
cp = configs_dir / f"{name}.json"
|
|
if not cp.exists():
|
|
raise HTTPException(status_code=404, detail=f"Experiment config not found: {name}")
|
|
config_paths.append(cp)
|
|
|
|
display_name = ", ".join(names)
|
|
task = _make_task(display_name, capital, start, end, year, no_trades)
|
|
task_id = task["task_id"]
|
|
|
|
log_dir = runs_dir / ".backtest_tasks"
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
log_file = log_dir / f"{task_id}.log"
|
|
|
|
# Save individual params files so last-params lookup works per experiment
|
|
created_at = task["created_at"]
|
|
for name in names:
|
|
pf = log_dir / f"{task_id}_{name[:30]}.params.json"
|
|
pf.write_text(json.dumps({
|
|
"experiment_name": name,
|
|
"capital": capital,
|
|
"start": start,
|
|
"end": end,
|
|
"year": year,
|
|
"no_trades": no_trades,
|
|
"created_at": created_at,
|
|
}, indent=2))
|
|
|
|
cmd = _build_cmd(config_paths, capital, start, end, year, no_trades, runs_dir)
|
|
|
|
with open(log_file, "wb") as log_fp:
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
stdout=log_fp,
|
|
stderr=subprocess.STDOUT,
|
|
cwd=str(project_root),
|
|
)
|
|
|
|
with _tasks_lock:
|
|
_tasks[task_id] = task
|
|
_persist_task(task)
|
|
|
|
thread = threading.Thread(
|
|
target=_watch_process,
|
|
args=(task_id, proc, runs_dir),
|
|
daemon=True,
|
|
)
|
|
thread.start()
|
|
|
|
return task
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Direct backtest — subprocess (survives server --reload restarts)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _watch_direct_process(task_id: str, proc: subprocess.Popen[bytes]) -> None:
|
|
"""Background thread: wait for the direct-runner subprocess and update task state."""
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
return
|
|
task["status"] = "running"
|
|
task["started_at"] = datetime.now(timezone.utc).isoformat()
|
|
task["pid"] = proc.pid
|
|
_persist_task(task)
|
|
|
|
returncode = proc.wait()
|
|
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
return
|
|
if task["status"] == "cancelled":
|
|
_persist_task(task)
|
|
return
|
|
task["finished_at"] = datetime.now(timezone.utc).isoformat()
|
|
task["pid"] = None
|
|
if returncode == 0 and _result_file(task_id).exists():
|
|
task["status"] = "completed"
|
|
task["has_direct_result"] = True
|
|
else:
|
|
task["status"] = "failed"
|
|
log_file = _log_dir() / f"{task_id}.direct.log"
|
|
task["error"] = _log_tail_error(log_file, returncode)
|
|
_persist_task(task)
|
|
|
|
|
|
def _launch_direct_backtest(req: BacktestRequest) -> dict[str, Any]:
|
|
"""Launch the direct backtest as a subprocess (survives uvicorn --reload)."""
|
|
configs_dir = get_configs_dir()
|
|
project_root = get_project_root()
|
|
config_path = configs_dir / f"{req.experiment_name}.json"
|
|
if not config_path.exists():
|
|
raise HTTPException(status_code=404, detail=f"Experiment config not found: {req.experiment_name}")
|
|
|
|
try:
|
|
start_date, end_date = _parse_dates(req.start, req.end, req.year)
|
|
except (ValueError, TypeError) as exc:
|
|
raise HTTPException(status_code=400, detail=f"Invalid date params: {exc}")
|
|
|
|
task = _make_task(
|
|
req.experiment_name, req.capital, req.start, req.end, req.year, req.no_trades,
|
|
mode="direct",
|
|
)
|
|
task_id = task["task_id"]
|
|
|
|
log_dir = _log_dir()
|
|
log_dir.mkdir(parents=True, exist_ok=True)
|
|
log_file = log_dir / f"{task_id}.direct.log"
|
|
|
|
cmd = [
|
|
sys.executable, "-m", "apps.web.direct_runner",
|
|
task_id,
|
|
str(config_path),
|
|
str(req.capital),
|
|
start_date.isoformat(),
|
|
end_date.isoformat(),
|
|
str(_result_file(task_id)),
|
|
]
|
|
|
|
with open(log_file, "wb") as log_fp:
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
stdout=log_fp,
|
|
stderr=subprocess.STDOUT,
|
|
cwd=str(project_root),
|
|
)
|
|
|
|
with _tasks_lock:
|
|
_tasks[task_id] = task
|
|
_persist_task(task)
|
|
|
|
thread = threading.Thread(
|
|
target=_watch_direct_process,
|
|
args=(task_id, proc),
|
|
daemon=True,
|
|
)
|
|
thread.start()
|
|
|
|
return task
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Endpoints
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@router.post("/submit")
|
|
def submit_backtest(req: BacktestRequest) -> dict[str, Any]:
|
|
"""Submit a single paper backtest task (CLI subprocess or in-process direct)."""
|
|
_ensure_tasks_loaded()
|
|
if req.mode == "direct":
|
|
return _launch_direct_backtest(req)
|
|
return _launch_backtest(req)
|
|
|
|
|
|
@router.post("/submit-batch")
|
|
def submit_batch(req: BatchBacktestRequest) -> dict[str, Any]:
|
|
"""Submit a single backtest task running all experiments in one subprocess."""
|
|
_ensure_tasks_loaded()
|
|
if not req.experiment_names:
|
|
raise HTTPException(status_code=400, detail="experiment_names is empty")
|
|
|
|
if len(req.experiment_names) == 1:
|
|
single = BacktestRequest(
|
|
experiment_name=req.experiment_names[0],
|
|
capital=req.capital,
|
|
start=req.start,
|
|
end=req.end,
|
|
year=req.year,
|
|
no_trades=req.no_trades,
|
|
)
|
|
task = _launch_backtest(single)
|
|
return {"tasks": [task]}
|
|
|
|
task = _launch_backtest_multi(
|
|
names=req.experiment_names,
|
|
capital=req.capital,
|
|
start=req.start,
|
|
end=req.end,
|
|
year=req.year,
|
|
no_trades=req.no_trades,
|
|
)
|
|
return {"tasks": [task]}
|
|
|
|
|
|
@router.get("/tasks")
|
|
def list_tasks() -> dict[str, Any]:
|
|
"""List all backtest tasks (most recent first)."""
|
|
_ensure_tasks_loaded()
|
|
with _tasks_lock:
|
|
tasks = sorted(
|
|
list(_tasks.values()),
|
|
key=lambda t: t["created_at"],
|
|
reverse=True,
|
|
)
|
|
return {"tasks": tasks}
|
|
|
|
|
|
@router.get("/tasks/{task_id}")
|
|
def get_task(task_id: str) -> dict[str, Any]:
|
|
"""Get status of a specific task."""
|
|
_ensure_tasks_loaded()
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task not found: {task_id}")
|
|
return task
|
|
|
|
|
|
@router.delete("/tasks/{task_id}")
|
|
def cancel_task(task_id: str) -> dict[str, Any]:
|
|
"""Cancel a running task or delete a finished task from memory."""
|
|
_ensure_tasks_loaded()
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task not found: {task_id}")
|
|
status = task["status"]
|
|
pid = task.get("pid")
|
|
|
|
if status in ("queued", "running"):
|
|
task["status"] = "cancelled"
|
|
task["finished_at"] = datetime.now(timezone.utc).isoformat()
|
|
_persist_task(task)
|
|
else:
|
|
del _tasks[task_id]
|
|
_delete_task_file(task_id)
|
|
return {"cancelled": True, "deleted": True}
|
|
|
|
if pid:
|
|
try:
|
|
os.kill(pid, signal.SIGTERM)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
return {"cancelled": True} # direct-mode tasks have no PID; thread runs to completion but result is discarded
|
|
|
|
|
|
@router.get("/tasks/{task_id}/log")
|
|
def get_task_log(task_id: str) -> dict[str, Any]:
|
|
"""Get stdout/stderr log for a task."""
|
|
_ensure_tasks_loaded()
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task not found: {task_id}")
|
|
|
|
runs_dir = get_runs_dir()
|
|
mode = task.get("mode", "cli")
|
|
suffix = ".direct.log" if mode == "direct" else ".log"
|
|
log_file = runs_dir / ".backtest_tasks" / f"{task_id}{suffix}"
|
|
if not log_file.exists():
|
|
return {"log": ""}
|
|
|
|
try:
|
|
content = log_file.read_text(errors="replace")
|
|
lines = content.splitlines()
|
|
return {"log": "\n".join(lines[-200:]), "total_lines": len(lines)}
|
|
except Exception as e:
|
|
return {"log": f"Error reading log: {e}"}
|
|
|
|
|
|
@router.get("/tasks/{task_id}/direct-result")
|
|
def get_direct_result(task_id: str) -> dict[str, Any]:
|
|
"""Get the in-process backtest result for a direct-mode task."""
|
|
_ensure_tasks_loaded()
|
|
with _tasks_lock:
|
|
task = _tasks.get(task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail=f"Task not found: {task_id}")
|
|
if task.get("mode") != "direct":
|
|
raise HTTPException(status_code=400, detail="Task is not a direct-mode task")
|
|
if task["status"] != "completed":
|
|
raise HTTPException(status_code=409, detail=f"Task not completed: {task['status']}")
|
|
|
|
result = _load_direct_result(task_id)
|
|
if result is None:
|
|
raise HTTPException(status_code=404, detail="Direct result not found on disk")
|
|
return result
|
|
|
|
|
|
|
|
@router.get("/last-params/{experiment_name}")
|
|
def get_last_params(experiment_name: str) -> dict[str, Any]:
|
|
"""Return parameters from the most recent run of this experiment."""
|
|
runs_dir = get_runs_dir()
|
|
log_dir = runs_dir / ".backtest_tasks"
|
|
if not log_dir.exists():
|
|
return {}
|
|
|
|
# Find most recent .params.json for this experiment
|
|
best: tuple[float, Path] | None = None
|
|
for f in log_dir.glob("*.params.json"):
|
|
try:
|
|
data = json.loads(f.read_text())
|
|
if data.get("experiment_name") != experiment_name:
|
|
continue
|
|
mtime = f.stat().st_mtime
|
|
if best is None or mtime > best[0]:
|
|
best = (mtime, f)
|
|
except Exception:
|
|
continue
|
|
|
|
if best is None:
|
|
return {}
|
|
|
|
try:
|
|
data = json.loads(best[1].read_text())
|
|
return {
|
|
"capital": data.get("capital"),
|
|
"start": data.get("start"),
|
|
"end": data.get("end"),
|
|
"year": data.get("year"),
|
|
}
|
|
except Exception:
|
|
return {}
|