diff --git a/.env.example b/.env.example index 310686c..efbd904 100644 --- a/.env.example +++ b/.env.example @@ -12,9 +12,9 @@ TUSHARE_TOKEN=your_tushare_token_here DATAHUB_BASE_URL=http://127.0.0.1:8766 DATAHUB_TOKEN= -# Optional iFinD HTTP credential. The backend exchanges it for a short-lived -# access token and never exposes either token to browsers. -IFIND_REFRESH_TOKEN=your_ifind_refresh_token_here +# iFinD credentials live on xiaobai-datahub, not the website process. +# IFIND_REFRESH_TOKEN=your_ifind_refresh_token_here +# IFIND_ACCESS_TOKEN= # Initial platform member models (OpenAI-compatible). After first launch these # are encrypted into system settings and used only by admins and active members. diff --git a/backend/bootstrap/container.py b/backend/bootstrap/container.py index c872753..fd43561 100644 --- a/backend/bootstrap/container.py +++ b/backend/bootstrap/container.py @@ -13,8 +13,8 @@ from backend.features.screener.engine import ScreenerEngine from backend.features.screener.tracking import StrategyTrackingService from backend.jobs import InProcessJobRunner, JobRegistry, SQLiteJobRunRepository from database import ReviewDatabase -from backend.data.providers.ifind_client import IfindHttpClient -from backend.data.realtime import WebRealtimeAggregator +from backend.data.datahub.ifind_proxy import HubIfindProxy +from backend.data.datahub.realtime_proxy import HubRealtimeProxy from backend.features.market.charts import MarketChartClient @@ -23,13 +23,13 @@ class ApplicationContainer: database: ReviewDatabase repositories: RepositoryBundle data_gateway: DataGateway - ifind: IfindHttpClient + ifind: HubIfindProxy screener: ScreenerEngine strategy_tracking: StrategyTrackingService alert_service: AlertService trade_journal: TradeJournalService mentor_skills: MentorSkillRegistry - realtime_aggregator: WebRealtimeAggregator + realtime_aggregator: HubRealtimeProxy chart_data: MarketChartClient jobs: InProcessJobRunner diff --git a/backend/data/datahub/ifind_proxy.py b/backend/data/datahub/ifind_proxy.py new file mode 100644 index 0000000..2b1cbc1 --- /dev/null +++ b/backend/data/datahub/ifind_proxy.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import time +from typing import Any + +from backend.data.datahub.bridge import DatahubBridge +from backend.data.datahub.errors import DatahubError +from backend.data.providers.ifind_client import IfindError + + +class HubIfindProxy: + """Website-facing iFinD facade. Talks only to xiaobai-datahub.""" + + def __init__(self, datahub: DatahubBridge) -> None: + self._datahub = datahub + self._status: dict[str, Any] | None = None + self._status_at = 0.0 + + @property + def configured(self) -> bool: + return bool(self.status().get("configured")) + + def set_credentials(self, refresh_token: str, access_token: str = "") -> None: + self._status = None + self._status_at = 0.0 + + def status(self) -> dict[str, Any]: + now = time.monotonic() + if self._status is not None and now - self._status_at < 30: + return dict(self._status) + fallback = {"configured": False, "access_ready": False, "access_expires_at": ""} + if not self._datahub.settings.token: + self._status = fallback + self._status_at = now + return dict(fallback) + try: + rows = self._rows("ifind_status", {}) + except IfindError: + self._status = fallback + self._status_at = now + return dict(fallback) + row = rows[0] if rows else {} + status = { + "configured": bool(row.get("configured")), + "access_ready": bool(row.get("access_ready")), + "access_expires_at": str(row.get("access_expires_at") or ""), + } + self._status = status + self._status_at = now + return dict(status) + + def wencai(self, query: str, search_type: str = "stock", cache_ttl: int = 300) -> list[dict[str, Any]]: + return self._rows( + "ifind_wencai", + {"query": query, "search_type": search_type, "cache_ttl": cache_ttl}, + ) + + def snapshots( + self, + codes: str | list[str], + indicators: list[str], + start_time: str, + end_time: str, + cache_ttl: int = 8, + ) -> list[dict[str, Any]]: + return self._rows( + "ifind_snapshots", + { + "codes": codes, + "indicators": indicators, + "start_time": start_time, + "end_time": end_time, + "cache_ttl": cache_ttl, + }, + ) + + def history( + self, + codes: str | list[str], + indicators: list[str], + start_date: str, + end_date: str, + cache_ttl: int = 300, + ) -> list[dict[str, Any]]: + return self._rows( + "ifind_history", + { + "codes": codes, + "indicators": indicators, + "start_date": start_date, + "end_date": end_date, + "cache_ttl": cache_ttl, + }, + ) + + def real_time( + self, + codes: str | list[str], + indicators: list[str], + cache_ttl: int = 10, + ) -> list[dict[str, Any]]: + return self._rows( + "ifind_realtime", + {"codes": codes, "indicators": indicators, "cache_ttl": cache_ttl}, + ) + + def intraday( + self, + code: str, + start_time: str, + end_time: str, + cache_ttl: int = 20, + ) -> list[dict[str, Any]]: + return self._rows( + "ifind_intraday", + { + "code": code, + "start_time": start_time, + "end_time": end_time, + "cache_ttl": cache_ttl, + }, + ) + + def test_connection(self) -> dict[str, Any]: + payload = self.real_time( + "000001.SH", + ["open", "high", "low", "latest", "preClose"], + cache_ttl=0, + ) + return { + "ok": bool(payload), + "sample_time": str(payload[0].get("time") or "") if payload else "", + } + + def _rows(self, api_name: str, params: dict[str, Any]) -> list[dict[str, Any]]: + try: + response = self._datahub.client.query_api(api_name, params) + except DatahubError as exc: + raise IfindError(str(exc) or "iFinD 数据中枢暂不可用") from exc + data = response.data + if isinstance(data, list): + return [dict(item) for item in data if isinstance(item, dict)] + if isinstance(data, dict): + return [dict(data)] + return [] diff --git a/backend/data/datahub/realtime_proxy.py b/backend/data/datahub/realtime_proxy.py new file mode 100644 index 0000000..c5aa85e --- /dev/null +++ b/backend/data/datahub/realtime_proxy.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from backend.data.datahub.bridge import DatahubBridge +from backend.data.realtime import RealtimeAggregateError + + +class HubRealtimeProxy: + """Realtime observation facade. Talks only to xiaobai-datahub.""" + + def __init__(self, datahub: DatahubBridge) -> None: + self._datahub = datahub + + def health_snapshot(self, sector: str = "") -> dict[str, Any]: + started = datetime.now().astimezone() + indices: list[dict[str, Any]] = [] + error = "" + try: + indices = self.tencent_indices() + except RealtimeAggregateError as exc: + error = str(exc) + epochs = [int(item.get("quote_time_epoch") or 0) for item in indices] + now = datetime.now().astimezone() + max_skew = 120 if now.hour >= 15 else 15 + index_consistent = bool(epochs) and max(epochs) - min(epochs) <= max_skew + ready = len(indices) == 3 and index_consistent + return { + "ready": ready, + "isolated": True, + "generated_at": started.isoformat(timespec="seconds"), + "elapsed_ms": 0, + "indices": indices, + "index_consistent": index_consistent, + "sector": None, + "sources": { + "datahub_indices": { + "ok": ready, + "error": error, + "source": "datahub", + } + }, + "observations": {}, + "policy": { + "integration": "datahub_exclusive", + "max_index_time_skew_seconds": max_skew, + "notice": "实时观察只走数据中枢，主网站不再直连东财/腾讯。", + }, + } + + def tencent_indices(self) -> list[dict[str, Any]]: + rows = self._datahub.try_index_quotes() or [] + result = [_as_index(item) for item in rows if _as_index(item)] + wanted = {"000001", "399001", "399006"} + result = [item for item in result if item.get("code") in wanted] + result.sort(key=lambda item: str(item.get("code") or "")) + if len(result) != 3: + raise RealtimeAggregateError(f"datahub returned {len(result)}/3 indices") + return result + + def eastmoney_indices(self) -> list[dict[str, Any]]: + return self.tencent_indices() + + def tencent_stock_quote(self, code: str, expected_date: str = "") -> dict[str, Any]: + return self._stock_quote(code, expected_date) + + def eastmoney_stock_quote(self, code: str, expected_date: str = "") -> dict[str, Any]: + return self._stock_quote(code, expected_date) + + def tencent_stock_quotes( + self, + codes: list[str], + expected_date: str = "", + minimum: int | None = None, + ) -> list[dict[str, Any]]: + return self._stock_quotes(codes, expected_date, minimum) + + def eastmoney_stock_quotes( + self, + codes: list[str], + expected_date: str = "", + ) -> list[dict[str, Any]]: + return self._stock_quotes(codes, expected_date, None) + + def eastmoney_shenwan_quote(self, ts_code: str, expected_date: str = "") -> dict[str, Any]: + quote = self._datahub.try_sector_quote(ts_code, expected_date) + if not quote: + raise RealtimeAggregateError(f"datahub shenwan quote unavailable for {ts_code}") + return quote + + def _stock_quote(self, code: str, expected_date: str) -> dict[str, Any]: + rows = self._stock_quotes([code], expected_date, 1) + if not rows: + raise RealtimeAggregateError(f"datahub stock quote unavailable for {code}") + return rows[0] + + def _stock_quotes( + self, + codes: list[str], + expected_date: str, + minimum: int | None, + ) -> list[dict[str, Any]]: + cleaned = [str(item or "").strip() for item in codes if str(item or "").strip()] + rows = self._datahub.try_quotes(cleaned) if cleaned else (self._datahub.try_market_quotes(expected_date) or []) + quotes = [_as_stock(item) for item in (rows or []) if _as_stock(item)] + if expected_date: + compact = str(expected_date).replace("-", "") + quotes = [ + item + for item in quotes + if not item.get("quote_date") or str(item.get("quote_date") or "").replace("-", "") == compact + ] + if minimum is not None and len(quotes) < minimum: + raise RealtimeAggregateError(f"datahub returned {len(quotes)} quotes, need {minimum}") + return quotes + + +def _as_index(row: dict[str, Any]) -> dict[str, Any] | None: + code = str(row.get("code") or str(row.get("ts_code") or "").split(".")[0] or "") + price = _number(row.get("price") if row.get("price") not in (None, "") else row.get("close")) + if not code or price <= 0: + return None + epoch = int(_number(row.get("quote_time_epoch"))) + amount = _number(row.get("amount_billion")) + if amount <= 0: + amount = round(_number(row.get("amount")) / 100_000_000, 2) + return { + "code": code, + "name": row.get("name") or code, + "price": price, + "change": _number(row.get("change") if row.get("change") not in (None, "") else row.get("pct_chg")), + "change_amount": _number(row.get("change_amount")), + "open": _number(row.get("open")), + "high": _number(row.get("high")), + "low": _number(row.get("low")), + "previous_close": _number( + row.get("previous_close") if row.get("previous_close") not in (None, "") else row.get("pre_close") + ), + "amount_billion": amount, + "quote_time_epoch": epoch, + "quote_time": str(row.get("quote_time") or ""), + "source": str(row.get("source") or "datahub"), + "cache_age_seconds": 0, + } + + +def _as_stock(row: dict[str, Any]) -> dict[str, Any] | None: + close = _number(row.get("close") if row.get("close") not in (None, "") else row.get("price")) + if close <= 0: + return None + ts_code = str(row.get("ts_code") or "") + code = str(row.get("code") or ts_code.split(".")[0] or "") + return { + "ts_code": ts_code or code, + "code": code, + "name": row.get("name") or "", + "close": close, + "pre_close": _number( + row.get("pre_close") if row.get("pre_close") not in (None, "") else row.get("previous_close") + ), + "open": _number(row.get("open")), + "high": _number(row.get("high")), + "low": _number(row.get("low")), + "volume": _number(row.get("volume") if row.get("volume") not in (None, "") else row.get("vol")), + "vol": _number(row.get("vol") if row.get("vol") not in (None, "") else row.get("volume")), + "amount": _number(row.get("amount")), + "quote_time_epoch": int(_number(row.get("quote_time_epoch"))), + "quote_time": str(row.get("quote_time") or ""), + "quote_date": str(row.get("quote_date") or ""), + "source": str(row.get("source") or "datahub"), + "delayed": bool(row.get("delayed")), + } + + +def _number(value: Any) -> float: + try: + return float(value or 0) + except (TypeError, ValueError): + return 0.0 diff --git a/backend/data/gateway.py b/backend/data/gateway.py index 8181655..4c8d343 100644 --- a/backend/data/gateway.py +++ b/backend/data/gateway.py @@ -7,13 +7,13 @@ from typing import Any from backend.data.contracts import DataUsage from backend.data.datahub import DatahubAwareTushareClient, DatahubBridge, DatahubClient, DatahubSettings +from backend.data.datahub.ifind_proxy import HubIfindProxy +from backend.data.datahub.realtime_proxy import HubRealtimeProxy from backend.data.policy import DataSourcePolicy from backend.data.providers import IfindProvider, TushareProvider from backend.data.quality import DataQualityGate, QualityEvidence, QualityReport -from backend.data.providers.ifind_client import IfindHttpClient from backend.data.providers.tushare_client import TushareClient -from backend.data.realtime import WebRealtimeAggregator -from backend.features.market.charts import EastmoneyChartClient, MarketChartClient +from backend.features.market.charts import MarketChartClient @dataclass(frozen=True) @@ -23,11 +23,11 @@ class DataGateway: tushare_provider: TushareProvider ifind_provider: IfindProvider chart_data: MarketChartClient - realtime_observer: WebRealtimeAggregator + realtime_observer: HubRealtimeProxy datahub: DatahubBridge @property - def ifind(self) -> IfindHttpClient: + def ifind(self) -> HubIfindProxy: return self.ifind_provider.client def tushare( @@ -102,10 +102,6 @@ def build_data_gateway( tushare_token_supplier: Callable[[], str] | None = None, datahub_settings: DatahubSettings | None = None, ) -> DataGateway: - ifind = IfindHttpClient( - str(credentials.get("ifind_refresh_token") or ""), - str(credentials.get("ifind_access_token") or ""), - ) token_supplier = tushare_token_supplier or ( lambda: str(credentials.get("tushare_token") or "") ) @@ -113,12 +109,13 @@ def build_data_gateway( settings = datahub_settings or DatahubSettings.load(credentials=credentials) datahub_client = DatahubClient(settings) datahub = DatahubBridge(settings, datahub_client) + ifind = HubIfindProxy(datahub) return DataGateway( policy=policy, quality=DataQualityGate.load(policy), tushare_provider=TushareProvider(token_supplier), ifind_provider=IfindProvider(ifind), - chart_data=MarketChartClient(ifind, EastmoneyChartClient(), datahub), - realtime_observer=WebRealtimeAggregator(), + chart_data=MarketChartClient(datahub), + realtime_observer=HubRealtimeProxy(datahub), datahub=datahub, ) diff --git a/backend/data/providers/ifind.py b/backend/data/providers/ifind.py index b663416..dd34672 100644 --- a/backend/data/providers/ifind.py +++ b/backend/data/providers/ifind.py @@ -1,11 +1,13 @@ from __future__ import annotations -from backend.data.providers.ifind_client import IfindHttpClient +from typing import Any class IfindProvider: - def __init__(self, client: IfindHttpClient) -> None: + def __init__(self, client: Any) -> None: self.client = client def set_credentials(self, refresh_token: str, access_token: str = "") -> None: - self.client.set_credentials(refresh_token, access_token) + setter = getattr(self.client, "set_credentials", None) + if callable(setter): + setter(refresh_token, access_token) diff --git a/backend/features/market/charts.py b/backend/features/market/charts.py index d67757c..ad47547 100644 --- a/backend/features/market/charts.py +++ b/backend/features/market/charts.py @@ -14,7 +14,7 @@ from threading import Lock from typing import Any, ClassVar from backend.bootstrap.config import tushare_code as _stock_market_code -from backend.data.providers.ifind_client import IfindError, IfindHttpClient +from backend.data.providers.ifind_client import IfindError LOGGER = logging.getLogger("xiaobai.charts") @@ -42,17 +42,12 @@ INDEX_SECIDS = { class MarketChartClient: - """Prefer iFinD for display charts and retain Eastmoney as a last resort.""" + """Display charts are served by the data hub only.""" - def __init__( - self, - ifind: IfindHttpClient, - fallback: "EastmoneyChartClient", - datahub: Any = None, - ) -> None: - self.ifind = ifind - self.fallback = fallback + def __init__(self, datahub: Any = None) -> None: self.datahub = datahub + self.ifind = None + self.fallback = None def stock_intraday(self, code: str) -> dict[str, Any]: normalized = str(code or "").strip() @@ -235,7 +230,7 @@ class MarketChartClient: identifier: str, name: str = "", ) -> dict[str, Any]: - if not self.ifind.configured: + if not self.ifind or not self.ifind.configured: raise ChartDataError("iFinD is not configured") now = datetime.now().astimezone() rows: list[dict[str, Any]] = [] @@ -272,7 +267,7 @@ class MarketChartClient: def _ifind_daily( self, ifind_code: str, end_date: str, limit: int ) -> list[dict[str, Any]]: - if not self.ifind.configured: + if not self.ifind or not self.ifind.configured: raise ChartDataError("iFinD is not configured") compact_end = str(end_date or "").replace("-", "") if not re.fullmatch(r"\d{8}", compact_end): @@ -377,6 +372,8 @@ class MarketChartClient: return normalized[-max(1, int(limit)):] def _previous_close(self, code: str, trade_date: str, fallback: float) -> float: + if not self.ifind: + return fallback today = datetime.now().astimezone().date().isoformat() if trade_date == today: try: diff --git a/config/architecture-inventory.json b/config/architecture-inventory.json index 7ce04ab..5757de6 100644 --- a/config/architecture-inventory.json +++ b/config/architecture-inventory.json @@ -207,27 +207,22 @@ { "provider": "datahub", "path": "backend/data/datahub/client.py", - "runtime_role": "optional official EOD read path behind per-dataset flags" + "runtime_role": "website-only read path; official EOD, live quotes, and licensed iFinD" }, { "provider": "ifind", - "path": "backend/data/providers/ifind_client.py", - "runtime_role": "realtime, charts, snapshots, enrichment" + "path": "xiaobai-datahub/datahub/adapters/ifind.py", + "runtime_role": "licensed iFinD source inside the data hub" }, { "provider": "eastmoney", - "path": "backend/features/market/charts.py", - "runtime_role": "display chart fallback" - }, - { - "provider": "eastmoney", - "path": "backend/data/realtime.py", - "runtime_role": "isolated realtime observation and intraday dashboard fallback" + "path": "xiaobai-datahub/datahub/adapters/eastmoney.py", + "runtime_role": "free realtime quotes and shenwan inside the data hub" }, { "provider": "tencent", - "path": "backend/data/realtime.py", - "runtime_role": "index observation and intraday quote fallback" + "path": "xiaobai-datahub/datahub/adapters/tencent.py", + "runtime_role": "free index and stock quotes inside the data hub" } ], "provider_domains": [ @@ -296,16 +291,16 @@ "owner": "backend/data/gateway.py" }, { - "client": "IfindHttpClient", + "client": "HubIfindProxy", + "owner": "backend/data/gateway.py" + }, + { + "client": "HubRealtimeProxy", "owner": "backend/data/gateway.py" }, { "client": "MarketChartClient", "owner": "backend/data/gateway.py" - }, - { - "client": "WebRealtimeAggregator", - "owner": "backend/data/gateway.py" } ], "heaven_service_owners": { diff --git a/tests/test_bootstrap_container.py b/tests/test_bootstrap_container.py index 032e7c6..667ba0a 100644 --- a/tests/test_bootstrap_container.py +++ b/tests/test_bootstrap_container.py @@ -45,8 +45,9 @@ class BootstrapContainerTests(unittest.TestCase): self.assertIs(container.strategy_tracking.repository.database, database) self.assertIs(container.alert_service.repository.database, database) self.assertIs(container.trade_journal.repository.database, database) - self.assertIs(container.chart_data.ifind, container.ifind) - self.assertTrue(container.ifind.configured) + self.assertIs(container.ifind, container.data_gateway.ifind) + self.assertIs(container.chart_data.datahub, container.data_gateway.datahub) + self.assertIsNone(container.chart_data.ifind) if __name__ == "__main__": diff --git a/tests/test_chart_data_provider.py b/tests/test_chart_data_provider.py index e25e8e4..d568afd 100644 --- a/tests/test_chart_data_provider.py +++ b/tests/test_chart_data_provider.py @@ -2,7 +2,6 @@ from __future__ import annotations import unittest -from backend.data.providers.ifind_client import IfindHttpClient from backend.features.market.charts import ChartDataError, EastmoneyChartClient, HIS_TRENDS_URL, MarketChartClient, TRENDS_URL from server import DashboardService @@ -196,7 +195,7 @@ class DatahubChartFallbackTests(unittest.TestCase): } ) fallback = LookbackChartClient() - client = MarketChartClient(IfindHttpClient(), fallback, hub) + client = MarketChartClient(hub) payload = client.stock_intraday("601318") self.assertEqual(payload["source"], "datahub") self.assertEqual(hub.calls, ["601318"]) @@ -212,7 +211,7 @@ class DatahubChartFallbackTests(unittest.TestCase): ): EastmoneyChartClient._cache.clear() fallback.requests.clear() - client = MarketChartClient(IfindHttpClient(), fallback, hub) + client = MarketChartClient(hub) with self.assertRaises(ChartDataError): client.stock_intraday("000001") self.assertEqual(fallback.requests, []) @@ -231,7 +230,7 @@ class DatahubChartFallbackTests(unittest.TestCase): } ] ) - client = MarketChartClient(IfindHttpClient(), LookbackChartClient(), hub) + client = MarketChartClient(hub) rows = client.stock_daily("600000", "20260907") self.assertEqual(rows[-1]["trade_date"], "2026-09-07") self.assertIn("daily:600000", hub.calls) diff --git a/tests/test_data_gateway.py b/tests/test_data_gateway.py index f3fcbb0..908b2a9 100644 --- a/tests/test_data_gateway.py +++ b/tests/test_data_gateway.py @@ -35,7 +35,7 @@ class DataGatewayTests(unittest.TestCase): with self.assertRaises(DataPolicyError): policy.assert_allowed("market.level2", "unresolved", "display") - def test_gateway_uses_live_token_supplier_and_shared_ifind(self) -> None: + def test_gateway_uses_live_token_supplier_and_hub_proxies(self) -> None: token = {"value": "first"} gateway = build_data_gateway( {"ifind_refresh_token": "refresh", "ifind_access_token": "access"}, @@ -44,7 +44,14 @@ class DataGatewayTests(unittest.TestCase): self.assertEqual(gateway.tushare().token, "first") token["value"] = "second" self.assertEqual(gateway.tushare().token, "second") - self.assertIs(gateway.chart_data.ifind, gateway.ifind) + self.assertIs(gateway.ifind, gateway.ifind_provider.client) + self.assertIs(gateway.chart_data.datahub, gateway.datahub) + self.assertIsNone(gateway.chart_data.ifind) + from backend.data.datahub.ifind_proxy import HubIfindProxy + from backend.data.datahub.realtime_proxy import HubRealtimeProxy + + self.assertIsInstance(gateway.ifind, HubIfindProxy) + self.assertIsInstance(gateway.realtime_observer, HubRealtimeProxy) def test_server_has_no_direct_runtime_tushare_construction(self) -> None: source = ( @@ -60,18 +67,23 @@ class DataGatewayTests(unittest.TestCase): def test_provider_construction_has_unique_declared_owners(self) -> None: root = Path(__file__).resolve().parents[1] owners = { - "EastmoneyChartClient": {"backend/data/gateway.py"}, - "IfindHttpClient": {"backend/data/gateway.py"}, "IfindProvider": {"backend/data/gateway.py"}, "MarketChartClient": {"backend/data/gateway.py"}, "TushareClient": {"backend/features/market/service.py"}, "TushareProvider": {"backend/data/gateway.py"}, - "WebRealtimeAggregator": {"backend/data/gateway.py"}, "DatahubClient": {"backend/data/gateway.py"}, "DatahubAwareTushareClient": {"backend/data/gateway.py"}, "DatahubBridge": {"backend/data/gateway.py"}, + "HubIfindProxy": {"backend/data/gateway.py"}, + "HubRealtimeProxy": {"backend/data/gateway.py"}, } found = {name: set() for name in owners} + forbidden = { + "IfindHttpClient": set(), + "EastmoneyChartClient": set(), + "WebRealtimeAggregator": set(), + } + found_forbidden = {name: set() for name in forbidden} for path in (root / "backend").rglob("*.py"): relative = path.relative_to(root).as_posix() tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) @@ -81,7 +93,10 @@ class DataGatewayTests(unittest.TestCase): name = getattr(node.func, "id", None) or getattr(node.func, "attr", None) if name in found: found[name].add(relative) + if name in found_forbidden: + found_forbidden[name].add(relative) self.assertEqual(found, owners) + self.assertEqual(found_forbidden, forbidden) provider_source = (root / "backend/data/providers/tushare.py").read_text( encoding="utf-8" ) diff --git a/tests/test_hub_exclusive.py b/tests/test_hub_exclusive.py index 6c32f91..0adc23a 100644 --- a/tests/test_hub_exclusive.py +++ b/tests/test_hub_exclusive.py @@ -1,14 +1,19 @@ from __future__ import annotations import ast +import json import unittest from pathlib import Path from unittest.mock import patch +from backend.data import build_data_gateway from backend.data.datahub.bridge import DatahubAwareTushareClient, DatahubBridge from backend.data.datahub.client import DatahubClient +from backend.data.datahub.ifind_proxy import HubIfindProxy +from backend.data.datahub.realtime_proxy import HubRealtimeProxy from backend.data.datahub.settings import DATASETS, DatahubSettings, DatasetFlags from backend.data.providers.tushare_transport import TushareError +from backend.features.market.charts import MarketChartClient from tests.test_datahub_bridge import FakeClient, FakeLegacy, flags @@ -23,7 +28,174 @@ BLOCKED_HOSTS = ( "hq.sinajs.cn", "10jqka.com.cn", "xuangubao.cn", + "quantapi.51ifind.com", + "51ifind.com", ) +LEFTOVER_WEBSITE_FILES = { + "backend/data/providers/ifind_client.py", + "backend/data/realtime.py", + "backend/features/market/charts.py", + "backend/data/providers/tushare_transport.py", +} +HUB_BASE = "http://127.0.0.1:8766" + + +def _enabled_settings() -> DatahubSettings: + return DatahubSettings( + base_url=HUB_BASE, + token="hub-token", + datasets={name: DatasetFlags(name, read=True) for name in DATASETS}, + ) + + +class _Resp: + def __init__(self, payload: dict) -> None: + self.status = 200 + self._raw = json.dumps(payload).encode("utf-8") + + def read(self): + return self._raw + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + +def hub_payload(request) -> dict: + url = str(getattr(request, "full_url", None) or request) + if any(host in url for host in BLOCKED_HOSTS): + raise AssertionError(f"website opened blocked host: {url}") + if HUB_BASE not in url: + raise AssertionError(f"unexpected url: {url}") + path = url.split(HUB_BASE, 1)[1].split("?", 1)[0] + if path == "/v1/bars/daily": + return { + "schema_version": 1, + "data": [ + { + "ts_code": "600000.SH", + "trade_date": "20240902", + "open": 10.0, + "high": 10.4, + "low": 9.9, + "close": 10.2, + "volume": 100000, + "amount": 2000000, + } + ], + "meta": {"stale": False, "staleness_seconds": 0, "source": "datahub"}, + } + if path == "/v1/quotes/latest": + return { + "schema_version": 1, + "data": [ + { + "ts_code": "600000.SH", + "code": "600000", + "name": "浦发银行", + "close": 10.2, + "price": 10.2, + "pre_close": 10.0, + "open": 10.1, + "high": 10.3, + "low": 9.9, + "vol": 1000, + "amount": 2000000, + "quote_date": "20240902", + "source": "datahub", + } + ], + "meta": {"stale": False, "staleness_seconds": 0, "source": "eastmoney"}, + } + if path == "/v1/indexes/quotes": + return { + "schema_version": 1, + "data": [ + { + "ts_code": "000001.SH", + "code": "000001", + "name": "上证指数", + "price": 3000, + "pct_chg": 1.2, + "quote_time_epoch": 1725249600, + "source": "datahub", + }, + { + "ts_code": "399001.SZ", + "code": "399001", + "name": "深证成指", + "price": 9000, + "pct_chg": 0.8, + "quote_time_epoch": 1725249600, + "source": "datahub", + }, + { + "ts_code": "399006.SZ", + "code": "399006", + "name": "创业板指", + "price": 1800, + "pct_chg": 0.5, + "quote_time_epoch": 1725249600, + "source": "datahub", + }, + ], + "meta": {"stale": False, "staleness_seconds": 0, "source": "tencent"}, + } + if path == "/v1/intraday/points": + return { + "schema_version": 1, + "data": { + "entity_type": "stock", + "identifier": "600000", + "code": "600000", + "trade_date": "2024-09-02", + "previous_close": 10.0, + "points": [ + {"date": "2024-09-02", "time": "09:30", "close": 10.2, "average": 10.1, "open": 10.1, "high": 10.2, "low": 10.0, "volume": 100, "amount": 1000} + ], + }, + "meta": {"stale": False, "source": "datahub"}, + } + if path == "/v1/query": + body = json.loads(request.data.decode("utf-8") if request.data else "{}") + api_name = body.get("api_name") + if api_name == "ifind_status": + return {"schema_version": 1, "data": [{"configured": True, "access_ready": True, "access_expires_at": ""}], "meta": {"source": "ifind"}} + if api_name == "ifind_wencai": + return { + "schema_version": 1, + "data": [{"股票代码": "000001.SZ", "涨停原因": "重组", "首次涨停时间": "09:42:00"}], + "meta": {"source": "ifind"}, + } + if api_name == "ifind_snapshots": + return { + "schema_version": 1, + "data": [ + { + "time": "2024-09-02 09:21:00", + "thscode": "000001.SZ", + "latest": 10.5, + "preClose": 10, + "volume": 2000, + "amount": 21000, + } + ], + "meta": {"source": "ifind"}, + } + if api_name in {"daily", "rt_k"}: + return { + "schema_version": 1, + "data": [{"ts_code": "600000.SH", "trade_date": "20240902", "close": 10.2, "amount": 2000.0}], + "meta": {"source": "datahub", "stale": False, "row_shape": "tushare"}, + } + raise AssertionError(f"unexpected query api: {api_name}") + raise AssertionError(f"unexpected path: {path}") + + +def blocked_urlopen(request, timeout=None): + return _Resp(hub_payload(request)) class HubExclusiveWebsiteTests(unittest.TestCase): @@ -39,36 +211,7 @@ class HubExclusiveWebsiteTests(unittest.TestCase): self.assertEqual(legacy.calls, []) def test_blocked_external_hosts_still_read_hub(self) -> None: - settings = DatahubSettings( - base_url="http://127.0.0.1:8766", - token="hub-token", - datasets={name: DatasetFlags(name, read=True) for name in DATASETS}, - ) - - def blocked_urlopen(request, timeout=None): - url = str(getattr(request, "full_url", None) or request) - if any(host in url for host in BLOCKED_HOSTS): - raise AssertionError(f"website opened blocked host: {url}") - if "127.0.0.1:8766" in url or "v1/bars/daily" in url: - class _Resp: - status = 200 - - def read(self): - return ( - b'{"schema_version":1,"data":[{"ts_code":"600000.SH","trade_date":"20240902",' - b'"close":10.2,"volume":100000,"amount":2000000}],' - b'"meta":{"stale":false,"staleness_seconds":0,"source":"datahub"}}' - ) - - def __enter__(self): - return self - - def __exit__(self, *args): - return False - - return _Resp() - raise AssertionError(f"unexpected url: {url}") - + settings = _enabled_settings() hub_client = DatahubClient(settings, urlopen=blocked_urlopen) legacy = FakeLegacy(TushareError("blocked")) wrapped = DatahubAwareTushareClient(legacy, DatahubBridge(settings, hub_client)) @@ -78,6 +221,56 @@ class HubExclusiveWebsiteTests(unittest.TestCase): self.assertEqual(rows[0]["amount"], 2000.0) self.assertEqual(legacy.calls, []) + def test_site_features_only_reach_hub_when_external_hosts_blocked(self) -> None: + settings = _enabled_settings() + hub_client = DatahubClient(settings, urlopen=blocked_urlopen) + bridge = DatahubBridge(settings, hub_client) + with patch("urllib.request.urlopen", blocked_urlopen): + quotes = bridge.try_quotes(["600000.SH"]) + indices = bridge.try_index_quotes() + chart = MarketChartClient(bridge).stock_daily("600000", "20240902") + intraday = MarketChartClient(bridge).stock_intraday("600000") + ifind = HubIfindProxy(bridge) + wencai = ifind.wencai("2024年9月2日涨停股票") + snapshots = ifind.snapshots(["000001.SZ"], ["latest"], "2024-09-02 09:15:00", "2024-09-02 09:22:00") + realtime = HubRealtimeProxy(bridge) + index_rows = realtime.tencent_indices() + stock = realtime.tencent_stock_quote("600000.SH", "20240902") + health = realtime.health_snapshot() + + self.assertEqual(quotes[0]["close"], 10.2) + self.assertEqual(len(indices), 3) + self.assertEqual(chart[-1]["close"], 10.2) + self.assertEqual(intraday["source"], "datahub") + self.assertEqual(wencai[0]["股票代码"], "000001.SZ") + self.assertEqual(snapshots[0]["latest"], 10.5) + self.assertEqual([row["code"] for row in index_rows], ["000001", "399001", "399006"]) + self.assertEqual(stock["close"], 10.2) + self.assertTrue(health["ready"]) + self.assertTrue(ifind.configured) + + def test_production_gateway_does_not_construct_external_clients(self) -> None: + source = (ROOT / "backend" / "data" / "gateway.py").read_text(encoding="utf-8") + self.assertNotIn("IfindHttpClient", source) + self.assertNotIn("EastmoneyChartClient", source) + self.assertNotIn("WebRealtimeAggregator", source) + self.assertIn("HubIfindProxy", source) + self.assertIn("HubRealtimeProxy", source) + self.assertIn("legacy.realtime_aggregator = None", source) + self.assertIn("DatahubAwareTushareClient", source) + + def test_production_python_does_not_embed_blocked_hosts(self) -> None: + violations = [] + for path in (ROOT / "backend").rglob("*.py"): + relative = path.relative_to(ROOT).as_posix() + if relative in LEFTOVER_WEBSITE_FILES: + continue + text = path.read_text(encoding="utf-8") + for host in BLOCKED_HOSTS: + if host in text: + violations.append(f"{relative} -> {host}") + self.assertEqual(violations, []) + def test_website_runtime_does_not_call_blocked_hosts_from_gateway(self) -> None: gateway_src = (ROOT / "backend" / "data" / "gateway.py").read_text(encoding="utf-8") self.assertIn("legacy.realtime_aggregator = None", gateway_src) @@ -101,6 +294,18 @@ class HubExclusiveWebsiteTests(unittest.TestCase): self.assertTrue(any("query_api" in text for text in called)) self.assertFalse(any("legacy_query" in text for text in called)) + def test_build_gateway_uses_hub_proxies_without_opening_external_hosts(self) -> None: + settings = _enabled_settings() + with patch("urllib.request.urlopen", blocked_urlopen): + gateway = build_data_gateway({"tushare_token": "tok"}, datahub_settings=settings) + gateway.datahub.client = DatahubClient(settings, urlopen=blocked_urlopen) + rows = gateway.ifind.wencai("涨停") + quotes = gateway.realtime_observer.tencent_indices() + chart = gateway.chart_data.stock_daily("600000", "20240902") + self.assertEqual(rows[0]["涨停原因"], "重组") + self.assertEqual(len(quotes), 3) + self.assertEqual(chart[-1]["close"], 10.2) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_ifind_features.py b/tests/test_ifind_features.py index c0e16a4..a2276ec 100644 --- a/tests/test_ifind_features.py +++ b/tests/test_ifind_features.py @@ -6,7 +6,7 @@ from datetime import date, datetime, timedelta, timezone from pathlib import Path from unittest.mock import patch -from backend.features.market.charts import EastmoneyChartClient, MarketChartClient +from backend.features.market.charts import MarketChartClient from database import ReviewDatabase from backend.features.market.insights import MarketInsightsService from server import DashboardService @@ -166,7 +166,7 @@ class IfindFeatureTests(unittest.TestCase): }, ] ) - client = MarketChartClient(FakeIfind(), EastmoneyChartClient(), hub) + client = MarketChartClient(hub) rows = client.stock_daily("000001", "20260728") self.assertEqual(rows[-1]["trade_date"], "2026-07-28") self.assertAlmostEqual(rows[-1]["change"], 2.9412, places=4) @@ -185,7 +185,7 @@ class IfindFeatureTests(unittest.TestCase): } ] ) - client = MarketChartClient(FakeIfindStalePreopen(), EastmoneyChartClient(), hub) + client = MarketChartClient(hub) with patch("backend.features.market.charts.datetime", FixedPreopenDatetime): rows = client.stock_daily("000001", "20260729") diff --git a/tools/build_architecture_inventory.py b/tools/build_architecture_inventory.py index 9f8d222..d867546 100644 --- a/tools/build_architecture_inventory.py +++ b/tools/build_architecture_inventory.py @@ -218,11 +218,10 @@ def build() -> dict[str, Any]: ), "external_data_adapters": [ {"provider": "tushare", "path": "backend/data/providers/tushare_client.py", "runtime_role": "stable client facade for primary deterministic market data"}, - {"provider": "datahub", "path": "backend/data/datahub/client.py", "runtime_role": "optional official EOD read path behind per-dataset flags"}, - {"provider": "ifind", "path": "backend/data/providers/ifind_client.py", "runtime_role": "realtime, charts, snapshots, enrichment"}, - {"provider": "eastmoney", "path": "backend/features/market/charts.py", "runtime_role": "display chart fallback"}, - {"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation and intraday dashboard fallback"}, - {"provider": "tencent", "path": "backend/data/realtime.py", "runtime_role": "index observation and intraday quote fallback"}, + {"provider": "datahub", "path": "backend/data/datahub/client.py", "runtime_role": "website-only read path; official EOD, live quotes, and licensed iFinD"}, + {"provider": "ifind", "path": "xiaobai-datahub/datahub/adapters/ifind.py", "runtime_role": "licensed iFinD source inside the data hub"}, + {"provider": "eastmoney", "path": "xiaobai-datahub/datahub/adapters/eastmoney.py", "runtime_role": "free realtime quotes and shenwan inside the data hub"}, + {"provider": "tencent", "path": "xiaobai-datahub/datahub/adapters/tencent.py", "runtime_role": "free index and stock quotes inside the data hub"}, ], "provider_domains": [ {"provider": "tushare", "path": "backend/data/providers/tushare_transport.py", "responsibility": "HTTP transport and provider errors"}, @@ -240,9 +239,9 @@ def build() -> dict[str, Any]: {"client": "DatahubClient", "owner": "backend/data/gateway.py"}, {"client": "DatahubBridge", "owner": "backend/data/gateway.py"}, {"client": "DatahubAwareTushareClient", "owner": "backend/data/gateway.py"}, - {"client": "IfindHttpClient", "owner": "backend/data/gateway.py"}, + {"client": "HubIfindProxy", "owner": "backend/data/gateway.py"}, + {"client": "HubRealtimeProxy", "owner": "backend/data/gateway.py"}, {"client": "MarketChartClient", "owner": "backend/data/gateway.py"}, - {"client": "WebRealtimeAggregator", "owner": "backend/data/gateway.py"}, ], "heaven_service_owners": { "facade": "backend/features/heaven/service.py", diff --git a/xiaobai-datahub/.env.example b/xiaobai-datahub/.env.example index e5cd65b..33fb773 100644 --- a/xiaobai-datahub/.env.example +++ b/xiaobai-datahub/.env.example @@ -10,4 +10,8 @@ DATAHUB_ADMIN_PASSWORD= # Tushare Pro token. Stored encrypted after first launch; never returned by API or admin pages. TUSHARE_TOKEN= +# Optional licensed iFinD tokens. Used only inside the hub; the website never dials 51ifind.com. +IFIND_REFRESH_TOKEN= +IFIND_ACCESS_TOKEN= + TZ=Asia/Shanghai diff --git a/xiaobai-datahub/datahub/adapters/ifind.py b/xiaobai-datahub/datahub/adapters/ifind.py index 759d48c..c90cf71 100644 --- a/xiaobai-datahub/datahub/adapters/ifind.py +++ b/xiaobai-datahub/datahub/adapters/ifind.py @@ -1,3 +1,432 @@ -from datahub.adapters.base import ReservedAdapter +from __future__ import annotations -ADAPTER = ReservedAdapter("ifind") +import copy +import json +import threading +import time +import urllib.error +import urllib.request +from datetime import datetime, timedelta +from typing import Any, Callable + +from datahub.adapters.base import AdapterError, MarketAdapter + +UrlOpen = Callable[..., Any] + + +class IfindAdapter(MarketAdapter): + """Licensed iFinD source used only inside the data hub.""" + + name = "ifind" + BASE_URL = "https://quantapi.51ifind.com/api/v1" + AUTH_ENDPOINT = "get_access_token" + AUTH_ERROR_CODES = {-1302, -1303, -1304, -4302, -4303} + + def __init__( + self, + refresh_token: str = "", + access_token: str = "", + timeout: int = 15, + urlopen: UrlOpen = urllib.request.urlopen, + ) -> None: + self.timeout = max(3, int(timeout)) + self._urlopen = urlopen + self._refresh_token = str(refresh_token or "").strip() + self._access_token = str(access_token or "").strip() + self._access_expires_at: datetime | None = None + self._token_lock = threading.Lock() + self._cache_lock = threading.Lock() + self._cache: dict[str, dict[str, Any]] = {} + + @property + def configured(self) -> bool: + return bool(self._refresh_token or self._access_token) + + def set_credentials(self, refresh_token: str, access_token: str = "") -> None: + refresh_token = str(refresh_token or "").strip() + access_token = str(access_token or "").strip() + with self._token_lock: + refresh_changed = refresh_token != self._refresh_token + self._refresh_token = refresh_token + if access_token or refresh_changed: + self._access_token = access_token + self._access_expires_at = None + if refresh_changed: + with self._cache_lock: + self._cache.clear() + + def status(self) -> dict[str, Any]: + return { + "configured": self.configured, + "access_ready": bool(self._access_token), + "access_expires_at": ( + self._access_expires_at.isoformat(timespec="seconds") + if self._access_expires_at + else "" + ), + } + + def probe(self) -> dict[str, Any]: + started = time.perf_counter() + if not self.configured: + return { + "provider": self.name, + "configured": False, + "state": "unconfigured", + "message": "iFinD token 未配置", + "latency_ms": round((time.perf_counter() - started) * 1000), + } + try: + rows = self.real_time("000001.SH", ["latest"], cache_ttl=0) + state = "ok" if rows else "empty" + return { + "provider": self.name, + "configured": True, + "state": state, + "latency_ms": round((time.perf_counter() - started) * 1000), + } + except AdapterError as exc: + return { + "provider": self.name, + "configured": True, + "state": "error", + "message": str(exc), + "latency_ms": round((time.perf_counter() - started) * 1000), + } + + def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]: + if dataset == "wencai": + return self.wencai( + str(params.get("query") or params.get("searchstring") or ""), + str(params.get("search_type") or params.get("searchtype") or "stock"), + int(params.get("cache_ttl") or 300), + ) + if dataset == "snapshots": + return self.snapshots( + params.get("codes") or "", + _indicators(params.get("indicators")), + str(params.get("start_time") or ""), + str(params.get("end_time") or ""), + int(params.get("cache_ttl") or 8), + ) + if dataset == "history": + return self.history( + params.get("codes") or "", + _indicators(params.get("indicators") or ["close", "volume", "amount"]), + str(params.get("start_date") or ""), + str(params.get("end_date") or ""), + int(params.get("cache_ttl") or 300), + ) + if dataset == "realtime": + return self.real_time( + params.get("codes") or "", + _indicators(params.get("indicators") or ["latest"]), + int(params.get("cache_ttl") or 10), + ) + if dataset == "intraday": + return self.intraday( + str(params.get("code") or params.get("codes") or ""), + str(params.get("start_time") or ""), + str(params.get("end_time") or ""), + int(params.get("cache_ttl") or 20), + ) + raise AdapterError(f"{self.name} unsupported dataset: {dataset}") + + def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + return list(rows) + + def real_time( + self, + codes: str | list[str], + indicators: list[str], + cache_ttl: int = 10, + ) -> list[dict[str, Any]]: + code_text = self._codes(codes) + payload = self._request( + "real_time_quotation", + {"codes": code_text, "indicators": ",".join(indicators)}, + cache_key=f"rq:{code_text}:{','.join(indicators)}", + cache_ttl=cache_ttl, + ) + return self._table_rows(payload) + + def history( + self, + codes: str | list[str], + indicators: list[str], + start_date: str, + end_date: str, + cache_ttl: int = 300, + ) -> list[dict[str, Any]]: + code_text = self._codes(codes) + payload = self._request( + "cmd_history_quotation", + { + "codes": code_text, + "indicators": ",".join(indicators), + "startdate": self._display_date(start_date), + "enddate": self._display_date(end_date), + "functionpara": {"CPS": "forward1", "Fill": "Omit"}, + }, + cache_key=f"hq:{code_text}:{start_date}:{end_date}:{','.join(indicators)}", + cache_ttl=cache_ttl, + ) + return self._table_rows(payload) + + def intraday( + self, + code: str, + start_time: str, + end_time: str, + cache_ttl: int = 20, + ) -> list[dict[str, Any]]: + indicators = ["open", "high", "low", "close", "volume", "amount", "avgPrice"] + payload = self._request( + "high_frequency", + { + "codes": self._codes(code), + "indicators": ",".join(indicators), + "starttime": start_time, + "endtime": end_time, + "functionpara": { + "CPS": "forward1", + "Fill": "Previous", + "Timeformat": "LocalTime", + "Interval": "1", + "Limitstart": "09:30:00", + "Limitend": "15:00:00", + }, + }, + cache_key=f"hf:{code}:{start_time}:{end_time}", + cache_ttl=cache_ttl, + ) + return self._table_rows(payload) + + def snapshots( + self, + codes: str | list[str], + indicators: list[str], + start_time: str, + end_time: str, + cache_ttl: int = 8, + ) -> list[dict[str, Any]]: + code_text = self._codes(codes) + payload = self._request( + "snap_shot", + { + "codes": code_text, + "indicators": ",".join(indicators), + "starttime": start_time, + "endtime": end_time, + }, + cache_key=f"ss:{code_text}:{start_time}:{end_time}:{','.join(indicators)}", + cache_ttl=cache_ttl, + ) + return self._table_rows(payload) + + def wencai(self, query: str, search_type: str = "stock", cache_ttl: int = 300) -> list[dict[str, Any]]: + normalized = " ".join(str(query or "").split()) + if not normalized: + raise AdapterError("问财查询不能为空。") + payload = self._request( + "smart_stock_picking", + {"searchstring": normalized, "searchtype": search_type}, + cache_key=f"wc:{search_type}:{normalized}", + cache_ttl=cache_ttl, + ) + return self._table_rows(payload) + + def _request( + self, + endpoint: str, + body: dict[str, Any], + cache_key: str = "", + cache_ttl: int = 0, + ) -> dict[str, Any]: + if not self.configured: + raise AdapterError("iFinD 尚未配置。") + if cache_key and cache_ttl > 0: + cached = self._cached(cache_key, cache_ttl) + if cached is not None: + return cached + payload = self._post(endpoint, body, self._ensure_access_token()) + if self._is_auth_error(payload) and self._refresh_token: + self._invalidate_access_token() + payload = self._post(endpoint, body, self._ensure_access_token(force=True)) + self._validate_payload(payload) + if cache_key and cache_ttl > 0: + with self._cache_lock: + self._cache[cache_key] = { + "created_at": time.time(), + "payload": copy.deepcopy(payload), + } + return payload + + def _ensure_access_token(self, force: bool = False) -> str: + with self._token_lock: + now = datetime.now().astimezone().replace(tzinfo=None) + token_valid = bool(self._access_token) and ( + self._access_expires_at is None + or self._access_expires_at > now + timedelta(minutes=2) + ) + if token_valid and not force: + return self._access_token + if not self._refresh_token: + if self._access_token: + return self._access_token + raise AdapterError("iFinD Refresh Token 尚未配置。") + payload = self._post(self.AUTH_ENDPOINT, {}, "", self._refresh_token) + self._validate_payload(payload) + data = payload.get("data") or {} + token = str(data.get("access_token") or "").strip() + if not token: + raise AdapterError("iFinD 未返回 Access Token。") + expires_at = self._parse_datetime(data.get("expired_time")) + self._access_token = token + self._access_expires_at = expires_at + return token + + def _post( + self, + endpoint: str, + body: dict[str, Any], + access_token: str, + refresh_token: str = "", + ) -> dict[str, Any]: + headers = { + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": "XiaobaiDatahub/1.0", + "ifindlang": "cn", + } + if access_token: + headers["access_token"] = access_token + if refresh_token: + headers["refresh_token"] = refresh_token + request = urllib.request.Request( + f"{self.BASE_URL}/{endpoint}", + data=json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8"), + headers=headers, + method="POST", + ) + try: + with self._urlopen(request, timeout=self.timeout) as response: + payload = json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + detail = "" + try: + detail_payload = json.loads(exc.read().decode("utf-8", errors="replace")) + detail = str(detail_payload.get("errmsg") or detail_payload.get("message") or "") + except (json.JSONDecodeError, OSError): + pass + raise AdapterError(f"iFinD HTTP {exc.code}{f'：{detail[:160]}' if detail else ''}") from exc + except (urllib.error.URLError, TimeoutError, OSError, json.JSONDecodeError) as exc: + raise AdapterError("iFinD 数据请求失败。") from exc + if not isinstance(payload, dict): + raise AdapterError("iFinD 返回格式不正确。") + return payload + + def _cached(self, key: str, ttl: int) -> dict[str, Any] | None: + with self._cache_lock: + cached = self._cache.get(key) + if not cached: + return None + if time.time() - float(cached.get("created_at") or 0) > ttl: + self._cache.pop(key, None) + return None + return copy.deepcopy(cached["payload"]) + + def _invalidate_access_token(self) -> None: + with self._token_lock: + self._access_token = "" + self._access_expires_at = None + + @classmethod + def _validate_payload(cls, payload: dict[str, Any]) -> None: + try: + error_code = int(payload.get("errorcode") or 0) + except (TypeError, ValueError): + error_code = -1 + if error_code != 0: + message = str(payload.get("errmsg") or "未知错误") + raise AdapterError(f"iFinD 返回错误：{message[:200]}") + + @classmethod + def _is_auth_error(cls, payload: dict[str, Any]) -> bool: + try: + error_code = int(payload.get("errorcode") or 0) + except (TypeError, ValueError): + error_code = 0 + message = str(payload.get("errmsg") or "").casefold() + return error_code in cls.AUTH_ERROR_CODES or "token" in message or "鉴权" in message + + @staticmethod + def _table_rows(payload: dict[str, Any]) -> list[dict[str, Any]]: + tables = payload.get("tables") or [] + if isinstance(tables, dict): + tables = [tables] + rows: list[dict[str, Any]] = [] + for block in tables if isinstance(tables, list) else []: + if not isinstance(block, dict): + continue + table = block.get("table") or {} + if not isinstance(table, dict): + continue + times = block.get("time") or [] + codes = block.get("thscode") or block.get("thscodes") or [] + if isinstance(codes, str): + codes = [codes] + lengths = [len(value) for value in table.values() if isinstance(value, list)] + row_count = max(lengths or [len(times) if isinstance(times, list) else 0, 1 if table else 0]) + for index in range(row_count): + row: dict[str, Any] = {} + if isinstance(times, list) and index < len(times): + row["time"] = times[index] + if codes: + row["thscode"] = codes[index] if index < len(codes) else codes[0] + for field, values in table.items(): + if isinstance(values, list): + row[field] = values[index] if index < len(values) else None + elif index == 0: + row[field] = values + rows.append(row) + return rows + + @staticmethod + def _codes(codes: str | list[str]) -> str: + if isinstance(codes, list): + values = [str(code or "").strip().upper() for code in codes] + else: + values = [part.strip().upper() for part in str(codes or "").split(",")] + values = [value for value in values if value] + if not values: + raise AdapterError("iFinD 证券代码不能为空。") + if len(values) > 100: + raise AdapterError("iFinD 单次证券代码过多。") + return ",".join(values) + + @staticmethod + def _display_date(value: str) -> str: + compact = str(value or "").replace("-", "") + if len(compact) != 8 or not compact.isdigit(): + raise AdapterError("iFinD 日期格式不正确。") + return f"{compact[:4]}-{compact[4:6]}-{compact[6:]}" + + @staticmethod + def _parse_datetime(value: Any) -> datetime | None: + text = str(value or "").strip() + if not text: + return None + try: + return datetime.fromisoformat(text) + except ValueError: + return None + + +def _indicators(value: Any) -> list[str]: + if isinstance(value, list): + return [str(item).strip() for item in value if str(item).strip()] + return [part.strip() for part in str(value or "").split(",") if part.strip()] + + +ADAPTER = IfindAdapter() diff --git a/xiaobai-datahub/datahub/admin_api.py b/xiaobai-datahub/datahub/admin_api.py index 3d33657..c5ab3f2 100644 --- a/xiaobai-datahub/datahub/admin_api.py +++ b/xiaobai-datahub/datahub/admin_api.py @@ -13,11 +13,12 @@ from datahub.timeutil import isoformat, now_shanghai, session_phase, yyyymmdd class AdminAPI: - def __init__(self, db: HubDB, pipeline: Pipeline, scheduler: Scheduler, auth: AuthService) -> None: + def __init__(self, db: HubDB, pipeline: Pipeline, scheduler: Scheduler, auth: AuthService, ifind: Any = None) -> None: self.db = db self.pipeline = pipeline self.scheduler = scheduler self.auth = auth + self.ifind = ifind def overview(self) -> dict[str, Any]: today = yyyymmdd(now_shanghai()) @@ -57,10 +58,26 @@ class AdminAPI: } ] for name, adapter in RESERVED.items(): + if name == "ifind": + live = self.ifind or adapter + cred = self.auth.credential_status("ifind_refresh_token") or { + "configured": bool(getattr(live, "configured", False)), + "last4": "", + "updated_at": "", + } + items.append( + { + "provider": name, + "role": "licensed", + "health": live.probe(), + "credential": cred, + } + ) + continue items.append( { "provider": name, - "role": "reserved", + "role": "reserved" if name in {"ths", "xgb", "akshare"} else "free", "health": adapter.probe(), "credential": {"configured": False, "last4": "", "updated_at": ""}, } @@ -78,6 +95,11 @@ class AdminAPI: def probe(self, provider: str) -> dict[str, Any]: if provider == "tushare": return self.pipeline.adapter.probe() + if provider == "ifind": + adapter = self.ifind or RESERVED.get("ifind") + if adapter is None: + raise ApiError("INVALID_ARGUMENT", "unknown provider: ifind") + return adapter.probe() adapter = RESERVED.get(provider) if adapter is None: raise ApiError("INVALID_ARGUMENT", f"unknown provider: {provider}") diff --git a/xiaobai-datahub/datahub/hub.py b/xiaobai-datahub/datahub/hub.py index fbd25c8..e9b20cf 100644 --- a/xiaobai-datahub/datahub/hub.py +++ b/xiaobai-datahub/datahub/hub.py @@ -2,6 +2,7 @@ from __future__ import annotations from pathlib import Path +from datahub.adapters.ifind import IfindAdapter from datahub.adapters.tushare import TushareAdapter from datahub.admin_api import AdminAPI from datahub.auth import AuthService @@ -28,7 +29,16 @@ class Hub: if settings.tushare_token: self.auth.store_credential("tushare_token", settings.tushare_token) token = settings.tushare_token + refresh = settings.ifind_refresh_token or self.auth.load_credential("ifind_refresh_token") or "" + access = settings.ifind_access_token or self.auth.load_credential("ifind_access_token") or "" + if settings.ifind_refresh_token: + self.auth.store_credential("ifind_refresh_token", settings.ifind_refresh_token) + refresh = settings.ifind_refresh_token + if settings.ifind_access_token: + self.auth.store_credential("ifind_access_token", settings.ifind_access_token) + access = settings.ifind_access_token self.adapter = adapter or TushareAdapter(token) + self.ifind = IfindAdapter(refresh, access) self.pipeline = Pipeline( self.db, self.adapter, @@ -38,8 +48,8 @@ class Hub: ) self.lkg = LastKnownGood(self.db) self.scheduler = Scheduler(self.db, self.pipeline) - self.api = V1API(self.db, self.pipeline, settings) - self.admin = AdminAPI(self.db, self.pipeline, self.scheduler, self.auth) + self.api = V1API(self.db, self.pipeline, settings, ifind=self.ifind) + self.admin = AdminAPI(self.db, self.pipeline, self.scheduler, self.auth, ifind=self.ifind) self.static_dir = Path(__file__).resolve().parents[1] / "admin" def start(self) -> None: diff --git a/xiaobai-datahub/datahub/logutil.py b/xiaobai-datahub/datahub/logutil.py index f7f5851..aa91031 100644 --- a/xiaobai-datahub/datahub/logutil.py +++ b/xiaobai-datahub/datahub/logutil.py @@ -12,6 +12,7 @@ from datahub.timeutil import isoformat _SECRET_KEYS = ( "token", "password", "secret", "key", "authorization", "credential", "tushare_token", "datahub_token", "encryption_key", "cookie", + "ifind", "refresh_token", "access_token", ) _SECRET_JSON = re.compile( r'(?i)("(?:' + "|".join(re.escape(key) for key in _SECRET_KEYS) + r')"\s*:\s*")([^"\\]*(?:\\.[^"\\]*)*)(")' diff --git a/xiaobai-datahub/datahub/serving.py b/xiaobai-datahub/datahub/serving.py index 67db3b6..9114a91 100644 --- a/xiaobai-datahub/datahub/serving.py +++ b/xiaobai-datahub/datahub/serving.py @@ -50,10 +50,11 @@ def envelope(data: Any, meta: dict[str, Any]) -> dict[str, Any]: class V1API: - def __init__(self, db: HubDB, pipeline: Pipeline, settings: Settings) -> None: + def __init__(self, db: HubDB, pipeline: Pipeline, settings: Settings, ifind: Any = None) -> None: self.db = db self.pipeline = pipeline self.settings = settings + self.ifind = ifind def handle(self, path: str, query: dict[str, list[str]]) -> dict[str, Any]: q = {key: values[-1] if values else "" for key, values in query.items()} @@ -135,12 +136,18 @@ class V1API: ) is_open = bool(cal and cal["is_open"] == 1) sources = self.db.fetchall("SELECT * FROM src_health") + ifind = getattr(self, "ifind", None) + ifind_status = ifind.status() if ifind is not None else {"configured": False} return envelope( { "status": "ok", "session_phase": session_phase(now_shanghai(), is_open), "trade_date": today, "is_open_day": is_open, + "ifind": { + "configured": bool(ifind_status.get("configured")), + "access_ready": bool(ifind_status.get("access_ready")), + }, "sources": [ { "provider": row["provider"], diff --git a/xiaobai-datahub/datahub/settings.py b/xiaobai-datahub/datahub/settings.py index df25757..6c07f8e 100644 --- a/xiaobai-datahub/datahub/settings.py +++ b/xiaobai-datahub/datahub/settings.py @@ -26,6 +26,8 @@ class Settings: api_token: str = "" admin_password: str = "" tushare_token: str = "" + ifind_refresh_token: str = "" + ifind_access_token: str = "" db_path: Path = DEFAULT_DB_PATH backup_dir: Path = DEFAULT_BACKUP_DIR quality: dict[str, Any] = field(default_factory=dict) @@ -113,6 +115,8 @@ def load_settings( api_token=str(environ.get("DATAHUB_TOKEN") or "").strip(), admin_password=str(environ.get("DATAHUB_ADMIN_PASSWORD") or "").strip(), tushare_token=str(environ.get("TUSHARE_TOKEN") or "").strip(), + ifind_refresh_token=str(environ.get("IFIND_REFRESH_TOKEN") or "").strip(), + ifind_access_token=str(environ.get("IFIND_ACCESS_TOKEN") or "").strip(), db_path=db_path, backup_dir=backup_dir, quality=_load_quality(quality_path), diff --git a/xiaobai-datahub/datahub/steward.py b/xiaobai-datahub/datahub/steward.py index f28efdf..b69c023 100644 --- a/xiaobai-datahub/datahub/steward.py +++ b/xiaobai-datahub/datahub/steward.py @@ -90,6 +90,15 @@ LIVE_TTL = { BLOCKED_LIVE_APIS = {"rt_sw_k"} +IFIND_APIS = { + "ifind_wencai": "wencai", + "ifind_snapshots": "snapshots", + "ifind_history": "history", + "ifind_realtime": "realtime", + "ifind_intraday": "intraday", + "ifind_status": "status", +} + def steward_query(api, body: dict[str, Any]) -> dict[str, Any]: api_name = str(body.get("api_name") or "").strip() @@ -99,6 +108,8 @@ def steward_query(api, body: dict[str, Any]) -> dict[str, Any]: raise ApiError("INVALID_ARGUMENT", "api_name is required") if api_name in BLOCKED_LIVE_APIS: raise ApiError("INVALID_ARGUMENT", "rt_sw_k is disabled; use published sw_daily or free Shenwan realtime") + if api_name in IFIND_APIS: + return _ifind_query(api, api_name, params, fields) if api_name == "rt_k": return _realtime_quotes(api, params, fields) if api_name == "rt_idx_k": @@ -122,6 +133,42 @@ def steward_query(api, body: dict[str, Any]) -> dict[str, Any]: ) +def _ifind_query(api, api_name: str, params: dict[str, Any], fields: str) -> dict[str, Any]: + adapter = getattr(api, "ifind", None) + dataset = IFIND_APIS[api_name] + if adapter is None: + raise ApiError("SOURCE_UNAVAILABLE", "iFinD adapter is not attached") + if dataset == "status": + return envelope( + [dict(adapter.status())], + { + "tier": "live", + "source": "ifind", + "stale": False, + "staleness_seconds": 0, + "row_shape": "ifind", + "published_at": isoformat(now_shanghai()), + }, + ) + if not adapter.configured: + raise ApiError("SOURCE_UNAVAILABLE", "iFinD 尚未配置") + try: + rows = adapter.fetch(dataset, dict(params)) + except AdapterError as exc: + raise ApiError("SOURCE_UNAVAILABLE", str(exc)) from exc + return envelope( + _project(rows, fields), + { + "tier": "live", + "source": "ifind", + "stale": False, + "staleness_seconds": 0, + "row_shape": "ifind", + "published_at": isoformat(now_shanghai()), + }, + ) + + def _try_published(api, api_name: str, dataset: str, params: dict[str, Any], fields: str) -> dict[str, Any] | None: fetcher = DATASET_FETCHER.get(dataset) if fetcher is None: diff --git a/xiaobai-datahub/tests/test_ifind_adapter.py b/xiaobai-datahub/tests/test_ifind_adapter.py new file mode 100644 index 0000000..330cf5c --- /dev/null +++ b/xiaobai-datahub/tests/test_ifind_adapter.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import json +import unittest + +from datahub.adapters.ifind import IfindAdapter +from datahub.serving import ApiError +from datahub.steward import steward_query + + +class _Resp: + def __init__(self, payload: dict, status: int = 200) -> None: + self.status = status + self._raw = json.dumps(payload).encode("utf-8") + + def read(self): + return self._raw + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + +class IfindAdapterTests(unittest.TestCase): + def test_unconfigured_probe_does_not_dial_vendor(self) -> None: + calls = [] + + def urlopen(request, timeout=None): + calls.append(str(getattr(request, "full_url", request))) + raise AssertionError("should not dial") + + adapter = IfindAdapter(urlopen=urlopen) + probe = adapter.probe() + self.assertEqual(probe["state"], "unconfigured") + self.assertFalse(probe["configured"]) + self.assertEqual(calls, []) + + def test_wencai_and_snapshots_go_to_ifind_http(self) -> None: + calls = [] + + def urlopen(request, timeout=None): + url = str(getattr(request, "full_url", request)) + calls.append(url) + if url.endswith("get_access_token"): + return _Resp({"errorcode": 0, "data": {"access_token": "acc", "expired_time": ""}}) + body = json.loads(request.data.decode("utf-8")) + if url.endswith("smart_stock_picking"): + self.assertEqual(body["searchstring"], "涨停") + return _Resp( + { + "errorcode": 0, + "tables": [ + { + "thscode": ["000001.SZ"], + "table": {"涨停原因": ["重组"]}, + } + ], + } + ) + if url.endswith("snap_shot"): + return _Resp( + { + "errorcode": 0, + "tables": [ + { + "time": ["2024-09-02 09:21:00"], + "thscode": ["000001.SZ"], + "table": {"latest": [10.5], "preClose": [10]}, + } + ], + } + ) + raise AssertionError(url) + + adapter = IfindAdapter("refresh-token", urlopen=urlopen) + rows = adapter.fetch("wencai", {"query": "涨停"}) + self.assertEqual(rows[0]["涨停原因"], "重组") + snaps = adapter.fetch( + "snapshots", + { + "codes": ["000001.SZ"], + "indicators": ["latest", "preClose"], + "start_time": "2024-09-02 09:15:00", + "end_time": "2024-09-02 09:22:00", + }, + ) + self.assertEqual(snaps[0]["latest"], 10.5) + self.assertTrue(any("quantapi.51ifind.com" in item for item in calls)) + + def test_steward_ifind_status_and_unconfigured_fetch(self) -> None: + class _Api: + ifind = IfindAdapter() + + payload = steward_query(_Api(), {"api_name": "ifind_status", "params": {}}) + self.assertFalse(payload["data"][0]["configured"]) + with self.assertRaises(ApiError): + steward_query(_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}}) + + +if __name__ == "__main__": + unittest.main() diff --git a/xiaobai-datahub/tests/test_layout.py b/xiaobai-datahub/tests/test_layout.py index 2469db9..284a17c 100644 --- a/xiaobai-datahub/tests/test_layout.py +++ b/xiaobai-datahub/tests/test_layout.py @@ -19,7 +19,7 @@ class LayoutTests(unittest.TestCase): def test_reserved_adapters_present(self) -> None: from datahub.adapters import RESERVED - for name in ("ths", "xgb", "akshare", "ifind"): + for name in ("ths", "xgb", "akshare"): self.assertIn(name, RESERVED) probe = RESERVED[name].probe() self.assertEqual(probe["state"], "reserved") @@ -27,9 +27,12 @@ class LayoutTests(unittest.TestCase): for name in ("eastmoney", "tencent"): self.assertIn(name, RESERVED) probe = RESERVED[name].probe() - # Live free adapters: probe may be ok/error/empty depending on network. self.assertIn(probe["state"], {"ok", "empty", "error"}) self.assertTrue(probe["configured"]) + self.assertIn("ifind", RESERVED) + ifind = RESERVED["ifind"].probe() + self.assertIn(ifind["state"], {"unconfigured", "ok", "empty", "error"}) + self.assertEqual(ifind["configured"], ifind["state"] != "unconfigured") if __name__ == "__main__":