qlib/backend/app/application/services/experiment_archive.py

"""Experiment 归档服务（AGENT.md §21「研究可复现」）：同步与异步两条路径共用。

谁在用：
- 异步 Job：`job_executor._execute_inner` 执行成功后归档；
- 同步研究端点：`api/research.py` 的 `POST /backtests`、`POST /factor-tests`
  （Phase 3 遗留的同步路径，此前只塞进程内存 `_LAST_*`，重启即丢，完全不落库）。

为什么单独成模块：`job_executor.py` 顶部依赖较重（研究服务、子进程调度），
把归档抽到此处既避免 `api/research.py` → `job_executor` 的循环导入风险，也让
「一次研究结果如何落库」只有一份实现（AGENT.md §40 简单可替换优先）。

归档内容：spec_json（复现依据）、完整结果 JSON、summary_text（人读摘要）、
code_version（git 短 rev）、**data_version（真实数据快照指纹，见 data_version 模块）**、
job_id（可追溯执行记录）、created_at。同一份结果**只在 experiment 存一份**：

- 异步 Job 路径下 `job.result_json` 不再重复写入（见 `job_executor`），
  `GET /api/jobs/{id}` 经 `job.experiment_id` 回读 experiment；
- 同步端点路径下不存在 Job 记录，`job_id=None`。

大小护栏（诚实 + 不炸库）：`experiment.result_json` 是 MEDIUMTEXT，**上限
16,777,215 字节**（MySQL 的 MEDIUMTEXT 按字节计，utf8mb4 下中文占 3 字节），
故预算按 UTF-8 字节执行，默认 12,000,000 字节（约上限的 71.5%，给行内其它
列、SQL 协议与未预料字段留余量）。超预算时按 |期末收益| 降序裁剪
`symbol_curves`，并把证据写进结果的 `archive_meta`（机器可读）与
`unimplemented`（人可读）——绝不静默丢弃。
"""

from __future__ import annotations

import json
import logging
import subprocess
import uuid
from datetime import datetime
from pathlib import Path

from app.domain.entities.research import ExperimentRecord

logger = logging.getLogger(__name__)

# backend/app/application/services/experiment_archive.py → parents[4] = 项目根
PROJECT_ROOT = Path(__file__).resolve().parents[4]

# 归档结果 JSON 的字节预算（可被 config research.archive_max_chars 覆盖）
DEFAULT_ARCHIVE_MAX_CHARS = 12_000_000

# MEDIUMTEXT 硬上限（字节），仅用于计算/说明默认预算的余量
MEDIUMTEXT_MAX_BYTES = 16_777_215


def new_id(prefix: str) -> str:
    """生成 `PREFIX-XXXXXXXX` 形式的业务主键（Job/Experiment/Selection 等共用）。"""
    return f"{prefix}-{uuid.uuid4().hex[:8].upper()}"


def _git_short_rev() -> str | None:
    """当前代码版本（git 短 rev）；无 git / 超时则如实返回 None。"""
    try:
        out = subprocess.run(
            ["git", "rev-parse", "--short", "HEAD"],
            cwd=PROJECT_ROOT,
            capture_output=True,
            text=True,
            timeout=5,
            check=False,
        )
        return out.stdout.strip() or None
    except Exception:  # noqa: BLE001
        return None


def _summary_text(kind: str, result) -> str | None:
    """人读摘要（列表页展示）；未知 kind 返回 None 而不是编造。"""
    from app.domain.entities.research import BacktestResult, FactorTestReport
    from app.domain.entities.selection import SelectionResult

    if kind == "backtest" and isinstance(result, BacktestResult):
        s = result.summary
        return (
            f"总收益 {s.total_return_pct:.2f}% · 年化 {s.annual_return_pct:.2f}% · "
            f"回撤 {s.max_drawdown_pct:.2f}%"
        )
    if isinstance(result, FactorTestReport):
        return (
            f"IC {result.ic_mean:.4f} · RankIC {result.rank_ic_mean:.4f} · "
            f"样本 {result.sample_days} 日"
        )
    if kind == "selection" and isinstance(result, SelectionResult):
        return (
            f"as_of {result.as_of_date} · 选出 {result.statistics.selected} / "
            f"评估 {result.statistics.evaluated}"
        )
    return None


def _archive_budget_bytes() -> int:
    """归档字节预算：config `research.archive_max_chars`（名义字符数）> 代码默认。"""
    from app.core.config import get_settings

    try:
        configured = get_settings().research_archive_max_chars
    except Exception as exc:  # noqa: BLE001 —— 配置读不到不能阻断归档
        logger.warning("归档预算配置读取失败，用默认值：%s: %s", type(exc).__name__, exc)
        configured = None
    value = int(configured) if configured else DEFAULT_ARCHIVE_MAX_CHARS
    return value if value > 0 else DEFAULT_ARCHIVE_MAX_CHARS


def _curve_sort_key(curve: dict):
    """个股曲线排序键：|期末收益| 降序（与引擎默认输出顺序一致，保证可预期）。"""
    try:
        return abs(float(curve.get("final_return_pct") or 0.0))
    except (TypeError, ValueError):
        return 0.0


def _truncation_note(stored: int, total: int, budget_bytes: int) -> str:
    """裁剪说明（人可读）：必须写清「实际存了多少 / 共多少 / 为什么」。"""
    return (
        f"归档体积超过预算（{budget_bytes:,} 字节，MEDIUMTEXT 上限 16MB）："
        f"个股收益曲线按 |期末收益| 降序仅存 {stored} / 共 {total} 只"
        "（完整明细见 trades / signal_history；archive_meta.truncated=true）"
    )


def _over_budget_note(budget_bytes: int, curves_total: int) -> str:
    """「曲线全裁掉仍超预算」的说明：主体（trades/positions/signal_history 等）
    无法裁剪，如实标注而不是假装达标。"""
    return (
        f"归档主体（trades / positions / signal_history 等，期内持有 {curves_total} 只）"
        f"已超过配置的体积预算（{budget_bytes:,} 字节）：个股收益曲线裁剪至 0 只仍无法达标，"
        "本次仍尝试完整落库；若 MEDIUMTEXT 写入失败会如实报错（archive_meta.over_budget=true）"
    )


def _fit_payload(payload: dict, *, budget_bytes: int) -> tuple[dict, str]:
    """给结果 payload 写入 `archive_meta`，必要时按预算裁剪 symbol_curves。

    返回 (payload, 已序列化的 result_json)。裁剪判定全部作用于「最终字符串的
    UTF-8 字节数」，不是估算：每次候选都真实序列化一次（超预算只发生在
    异常巨大的归档上，此时多几次序列化的代价可接受）。
    """
    has_curves = isinstance(payload.get("symbol_curves"), list)
    curves: list[dict] = list(payload.get("symbol_curves") or [])
    total = len(curves)
    ordered = sorted(curves, key=_curve_sort_key, reverse=True)
    notes = list(payload.get("unimplemented") or [])

    def build(kept: list[dict], *, over_budget: bool = False) -> tuple[dict, str, dict]:
        """按 kept 组装候选 payload 并序列化（含 archive_meta 自收敛）。"""
        truncated = has_curves and len(kept) < total
        candidate = dict(payload)
        extra_notes: list[str] = []
        if has_curves:
            candidate["symbol_curves"] = kept
        if truncated:
            extra_notes.append(_truncation_note(len(kept), total, budget_bytes))
        if over_budget:
            extra_notes.append(_over_budget_note(budget_bytes, total))
        if extra_notes:
            candidate["unimplemented"] = [*notes, *extra_notes]
        meta = {
            "budget_chars": budget_bytes,
            "budget_bytes": budget_bytes,
            "over_budget": over_budget,
            "result_chars": 0,
            "result_bytes": 0,
        }
        if has_curves:
            meta = {
                "curves_stored": len(kept),
                "curves_total": total,
                "truncated": truncated,
                **meta,
            }
        text = ""
        # result_chars/result_bytes 会改变自身长度，迭代至收敛（通常 2 轮内）
        for _ in range(8):
            meta["result_chars"] = len(text)
            meta["result_bytes"] = len(text.encode("utf-8"))
            candidate["archive_meta"] = meta
            text = json.dumps(candidate, ensure_ascii=False)
            if meta["result_chars"] == len(text) and meta["result_bytes"] == len(
                text.encode("utf-8")
            ):
                break
        return candidate, text, meta

    if not has_curves:  # 因子测试 / 选股等无曲线结果：不裁剪，只记录体积
        fitted, text, meta = build([])
        if meta["result_bytes"] > budget_bytes:
            # 无曲线可裁：如实标注超预算，仍尝试落库（写入失败会如实报错）
            logger.warning(
                "归档超过预算且无曲线可裁剪：%d 字节 > %d 字节", meta["result_bytes"], budget_bytes
            )
            fitted, text, meta = build([], over_budget=True)
        return fitted, text

    fitted, text, meta = build(ordered)
    if meta["result_bytes"] <= budget_bytes:
        return fitted, text

    # 超预算：二分找「能放进预算的最大曲线数」（降序保留收益绝对值最大的那些）
    lo, hi, best = 0, total, 0
    while lo <= hi:
        mid = (lo + hi) // 2
        _, candidate_text, _ = build(ordered[:mid])
        if len(candidate_text.encode("utf-8")) <= budget_bytes:
            best = mid
            lo = mid + 1
        else:
            hi = mid - 1
    fitted, text, meta = build(ordered[:best])
    if meta["result_bytes"] > budget_bytes:
        # 连 0 条曲线都放不下：主体（trades/positions/…）超预算，如实标注
        fitted, text, meta = build(ordered[:best], over_budget=True)
    logger.warning(
        "归档超过预算：symbol_curves 裁剪为 %d/%d（预算 %d 字节，实际 %d 字节）",
        meta.get("curves_stored"),
        total,
        budget_bytes,
        meta.get("result_bytes"),
    )
    return fitted, text


def archive_experiment(
    *,
    session,
    kind: str,
    spec_json: str,
    result,
    job_id: str | None = None,
    experiment_repo=None,
) -> ExperimentRecord:
    """把一次研究结果落为 Experiment 归档，返回归档记录（已 save + commit）。

    参数：
    - `session`：SQLAlchemy Session（由调用方持有生命周期；本函数负责 commit）；
    - `kind`：`backtest` / `factor_test` / `selection`；
    - `spec_json`：完整研究 spec（复现依据）；
    - `result`：pydantic 结果对象（`model_dump(mode="json")` 序列化）；
    - `job_id`：来源 Job（同步端点没有 Job，留 None）；
    - `experiment_repo`：ExperimentRepository（不传则用 SqlAlchemy 实现；测试可注入假仓储）。

    `data_version` 由 `compute_data_version(session)` 计算（真实数据指纹，取不到则
    如实降级为 `unavailable`，见该模块 docstring）。
    """
    from app.infrastructure.persistence.sqlalchemy.data_version import compute_data_version

    payload = result.model_dump(mode="json")
    budget_bytes = _archive_budget_bytes()
    payload, result_json = _fit_payload(payload, budget_bytes=budget_bytes)
    meta = payload.get("archive_meta") or {}

    # 把归档元数据同步写回结果对象：同步端点的响应体因此也如实标注裁剪/超预算
    #（AGENT.md §24：不假装支持 / 不静默降级）。没有该字段的结果类型跳过。
    if hasattr(result, "archive_meta"):
        result.archive_meta = meta
    archived_notes = payload.get("unimplemented")
    if isinstance(archived_notes, list) and hasattr(result, "unimplemented"):
        # 归档 JSON 里的 unimplemented 才是权威（含归档侧追加的裁剪/超预算说明），
        # 让同步响应体与归档内容保持一字不差
        result.unimplemented = list(archived_notes)

    experiment = ExperimentRecord(
        id=new_id("EXP"),
        kind=kind,
        spec_json=spec_json,
        result_json=result_json,
        summary_text=_summary_text(kind, result),
        code_version=_git_short_rev(),
        data_version=compute_data_version(session),
        job_id=job_id,
        created_at=datetime.now(),
    )
    repo = experiment_repo if experiment_repo is not None else _default_experiment_repo(session)
    repo.save(experiment)
    session.commit()
    return experiment


def _default_experiment_repo(session):
    from app.infrastructure.persistence.sqlalchemy.repositories.jobs_impl import (
        SqlAlchemyExperimentRepository,
    )

    return SqlAlchemyExperimentRepository(session)