qlib/backend/tests/test_migrations.py

"""Alembic 迁移测试：全新数据库 upgrade head 后应包含全部 Phase 1 表。"""

from __future__ import annotations

import json
import sqlite3
from pathlib import Path

from alembic import command
from alembic.config import Config

BACKEND_ROOT = Path(__file__).resolve().parents[1]


def _alembic_config(db_path: Path) -> Config:
    cfg = Config(str(BACKEND_ROOT / "alembic.ini"))
    cfg.set_main_option("script_location", "app/infrastructure/persistence/migrations")
    cfg.set_main_option("sqlalchemy.url", f"sqlite:///{db_path}")
    return cfg


def test_upgrade_head_creates_phase1_tables(tmp_path) -> None:
    db_path = tmp_path / "fresh.db"
    command.upgrade(_alembic_config(db_path), "head")

    con = sqlite3.connect(db_path)
    try:
        tables = {
            row[0]
            for row in con.execute("select name from sqlite_master where type='table'").fetchall()
        }
    finally:
        con.close()

    expected = {
        "stock",
        "stock_daily",
        "adjust_factor",
        "trading_calendar",
        "financial_indicator",
        "sync_log",
        "condition_field",
        "alembic_version",
    }
    assert expected <= tables

    # 关键防未来函数列存在
    con = sqlite3.connect(db_path)
    try:
        fin_cols = {
            row[1] for row in con.execute("pragma table_info(financial_indicator)").fetchall()
        }
        daily_cols = {row[1] for row in con.execute("pragma table_info(stock_daily)").fetchall()}
    finally:
        con.close()
    assert {"report_date", "announce_date"} <= fin_cols
    assert {"symbol", "trade_date", "close"} <= daily_cols
    # 来源标记列（新浪校验兜底行需与 Tushare 首选行区分）
    assert "source" in fin_cols
    assert {"source", "adjust"} <= daily_cols


def test_upgrade_head_idempotent(tmp_path) -> None:
    db_path = tmp_path / "again.db"
    cfg = _alembic_config(db_path)
    command.upgrade(cfg, "head")
    command.upgrade(cfg, "head")  # 二次执行不报错


def test_refresh_stale_strategy_descriptions(tmp_path) -> None:
    """存量策略的陈旧说明被重算，人工说明保留，空说明补全（b4c5… → c5d6… 全链路）。

    模拟真实历史：在「策略表已建、还没做组合重构」的版本上插入带全套回测参数的旧行，
    一路 upgrade head —— 既验证 b4c5d6e7f8a9 剥掉配置键，也验证 c5d6e7f8a9 重算说明。
    """
    db_path = tmp_path / "refresh.db"
    cfg = _alembic_config(db_path)
    command.upgrade(cfg, "e1f2a3b4c5d6")  # strategy 表建好、尚未重构

    stale_desc = (
        "全市场（剔除 ST），按股息率排序取出前 20 只等权持有，每 6 个月重新择股、"
        "每 6 个月调仓，后复权口径、按调仓日收盘价成交（含佣金 0.03%／印花税 0.05%／滑点 0.1%）。"
    )
    base_cfg = {
        "universe": {"market": "CN_A", "exclude_st": True},
        "factors": [{"name": "dividend_yield", "weight": 1}],
        "conditions": [],
    }
    legacy_cfg = {**base_cfg, "selection": {"top_n": 20}, "rebalance": "monthly",
                  "costs": {"commission_rate": 0.0003}, "price_adjustment": "hfq"}
    con = sqlite3.connect(db_path)
    con.execute(
        "INSERT INTO strategy (id,name,description,spec_type,config_json,version,created_at) "
        "VALUES (?,?,?,?,?,?,?)",
        ("STG-STALE", "高股息 Top20（案例口径）", stale_desc, "backtest",
         json.dumps(legacy_cfg, ensure_ascii=False), "1", "2026-01-01 00:00:00"),
    )
    con.execute(
        "INSERT INTO strategy (id,name,description,spec_type,config_json,version,created_at) "
        "VALUES (?,?,?,?,?,?,?)",
        ("STG-HUMAN", "我的成长股", "只看 ROE 与动量，人工撰写的说明不要被覆盖。", "backtest",
         json.dumps({**base_cfg, "factors": [{"name": "momentum_60", "weight": 1}]},
                    ensure_ascii=False), "1", "2026-01-02 00:00:00"),
    )
    con.execute(
        "INSERT INTO strategy (id,name,description,spec_type,config_json,version,created_at) "
        "VALUES (?,?,?,?,?,?,?)",
        ("STG-EMPTY", "空说明策略", "", "backtest",
         json.dumps(base_cfg, ensure_ascii=False), "1", "2026-01-03 00:00:00"),
    )
    con.commit()
    con.close()

    command.upgrade(cfg, "head")

    con = sqlite3.connect(db_path)
    try:
        rows = {
            r[0]: {"desc": r[1], "spec": r[2], "cfg": r[3]}
            for r in con.execute(
                "SELECT id, description, spec_type, config_json FROM strategy"
            ).fetchall()
        }
        tables = {
            r[0] for r in con.execute(
                "select name from sqlite_master where type='table'"
            ).fetchall()
        }
    finally:
        con.close()

    # ① 陈旧自动说明被重算：不再提佣金/印花税/滑点/调仓，且是新口径文案
    stale = rows["STG-STALE"]
    assert stale["desc"] != stale_desc
    for marker in ("佣金", "印花税", "滑点", "调仓", "择股", "复权口径"):
        assert marker not in stale["desc"], f"陈旧说明仍含 {marker}: {stale['desc']}"
    assert "选股策略" in stale["desc"] and "回测组合" in stale["desc"]

    # ② 人工撰写的说明原样保留（迁移不覆盖用户文本）
    assert rows["STG-HUMAN"]["desc"] == "只看 ROE 与动量，人工撰写的说明不要被覆盖。"

    # ③ 空说明按当前口径补全
    assert rows["STG-EMPTY"]["desc"].strip()
    assert "选股策略" in rows["STG-EMPTY"]["desc"]

    # ④ b4c5… 的职责仍在：旧回测参数从 config_json 剥掉、spec_type 收敛
    stripped = json.loads(rows["STG-STALE"]["cfg"])
    assert "costs" not in stripped and "rebalance" not in stripped and "selection" not in stripped
    assert stripped["factors"][0]["name"] == "dividend_yield"
    assert rows["STG-STALE"]["spec"] == "selection"
    assert {"global_config", "backtest_combo"} <= tables


def test_condition_field_table_columns(tmp_path) -> None:
    """字段库表结构：name 主键 + 中文名/含义/类型/分组/来源/启用状态（2026-10）。"""
    db_path = tmp_path / "fields.db"
    command.upgrade(_alembic_config(db_path), "head")

    con = sqlite3.connect(db_path)
    try:
        cols = {row[1] for row in con.execute("pragma table_info(condition_field)").fetchall()}
        pk = [
            row[1]
            for row in con.execute("pragma table_info(condition_field)").fetchall()
            if row[5]
        ]
    finally:
        con.close()

    assert {
        "name", "label", "description", "kind", "group_name",
        "unit", "source", "enabled", "sort_order", "created_at", "updated_at",
    } <= cols
    assert pk == ["name"]