"""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()