qlib/backend/app/api/research.py

"""研究执行 API：/api/factor-tests 与 /api/backtests。

Phase 3 为同步执行（样本有限）；Phase 4 将改为 Job + SSE 异步（接口契约不变）。

**归档（2026-09 补齐）**：同步端点此前只把结果塞进进程内存 `_LAST_*`，重启即丢，
完全不落库 —— 与「研究可复现」（AGENT.md §21）矛盾。现改为：返回结果前调用
`archive_experiment` 落库（复用与异步 Job 完全相同的归档实现），并通过响应头
`X-Experiment-Id` 暴露归档 id。body 形状保持不变（前端与既有测试依赖它）。
"""

from __future__ import annotations

import logging
from datetime import date

from fastapi import APIRouter, HTTPException, Response
from pydantic import BaseModel, Field

from app.api.deps import DbSession, ExperimentRepoDep, ResearchServiceDep
from app.application.services.experiment_archive import archive_experiment
from app.domain.entities.research import (
    BacktestResult,
    FactorCorrelationReport,
    FactorSpec,
    FactorTestReport,
    ResearchSpec,
    SelectionSpec,
    UniverseSpec,
)
from app.quant.factors import FactorError

logger = logging.getLogger(__name__)

router = APIRouter(tags=["research"])

# 内存中的最近结果（仅作「最近一次」便捷读取；持久化归档见 archive_experiment）
_LAST_BACKTEST: dict[str, BacktestResult] = {}
_LAST_FACTOR_TEST: dict[str, FactorTestReport] = {}


def _archive_or_expose(
    response: Response,
    *,
    session,
    kind: str,
    spec_json: str,
    result,
    experiment_repo,
) -> None:
    """归档同步端点的计算结果，并把归档结果如实反映到响应头。

    取舍（AGENT.md §7「不静默」+ §24）：研究结果是用户真实等待数十秒得到的产出，
    归档是副产物。若数据库故障导致归档失败，直接抛 500 会把**已经算出来的可用结果**
    一并丢掉；故此处捕获异常、接口仍然 200 返回完整结果，同时：
    - `logger.warning` 落盘（服务端可观测）；
    - 响应头 `X-Archive-Error: <Type>: <msg>` 如实暴露失败原因（客户端可判读）。
    即「结果不丢 + 失败不静默」两者兼顾；成功时给 `X-Experiment-Id`。
    """
    try:
        experiment = archive_experiment(
            session=session,
            kind=kind,
            spec_json=spec_json,
            result=result,
            experiment_repo=experiment_repo,
        )
    except Exception as exc:  # noqa: BLE001 —— 归档失败不得吞掉计算结果
        logger.warning("同步端点归档失败（kind=%s）：%s: %s", kind, type(exc).__name__, exc)
        # HTTP 头只能承载 latin-1：中文错误信息降级为 ASCII 转义，避免编码异常掩盖原因
        reason = f"{type(exc).__name__}: {exc}"[:180]
        response.headers["X-Archive-Error"] = reason.encode("ascii", "backslashreplace").decode(
            "ascii"
        )
        return
    response.headers["X-Experiment-Id"] = experiment.id


@router.post("/factor-tests", response_model=FactorTestReport, summary="运行单因子测试（同步）")
def run_factor_test(
    spec: ResearchSpec,
    service: ResearchServiceDep,
    session: DbSession,
    experiment_repo: ExperimentRepoDep,
    response: Response,
) -> FactorTestReport:
    try:
        report = service.run_factor_test(spec)
    except (ValueError, FactorError) as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    _LAST_FACTOR_TEST["default"] = report
    _archive_or_expose(
        response,
        session=session,
        kind="factor_test",
        spec_json=spec.model_dump_json(),
        result=report,
        experiment_repo=experiment_repo,
    )
    return report


@router.post("/backtests", response_model=BacktestResult, summary="运行回测（同步）")
def run_backtest(
    spec: ResearchSpec,
    service: ResearchServiceDep,
    session: DbSession,
    experiment_repo: ExperimentRepoDep,
    response: Response,
) -> BacktestResult:
    try:
        result = service.run_backtest(spec)
    except (ValueError, FactorError) as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    _LAST_BACKTEST["default"] = result
    _archive_or_expose(
        response,
        session=session,
        kind="backtest",
        spec_json=spec.model_dump_json(),
        result=result,
        experiment_repo=experiment_repo,
    )
    return result


@router.get("/backtests/last", response_model=BacktestResult, summary="最近一次回测结果")
def last_backtest() -> BacktestResult:
    if "default" not in _LAST_BACKTEST:
        raise HTTPException(status_code=404, detail="尚无回测结果，请先 POST /api/backtests")
    return _LAST_BACKTEST["default"]


class FactorCorrelationRequest(BaseModel):
    """因子相关性分析请求（v3 §12）。"""

    universe: UniverseSpec = UniverseSpec()
    factors: list[FactorSpec] = Field(min_length=2)
    period: tuple[date, date]
    price_adjustment: str = Field(default="none", pattern="^(none|qfq)$")


@router.post(
    "/factor-correlations",
    response_model=FactorCorrelationReport,
    summary="多因子两两相关（横截面 Spearman）",
)
def run_factor_correlation(
    req: FactorCorrelationRequest,
    service: ResearchServiceDep,
) -> FactorCorrelationReport:
    if req.period[0] >= req.period[1]:
        raise HTTPException(status_code=400, detail="period 必须满足 start < end")
    spec = ResearchSpec(
        type="factor_test",
        universe=req.universe,
        price_adjustment=req.price_adjustment,
        factors=req.factors,
        selection=SelectionSpec(top_n=10),
        rebalance="monthly",
        period=req.period,
    )
    try:
        return service.run_factor_correlation(spec)
    except (ValueError, FactorError) as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc