"""Tests for verify_increment.py — 安全闸门。"""
from __future__ import annotations

import os
import sys
import time
from datetime import datetime

import pandas as pd
import pytest

_HERE = os.path.dirname(os.path.abspath(__file__))
_SCRIPT_DIR = os.path.abspath(os.path.join(_HERE, "..", "..", "scripts", "data_platform"))
if _SCRIPT_DIR not in sys.path:
    sys.path.insert(0, _SCRIPT_DIR)

import verify_increment as vi  # noqa: E402


# ---------- helpers ----------

def _good_df(dates: list[str]) -> pd.DataFrame:
    """合法日线（通过 DataValidator 所有 fatal）。"""
    n = len(dates)
    return pd.DataFrame({
        "date": pd.to_datetime(dates),
        "open": [10.0 + i for i in range(n)],
        "high": [10.5 + i for i in range(n)],
        "low": [9.8 + i for i in range(n)],
        "close": [10.2 + i for i in range(n)],
        "volume": [10000 + i for i in range(n)],
    })


def _write_staging(staging_root: str, year: str, fname: str, df: pd.DataFrame) -> None:
    ydir = os.path.join(staging_root, year)
    os.makedirs(ydir, exist_ok=True)
    df.to_parquet(os.path.join(ydir, fname), index=False)


def _recent_dates(n: int = 3) -> list[str]:
    """最近 n 个工作日（保证 fresh，含今天/最近交易日）。"""
    today = pd.Timestamp(time.strftime("%Y-%m-%d"))
    dates = pd.bdate_range(end=today, periods=n).strftime("%Y-%m-%d").tolist()
    return dates


# ---------- symbol_from_filename ----------

def test_symbol_from_filename():
    assert vi.symbol_from_filename("sh600000_daily.parquet") == "600000"
    assert vi.symbol_from_filename("sz000001_daily.parquet") == "000001"
    assert vi.symbol_from_filename("bj920000_daily.parquet") == "920000"


# ---------- all good → passed ----------

def test_verify_all_good_passes(tmp_path):
    """staging 全合法且 fresh → passed=True。"""
    staging = tmp_path / "staging"
    dates = _recent_dates(3)
    # 5 只正常 + 1 只北交所（应被扣分母，不影响通过）
    for sym in ("sh600000", "sh600004", "sz000001", "sz300001", "sh688001"):
        _write_staging(str(staging), "2026", f"{sym}_daily.parquet", _good_df(dates))
    _write_staging(str(staging), "2026", "bj920000_daily.parquet", _good_df(dates))

    start = dates[0]
    result = vi.verify(str(staging), start)

    assert result["passed"] is True
    assert result["total"] == 6
    assert result["unsupported_skipped"] == 1   # 北交所 920
    assert result["success"] == 5
    assert result["success_rate"] == 1.0
    assert result["fresh_rate"] == 1.0
    assert result["failed_symbols"] == []


# ---------- fatal cases → failed ----------

def test_verify_empty_file_fails(tmp_path):
    """空 df（DataValidator 直接判 fatal '数据为空'）→ 该 symbol 失败。"""
    staging = tmp_path / "staging"
    dates = _recent_dates(3)

    # 4 只好 + 1 只空
    for sym in ("sh600000", "sh600004", "sz000001", "sz300001"):
        _write_staging(str(staging), "2026", f"{sym}_daily.parquet", _good_df(dates))
    _write_staging(str(staging), "2026", "sh688001_daily.parquet", pd.DataFrame(
        {"date": [], "open": [], "high": [], "low": [], "close": [], "volume": []}
    ))

    result = vi.verify(str(staging), dates[0])
    # 4 好 / 5 总 = 0.8 < 0.95 → fail
    assert result["passed"] is False
    assert result["success"] == 4
    assert result["total"] == 5
    assert "688001" in result["failed_symbols"]
    assert result["success_rate"] < vi.MIN_SUCCESS_RATE
    # fatal 样本里有 688001
    assert any(s["symbol"] == "688001" for s in result["fatal_samples"])


def test_verify_zero_price_fails(tmp_path):
    """价格<=0（D1 fatal）→ 该 symbol 失败。"""
    staging = tmp_path / "staging"
    dates = _recent_dates(3)

    for sym in ("sh600000", "sh600004", "sz000001"):
        _write_staging(str(staging), "2026", f"{sym}_daily.parquet", _good_df(dates))
    # 构造 close<=0 的坏 df
    bad = pd.DataFrame({
        "date": pd.to_datetime(dates),
        "open": [0.0, 0.0, 0.0], "high": [0.0, 0.0, 0.0],
        "low": [0.0, 0.0, 0.0], "close": [0.0, 0.0, 0.0],
        "volume": [100, 200, 300],
    })
    _write_staging(str(staging), "2026", "sz300001_daily.parquet", bad)

    result = vi.verify(str(staging), dates[0])
    # 3 好 / 4 总 = 0.75 < 0.95 → fail
    assert result["passed"] is False
    assert "300001" in result["failed_symbols"]
    # 样本错误里有 D1
    sample = next(s for s in result["fatal_samples"] if s["symbol"] == "300001")
    assert any("D1" in e for e in sample["errors"])


def test_verify_bse_excluded_from_denominator(tmp_path):
    """北交所码（920/921/83/87）从分母扣——不算失败也不算成功。"""
    staging = tmp_path / "staging"
    dates = _recent_dates(3)
    # 3 只全北交所 → denom=0 → passed=False（denom=0 算不通过，因为没有有效样本可验）
    for sym in ("bj920000", "bj920001", "sz830001", "sz870001"):
        _write_staging(str(staging), "2026", f"{sym}_daily.parquet", _good_df(dates))

    result = vi.verify(str(staging), dates[0])
    assert result["unsupported_skipped"] == 4
    assert result["denom"] == 0
    assert result["passed"] is False  # denom=0 → 不通过


def test_verify_missing_year_dir_raises(tmp_path):
    """staging 的 year 目录不存在 → FileNotFoundError。"""
    staging = tmp_path / "staging"
    staging.mkdir()
    with pytest.raises(FileNotFoundError):
        vi.verify(str(staging), "2099-01-01")


# ---------- _latest_available_trading_day（收盘感知） ----------
# 2026-07-13=Mon, 07-14=Tue, 07-10=Fri, 07-11=Sat, 07-12=Sun

def test_latest_available_premarket_weekday():
    """工作日盘前（<15:00）→ 上一交易日。"""
    now = datetime(2026, 7, 14, 4, 12)  # 周二 04:12
    assert vi._latest_available_trading_day(now) == "2026-07-13"


def test_latest_available_after_market_weekday():
    """工作日盘后（>=15:00）→ 今天。"""
    now = datetime(2026, 7, 14, 16, 0)  # 周二 16:00
    assert vi._latest_available_trading_day(now) == "2026-07-14"


def test_latest_available_at_15_exact():
    """15:00 整点算盘后（>= 15:00 → 今天）。"""
    now = datetime(2026, 7, 14, 15, 0)  # 周二 15:00
    assert vi._latest_available_trading_day(now) == "2026-07-14"


def test_latest_available_weekend():
    """周末 → 上周五。"""
    assert vi._latest_available_trading_day(datetime(2026, 7, 11, 10, 0)) == "2026-07-10"  # Sat
    assert vi._latest_available_trading_day(datetime(2026, 7, 12, 20, 0)) == "2026-07-10"  # Sun


def test_latest_available_monday_premarket():
    """周一盘前 → 上周五（回退周末）。"""
    now = datetime(2026, 7, 13, 4, 0)  # 周一 04:00
    assert vi._latest_available_trading_day(now) == "2026-07-10"


def test_latest_available_default_now():
    """不传 now → 返回字符串（不抛异常）。"""
    result = vi._latest_available_trading_day()
    assert isinstance(result, str)
    assert len(result) == 10  # YYYY-MM-DD


# ---------- verify 收盘感知场景（跨夜 catch-up 核心 bug） ----------

def test_verify_premarket_overnight_passes(tmp_path):
    """用例1 跨夜/盘前：now=周二 04:12, staging max=周一 07-13 → 目标=07-13 → fresh → passed."""
    staging = tmp_path / "staging"
    dates = ["2026-07-09", "2026-07-10", "2026-07-13"]  # max=周一
    for sym in ("sh600000", "sh600004", "sz000001", "sz300001", "sh688001"):
        _write_staging(str(staging), "2026", f"{sym}_daily.parquet", _good_df(dates))
    _write_staging(str(staging), "2026", "bj920000_daily.parquet", _good_df(dates))

    now = datetime(2026, 7, 14, 4, 12)  # 周二 04:12 盘前
    result = vi.verify(str(staging), "2026-07-06", _now=now)

    assert result["latest_trading_day"] == "2026-07-13"
    assert result["fresh_rate"] == 1.0
    assert result["passed"] is True


def test_verify_after_market_fails_if_stale(tmp_path):
    """用例2 盘后：now=周二 16:00, staging max=周一 07-13 → 目标=07-14 → fresh_rate=0 → fail."""
    staging = tmp_path / "staging"
    dates = ["2026-07-09", "2026-07-10", "2026-07-13"]  # max=周一
    for sym in ("sh600000", "sh600004", "sz000001", "sz300001", "sh688001"):
        _write_staging(str(staging), "2026", f"{sym}_daily.parquet", _good_df(dates))
    _write_staging(str(staging), "2026", "bj920000_daily.parquet", _good_df(dates))

    now = datetime(2026, 7, 14, 16, 0)  # 周二 16:00 盘后
    result = vi.verify(str(staging), "2026-07-06", _now=now)

    assert result["latest_trading_day"] == "2026-07-14"
    assert result["fresh_rate"] == 0.0
    assert result["passed"] is False


def test_verify_weekend_passes(tmp_path):
    """用例3 周末：now=周六, staging max=周五 07-10 → 目标=07-10 → fresh → passed."""
    staging = tmp_path / "staging"
    dates = ["2026-07-08", "2026-07-09", "2026-07-10"]  # max=周五
    for sym in ("sh600000", "sh600004", "sz000001", "sz300001", "sh688001"):
        _write_staging(str(staging), "2026", f"{sym}_daily.parquet", _good_df(dates))
    _write_staging(str(staging), "2026", "bj920000_daily.parquet", _good_df(dates))

    now = datetime(2026, 7, 11, 10, 0)  # 周六 10:00
    result = vi.verify(str(staging), "2026-07-06", _now=now)

    assert result["latest_trading_day"] == "2026-07-10"
    assert result["fresh_rate"] == 1.0
    assert result["passed"] is True