qlib/backend/app/api/combos.py

"""回测组合 API：/api/combos（CRUD + 运行）。

POST /api/combos            保存组合（name 唯一）
GET  /api/combos            列表
GET  /api/combos/{id}       详情
PUT  /api/combos/{id}       原地更新
DELETE /api/combos/{id}     删除
POST /api/combos/{id}/run   提交已保存组合为异步 Job（kind=combo）
POST /api/combos/run        提交临时组合（不保存）为异步 Job

运行时：按 combo.strategy_ids 取齐选股策略 + 读公共配置 → ComboService.run → 归档。
费率/复权来自公共配置，快照进归档 config_snapshot（可复现，AGENT.md §21）。

依赖注入说明：建 Job 与策略校验都走 FastAPI 注入的 session/repo（而非直接 SessionLocal），
这样测试用 dependency_overrides 替换数据库时也能命中同一份库，行为一致可测。
"""

from __future__ import annotations

from datetime import datetime

from fastapi import APIRouter, BackgroundTasks, HTTPException

from app.api.deps import ComboRepoDep, DbSession, StrategyRepoDep
from app.application.services.job_executor import new_id, run_job_background
from app.domain.entities.combo import BacktestCombo
from app.domain.entities.research import JobRecord, JobStatus
from app.infrastructure.persistence.sqlalchemy.repositories.jobs_impl import (
    SqlAlchemyJobRepository,
)

router = APIRouter(prefix="/combos", tags=["combos"])


def _ensure_strategies_exist(combo: BacktestCombo, strategy_repo) -> None:
    """引用的选股策略必须都存在；缺任何一个即 400（提前失败，不等后台 Job 才暴露）。"""
    for sid in combo.strategy_ids:
        if strategy_repo.get(sid) is None:
            raise HTTPException(status_code=400, detail=f"组合引用的选股策略 {sid} 不存在")


@router.post("", response_model=BacktestCombo, summary="保存回测组合")
def create_combo(
    combo: BacktestCombo,
    repo: ComboRepoDep,
    session: DbSession,
) -> BacktestCombo:
    try:
        saved = repo.save(combo.model_copy(update={"id": new_id("CMB")}))
        session.commit()
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return repo.get(saved.id) or saved


@router.get("", response_model=list[BacktestCombo], summary="回测组合列表")
def list_combos(repo: ComboRepoDep) -> list[BacktestCombo]:
    return repo.list()


@router.get("/{combo_id}", response_model=BacktestCombo, summary="读取回测组合")
def get_combo(combo_id: str, repo: ComboRepoDep) -> BacktestCombo:
    row = repo.get(combo_id)
    if row is None:
        raise HTTPException(status_code=404, detail=f"回测组合 {combo_id} 不存在")
    return row


@router.put("/{combo_id}", response_model=BacktestCombo, summary="原地更新回测组合")
def update_combo(
    combo_id: str,
    combo: BacktestCombo,
    repo: ComboRepoDep,
    session: DbSession,
) -> BacktestCombo:
    existing = repo.get(combo_id)
    if existing is None:
        raise HTTPException(status_code=404, detail=f"回测组合 {combo_id} 不存在")
    payload = combo.model_copy(update={"id": combo_id, "created_at": existing.created_at})
    try:
        saved = repo.save(payload)
        session.commit()
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return repo.get(saved.id) or saved


@router.delete("/{combo_id}", summary="删除回测组合")
def delete_combo(combo_id: str, repo: ComboRepoDep, session: DbSession) -> dict:
    if not repo.delete(combo_id):
        raise HTTPException(status_code=404, detail=f"回测组合 {combo_id} 不存在")
    session.commit()
    return {"deleted": combo_id}


def _submit_combo_job(
    combo: BacktestCombo,
    *,
    strategy_repo,
    session,
    background: BackgroundTasks,
) -> dict:
    """校验策略 → 建 Job（kind=combo）→ 入后台执行。

    spec_json 存 BacktestCombo JSON；执行端（job_executor）识别 kind="combo"，
    再按 strategy_ids 取策略 + 读公共配置后调 ComboService。Job 表只存组合本身，
    策略/配置的「当时快照」由 ComboService 写进归档 config_snapshot（可复现）。
    校验与建 Job 都用注入的 session/repo，保证与测试覆写一致。
    """
    _ensure_strategies_exist(combo, strategy_repo)
    job = JobRecord(
        id=new_id("JOB"), kind="combo", status=JobStatus.QUEUED,
        spec_json=combo.model_dump_json(), created_at=datetime.now(),
    )
    SqlAlchemyJobRepository(session).create(job)
    session.commit()
    background.add_task(run_job_background, job.id)
    return {"job_id": job.id, "status": job.status}


@router.post("/{combo_id}/run", summary="运行已保存的回测组合（异步 Job）")
def run_saved_combo(
    combo_id: str,
    repo: ComboRepoDep,
    strategy_repo: StrategyRepoDep,
    session: DbSession,
    background: BackgroundTasks,
) -> dict:
    combo = repo.get(combo_id)
    if combo is None:
        raise HTTPException(status_code=404, detail=f"回测组合 {combo_id} 不存在")
    return _submit_combo_job(combo, strategy_repo=strategy_repo, session=session, background=background)


@router.post("/run", summary="运行临时回测组合（不保存，异步 Job）")
def run_adhoc_combo(
    combo: BacktestCombo,
    strategy_repo: StrategyRepoDep,
    session: DbSession,
    background: BackgroundTasks,
) -> dict:
    return _submit_combo_job(combo, strategy_repo=strategy_repo, session=session, background=background)