diff --git a/backend/bootstrap/container.py b/backend/bootstrap/container.py index 4409557..0060ad6 100644 --- a/backend/bootstrap/container.py +++ b/backend/bootstrap/container.py @@ -2,9 +2,11 @@ from __future__ import annotations from dataclasses import dataclass from pathlib import Path +from collections.abc import Callable from alert_service import AlertService -from chart_data_provider import EastmoneyChartClient, MarketChartClient +from backend.data import DataGateway, build_data_gateway +from chart_data_provider import MarketChartClient from database import ReviewDatabase from ifind_client import IfindHttpClient from mentor_agent import MentorSkillRegistry @@ -17,6 +19,7 @@ from trade_journal import TradeJournalService @dataclass(frozen=True) class ApplicationContainer: database: ReviewDatabase + data_gateway: DataGateway ifind: IfindHttpClient screener: ScreenerEngine strategy_tracking: StrategyTrackingService @@ -32,19 +35,18 @@ def build_application_container( credentials: dict[str, object], mentor_skills_dir: Path, private_mentor_skills_dir: Path, + tushare_token_supplier: Callable[[], str] | None = None, ) -> ApplicationContainer: - ifind = IfindHttpClient( - str(credentials.get("ifind_refresh_token") or ""), - str(credentials.get("ifind_access_token") or ""), - ) + data_gateway = build_data_gateway(credentials, tushare_token_supplier) return ApplicationContainer( database=database, - ifind=ifind, + data_gateway=data_gateway, + ifind=data_gateway.ifind, screener=ScreenerEngine(database), strategy_tracking=StrategyTrackingService(database), alert_service=AlertService(database), trade_journal=TradeJournalService(database), mentor_skills=MentorSkillRegistry(mentor_skills_dir, private_mentor_skills_dir), - realtime_aggregator=WebRealtimeAggregator(), - chart_data=MarketChartClient(ifind, EastmoneyChartClient()), + realtime_aggregator=data_gateway.realtime_observer, + chart_data=data_gateway.chart_data, ) diff --git a/backend/data/__init__.py b/backend/data/__init__.py new file mode 100644 index 0000000..94a309e --- /dev/null +++ b/backend/data/__init__.py @@ -0,0 +1,4 @@ +from .gateway import DataGateway, build_data_gateway +from .policy import DataPolicyError, DataSourcePolicy + +__all__ = ["DataGateway", "DataPolicyError", "DataSourcePolicy", "build_data_gateway"] diff --git a/backend/data/contracts.py b/backend/data/contracts.py new file mode 100644 index 0000000..362ee61 --- /dev/null +++ b/backend/data/contracts.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + + +DataUsage = Literal["display", "calculation"] + + +@dataclass(frozen=True) +class ProviderContract: + id: str + provider_class: str + calculation_allowed: bool + + +@dataclass(frozen=True) +class DatasetContract: + id: str + entity: str + frequency: str + primary: str + fallbacks: tuple[str, ...] + usage: str + fields: tuple[str, ...] + + @property + def providers(self) -> tuple[str, ...]: + return (self.primary, *self.fallbacks) diff --git a/backend/data/gateway.py b/backend/data/gateway.py new file mode 100644 index 0000000..f79fdff --- /dev/null +++ b/backend/data/gateway.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass + +from backend.data.contracts import DataUsage +from backend.data.policy import DataSourcePolicy +from backend.data.providers import IfindProvider, TushareProvider +from chart_data_provider import EastmoneyChartClient, MarketChartClient +from ifind_client import IfindHttpClient +from realtime_aggregator import WebRealtimeAggregator +from tushare_client import TushareClient + + +@dataclass(frozen=True) +class DataGateway: + policy: DataSourcePolicy + tushare_provider: TushareProvider + ifind_provider: IfindProvider + chart_data: MarketChartClient + realtime_observer: WebRealtimeAggregator + + @property + def ifind(self) -> IfindHttpClient: + return self.ifind_provider.client + + def tushare( + self, + dataset_id: str = "", + usage: DataUsage = "calculation", + ) -> TushareClient: + if dataset_id: + self.policy.assert_allowed(dataset_id, "tushare", usage) + return self.tushare_provider.client() + + def assert_source(self, dataset_id: str, provider_id: str, usage: DataUsage) -> None: + self.policy.assert_allowed(dataset_id, provider_id, usage) + + +def build_data_gateway( + credentials: dict[str, object], + tushare_token_supplier: Callable[[], str] | 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 "") + ) + return DataGateway( + policy=DataSourcePolicy.load(), + tushare_provider=TushareProvider(token_supplier), + ifind_provider=IfindProvider(ifind), + chart_data=MarketChartClient(ifind, EastmoneyChartClient()), + realtime_observer=WebRealtimeAggregator(), + ) diff --git a/backend/data/policy.py b/backend/data/policy.py new file mode 100644 index 0000000..853f123 --- /dev/null +++ b/backend/data/policy.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from app_config import APP_DIR +from backend.data.contracts import DataUsage, DatasetContract, ProviderContract + + +class DataPolicyError(RuntimeError): + pass + + +class DataSourcePolicy: + def __init__( + self, + providers: dict[str, ProviderContract], + datasets: dict[str, DatasetContract], + ) -> None: + self.providers = dict(providers) + self.datasets = dict(datasets) + + @classmethod + def load(cls, path: Path | None = None) -> "DataSourcePolicy": + config_path = path or APP_DIR / "config" / "data-fields.config.json" + payload = json.loads(config_path.read_text(encoding="utf-8")) + providers = { + provider_id: ProviderContract( + id=provider_id, + provider_class=str(item["class"]), + calculation_allowed=bool(item["calculation_allowed"]), + ) + for provider_id, item in payload["providers"].items() + } + datasets = { + item["id"]: DatasetContract( + id=str(item["id"]), + entity=str(item["entity"]), + frequency=str(item["frequency"]), + primary=str(item["primary"]), + fallbacks=tuple(str(value) for value in item.get("fallbacks", [])), + usage=str(item["usage"]), + fields=tuple(str(value) for value in item.get("fields", [])), + ) + for item in payload["datasets"] + } + return cls(providers, datasets) + + def dataset(self, dataset_id: str) -> DatasetContract: + try: + return self.datasets[dataset_id] + except KeyError as exc: + raise DataPolicyError(f"Unregistered dataset: {dataset_id}") from exc + + def assert_allowed( + self, + dataset_id: str, + provider_id: str, + usage: DataUsage, + ) -> DatasetContract: + dataset = self.dataset(dataset_id) + if dataset.usage == "blocked": + raise DataPolicyError(f"Dataset is blocked: {dataset_id}") + if provider_id not in dataset.providers: + raise DataPolicyError( + f"Provider {provider_id} is not registered for dataset {dataset_id}" + ) + try: + provider = self.providers[provider_id] + except KeyError as exc: + raise DataPolicyError(f"Unregistered provider: {provider_id}") from exc + if usage == "calculation": + if dataset.usage != "calculation" or not provider.calculation_allowed: + raise DataPolicyError( + f"Provider {provider_id} cannot calculate dataset {dataset_id}" + ) + return dataset diff --git a/backend/data/providers/__init__.py b/backend/data/providers/__init__.py new file mode 100644 index 0000000..7bdeb09 --- /dev/null +++ b/backend/data/providers/__init__.py @@ -0,0 +1,4 @@ +from .ifind import IfindProvider +from .tushare import TushareProvider + +__all__ = ["IfindProvider", "TushareProvider"] diff --git a/backend/data/providers/ifind.py b/backend/data/providers/ifind.py new file mode 100644 index 0000000..c64a4dd --- /dev/null +++ b/backend/data/providers/ifind.py @@ -0,0 +1,11 @@ +from __future__ import annotations + +from ifind_client import IfindHttpClient + + +class IfindProvider: + def __init__(self, client: IfindHttpClient) -> None: + self.client = client + + def set_credentials(self, refresh_token: str, access_token: str = "") -> None: + self.client.set_credentials(refresh_token, access_token) diff --git a/backend/data/providers/tushare.py b/backend/data/providers/tushare.py new file mode 100644 index 0000000..a2eeda0 --- /dev/null +++ b/backend/data/providers/tushare.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from collections.abc import Callable + +from tushare_client import TushareClient + + +class TushareProvider: + def __init__( + self, + token_supplier: Callable[[], str], + client_factory: Callable[[str], TushareClient] = TushareClient, + ) -> None: + self._token_supplier = token_supplier + self._client_factory = client_factory + + def client(self) -> TushareClient: + return self._client_factory(str(self._token_supplier() or "").strip()) diff --git a/docs/governance/architecture-inventory.json b/docs/governance/architecture-inventory.json index 65fe985..c9ca986 100644 --- a/docs/governance/architecture-inventory.json +++ b/docs/governance/architecture-inventory.json @@ -265,8 +265,8 @@ }, { "path": "server.py", - "bytes": 267525, - "lines": 5938 + "bytes": 267835, + "lines": 5947 }, { "path": "static/redesign-v2.css", diff --git a/docs/governance/stage-06-data-gateway.md b/docs/governance/stage-06-data-gateway.md new file mode 100644 index 0000000..83bd32c --- /dev/null +++ b/docs/governance/stage-06-data-gateway.md @@ -0,0 +1,29 @@ +# Stage 06: Unified Data Gateway + +Date: 2026-07-29 + +## Result + +- Added canonical provider and dataset contracts under `backend/data`. +- Added `DataSourcePolicy`, loaded from the Stage 04 data registry. +- Added provider adapters for Tushare and iFinD. +- Added one `DataGateway` that owns Tushare creation, the shared iFinD client, chart routing, + and isolated Eastmoney/Tencent realtime observation. +- Replaced all real `DashboardService` Tushare construction paths with the gateway. +- Kept one compatibility constructor for unit tests that instantiate an incomplete service with + `__new__`; production instances never use it. +- The Tushare token is supplied lazily, so administrator credential changes do not leave a + stale client in memory. + +## Enforcement Introduced + +- Unregistered datasets fail. +- Blocked datasets fail. +- Public-web providers cannot be promoted to calculation inputs through a fallback call. +- Display chart fallbacks remain distinct from deterministic calculation datasets. + +## Deferred to Stage 07 + +Stage 06 centralizes provider access but does not yet attach freshness, coverage, unit, and +point-in-time quality evidence to every returned observation. Stage 07 introduces those gates +without changing provider routing again. diff --git a/server.py b/server.py index f5cd335..2304088 100644 --- a/server.py +++ b/server.py @@ -180,7 +180,9 @@ class DashboardService: self._system_credentials, MENTOR_SKILLS_DIR, PRIVATE_MENTOR_SKILLS_DIR, + lambda: self.token, ) + self.data_gateway = self.container.data_gateway self.ifind = self.container.ifind self.screener = self.container.screener self.strategy_tracking = self.container.strategy_tracking @@ -198,6 +200,13 @@ class DashboardService: ) self._background_thread.start() + def _tushare_client(self) -> TushareClient: + gateway = getattr(self, "data_gateway", None) + if gateway is not None: + return gateway.tushare() + # Compatibility for isolated legacy unit-test service stubs. + return TushareClient(self.token) + def _load_system_credentials(self, environment: dict[str, str]) -> dict[str, Any]: encrypted = self.database.get_system_setting("credentials") current = self.vault.decrypt_json(encrypted) if encrypted else {} @@ -1047,7 +1056,7 @@ class DashboardService: try: if not self.configured: raise TushareError("公共行情尚未配置") - dashboard = TushareClient(self.token).dashboard(normalized_date) + dashboard = self._tushare_client().dashboard(normalized_date) dashboard["meta"]["source"] = source dashboard["meta"]["requested_date"] = self._display_compact_date(normalized_date) @@ -1207,7 +1216,7 @@ class DashboardService: ts_code = f"{raw_code}.SH" else: ts_code = f"{raw_code}.SZ" - client = TushareClient(self.token) + client = self._tushare_client() try: industry = client.sw_stock_industry(ts_code, actual_date) sector_code = str(industry.get("l2_code") or "") @@ -1295,7 +1304,7 @@ class DashboardService: raise ValueError("行情数据尚未配置。") return MarketInsightsService( self.database, - TushareClient(self.token), + self._tushare_client(), ifind=self.ifind, ) @@ -1461,7 +1470,7 @@ class DashboardService: notice = "" if self.configured: try: - FactorDataService(self.database, TushareClient(self.token)).sync( + FactorDataService(self.database, self._tushare_client()).sync( normalized_date, 15 ) except TushareError: @@ -1513,7 +1522,7 @@ class DashboardService: resolved_date = normalized_date if self.configured: try: - client = TushareClient(self.token) + client = self._tushare_client() resolved_date, _ = client.resolve_trade_context(normalized_date) history = self.database.watchlist_price_history( [str(item["code"]) for item in items], resolved_date @@ -1703,7 +1712,7 @@ class DashboardService: normalized_date = normalize_date(trade_date) lookback = max(25, min(260, int(lookback))) with self.sync_lock: - return FactorDataService(self.database, TushareClient(self.token)).sync( + return FactorDataService(self.database, self._tushare_client()).sync( normalized_date, lookback ) @@ -1759,7 +1768,7 @@ class DashboardService: ) try: factor_sync = FactorDataService( - self.database, TushareClient(self.token) + self.database, self._tushare_client() ).sync(normalized_date, 260) factor_dates = self.database.factor_dates(normalized_date, 300) if not factor_dates or factor_dates[-1] != normalized_date: @@ -2511,7 +2520,7 @@ class DashboardService: exact = [item for item in candidates if str(item.get("name") or "").casefold() == raw.casefold()] if not exact and self.configured: try: - rows = TushareClient(self.token).query( + rows = self._tushare_client().query( "stock_basic", {"name": raw, "list_status": "L"}, "ts_code,symbol,name,industry,market,list_date", @@ -2730,7 +2739,7 @@ class DashboardService: if market_mode == "intraday": if self.configured: try: - quote = TushareClient(self.token).realtime_stock_quote( + quote = self._tushare_client().realtime_stock_quote( tushare_code(stock_code), trade_date, ) @@ -3217,7 +3226,7 @@ class DashboardService: error = "Tushare Token 未配置" else: try: - client = TushareClient(self.token) + client = self._tushare_client() if market_mode == "intraday": payload = self._aggregate_index_context(trade_date) payload["schema_version"] = 3 @@ -3277,7 +3286,7 @@ class DashboardService: "399001": "399001.SZ", "399006": "399006.SZ", } - client = TushareClient(self.token) + client = self._tushare_client() indices = [] start_date = ( datetime.strptime(trade_date, "%Y%m%d") - timedelta(days=20) @@ -3369,7 +3378,7 @@ class DashboardService: if not self.configured: return None try: - payload = TushareClient(self.token).sw_sector_snapshot( + payload = self._tushare_client().sw_sector_snapshot( tushare_code(identifier), trade_date, realtime_expected=market_mode == "intraday", @@ -3696,7 +3705,7 @@ class DashboardService: dashboard = self.get_dashboard(trade_date) if self.configured and dashboard.get("meta", {}).get("realtime"): try: - realtime_snapshot = TushareClient(self.token).realtime_factor_snapshot(trade_date) + realtime_snapshot = self._tushare_client().realtime_factor_snapshot(trade_date) except TushareError as exc: raise ValueError(f"实时选股行情不可用，已停止筛选：{exc}") from exc result = self.screener.screen( @@ -3716,7 +3725,7 @@ class DashboardService: return cached if self.configured: try: - payload = TushareClient(self.token).hot_money_profiles() + payload = self._tushare_client().hot_money_profiles() except TushareError: if cached: cached["meta"] = { @@ -3781,7 +3790,7 @@ class DashboardService: return cached if self.configured: try: - payload = TushareClient(self.token).dragon_tiger(normalized_date) + payload = self._tushare_client().dragon_tiger(normalized_date) except TushareError as exc: return { "meta": { @@ -3845,7 +3854,7 @@ class DashboardService: return cached_items try: - rows = TushareClient(self.token).query( + rows = self._tushare_client().query( "ths_index", {}, "ts_code,name,count,exchange,list_date,type", @@ -4023,7 +4032,7 @@ class DashboardService: def _ths_search_detail( self, basic: dict[str, Any], trade_date: str ) -> dict[str, Any]: - client = TushareClient(self.token) + client = self._tushare_client() resolved_date, _ = client.resolve_trade_context(trade_date) end = datetime.strptime(resolved_date, "%Y%m%d") start_date = (end - timedelta(days=190)).strftime("%Y%m%d") @@ -4112,7 +4121,7 @@ class DashboardService: def _index_search_detail( self, basic: dict[str, Any], trade_date: str ) -> dict[str, Any]: - client = TushareClient(self.token) + client = self._tushare_client() resolved_date, _ = client.resolve_trade_context(trade_date) payload = ( client.realtime_market_indices(resolved_date) @@ -4200,7 +4209,7 @@ class DashboardService: source = "tushare" if self.configured: try: - payload = TushareClient(self.token).stock_detail( + payload = self._tushare_client().stock_detail( tushare_code(code), normalized_date ) if not payload.get("prices"): @@ -4280,7 +4289,7 @@ class DashboardService: if quote and self._valid_realtime_stock_quote(quote, today): self._merge_realtime_stock_detail(result, quote, requested_date) elif self.configured and actual_date < today: - client = TushareClient(self.token) + client = self._tushare_client() try: resolved_date, _ = client.resolve_trade_context(requested_date) if resolved_date == today: diff --git a/tests/test_data_gateway.py b/tests/test_data_gateway.py new file mode 100644 index 0000000..b409e3d --- /dev/null +++ b/tests/test_data_gateway.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +import unittest + +from backend.data import DataPolicyError, DataSourcePolicy, build_data_gateway + + +class DataGatewayTests(unittest.TestCase): + def test_policy_allows_registered_calculation_source(self) -> None: + policy = DataSourcePolicy.load() + contract = policy.assert_allowed( + "market.stock_daily", "tushare", "calculation" + ) + self.assertEqual(contract.primary, "tushare") + + def test_policy_rejects_public_web_source_for_calculation(self) -> None: + policy = DataSourcePolicy.load() + with self.assertRaises(DataPolicyError): + policy.assert_allowed( + "observation.realtime_indices", "eastmoney", "calculation" + ) + + def test_policy_rejects_blocked_dataset(self) -> None: + policy = DataSourcePolicy.load() + with self.assertRaises(DataPolicyError): + policy.assert_allowed("market.level2", "unresolved", "display") + + def test_gateway_uses_live_token_supplier_and_shared_ifind(self) -> None: + token = {"value": "first"} + gateway = build_data_gateway( + {"ifind_refresh_token": "refresh", "ifind_access_token": "access"}, + lambda: token["value"], + ) + self.assertEqual(gateway.tushare().token, "first") + token["value"] = "second" + self.assertEqual(gateway.tushare().token, "second") + self.assertIs(gateway.chart_data.ifind, gateway.ifind) + + def test_server_has_no_direct_runtime_tushare_construction(self) -> None: + from pathlib import Path + + source = (Path(__file__).resolve().parents[1] / "server.py").read_text(encoding="utf-8") + self.assertEqual(source.count("TushareClient(self.token)"), 1) + self.assertIn("return gateway.tushare()", source) + + +if __name__ == "__main__": + unittest.main()