"""单位换算与筛选滤网测试。

这里覆盖的都是**开发过程中真实踩到过**的坑，每个测试对应一次静默错误：

1. ``total_mv`` 是万元而阈值配的是元 → 筛选结果为空（不报错）；
2. 拿季报 ROE（年初至今累计）去比「5 年年均 ROE」→ 好公司全被误杀；
3. 银行负债率天然 90%+ → 整个金融板块被误杀；
4. 分红除权晚于 asof → 稳定分红公司被误判为「连续分红 0 年」。
"""

from __future__ import annotations

from datetime import date

import pandas as pd
import pytest

from hdiv.data.units import (
    normalize_financial_panel,
    normalize_market_panel,
    pct_to_ratio,
    verify_market_units,
    vol_shou_to_shares,
    wan_to_shares,
    wan_to_yuan,
)


# ---------------------------------------------------------------------------
# 单位换算
# ---------------------------------------------------------------------------


def test_wan_to_yuan() -> None:
    assert wan_to_yuan(pd.Series([1.0, 100.0])).tolist() == [1e4, 1e6]


def test_wan_to_shares() -> None:
    assert wan_to_shares(pd.Series([125619.78])).iloc[0] == pytest.approx(1.2561978e9)


def test_pct_to_ratio() -> None:
    assert pct_to_ratio(pd.Series([5.08])).iloc[0] == pytest.approx(0.0508)


def test_vol_shou_to_shares() -> None:
    assert vol_shou_to_shares(pd.Series([100])).iloc[0] == 10000


def test_normalize_market_panel_units() -> None:
    """茅台 2024-06-28 实测值：总市值 1.84 万亿元，总股本 12.56 亿股。"""
    df = pd.DataFrame(
        {
            "symbol": ["600519.SH"],
            "close": [1467.39],
            "total_share": [125619.78],   # 万股
            "total_mv": [184333208.97],   # 万元
            "circ_mv": [184333208.97],
            "dv_ttm": [5.1720],           # 百分数
            "turnover_rate": [0.25],
        }
    )
    out = normalize_market_panel(df)
    assert out["total_mv"].iloc[0] == pytest.approx(1.8433320897e12, rel=1e-6)
    assert out["total_share"].iloc[0] == pytest.approx(1.2561978e9, rel=1e-9)
    assert out["dv_ttm"].iloc[0] == pytest.approx(0.051720)
    assert out["_units"].iloc[0] == "yuan/shares/ratio"


def test_normalize_market_panel_does_not_mutate_input() -> None:
    df = pd.DataFrame({"total_mv": [100.0], "total_share": [10.0]})
    before = df.copy()
    normalize_market_panel(df)
    pd.testing.assert_frame_equal(df, before)


def test_normalize_financial_panel_converts_pct() -> None:
    df = pd.DataFrame({"roe": [15.13], "debt_to_assets": [90.23], "total_revenue": [1.78e11]})
    out = normalize_financial_panel(df)
    assert out["roe"].iloc[0] == pytest.approx(0.1513)
    assert out["debt_to_assets"].iloc[0] == pytest.approx(0.9023)
    assert out["total_revenue"].iloc[0] == 1.78e11, "金额列本就以元计，不得被换算"


def test_verify_market_units_detects_correct_data() -> None:
    df = pd.DataFrame(
        {
            "close": [10.0, 20.0],
            "total_share": [1e9, 5e8],
            "total_mv": [1e10, 1e10],
        }
    )
    v = verify_market_units(df)
    assert v["checked"] == 2 and v["bad"] == 0
    assert v["median_ratio"] == pytest.approx(1.0)


def test_identity_alone_cannot_detect_wan_vs_yuan() -> None:
    """恒等式对「整体万元/元混淆」无效 —— 因为 元/股 × 万股 = 万元。

    这是一个容易误以为「有了恒等式检查就安全」的陷阱，必须显式记录：
    原始单位下比值仍然恰好是 1。
    """
    raw = pd.DataFrame(
        {"close": [1467.39], "total_share": [125619.78], "total_mv": [184333208.97]}
    )
    v = verify_market_units(raw)
    assert v["identity_ok"] is True, "恒等式在原始单位下同样成立"
    assert v["unit_ok"] is False, "但绝对量级检查必须发现单位错误"
    assert v["detected_unit"] == "wan"


def test_verify_market_units_accepts_normalized_yuan() -> None:
    norm = normalize_market_panel(
        pd.DataFrame(
            {"close": [1467.39], "total_share": [125619.78], "total_mv": [184333208.97]}
        )
    )
    v = verify_market_units(norm)
    assert v["unit_ok"] is True and v["identity_ok"] is True
    assert v["detected_unit"] == "yuan"


def test_verify_market_units_detects_partial_conversion() -> None:
    """只换算一个字段（常见疏漏）也必须被抓出来。"""
    df = pd.DataFrame(
        {"close": [10.0], "total_share": [1e9], "total_mv": [1e10 / 1e4]}  # 市值漏换算
    )
    v = verify_market_units(df)
    assert v["bad"] == 1


def test_verify_market_units_empty() -> None:
    assert verify_market_units(pd.DataFrame())["checked"] == 0


# ---------------------------------------------------------------------------
# 滤网：行业豁免
# ---------------------------------------------------------------------------


class _FakeRepo:
    def __init__(self, st: set[str] | None = None, susp: set[str] | None = None) -> None:
        self._st = st or set()
        self._susp = susp or set()

    def st_symbols(self, asof, include_delisting=True):  # noqa: ANN001, ARG002
        return self._st

    def suspended_on(self, asof):  # noqa: ANN001, ARG002
        return self._susp


def _frame(**kwargs) -> pd.DataFrame:
    base = {
        "symbol": ["600036.SH", "601088.SH"],
        "name": ["招商银行", "中国神华"],
        "industry": ["银行", "煤炭开采"],
        "market": ["主板", "主板"],
        "exchange": ["SSE", "SSE"],
        "listed_years": [30.0, 17.0],
        "is_fresh": [True, True],
        "total_mv": [8.6e11, 8.8e11],
        "circ_mv": [7.0e11, 7.3e11],
        "avg_amount": [1e9, 8e8],
        "close": [34.19, 44.37],
        "debt_ratio": [0.902, 0.236],
        "total_hldr_eqy_exc_min_int": [3.5e11, 4.0e11],
    }
    base.update(kwargs)
    return pd.DataFrame(base)


def test_risk_filter_exempts_banks_from_leverage() -> None:
    from hdiv.core.config import RiskFilterConfig
    from hdiv.universe.filters.risk import RiskFilter

    f = RiskFilter(RiskFilterConfig(max_debt_to_assets=0.80), exempt_leverage=["银行"])
    out = f.compute(_frame(), _FakeRepo(), date(2024, 6, 28))
    assert bool(out.passed.iloc[0]) is True, "银行必须豁免负债率上限"
    assert bool(out.passed.iloc[1]) is True, "低负债公司自然通过"
    assert out.values["600036.SH"]["debt_ratio_exempt"] is True


def test_risk_filter_still_rejects_high_leverage_non_exempt() -> None:
    from hdiv.core.config import RiskFilterConfig
    from hdiv.universe.filters.risk import RiskFilter

    f = RiskFilter(RiskFilterConfig(max_debt_to_assets=0.80), exempt_leverage=["银行"])
    df = _frame(industry=["房地产", "煤炭开采"])
    out = f.compute(df, _FakeRepo(), date(2024, 6, 28))
    assert bool(out.passed.iloc[0]) is False
    assert "资产负债率" in out.reasons["600036.SH"]


def test_quality_filter_exempts_banks_from_fcf_and_leverage() -> None:
    from hdiv.core.config import QualityFilterConfig
    from hdiv.universe.filters.quality import FinancialQualityFilter

    cfg = QualityFilterConfig(min_ocf_to_profit=0.60, max_debt_to_assets=0.80)
    f = FinancialQualityFilter(cfg, exempt_leverage=["银行"], exempt_fcf=["银行"])
    df = _frame(roe_avg=[0.1513, 0.1406], ocf_to_profit=[-0.03, 1.80])
    out = f.compute(df, _FakeRepo(), date(2024, 6, 28))
    assert bool(out.passed.iloc[0]) is True, "银行豁免 FCF 与负债率"


def test_quality_filter_uses_annual_average_not_quarterly() -> None:
    """季报 ROE 3.47% 不该被拿去比年均 8% 的阈值。"""
    from hdiv.core.config import QualityFilterConfig
    from hdiv.universe.filters.quality import FinancialQualityFilter

    cfg = QualityFilterConfig(min_roe_5y_avg=0.08, min_ocf_to_profit=None)
    f = FinancialQualityFilter(cfg)
    df = _frame(
        industry=["白酒", "煤炭开采"],
        roe=[0.0347, 0.0380],          # 季报累计值（会被 avg 覆盖）
        roe_avg=[0.1513, 0.1406],      # 年报 5 年平均
    )
    out = f.compute(df, _FakeRepo(), date(2024, 6, 28))
    assert bool(out.passed.iloc[0]) is True, "应使用 roe_avg 而非季报 roe"
    assert out.values["600036.SH"]["roe"] == pytest.approx(0.1513)


def test_quality_filter_falls_back_to_latest_when_no_average() -> None:
    from hdiv.core.config import QualityFilterConfig
    from hdiv.universe.filters.quality import FinancialQualityFilter

    cfg = QualityFilterConfig(min_roe_5y_avg=0.08, min_ocf_to_profit=None)
    f = FinancialQualityFilter(cfg)
    df = _frame(industry=["白酒", "煤炭开采"], roe=[0.20, 0.03])  # 无 roe_avg 列
    out = f.compute(df, _FakeRepo(), date(2024, 6, 28))
    assert bool(out.passed.iloc[0]) is True
    assert bool(out.passed.iloc[1]) is False


# ---------------------------------------------------------------------------
# 分红连续性：一年宽限期
# ---------------------------------------------------------------------------


def _div(symbol: str, end_year: int, ex_year: int, dps: float = 1.0, month: int = 7) -> dict:
    return {
        "symbol": symbol,
        "end_date": date(end_year, 12, 31),
        "imp_ann_date": date(ex_year, month - 1 if month > 1 else 12, 1),
        "div_proc": "实施",
        "cash_div_tax": dps,
        "cash_div": dps,
        "stk_div": None,
        "base_share": 10000.0,
        "ex_date": date(ex_year, month, 15),
    }


def test_continuity_within_target_year() -> None:
    """FY2023 分红已在 2024-05 除权 → target=2023，连续 5 年。"""
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    recs = {2023: 2024, 2022: 2023, 2021: 2022, 2020: 2021, 2019: 2020}
    rows = [_div("600036.SH", fy, ex, month=5) for fy, ex in recs.items()]
    s = DividendFilter._stats(rows, target_year=2023, asof=date(2024, 6, 28),
                              cfg=DividendFilterConfig())
    assert s["dividend_continuity_years"] == 5
    assert s["continuity_grace_used"] is False


def test_continuity_grace_when_ex_date_lags() -> None:
    """神华实测情形：FY2023 分红要 2024-07 才除权，asof=2024-06-28 时不可见。

    此时必须用一年宽限期从 FY2022 起算，而不是判定为「中断分红」。
    """
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    rows = [_div("601088.SH", fy, fy + 1, month=7) for fy in (2022, 2021, 2020, 2019, 2018)]
    s = DividendFilter._stats(rows, target_year=2023, asof=date(2024, 6, 28),
                              cfg=DividendFilterConfig())
    assert s["latest_dividend_year"] == 2022
    assert s["continuity_start_year"] == 2022
    assert s["continuity_grace_used"] is True
    assert s["dividend_continuity_years"] == 5, "不应因为除权晚而误判中断"


def test_continuity_zero_when_genuinely_stopped() -> None:
    """最近可见分红比应考核财年早两年以上 → 视为真的中断。"""
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    rows = [_div("000002.SZ", fy, fy + 1) for fy in (2019, 2018, 2017, 2016, 2015)]
    s = DividendFilter._stats(rows, target_year=2023, asof=date(2024, 6, 28),
                              cfg=DividendFilterConfig())
    assert s["dividend_continuity_years"] == 0


def test_continuity_breaks_on_gap() -> None:
    """有断档：2022 有、2021 无 → 连续 1 年。"""
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    rows = [_div("X.SZ", fy, fy + 1) for fy in (2022, 2020, 2019)]
    s = DividendFilter._stats(rows, target_year=2023, asof=date(2024, 6, 28),
                              cfg=DividendFilterConfig())
    assert s["dividend_continuity_years"] == 1


def test_ttm_dps_only_counts_ex_date_in_window() -> None:
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    rows = [
        _div("X.SZ", 2023, 2024, dps=1.5, month=4),   # 窗口内
        _div("X.SZ", 2022, 2023, dps=1.2, month=7),   # 窗口内（>2023-06-28）
        _div("X.SZ", 2021, 2022, dps=1.0, month=7),   # 窗口外
    ]
    s = DividendFilter._stats(rows, target_year=2023, asof=date(2024, 6, 28),
                              cfg=DividendFilterConfig())
    assert s["ttm_dps"] == pytest.approx(2.7)


def test_shift_year_handles_leap_day() -> None:
    from hdiv.universe.filters.dividend import _shift_year

    assert _shift_year(date(2024, 2, 29), -1) == date(2023, 2, 28)
    assert _shift_year(date(2024, 6, 28), -1) == date(2023, 6, 28)


def test_target_year_uses_latest_annual_report() -> None:
    """最新年报为 FY2023 时，考核目标年应为 2023。"""
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    df = pd.DataFrame(
        {
            "symbol": ["A.SH", "B.SH"],
            "fin_end_date": [date(2024, 3, 31), date(2023, 12, 31)],
        }
    )
    t = DividendFilter._target_years(df, date(2024, 6, 28))
    # 能看到 2024Q1 报，说明 FY2023 年报必然已披露 → 目标年 2023
    assert t["A.SH"] == 2023, "有 2024Q1 报 → FY2023 年报已出"
    assert t["B.SH"] == 2023, "有 FY2023 年报 → 目标是 2023"

    df2 = pd.DataFrame({"symbol": ["C.SH"], "fin_end_date": [date(2023, 9, 30)]})
    assert DividendFilter._target_years(df2, date(2024, 6, 28))["C.SH"] == 2022, (
        "只看到 2023Q3 → FY2023 年报未出，退回到 2022"
    )


def test_payout_ratio_uses_base_share() -> None:
    """总现金分红 = 每股分红 × 基准股本（万股→股）。"""
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    rows = [
        {
            "symbol": "X.SH", "end_date": date(2023, 12, 31), "ex_date": date(2024, 6, 1),
            "div_proc": "实施", "cash_div_tax": 2.0, "base_share": 10000.0,  # 1 亿股
        }
    ]
    row = pd.Series({"n_income_attr_p": 4e8, "free_cashflow": 8e8})
    out = DividendFilter._payout_and_cover(rows, row, date(2024, 6, 28),
                                           DividendFilterConfig())
    assert out["total_cash_dividend"] == pytest.approx(2.0 * 10000.0 * 1e4)
    assert out["payout_ratio"] == pytest.approx(0.5)
    # 总现金分红 = 2.0 元 × 10000 万股 × 1e4 = 2e8 元；FCF 8e8 → 覆盖 4 倍
    assert out["fcf_dividend_cover"] == pytest.approx(4.0)


def test_fcf_fallback_when_tushare_missing() -> None:
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    rows = [{
        "symbol": "X.SH", "end_date": date(2023, 12, 31), "ex_date": date(2024, 6, 1),
        "div_proc": "实施", "cash_div_tax": 1.0, "base_share": 1000.0,
    }]
    row = pd.Series({
        "n_income_attr_p": 5e6, "free_cashflow": None,
        "n_cashflow_act": 1e7, "c_pay_dist_dpcp_int_exp": 2e6,
    })
    out = DividendFilter._payout_and_cover(rows, row, date(2024, 6, 28),
                                           DividendFilterConfig())
    assert out["fcf_source"] == "ocf_minus_dist"
    assert out["free_cashflow"] == pytest.approx(8e6)


# ---------------------------------------------------------------------------
# 滤网结果索引契约（曾经把 DataFrame 索引当成 symbol 用的真实 bug）
# ---------------------------------------------------------------------------


def test_filter_outcome_index_is_dataframe_index_not_symbol() -> None:
    """滤网的 passed 索引必须与传入 DataFrame 的索引一致。

    选择器据此用 ``live.at[i, "symbol"]`` 映射；
    若误把索引当 symbol，会导致「全部淘汰」（本项目开发中确实发生过）。
    """
    from hdiv.core.config import MarketFilterConfig
    from hdiv.universe.filters.market import MarketFilter

    df = _frame()
    df.index = [10, 20]  # 非默认索引
    f = MarketFilter(MarketFilterConfig(min_market_cap=1e10))
    out = f.compute(df, _FakeRepo(), date(2024, 6, 28))
    assert list(out.passed.index) == [10, 20]
    assert set(out.passed.values) <= {True, False}


# ---------------------------------------------------------------------------
# 数据缺失策略（安全边际策略的关键取舍）
# ---------------------------------------------------------------------------


def _div_cfg(**kw):
    from hdiv.core.config import DividendFilterConfig

    base = {
        "min_dividend_yield": None,
        "min_continuous_years": 0,
        "min_dividend_years_in_window": 0,
        "max_payout_ratio": None,
        "require_positive_fcf": True,
        "min_fcf_dividend_cover": None,
    }
    base.update(kw)
    return DividendFilterConfig(**base)


def test_missing_fcf_passes_by_default() -> None:
    """默认宽松：数据缺失放行，避免因未同步而误杀。"""
    from hdiv.universe.filters.dividend import DividendFilter

    cfg = _div_cfg(on_missing_data="pass")
    stats = {"free_cashflow": None, "dividend_continuity_years": 0,
             "dividend_years_in_window": 0, "dividend_yield": 0.05}
    assert DividendFilter._reject_reason(stats, cfg) is None


def test_missing_fcf_rejected_when_strict() -> None:
    """严格模式：无法验证现金流即淘汰 —— 忠于「安全边际」的策略逻辑。"""
    from hdiv.universe.filters.dividend import DividendFilter

    cfg = _div_cfg(on_missing_data="reject")
    stats = {"free_cashflow": None, "dividend_continuity_years": 0,
             "dividend_years_in_window": 0, "dividend_yield": 0.05}
    why = DividendFilter._reject_reason(stats, cfg)
    assert why is not None and "缺失" in why


def test_negative_fcf_rejected_in_both_modes() -> None:
    """FCF 为负时两种模式都必须淘汰 —— 宽松不等于放行已知风险。"""
    from hdiv.universe.filters.dividend import DividendFilter

    stats = {"free_cashflow": -1e8, "dividend_continuity_years": 0,
             "dividend_years_in_window": 0, "dividend_yield": 0.05}
    for mode in ("pass", "reject"):
        why = DividendFilter._reject_reason(stats, _div_cfg(on_missing_data=mode))
        assert why is not None and "负" in why, f"{mode} 模式必须拒绝负 FCF"


def test_missing_coverage_strict_only() -> None:
    from hdiv.universe.filters.dividend import DividendFilter

    stats = {"free_cashflow": 1e8, "fcf_dividend_cover": None,
             "dividend_continuity_years": 0, "dividend_years_in_window": 0,
             "dividend_yield": 0.05}
    assert DividendFilter._reject_reason(
        stats, _div_cfg(min_fcf_dividend_cover=1.0, on_missing_data="pass")
    ) is None
    assert DividendFilter._reject_reason(
        stats, _div_cfg(min_fcf_dividend_cover=1.0, on_missing_data="reject")
    ) is not None


def test_on_missing_data_invalid_value_rejected() -> None:
    from hdiv.core.errors import SchemaValidationError

    with pytest.raises(Exception) as ei:
        _div_cfg(on_missing_data="maybe")
    assert "on_missing_data" in str(ei.value) or "Input should be" in str(ei.value)
    del SchemaValidationError


# ---------------------------------------------------------------------------
# 分红支付率必须与分红**同财年**（真实踩到的错误）
# ---------------------------------------------------------------------------


def test_payout_uses_same_fiscal_year_not_latest_quarter() -> None:
    """回归：曾用「FY2023 分红 ÷ 2024Q1 净利润」算出 230.9% 的荒谬支付率。

    正确口径下美的集团 FY2023 为：分红 207.8 亿 ÷ 净利 337.2 亿 = 61.63%。
    """
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    recs = [{
        "symbol": "000333.SZ", "end_date": date(2023, 12, 31),
        "ex_date": date(2024, 5, 15), "div_proc": "实施",
        "cash_div_tax": 3.0, "base_share": 692675.9241,
    }]
    # 同财年（FY2023）财务
    fy_row = pd.Series({
        "n_income_attr_p": 3.372e10, "free_cashflow": 7.16e10,
    })
    out = DividendFilter._payout_and_cover(
        recs, fy_row, date(2024, 6, 28), DividendFilterConfig()
    )
    assert out["financial_year"] == 2023
    assert out["payout_ratio"] == pytest.approx(207.8 / 337.2, abs=0.01)
    assert out["payout_basis"] == "same_fiscal_year"


def test_payout_not_computed_without_same_year_row() -> None:
    """缺少同财年财务时必须返回「不可得」，而不是退回到最新季报。"""
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    recs = [{
        "symbol": "X.SZ", "end_date": date(2023, 12, 31),
        "ex_date": date(2024, 5, 15), "div_proc": "实施",
        "cash_div_tax": 1.0, "base_share": 10000.0,
    }]
    out = DividendFilter._payout_and_cover(
        recs, None, date(2024, 6, 28), DividendFilterConfig()
    )
    assert out["payout_ratio"] is None
    assert out["fcf_dividend_cover"] is None
    assert out["payout_basis"] == "missing_same_year_financials"


def test_total_cash_dividend_uses_base_share() -> None:
    """base_share 是**万股**，漏乘 1e4 会把支付率缩小一万倍。"""
    from hdiv.core.config import DividendFilterConfig
    from hdiv.universe.filters.dividend import DividendFilter

    recs = [{
        "symbol": "X.SZ", "end_date": date(2023, 12, 31),
        "ex_date": date(2024, 5, 15), "div_proc": "实施",
        "cash_div_tax": 2.0, "base_share": 10000.0,  # 1 亿股
    }]
    row = pd.Series({"n_income_attr_p": 4e8, "free_cashflow": 8e8})
    out = DividendFilter._payout_and_cover(
        recs, row, date(2024, 6, 28), DividendFilterConfig()
    )
    assert out["total_cash_dividend"] == pytest.approx(2.0 * 10000.0 * 1e4)
    assert out["payout_ratio"] == pytest.approx(0.5)


def test_dividend_records_include_base_share() -> None:
    """回归：repo.dividend_records 曾漏选 base_share，

    导致 payout_ratio 与 fcf_dividend_cover 在全库范围内静默为 NULL，
    进而使 max_payout_ratio / min_fcf_dividend_cover 两个筛选条件从未生效。
    """
    import inspect

    from hdiv.data.repo import Repo

    src = inspect.getsource(Repo.dividend_records)
    assert "base_share" in src, "dividend_records 必须选出 base_share"


# ---------------------------------------------------------------------------
# 重跑覆盖同一条记录
# ---------------------------------------------------------------------------


def test_universe_run_id_is_deterministic() -> None:
    """回归：run_id 不得含时间戳，否则同参数重跑会不断累积重复记录。

    早期实现把 datetime.now() 编进指纹，同一 asof 最多累积了 11 条内容相同的记录。
    现在的语义是「同一份配置 + 同一时点 → 同一个 run_id → 重跑原地覆盖」。
    """
    import inspect

    from hdiv.universe import selector

    src = inspect.getsource(selector.UniverseSelector)
    i = src.find("run_id = stable_id(")
    assert i != -1, "未找到 run_id 生成处"
    # 取到该语句结束的分号行（不能用第一个 ')'，那会截断在 config_hash(self.config) 里）
    end = src.find("\n        )", i)
    assert end != -1, "未找到 run_id 语句结尾"
    block = src[i:end]
    assert "datetime.now" not in block, f"run_id 指纹仍含时间戳：{block}"
    for must in ("config_hash", "effective", "self.config.name"):
        assert must in block, f"run_id 指纹缺少 {must}：{block}"


@pytest.mark.db
def test_universe_rerun_overwrites_same_record() -> None:
    """同参数重跑不新增记录，且成员行数等于候选数（无重复堆积）。"""
    from hdiv.core.config import load_config
    from hdiv.data import db
    from hdiv.data.sync.base import stable_id

    db.load_dotenv_once()
    cfg = load_config("datasource")
    df = db.read_sql(
        "SELECT r.run_id, r.candidate_count, r.asof_date, r.config_hash, r.name, "
        "       COUNT(m.id) AS member_rows "
        "FROM hd_universe_run r LEFT JOIN hd_universe_member m ON m.run_id = r.run_id "
        "GROUP BY r.run_id HAVING member_rows > 0 "
        "ORDER BY r.created_at DESC LIMIT 5",
        cfg=cfg,
    )
    if df.empty:
        pytest.skip("没有筛选记录")
    checked = 0
    for _, r in df.iterrows():
        expect = stable_id("universe", r["name"], str(r["asof_date"]), r["config_hash"])
        if r["run_id"] != expect:
            continue  # 确定化之前的历史记录，跳过
        checked += 1
        assert int(r["member_rows"]) == int(r["candidate_count"]), (
            f"run_id={r['run_id'][:10]} 成员行数 {r['member_rows']} "
            f"应等于候选数 {r['candidate_count']}（出现重复堆积）"
        )
    assert checked > 0, "未找到确定化之后生成的筛选记录，无法验证"