from __future__ import annotations import http.client import json import re import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass from threading import Lock from typing import Any, ClassVar class ChartDataError(RuntimeError): pass TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get" BOARD_LIST_URL = "https://push2delay.eastmoney.com/api/qt/clist/get" BROWSER_USER_AGENT = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/138.0.0.0 Safari/537.36" ) INDEX_SECIDS = { "000001.SH": "1.000001", "399001.SZ": "0.399001", "399006.SZ": "0.399006", } @dataclass class EastmoneyChartClient: """Isolated display-only minute chart source. The returned data must not be used by market snapshots, scoring, screening, or divination. Its only consumer is a chart-rendering endpoint. """ timeout: int = 6 cache_ttl_seconds: int = 20 retry_attempts: int = 2 _cache: ClassVar[dict[str, dict[str, Any]]] = {} _cache_lock: ClassVar[Lock] = Lock() _board_catalog: ClassVar[dict[str, dict[str, str]]] = {} _board_catalog_at: ClassVar[float] = 0.0 _board_catalog_lock: ClassVar[Lock] = Lock() def stock_intraday(self, code: str) -> dict[str, Any]: normalized = str(code or "").strip() if not re.fullmatch(r"\d{6}", normalized): raise ChartDataError("Invalid stock code") market = "1" if normalized.startswith(("5", "6", "9")) else "0" return self._intraday(f"{market}.{normalized}", "stock", normalized) def index_intraday(self, identifier: str) -> dict[str, Any]: normalized = str(identifier or "").strip().upper() secid = INDEX_SECIDS.get(normalized) if not secid: raise ChartDataError("Unsupported index") return self._intraday(secid, "index", normalized) def board_intraday(self, identifier: str, name: str = "") -> dict[str, Any]: normalized = str(identifier or "").strip().upper() if re.fullmatch(r"BK\d{4}", normalized): board_code = normalized else: board_code = self._resolve_board_code(name or identifier) return self._intraday(f"90.{board_code}", "board", board_code) def _intraday(self, secid: str, entity_type: str, identifier: str) -> dict[str, Any]: cache_key = f"{entity_type}:{identifier}" cached = self._get_cached(cache_key) if cached is not None: return cached payload = self._request_json( TRENDS_URL, { "secid": secid, "fields1": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13", "fields2": "f51,f52,f53,f54,f55,f56,f57,f58", "iscr": "0", "ndays": "1", }, "https://quote.eastmoney.com/", ) data = payload.get("data") or {} points = [point for raw in data.get("trends") or [] if (point := _parse_trend(raw))] if not points: raise ChartDataError("No intraday chart data returned") result = { "entity_type": entity_type, "identifier": identifier, "name": str(data.get("name") or ""), "code": str(data.get("code") or identifier), "trade_date": points[-1]["date"], "previous_close": _number(data.get("preClose")), "points": points, } with self._cache_lock: self._cache[cache_key] = {"created_at": time.time(), "payload": result} return result def _get_cached(self, cache_key: str) -> dict[str, Any] | None: with self._cache_lock: cached = self._cache.get(cache_key) if not cached: return None if time.time() - float(cached.get("created_at") or 0) > self.cache_ttl_seconds: with self._cache_lock: self._cache.pop(cache_key, None) return None return dict(cached["payload"]) def _resolve_board_code(self, name: str) -> str: normalized = _normalize_name(name) if not normalized: raise ChartDataError("Board name is required") catalog = self._load_board_catalog() item = catalog.get(normalized) if not item: raise ChartDataError("No matching chart board") return item["code"] def _load_board_catalog(self) -> dict[str, dict[str, str]]: now = time.time() with self._board_catalog_lock: if self._board_catalog and now - self._board_catalog_at < 6 * 60 * 60: return dict(self._board_catalog) rows: list[dict[str, Any]] = [] for board_type in ("1", "2", "3"): for page in range(1, 6): payload = self._request_json( BOARD_LIST_URL, { "pn": str(page), "pz": "100", "po": "1", "np": "1", "fltt": "2", "invt": "2", "fid": "f3", "fs": f"m:90+t:{board_type}", "fields": "f12,f14", }, "https://quote.eastmoney.com/center/boardlist.html", ) page_rows = (payload.get("data") or {}).get("diff") or [] rows.extend(page_rows) if len(page_rows) < 100: break catalog: dict[str, dict[str, str]] = {} for row in rows: code = str(row.get("f12") or "").strip().upper() board_name = str(row.get("f14") or "").strip() if re.fullmatch(r"BK\d{4}", code) and board_name: catalog.setdefault(_normalize_name(board_name), {"code": code, "name": board_name}) if not catalog: raise ChartDataError("Board chart directory is unavailable") with self._board_catalog_lock: type(self)._board_catalog = catalog type(self)._board_catalog_at = now return dict(catalog) def _request_json( self, url: str, params: dict[str, str], referer: str ) -> dict[str, Any]: request_url = f"{url}?{urllib.parse.urlencode(params)}" last_error: Exception | None = None for attempt in range(max(1, int(self.retry_attempts))): request = urllib.request.Request( request_url, headers={ "Accept": "application/json,text/plain,*/*", "Connection": "close", "Referer": referer, "User-Agent": BROWSER_USER_AGENT, }, ) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: payload = json.loads(response.read().decode("utf-8")) if not isinstance(payload, dict): raise ChartDataError("Invalid intraday chart response") return payload except ( urllib.error.URLError, TimeoutError, ConnectionError, OSError, http.client.HTTPException, json.JSONDecodeError, ChartDataError, ) as exc: last_error = exc if attempt + 1 < self.retry_attempts: time.sleep(0.12) raise ChartDataError("Intraday chart request failed") from last_error def _parse_trend(raw: Any) -> dict[str, Any] | None: fields = str(raw or "").split(",") if len(fields) < 8 or " " not in fields[0]: return None stamp = fields[0].strip() trade_date, trade_time = stamp.split(" ", 1) close = _number(fields[2]) if close <= 0: return None return { "date": trade_date, "time": trade_time[:5], "open": _number(fields[1]), "close": close, "high": _number(fields[3]), "low": _number(fields[4]), "volume": _number(fields[5]), "amount": _number(fields[6]), "average": _number(fields[7]), } def _number(value: Any) -> float: try: return float(value or 0) except (TypeError, ValueError): return 0.0 def _normalize_name(value: Any) -> str: normalized = re.sub(r"[\s·・（）()\-_/]", "", str(value or "")).casefold() return re.sub(r"(?:概念|行业|[ⅠⅡⅢ])$", "", normalized)