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.
1320 lines
47 KiB
Python
1320 lines
47 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
|
|
from apps.web.experiment_baked_sleeves import read_effective_baked_sleeves
|
|
from libs.backtest.domain import (
|
|
FORM4_CAPTURE_SLEEVE_PRESETS,
|
|
IDLE_ALPHA_SLEEVE_PRESETS,
|
|
OWNERSHIP_CAPTURE_SLEEVE_PRESETS,
|
|
PARKING_PRESETS,
|
|
RISK_OFF_ALPHA_SLEEVE_PRESETS,
|
|
)
|
|
|
|
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 and start.isdigit():
|
|
start_date = dt.date(int(start), 1, 1)
|
|
elif len(start) == 7 and start[4] == "-":
|
|
# YYYY-MM → first day of that month
|
|
y, m = int(start[:4]), int(start[5:])
|
|
start_date = dt.date(y, m, 1)
|
|
else:
|
|
start_date = dt.date.fromisoformat(start)
|
|
|
|
if end:
|
|
if len(end) == 4 and end.isdigit():
|
|
end_date = dt.date(int(end), 12, 31)
|
|
elif len(end) == 7 and end[4] == "-":
|
|
import calendar
|
|
y, m = int(end[:4]), int(end[5:])
|
|
end_date = dt.date(y, m, calendar.monthrange(y, m)[1])
|
|
else:
|
|
end_date = dt.date.fromisoformat(end)
|
|
else:
|
|
end_date = dt.date.today()
|
|
|
|
return start_date, end_date
|
|
|
|
|
|
def _log_dir() -> Path:
|
|
return get_runs_dir() / ".backtest_tasks"
|
|
|
|
|
|
def _snapshot_meta(snapshot_id: str) -> dict[str, Any]:
|
|
"""Read coverage_end_date and last_refresh_utc from the snapshot manifest."""
|
|
try:
|
|
from libs.backtest.snapshots import resolve_snapshot
|
|
resolution = resolve_snapshot(snapshot_id)
|
|
snap_path = Path("data/parquet") / resolution.canonical_snapshot_id / "manifest.json"
|
|
if snap_path.exists():
|
|
m = json.loads(snap_path.read_text())
|
|
return {
|
|
"snapshot_coverage_end_date": m.get("coverage_end_date"),
|
|
"snapshot_last_refresh_utc": m.get("last_refresh_utc"),
|
|
}
|
|
except Exception:
|
|
pass
|
|
return {"snapshot_coverage_end_date": None, "snapshot_last_refresh_utc": None}
|
|
|
|
|
|
def _natural_sort_key(value: str) -> list[Any]:
|
|
return [int(part) if part.isdigit() else part.lower() for part in re.split(r"(\d+)", value)]
|
|
|
|
|
|
def _auto_preset_response(group_label: str, available_presets: dict[str, Any]) -> dict[str, Any]:
|
|
presets = sorted(available_presets.keys(), key=_natural_sort_key)
|
|
return {"presets": presets, "groups": {group_label: presets}}
|
|
|
|
|
|
def _result_summary_from_run_id(run_id: str) -> dict[str, Any] | None:
|
|
"""Load result_summary from a CLI run's metrics_summary.json."""
|
|
try:
|
|
runs_dir = get_runs_dir()
|
|
metrics_path = runs_dir / run_id / "metrics" / "metrics_summary.json"
|
|
if not metrics_path.exists():
|
|
return None
|
|
m = json.loads(metrics_path.read_text())
|
|
win_rate = m.get("win_rate")
|
|
# MetricsBundle stores win_rate as fraction 0-1; convert to percentage
|
|
if win_rate is not None:
|
|
win_rate = float(win_rate) * 100
|
|
|
|
# Read snapshot coverage info from the run manifest
|
|
snap_meta: dict[str, Any] = {"snapshot_coverage_end_date": None, "snapshot_last_refresh_utc": None}
|
|
run_manifest_path = runs_dir / run_id / "manifest.json"
|
|
if run_manifest_path.exists():
|
|
snap_id = json.loads(run_manifest_path.read_text()).get("dataset_snapshot_id")
|
|
if snap_id:
|
|
snap_meta = _snapshot_meta(snap_id)
|
|
|
|
return {
|
|
"return_pct": m.get("total_return_pct"),
|
|
"simple_return_pct": m.get("simple_return_pct"),
|
|
"max_dd_pct": m.get("max_drawdown_pct"),
|
|
"sharpe": m.get("sharpe_ratio"),
|
|
"win_rate": win_rate,
|
|
"trade_count": m.get("trade_count"),
|
|
**snap_meta,
|
|
}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
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
|
|
summary = _result_summary_from_run_id(run_id)
|
|
if summary:
|
|
task["result_summary"] = summary
|
|
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
|
|
if not data.get("result_summary"):
|
|
result = _load_direct_result(task_id)
|
|
if result:
|
|
s = result.get("summary", {})
|
|
mb = result.get("metrics_bundle", {})
|
|
data["result_summary"] = {
|
|
"return_pct": s.get("return_pct"),
|
|
"simple_return_pct": mb.get("simple_return_pct"),
|
|
"max_dd_pct": s.get("max_dd_pct"),
|
|
"sharpe": s.get("sharpe"),
|
|
"win_rate": s.get("win_rate"),
|
|
"trade_count": s.get("trade_count"),
|
|
"snapshot_coverage_end_date": result.get("snapshot_coverage_end_date"),
|
|
"snapshot_last_refresh_utc": result.get("snapshot_last_refresh_utc"),
|
|
}
|
|
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)
|
|
parking: str | None = None # cash parking preset name
|
|
idle_alpha: str | None = None # idle alpha sleeve preset name
|
|
form4_sleeve: str | None = None # Form 4 residual-cash sleeve preset name
|
|
ownership_sleeve: str | None = None # 13D/13G ownership sleeve preset name
|
|
risk_off_sleeve: str | None = None # risk-off alpha sleeve preset name
|
|
non_core_allocator_v2_mode: str | None = None # shadow|live
|
|
snapshot_id: str | None = None # override dataset_snapshot_id (e.g. for OOT periods)
|
|
fixed_capital: bool = False # non-compounding: size positions using initial capital
|
|
|
|
|
|
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
|
|
parking: str | None = None
|
|
idle_alpha: str | None = None
|
|
form4_sleeve: str | None = None
|
|
ownership_sleeve: str | None = None
|
|
risk_off_sleeve: str | None = None
|
|
non_core_allocator_v2_mode: str | None = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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",
|
|
parking: str | None = None,
|
|
idle_alpha: str | None = None,
|
|
form4_sleeve: str | None = None,
|
|
ownership_sleeve: str | None = None,
|
|
risk_off_sleeve: str | None = None,
|
|
non_core_allocator_v2_mode: str | None = None,
|
|
snapshot_id: str | None = None,
|
|
fixed_capital: bool = False,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"task_id": str(uuid.uuid4()),
|
|
"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,
|
|
"fixed_capital": fixed_capital,
|
|
"mode": mode,
|
|
"has_direct_result": False,
|
|
"parking": parking,
|
|
"idle_alpha": idle_alpha,
|
|
"form4_sleeve": form4_sleeve,
|
|
"ownership_sleeve": ownership_sleeve,
|
|
"risk_off_sleeve": risk_off_sleeve,
|
|
"non_core_allocator_v2_mode": non_core_allocator_v2_mode,
|
|
"snapshot_id": snapshot_id,
|
|
}
|
|
|
|
|
|
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,
|
|
parking: str | None = None,
|
|
idle_alpha: str | None = None,
|
|
form4_sleeve: str | None = None,
|
|
ownership_sleeve: str | None = None,
|
|
risk_off_sleeve: str | None = None,
|
|
non_core_allocator_v2_mode: str | None = None,
|
|
snapshot_id: str | None = None,
|
|
) -> 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")
|
|
if parking:
|
|
cmd += ["--parking", parking]
|
|
if idle_alpha:
|
|
cmd += ["--idle-alpha", idle_alpha]
|
|
if form4_sleeve:
|
|
cmd += ["--form4-sleeve", form4_sleeve]
|
|
if ownership_sleeve:
|
|
cmd += ["--ownership-sleeve", ownership_sleeve]
|
|
if risk_off_sleeve:
|
|
cmd += ["--risk-off-sleeve", risk_off_sleeve]
|
|
if non_core_allocator_v2_mode:
|
|
cmd += ["--non-core-allocator-v2", "--non-core-allocator-v2-mode", non_core_allocator_v2_mode]
|
|
if snapshot_id:
|
|
cmd += ["--snapshot-id", snapshot_id]
|
|
return cmd
|
|
|
|
|
|
def _read_baked_sleeves(config_path: Path) -> dict[str, str | None]:
|
|
"""Read effective baked sleeves from the manifest/base_config chain."""
|
|
return read_effective_baked_sleeves(config_path, get_project_root())
|
|
|
|
|
|
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}")
|
|
|
|
# Merge request params with baked-in overrides (request takes priority)
|
|
baked = _read_baked_sleeves(config_path)
|
|
eff_parking = req.parking or baked.get("parking")
|
|
eff_idle_alpha = req.idle_alpha or baked.get("idle_alpha")
|
|
eff_form4 = req.form4_sleeve or baked.get("form4_sleeve")
|
|
eff_ownership = req.ownership_sleeve or baked.get("ownership_sleeve")
|
|
eff_risk_off = req.risk_off_sleeve or baked.get("risk_off_sleeve")
|
|
eff_non_core_allocator_v2_mode = req.non_core_allocator_v2_mode
|
|
|
|
if eff_parking and eff_parking not in PARKING_PRESETS:
|
|
raise HTTPException(status_code=400, detail=f"Unknown parking preset: {eff_parking}")
|
|
if eff_idle_alpha and eff_idle_alpha not in IDLE_ALPHA_SLEEVE_PRESETS:
|
|
raise HTTPException(status_code=400, detail=f"Unknown idle alpha preset: {eff_idle_alpha}")
|
|
if eff_form4 and eff_form4 not in FORM4_CAPTURE_SLEEVE_PRESETS:
|
|
raise HTTPException(status_code=400, detail=f"Unknown Form 4 sleeve preset: {eff_form4}")
|
|
if eff_ownership and eff_ownership not in OWNERSHIP_CAPTURE_SLEEVE_PRESETS:
|
|
raise HTTPException(status_code=400, detail=f"Unknown ownership sleeve preset: {eff_ownership}")
|
|
if eff_risk_off and eff_risk_off not in RISK_OFF_ALPHA_SLEEVE_PRESETS:
|
|
raise HTTPException(status_code=400, detail=f"Unknown risk-off sleeve preset: {eff_risk_off}")
|
|
if eff_non_core_allocator_v2_mode and eff_non_core_allocator_v2_mode not in {"shadow", "live"}:
|
|
raise HTTPException(status_code=400, detail=f"Unknown non-core allocator v2 mode: {eff_non_core_allocator_v2_mode}")
|
|
|
|
task = _make_task(
|
|
req.experiment_name,
|
|
req.capital,
|
|
req.start,
|
|
req.end,
|
|
req.year,
|
|
req.no_trades,
|
|
parking=eff_parking,
|
|
idle_alpha=eff_idle_alpha,
|
|
form4_sleeve=eff_form4,
|
|
ownership_sleeve=eff_ownership,
|
|
risk_off_sleeve=eff_risk_off,
|
|
non_core_allocator_v2_mode=eff_non_core_allocator_v2_mode,
|
|
snapshot_id=req.snapshot_id,
|
|
)
|
|
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,
|
|
"parking": eff_parking,
|
|
"idle_alpha": eff_idle_alpha,
|
|
"form4_sleeve": eff_form4,
|
|
"ownership_sleeve": eff_ownership,
|
|
"risk_off_sleeve": eff_risk_off,
|
|
"non_core_allocator_v2_mode": eff_non_core_allocator_v2_mode,
|
|
"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,
|
|
eff_parking,
|
|
eff_idle_alpha,
|
|
eff_form4,
|
|
eff_ownership,
|
|
eff_risk_off,
|
|
eff_non_core_allocator_v2_mode,
|
|
req.snapshot_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_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,
|
|
parking: str | None = None,
|
|
idle_alpha: str | None = None,
|
|
form4_sleeve: str | None = None,
|
|
ownership_sleeve: str | None = None,
|
|
risk_off_sleeve: str | None = None,
|
|
non_core_allocator_v2_mode: str | None = None,
|
|
) -> 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)
|
|
if parking and parking not in PARKING_PRESETS:
|
|
raise HTTPException(status_code=400, detail=f"Unknown parking preset: {parking}")
|
|
if idle_alpha and idle_alpha not in IDLE_ALPHA_SLEEVE_PRESETS:
|
|
raise HTTPException(status_code=400, detail=f"Unknown idle alpha preset: {idle_alpha}")
|
|
if form4_sleeve and form4_sleeve not in FORM4_CAPTURE_SLEEVE_PRESETS:
|
|
raise HTTPException(status_code=400, detail=f"Unknown Form 4 sleeve preset: {form4_sleeve}")
|
|
if ownership_sleeve and ownership_sleeve not in OWNERSHIP_CAPTURE_SLEEVE_PRESETS:
|
|
raise HTTPException(status_code=400, detail=f"Unknown ownership sleeve preset: {ownership_sleeve}")
|
|
if risk_off_sleeve and risk_off_sleeve not in RISK_OFF_ALPHA_SLEEVE_PRESETS:
|
|
raise HTTPException(status_code=400, detail=f"Unknown risk-off sleeve preset: {risk_off_sleeve}")
|
|
if non_core_allocator_v2_mode and non_core_allocator_v2_mode not in {"shadow", "live"}:
|
|
raise HTTPException(status_code=400, detail=f"Unknown non-core allocator v2 mode: {non_core_allocator_v2_mode}")
|
|
|
|
display_name = ", ".join(names)
|
|
task = _make_task(
|
|
display_name,
|
|
capital,
|
|
start,
|
|
end,
|
|
year,
|
|
no_trades,
|
|
parking=parking,
|
|
idle_alpha=idle_alpha,
|
|
form4_sleeve=form4_sleeve,
|
|
ownership_sleeve=ownership_sleeve,
|
|
risk_off_sleeve=risk_off_sleeve,
|
|
non_core_allocator_v2_mode=non_core_allocator_v2_mode,
|
|
)
|
|
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,
|
|
"parking": parking,
|
|
"idle_alpha": idle_alpha,
|
|
"form4_sleeve": form4_sleeve,
|
|
"ownership_sleeve": ownership_sleeve,
|
|
"risk_off_sleeve": risk_off_sleeve,
|
|
"non_core_allocator_v2_mode": non_core_allocator_v2_mode,
|
|
"created_at": created_at,
|
|
}, indent=2))
|
|
|
|
cmd = _build_cmd(
|
|
config_paths,
|
|
capital,
|
|
start,
|
|
end,
|
|
year,
|
|
no_trades,
|
|
runs_dir,
|
|
parking,
|
|
idle_alpha,
|
|
form4_sleeve,
|
|
ownership_sleeve,
|
|
risk_off_sleeve,
|
|
non_core_allocator_v2_mode,
|
|
)
|
|
|
|
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
|
|
result = _load_direct_result(task_id)
|
|
if result:
|
|
s = result.get("summary", {})
|
|
mb = result.get("metrics_bundle", {})
|
|
task["result_summary"] = {
|
|
"return_pct": s.get("return_pct"),
|
|
"simple_return_pct": mb.get("simple_return_pct"),
|
|
"max_dd_pct": s.get("max_dd_pct"),
|
|
"sharpe": s.get("sharpe"),
|
|
"win_rate": s.get("win_rate"),
|
|
"trade_count": s.get("trade_count"),
|
|
"snapshot_coverage_end_date": result.get("snapshot_coverage_end_date"),
|
|
"snapshot_last_refresh_utc": result.get("snapshot_last_refresh_utc"),
|
|
}
|
|
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}")
|
|
|
|
# Merge request params with baked-in overrides (request takes priority)
|
|
baked = _read_baked_sleeves(config_path)
|
|
eff_parking = req.parking or baked.get("parking")
|
|
eff_idle_alpha = req.idle_alpha or baked.get("idle_alpha")
|
|
eff_form4 = req.form4_sleeve or baked.get("form4_sleeve")
|
|
eff_ownership = req.ownership_sleeve or baked.get("ownership_sleeve")
|
|
eff_risk_off = req.risk_off_sleeve or baked.get("risk_off_sleeve")
|
|
eff_non_core_allocator_v2_mode = req.non_core_allocator_v2_mode
|
|
|
|
task = _make_task(
|
|
req.experiment_name, req.capital, req.start, req.end, req.year, req.no_trades,
|
|
mode="direct",
|
|
parking=eff_parking,
|
|
idle_alpha=eff_idle_alpha,
|
|
form4_sleeve=eff_form4,
|
|
ownership_sleeve=eff_ownership,
|
|
risk_off_sleeve=eff_risk_off,
|
|
non_core_allocator_v2_mode=eff_non_core_allocator_v2_mode,
|
|
snapshot_id=req.snapshot_id,
|
|
fixed_capital=req.fixed_capital,
|
|
)
|
|
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)),
|
|
]
|
|
if eff_parking:
|
|
cmd += ["--parking", eff_parking]
|
|
if eff_idle_alpha:
|
|
cmd += ["--idle-alpha", eff_idle_alpha]
|
|
if eff_form4:
|
|
cmd += ["--form4-sleeve", eff_form4]
|
|
if eff_ownership:
|
|
cmd += ["--ownership-sleeve", eff_ownership]
|
|
if eff_risk_off:
|
|
cmd += ["--risk-off-sleeve", eff_risk_off]
|
|
if eff_non_core_allocator_v2_mode:
|
|
cmd += ["--non-core-allocator-v2-mode", eff_non_core_allocator_v2_mode]
|
|
if req.snapshot_id:
|
|
cmd += ["--snapshot-id", req.snapshot_id]
|
|
if req.fixed_capital:
|
|
cmd += ["--fixed-capital"]
|
|
|
|
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,
|
|
parking=req.parking,
|
|
idle_alpha=req.idle_alpha,
|
|
form4_sleeve=req.form4_sleeve,
|
|
ownership_sleeve=req.ownership_sleeve,
|
|
risk_off_sleeve=req.risk_off_sleeve,
|
|
non_core_allocator_v2_mode=req.non_core_allocator_v2_mode,
|
|
)
|
|
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,
|
|
parking=req.parking,
|
|
idle_alpha=req.idle_alpha,
|
|
form4_sleeve=req.form4_sleeve,
|
|
ownership_sleeve=req.ownership_sleeve,
|
|
risk_off_sleeve=req.risk_off_sleeve,
|
|
non_core_allocator_v2_mode=req.non_core_allocator_v2_mode,
|
|
)
|
|
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,
|
|
)
|
|
# Backfill result_summary for tasks that completed before this field was added
|
|
for task in tasks:
|
|
if task.get("status") != "completed" or task.get("result_summary") is not None:
|
|
continue
|
|
summary: dict[str, Any] | None = None
|
|
if task.get("run_id") and task.get("mode", "cli") == "cli":
|
|
# CLI mode: load from metrics_summary.json
|
|
summary = _result_summary_from_run_id(task["run_id"])
|
|
elif task.get("mode") == "direct" and _result_file(task["task_id"]).exists():
|
|
# Direct mode: load from result JSON
|
|
result = _load_direct_result(task["task_id"])
|
|
if result:
|
|
s = result.get("summary", {})
|
|
summary = {
|
|
"return_pct": s.get("return_pct"),
|
|
"max_dd_pct": s.get("max_dd_pct"),
|
|
"sharpe": s.get("sharpe"),
|
|
"win_rate": s.get("win_rate"),
|
|
"trade_count": s.get("trade_count"),
|
|
"snapshot_coverage_end_date": result.get("snapshot_coverage_end_date"),
|
|
"snapshot_last_refresh_utc": result.get("snapshot_last_refresh_utc"),
|
|
}
|
|
if summary:
|
|
task["result_summary"] = summary
|
|
with _tasks_lock:
|
|
if task["task_id"] in _tasks:
|
|
_tasks[task["task_id"]]["result_summary"] = summary
|
|
_persist_task(task)
|
|
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}")
|
|
# Augment snapshot info on-the-fly if missing (handles old tasks stored without it)
|
|
rs = task.get("result_summary")
|
|
if rs is not None and rs.get("snapshot_coverage_end_date") is None:
|
|
exp_name = task.get("experiment_name")
|
|
if exp_name:
|
|
try:
|
|
config_path = get_configs_dir() / f"{exp_name}.json"
|
|
if config_path.exists():
|
|
exp_cfg = json.loads(config_path.read_text())
|
|
snap_id = exp_cfg.get("canonical_snapshot_id") or exp_cfg.get("dataset_snapshot_id")
|
|
if snap_id:
|
|
task = dict(task)
|
|
task["result_summary"] = dict(rs, **_snapshot_meta(snap_id))
|
|
except Exception:
|
|
pass
|
|
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 {}
|
|
|
|
|
|
@router.get("/snapshots")
|
|
def list_snapshots() -> dict[str, Any]:
|
|
"""Return available dataset snapshots with date ranges.
|
|
|
|
Includes both data/datasets/snapshots/ (primary) and data/parquet/ OOT snapshots.
|
|
"""
|
|
import pandas as pd # type: ignore
|
|
|
|
def _read_snap(snap_dir: Path) -> dict[str, Any] | None:
|
|
all_dfs = []
|
|
for split in ("train", "valid", "test"):
|
|
p = snap_dir / f"{split}.parquet"
|
|
if p.exists():
|
|
try:
|
|
all_dfs.append(pd.read_parquet(p, columns=["event_date"]))
|
|
except Exception:
|
|
pass
|
|
if not all_dfs:
|
|
return None
|
|
try:
|
|
df = pd.concat(all_dfs, ignore_index=True)
|
|
return {
|
|
"id": snap_dir.name,
|
|
"start": str(df["event_date"].min()),
|
|
"end": str(df["event_date"].max()),
|
|
"rows": len(df),
|
|
}
|
|
except Exception:
|
|
return {"id": snap_dir.name, "start": None, "end": None, "rows": None}
|
|
|
|
seen: set[str] = set()
|
|
results = []
|
|
|
|
# Primary: data/datasets/snapshots/
|
|
for snap_dir in sorted(Path("data/datasets/snapshots").iterdir()) if Path("data/datasets/snapshots").exists() else []:
|
|
if not snap_dir.is_dir():
|
|
continue
|
|
info = _read_snap(snap_dir)
|
|
if info:
|
|
seen.add(snap_dir.name)
|
|
results.append(info)
|
|
|
|
# Secondary: data/parquet/ OOT snapshots (not already seen)
|
|
parquet_dir = Path("data/parquet")
|
|
if parquet_dir.exists():
|
|
for snap_dir in sorted(parquet_dir.iterdir()):
|
|
if snap_dir.name in seen:
|
|
continue
|
|
# Only include OOT / historical / test / fix snapshots (skip active live ones)
|
|
if "oot" not in snap_dir.name and "2020" not in snap_dir.name and "2021" not in snap_dir.name and "test" not in snap_dir.name and "fix" not in snap_dir.name:
|
|
continue
|
|
if not snap_dir.is_dir():
|
|
continue
|
|
info = _read_snap(snap_dir)
|
|
if info:
|
|
seen.add(snap_dir.name)
|
|
results.append(info)
|
|
|
|
# Tertiary: registry-managed canonicals not yet built (show as available but unbuilt)
|
|
try:
|
|
from libs.backtest.snapshots import load_snapshot_registry
|
|
registry = load_snapshot_registry()
|
|
for canonical_id in sorted(registry.get("canonicals", {}).keys()):
|
|
if canonical_id in seen:
|
|
continue
|
|
results.append({"id": canonical_id, "start": None, "end": None, "rows": None})
|
|
seen.add(canonical_id)
|
|
except Exception:
|
|
pass
|
|
|
|
return {"snapshots": results}
|
|
|
|
|
|
@router.get("/parking-presets")
|
|
def get_parking_presets() -> dict[str, Any]:
|
|
"""Return all available parking presets for UI selection."""
|
|
return _auto_preset_response("Parking Presets", PARKING_PRESETS)
|
|
|
|
|
|
@router.get("/idle-alpha-presets")
|
|
def get_idle_alpha_presets() -> dict[str, Any]:
|
|
"""Return all available idle-alpha sleeve presets."""
|
|
return _auto_preset_response("Idle Alpha Sleeve Presets", IDLE_ALPHA_SLEEVE_PRESETS)
|
|
|
|
|
|
@router.get("/form4-capture-presets")
|
|
def get_form4_capture_presets() -> dict[str, Any]:
|
|
"""Return all available Form 4 residual-cash sleeve presets."""
|
|
return _auto_preset_response("Form 4 Sleeve Presets", FORM4_CAPTURE_SLEEVE_PRESETS)
|
|
|
|
|
|
@router.get("/ownership-capture-presets")
|
|
def get_ownership_capture_presets() -> dict[str, Any]:
|
|
"""Return all available 13D/13G residual-cash sleeve presets."""
|
|
return _auto_preset_response("Ownership Sleeve Presets", OWNERSHIP_CAPTURE_SLEEVE_PRESETS)
|
|
|
|
|
|
@router.get("/risk-off-alpha-presets")
|
|
def get_risk_off_alpha_presets() -> dict[str, Any]:
|
|
"""Return all available risk-off alpha sleeve presets."""
|
|
return _auto_preset_response("Risk-Off Alpha Sleeve Presets", RISK_OFF_ALPHA_SLEEVE_PRESETS)
|