qlib/backend/app/cli/prune_experiments.py

"""实验归档保留策略 CLI（归档体积治理）。

用法（cd backend）：
    # 只看会删什么（**默认就是 dry-run**，不写库）
    uv run python -m app.cli.prune_experiments --keep 30
    uv run python -m app.cli.prune_experiments --keep 20 --kind backtest
    uv run python -m app.cli.prune_experiments --older-than 180          # 180 天前的归档
    # 真正删除（必须显式加 --apply）
    uv run python -m app.cli.prune_experiments --keep 30 --apply

为什么默认 dry-run：归档是**研究结论的唯一凭据**（结果 + spec + 代码/数据版本），
误删不可恢复。因此：
- 默认只打印候选清单（含 id / 类型 / 时间 / 体积 / 摘要）与合计释放空间，不写库；
- 只有显式 `--apply` 才执行删除，且删除前**再打印一次**清单；
- 按 `created_at` **由新到旧保留**：先按过滤条件选出候选集合，再保留最新的 N 条，
  其余删除（`--keep` 与 `--older-than` 可同时给，取交集）；
- 只删 `experiment` 表（归档本身），**不动 `job_record`**（执行历史仍可追溯）；
  但要注意：完整结果只存归档一份（`job.result_json` 对新记录为 NULL），
  因此**删除归档 = 该次回测结果不可再查看**。要留底先导出（归档页「导出完整 JSON」）；
- `--kind` 可只清理某一类（如只清 `factor_test`，保留回测结论）。

本模块是组装层（composition root）：装配 Session 与 Repository，删除逻辑走仓储。
"""

from __future__ import annotations

import argparse
import sys
from datetime import datetime, timedelta

from app.infrastructure.persistence.sqlalchemy.repositories.jobs_impl import (
    SqlAlchemyExperimentRepository,
)
from app.infrastructure.persistence.sqlalchemy.session import SessionLocal


def select_prune_candidates(
    rows: list,
    *,
    keep: int | None,
    older_than_days: int | None,
    now: datetime | None = None,
) -> list:
    """从归档列表里挑出**该删除**的那些（纯函数，便于单测）。

    `rows`：任意具备 `id` / `created_at` 的归档摘要（不依赖 ORM 与数据库）。
    规则（两个条件同时给出时取交集）：
    - `keep=N`：按 `created_at` 由新到旧排序后，**保留最新 N 条**，其余为候选；
    - `older_than_days=D`：创建时间早于 `now - D 天` 的才是候选；
    - `created_at` 为空的归档视为**最旧**（排序末位）——缺失时间时宁可被列为候选，
      也不静默把它当作「最新」而永久留在库里。
    """
    ordered = sorted(rows, key=lambda e: (e.created_at or datetime.min), reverse=True)
    candidates = list(ordered)
    if older_than_days is not None:
        cutoff = (now or datetime.now()) - timedelta(days=older_than_days)
        candidates = [e for e in candidates if (e.created_at or datetime.min) < cutoff]
    if keep is not None:
        keep_ids = {e.id for e in ordered[: max(keep, 0)]}
        candidates = [e for e in candidates if e.id not in keep_ids]
    # 输出仍按新 → 旧，便于人核对
    candidates.sort(key=lambda e: (e.created_at or datetime.min), reverse=True)
    return candidates


def _parse_args(argv: list[str]) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="python -m app.cli.prune_experiments",
        description="实验归档保留策略（默认 dry-run，需 --apply 才真正删除）",
    )
    p.add_argument("--keep", type=int, default=None, help="按时间由新到旧保留的最新条数")
    p.add_argument("--kind", default=None, help="只处理某类归档（backtest/factor_test/selection）")
    p.add_argument(
        "--older-than",
        type=int,
        default=None,
        metavar="DAYS",
        help="只处理创建时间早于 N 天的归档",
    )
    p.add_argument("--apply", action="store_true", help="真正执行删除（否则仅预览）")
    args = p.parse_args(argv)
    if args.keep is None and args.older_than is None:
        p.error("至少给出 --keep 或 --older-than 之一，避免误删全部归档")
    if args.keep is not None and args.keep < 0:
        p.error("--keep 不能为负数")
    return args


def main(argv: list[str] | None = None) -> int:
    args = _parse_args(sys.argv[1:] if argv is None else argv)
    with SessionLocal() as session:
        repo = SqlAlchemyExperimentRepository(session)
        # 列表查询刻意不取 result_json，体积用库侧算出的 result_bytes（单位：字符）
        rows = repo.list_filtered(kind=args.kind, limit=1_000_000, offset=0)
        candidates = select_prune_candidates(
            rows, keep=args.keep, older_than_days=args.older_than
        )

        total_bytes = sum(e.result_bytes or 0 for e in candidates)
        print(f"归档总数 {len(rows)} 条，命中删除条件 {len(candidates)} 条，"
              f"预计释放约 {total_bytes / 1024 / 1024:.1f} MB")
        for e in candidates:
            created = e.created_at.strftime("%Y-%m-%d %H:%M") if e.created_at else "未知时间"
            print(
                f"  - {e.id}  {e.kind:<12} {created}  "
                f"{((e.result_bytes or 0) / 1024):>8.0f} KB  {(e.summary_text or '')[:60]}"
            )
        if not candidates:
            print("没有需要删除的归档。")
            return 0
        if not args.apply:
            print("\n[dry-run] 未删除任何记录；确认无误后加 --apply 执行。")
            return 0

        deleted = 0
        for e in candidates:
            if repo.delete(e.id):
                deleted += 1
        session.commit()
        print(f"\n已删除 {deleted} 条归档（job_record 未改动）。")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())