xiaobai-review/backend/data/datahub/ifind_proxy.py

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 []