path: root/tests/worker/test_data.py

"""Data adapter / instrument-search tests — mocked provider transport, no network."""
import json
import sys
from pathlib import Path

import pandas as pd
import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
sys.path.insert(0, str(Path(__file__).resolve().parent))

import worker.data as data  # noqa: E402
from fixtures_synth import synthetic_daily, SYNTH  # noqa: E402


# ---- live-captured provider reply shapes (see docs/search-fix.md probes) ----

def _em_payload(rows):
    return {"QuotationCodeTable": {"Data": rows, "Status": 0, "Message": "成功",
                                   "TotalCount": len(rows)},
            }


def _em_row(code, name, classify, mktnum, sec_type_name):
    return {"Code": code, "Name": name, "Classify": classify, "MktNum": mktnum,
            "SecurityTypeName": sec_type_name, "MarketType": mktnum,
            "QuoteID": f"{mktnum}.{code}"}


def test_search_stock_real_shape(monkeypatch):
    """q=600000 → real-shape eastmoney AStock row classifies as stock, identity validated."""
    body = _em_payload([_em_row("600000", "浦发银行", "AStock", "1", "沪A")])
    monkeypatch.setattr(data, "_http_json", lambda *a, **k: body)

    def boom(*a, **k):
        raise RuntimeError("tencent not called in this test")
    monkeypatch.setattr(data, "_http_text", boom)
    res = data.search_instruments("600000")
    assert res["status"] == "ready"
    item = res["items"][0]
    assert item["symbol"] == "600000" and item["name"] == "浦发银行"
    assert item["asset_type"] == "stock" and item["market"] == "cn"
    assert item["canonical_symbol"] == "SH#600000"
    assert item["source"] == "eastmoney"


def test_search_etf_real_shape_with_tencent_confirmation(monkeypatch):
    """q=510300: eastmoney Classify=Fund only counts as ETF when tencent confirms."""
    em = _em_payload([_em_row("510300", "沪深300ETF华泰柏瑞", "Fund", "1", "基金")])
    monkeypatch.setattr(data, "_http_json", lambda *a, **k: em)
    tx = 'v_hint="sh~510300~\\u6caa\\u6df1300ETF\\u534e\\u6cf0\\u67cf\\u745e~hs300etfhtbr~ETF"'
    monkeypatch.setattr(data, "_http_text", lambda *a, **k: tx)
    res = data.search_instruments("510300")
    etfs = [i for i in res["items"] if i["symbol"] == "510300"]
    assert etfs and all(i["asset_type"] == "etf" for i in etfs)
    assert {i["source"] for i in etfs} == {"eastmoney", "tencent"}
    assert any(i["canonical_symbol"] == "SH#510300" for i in etfs)


def test_search_etf_ambiguous_fund_excluded_without_tencent_confirmation(monkeypatch):
    """Eastmoney Classify=Fund alone cannot distinguish ETF from LOF (probed):
    without a tencent ETF confirmation the item must not surface, ever."""
    em = _em_payload([_em_row("160706", "沪深300LOF", "Fund", "0", "基金")])
    monkeypatch.setattr(data, "_http_json", lambda *a, **k: em)
    monkeypatch.setattr(
        data, "_http_text",
        lambda *a, **k: 'v_hint="sz~160706~\\u6caa\\u6df1300LOF~hs300lof~LOF"')
    res = data.search_instruments("160706")
    assert res["status"] == "ready"
    assert res["items"] == []  # LOF must never surface as an etf


def test_search_lof_not_present_when_eastmoney_only(monkeypatch):
    em = _em_payload([_em_row("160706", "沪深300LOF", "Fund", "0", "基金")])
    monkeypatch.setattr(data, "_http_json", lambda *a, **k: em)

    def boom(*a, **k):
        raise RuntimeError("tencent down")
    monkeypatch.setattr(data, "_http_text", boom)
    res = data.search_instruments("160706")
    # tencent failed but eastmoney succeeded → still ready; LOF suppressed
    assert res["status"] == "ready"
    assert res["items"] == []
    assert res["providers"]["tencent"].startswith("failed")


def test_search_index_from_chinese_query_and_class_not_code(monkeypatch):
    """q=沪深300 → indexes come from provider class (Classify=Index / ZS tag);
    a numeric code like 000300 is NEVER guessed to be an index by shape."""
    em = _em_payload([
        _em_row("000300", "沪深300", "Index", "1", "指数"),
        _em_row("399300", "沪深300", "Index", "0", "指数"),
        _em_row("510300", "沪深300ETF华泰柏瑞", "Fund", "1", "基金"),
    ])
    monkeypatch.setattr(data, "_http_json", lambda *a, **k: em)
    tx = ('v_hint="sh~000300~\\u6caa\\u6df1300~hs300~ZS^sz~399300~\\u6caa\\u6df1300~hs300~ZS'
          '^sh~510300~\\u6caa\\u6df1300ETF\\u534e\\u6cf0\\u67cf\\u745e~hs300etfhtbr~ETF^'
          'sz~160706~\\u6caa\\u6df1300LOF~hs300lof~LOF"')
    monkeypatch.setattr(data, "_http_text", lambda *a, **k: tx)
    res = data.search_instruments("沪深300")
    by_key = {(i["canonical_symbol"], i["asset_type"], i["source"]) for i in res["items"]}
    assert ("SH#000300", "index", "eastmoney") in by_key
    assert ("SZ#399300", "index", "eastmoney") in by_key
    assert ("SH#000300", "index", "tencent") in by_key
    assert all(i["asset_type"] in ("stock", "etf", "index") for i in res["items"])
    assert not any("160706" == i["symbol"] and i["asset_type"] == "etf" for i in res["items"])


def test_search_numeric_code_not_auto_index(monkeypatch):
    """000300 with a provider stock-ish class (hypothetical) must be honored as
    whatever the provider states — classification is never inferred."""
    em = _em_payload([_em_row("000300", "某标的", "AStock", "0", "深A")])
    monkeypatch.setattr(data, "_http_json", lambda *a, **k: em)
    monkeypatch.setattr(
        data, "_http_text",
        lambda *a, **k: 'v_hint="sz~000300~\\u67d0\\u6807\\u7684~m?~GP-A"')
    res = data.search_instruments("000300")
    assert all(i["asset_type"] == "stock" for i in res["items"])


def test_search_excludes_unsupported_classes_real_rows(monkeypatch):
    """Bond/HK/OTC/LOF rows and tencent LOF tags are dropped, no guessing."""
    em = _em_payload([
        _em_row("160706", "20天津38", "Bond", "1", "债券"),
        _em_row("00700", "腾讯控股", "HK", "116", "港股"),
        _em_row("160706", "嘉实沪深300ETF联接A", "OTCFUND", "0", "场外基金"),
        _em_row("600000", "浦发银行", "AStock", "1", "沪A"),
    ])
    monkeypatch.setattr(data, "_http_json", lambda *a, **k: em)
    monkeypatch.setattr(
        data, "_http_text",
        lambda *a, **k: 'v_hint="sz~160706~\\u6caa\\u6df1300LOF~hs300lof~LOF"')
    res = data.search_instruments("x")
    assert res["status"] == "ready"
    symbols = [i["symbol"] for i in res["items"]]
    assert symbols == ["600000"]  # bond/HK/OTCFUND dropped; tencent LOF tag excluded and eastmoney Fund lacks ETF confirmation
    assert res["items"][0]["asset_type"] == "stock"


def test_search_chinese_name_query_matches_provider(monkeypatch):
    em = _em_payload([_em_row("600000", "浦发银行", "AStock", "1", "沪A")])
    captured = {}
    monkeypatch.setattr(data, "_http_json", lambda url, params, **k: captured.update(params) or em)

    def boom(*a, **k):
        raise RuntimeError("tencent down")
    monkeypatch.setattr(data, "_http_text", boom)
    res = data.search_instruments("浦发银行")
    assert captured["input"] == "浦发银行"  # provider receives the raw CJK query
    assert res["items"][0]["name"] == "浦发银行"


def test_search_provider_failure_is_explicit_failed_status(monkeypatch):
    def boom(*a, **k):
        raise RuntimeError("network down")
    monkeypatch.setattr(data, "_http_json", boom)
    monkeypatch.setattr(data, "_http_text", boom)
    res = data.search_instruments("600000")
    assert res["status"] == "failed"
    assert res["items"] == []
    assert res["error"]["code"] == "providers_unavailable"
    assert "eastmoney" in res["error"]["message"]
    assert "tencent" in res["error"]["message"]


def test_search_provider_timeout_is_explicit_failed_status(monkeypatch):
    import requests as _req
    def timeout(*a, **k):
        raise _req.exceptions.Timeout("read timed out")
    monkeypatch.setattr(data, "_http_json", timeout)
    monkeypatch.setattr(data, "_http_text", timeout)
    res = data.search_instruments("600000")
    assert res["status"] == "failed"
    assert "Timeout" in res["error"]["message"]


def test_search_partial_failure_is_ready_with_warning(monkeypatch):
    def boom(*a, **k):
        raise RuntimeError("tencent down")
    monkeypatch.setattr(
        data, "_http_json",
        lambda *a, **k: _em_payload([_em_row("600000", "浦发银行", "AStock", "1", "沪A")]))
    monkeypatch.setattr(data, "_http_text", boom)
    res = data.search_instruments("600000")
    assert res["status"] == "ready"
    assert res["providers"]["tencent"].startswith("failed")
    assert res["items"][0]["symbol"] == "600000"


def test_suggest_limits_are_bounded():
    import inspect
    src = inspect.getsource(data)
    assert "SUGGEST_TIMEOUT_SECS = 4.0" in src  # per-source timeout applies to the actual calls
    # the catalog paths are gone: every query must be a bounded provider call
    assert "_stock_catalog" not in src
    assert "_catalog_frame" not in src


def test_limit_is_passed_to_provider_bounded(monkeypatch):
    seen = {}
    rows = [_em_row(str(600000 + i), f"n{i}", "AStock", "1", "沪A") for i in range(8)]
    monkeypatch.setattr(data, "_http_json",
                        lambda url, params, **k: seen.update(params) or _em_payload(rows[:params["count"]]))
    monkeypatch.setattr(data, "_http_text", lambda *a, **k: "")
    res = data.search_instruments("600", limit=4)
    assert seen["count"] == 5  # bounded floor 5, well under the old full catalogs
    assert len(res["items"]) <= 4


def test_fetch_rejects_unknown_asset_type():
    with pytest.raises(data.DataError):
        data.fetch_source({"symbol": "X#1", "market": "cn", "asset_type": "crypto"},
                          "2024-01-01", "2024-02-01", "daily", "none", ["open", "close"])


# ---- sina listed-ETF adapter (recovery-01: eastmoney down, tencent has no
# ETF adapter; live evidence in docs/recovery-01-plan.md). The stock-like frame
# below is a SYNTHETIC test fixture mimicking the live-captured sina reply shape. ----

def _sina_synth_frame():
    """Synthetic fixture in the exact live-captured shape of
    ak.fund_etf_hist_sina(symbol='sz159399') (recovered live 2026-09-17:
    381 rows 2025-02-27..2026-09-16, columns date/open/.../amount/postVol/postAmt)."""
    rows = [
        {"date": "2025-12-30", "open": 1.001, "high": 1.006, "low": 0.999,
         "close": 1.006, "volume": 695982432, "amount": 696910071.0,
         "postVol": float("nan"), "postAmt": float("nan")},
        {"date": "2026-01-02", "open": 1.007, "high": 1.008, "low": 0.994,
         "close": 0.997, "volume": 505082425, "amount": 506417752.0,
         "postVol": float("nan"), "postAmt": float("nan")},
        {"date": "2026-09-16", "open": 1.001, "high": 1.001, "low": 0.982,
         "close": 0.992, "volume": 144526400, "amount": 142914764.0,
         "postVol": 300.0, "postAmt": 298.0},
    ]
    df = pd.DataFrame(rows)
    df.attrs["synthetic"] = True
    return df


def test_sina_etf_serves_159399_with_honest_provenance(monkeypatch):
    inst = {"symbol": "159399", "market": "cn", "asset_type": "etf"}
    monkeypatch.setattr(data, "ak",
                        type("M", (), {"fund_etf_hist_sina":
                                       staticmethod(lambda **kw: _sina_synth_frame())})())
    res = data.fetch_source(inst, "2025-12-31", "2026-09-17", "daily", "none")
    df, endpoint, params = res
    assert res.provider == "sina"
    assert endpoint == "fund_etf_hist_sina"
    assert params == {"symbol": "sz159399"}
    assert set(df["symbol"]) == {"SZ#159399"}
    dates = df["date"].tolist()
    assert dates[0] >= "2025-12-31" and dates[-1] <= "2026-09-17"
    assert "2026-09-17" not in dates  # no fabricated end-date bar
    warns = " ".join(df.attrs.get("source_warnings", []))
    assert "股" in warns and "unadjusted" in warns
    assert "159399" not in warns  # numbers only, identity preserved


def test_sina_rejects_adjustment_and_range_semantics(monkeypatch):
    inst = {"symbol": "159399", "market": "cn", "asset_type": "etf"}
    monkeypatch.setattr(data, "ak",
                        type("M", (), {"fund_etf_hist_sina":
                                       staticmethod(lambda **kw: _sina_synth_frame())})())
    with pytest.raises(data.DataError) as e:
        data.fetch_source(inst, "2025-12-31", "2026-09-17", "daily", "qfq", source="sina")
    assert e.value.code == "unsupported_adjustment"


def test_sina_rejects_stock_and_index():
    for asset in ("stock", "index"):
        inst = {"symbol": "600000" if asset == "stock" else "000300",
                "market": "cn", "asset_type": asset}
        with pytest.raises(data.DataError) as e:
            data.fetch_source(inst, "2025-12-31", "2026-09-17", "daily", "none",
                              source="sina")
        assert e.value.code == "source_unavailable"


def test_auto_etf_falls_back_from_eastmoney_to_sina_honestly(monkeypatch):
    inst = {"symbol": "159399", "market": "cn", "asset_type": "etf"}

    def dis(*a, **k):
        raise ConnectionError("RemoteDisconnected('Remote end closed connection')")
    monkeypatch.setattr(data, "ak",
                        type("M", (), {"fund_etf_hist_em": staticmethod(dis),
                                       "fund_etf_hist_sina":
                                       staticmethod(lambda **kw: _sina_synth_frame())})())
    res, warns = data.fetch_source_with_warnings(
        inst, "2025-12-31", "2026-09-17", "daily", "none", source="auto")
    assert res.provider == "sina"  # actual serving provider, never disguised
    joined = " ".join(warns)
    assert "provider_fallback" in joined and "eastmoney" in joined and "sina" in joined
    assert set(res.df["symbol"]) == {"SZ#159399"}


def test_auto_etf_fails_when_no_provider_is_viable(monkeypatch):
    inst = {"symbol": "159399", "market": "cn", "asset_type": "etf"}

    def dis(*a, **k):
        raise ConnectionError("RemoteDisconnected")
    monkeypatch.setattr(data, "ak",
                        type("M", (), {"fund_etf_hist_em": staticmethod(dis),
                                       "fund_etf_hist_sina": staticmethod(dis)})())
    with pytest.raises(data.DataError) as e:
        data.fetch_source(inst, "2025-12-31", "2026-09-17", "daily", "none", source="auto")
    assert e.value.code == "provider_unavailable"
    assert "eastmoney" in str(e.value) and "sina" in str(e.value)


def test_tencent_etf_rejection_message_unchanged():
    inst = {"symbol": "159399", "market": "cn", "asset_type": "etf"}
    with pytest.raises(data.DataError) as e:
        data.fetch_source(inst, "2025-12-31", "2026-09-17", "daily", "none",
                          source="tencent")
    assert e.value.code == "source_unavailable"
    assert "tencent source has no listed-ETF daily adapter" in str(e.value)


def test_sina_registered_in_explicit_source_surface():
    assert "sina" in data.SUPPORTED_SOURCES


def test_fetch_columns_subset_requested():
    req = {"symbol": "600000", "market": "cn", "asset_type": "stock"}
    called = {}

    def fake_hist(symbol, **kw):
        called["params"] = kw
        return synthetic_daily("SH#600000")
    monkeypatch = pytest.MonkeyPatch()
    monkeypatch.setattr(data, "ak", type("M", (), {"stock_zh_a_hist": staticmethod(fake_hist)})())
    df, endpoint, params = data.fetch_source(req, "2024-01-01", "2024-02-01",
                                             "daily", "none", ["open", "close", "volume"])
    assert endpoint == "stock_zh_a_hist"
    assert all(c in df.columns for c in ("date", "symbol", "open", "close", "volume"))
    monkeypatch.undo()


# ---- identity contract: the server persists instruments with the user-facing
# market field (SH/SZ/BJ) exactly as the frontend manual-entry form sends them
# (159399 QA regression 2026-09-17: market:"SZ" used to fail with
# unsupported_market before any network attempt). Canonical "SH#600000" and
# market:"cn" bare-code inference remain the primary, unchanged contracts. ----

def test_split_identity_accepts_bare_code_with_market_sz_matches_device_inference():
    exch, code = data.split_identity({"symbol": "159399", "market": "SZ"})
    assert (exch, code) == ("SZ", "159399")


def test_split_identity_accepts_bare_code_with_market_sh():
    exch, code = data.split_identity({"symbol": "600000", "market": "SH"})
    assert (exch, code) == ("SH", "600000")


def test_split_identity_accepts_bare_code_with_market_cn_inference_unchanged():
    exch, code = data.split_identity({"symbol": "159399", "market": "cn"})
    assert (exch, code) == ("SZ", "159399")


def test_etf_fetch_passes_market_sz_instrument_to_provider(monkeypatch):
    """Full fetch_source identity path: market:"SZ" (as stored by the server
    from the frontend manual-entry form) must reach the sina provider."""
    captured = {}
    real_frame = _sina_synth_frame()

    def fake_em(inst, s, e, f, adj):
        raise data.DataError("source_unavailable", "eastmoney down (test)")

    def fake_sina(inst, s, e, f, adj):
        captured["instrument"] = inst
        return data.FetchResult(real_frame.copy(), "fund_etf_hist_sina",
                                {"symbol": "sz159399"}, real_frame.copy(), "sina")

    monkeypatch.setattr(data, "_fetch_eastmoney", fake_em)
    monkeypatch.setattr(data, "_fetch_sina", fake_sina)
    inst = {"symbol": "159399", "market": "SZ", "asset_type": "etf", "name": "国泰自由现金流"}
    res, warns = data.fetch_source_with_warnings(inst, "2025-12-31", "2026-09-17",
                                                 "daily", "none",
                                                 ["open", "high", "low", "close", "volume"],
                                                 source="auto")
    assert captured["instrument"] is inst
    assert res.provider == "sina"
    assert any(w.startswith("provider_fallback") for w in warns)


def test_split_identity_rejects_unknown_market_prefixes():
    with pytest.raises(data.DataError):
        data.split_identity({"symbol": "1", "market": "adlhkj"})