100752f43c
生产 gateway 不再实例化 iFinD、东财图和免费实时聚合器；问财与竞价快照作为中枢内部数据源。全站阻断外源测试覆盖日K、报价、图表、问财和竞价快照。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
433 lines
16 KiB
Python
433 lines
16 KiB
Python
from __future__ import annotations
|
||
|
||
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()
|