"""新浪财经 Provider —— 备用数据源。 通道（公开接口方案参考 cc-cursor/finance/data/sources/sina_source.py）： 1. 财务：quotes.sina.cn CompanyFinanceService.getFinanceReport2022（source=gjzb， 匿名免费、一次多期），含披露日 publish_date → FinancialIndicator （symbol / report_date=end_date / announce_date=publish_date），schema 与 Tushare fina_indicator 一致 —— 用于财务兜底（保留防未来函数所需的公告日）。 2. 日 K：quotes.sina.cn getKLineData（jsonp，**前复权**）。新浪无「不复权 + 独立复权 因子」，因此日线兜底行标记 source=sina、adjust=qfq，与主口径区分；Tushare 恢复 后 --resume 会按日覆盖回不复权行。 能力边界（其余接口新浪不支持 → DataSourceNotSupported）： get_stock_basic / get_trade_cal / get_adjust_factor。 """ from __future__ import annotations import json import re import time import urllib.parse import urllib.request from datetime import date, datetime from decimal import Decimal from typing import Any from app.domain.entities.market import DailyBar, FinancialIndicator from app.infrastructure.data_sources.errors import ( DataSourceError, DataSourceNotSupported, ) _UA = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36" ) _KLINE_JSONP = ( "https://quotes.sina.cn/cn/api/jsonp_v2.php/var%20data=/CN_MarketDataService" ".getKLineData?symbol={sina_symbol}&scale=240&ma=no&datalen={datalen}" ) _FIN_BASE = "https://quotes.sina.cn/cn/api/openapi.php/CompanyFinanceService.getFinanceReport2022" # 新浪「关键指标」中文项名 → 本项目 FinancialIndicator 字段（None 表示已具备/忽略） _FIN_FIELD_MAP = { "基本每股收益": "eps", "净资产收益率(ROE)": "roe", "加权净资产收益率": "roe", "销售毛利率": "gross_margin", "毛利率": "gross_margin", "营业总收入": "total_revenue", "净利润": "net_profit", } def _to_sina_symbol(symbol: str) -> str: """600519.SH -> sh600519；000001.SZ -> sz000001；无后缀时按规则猜测。""" code = symbol.strip().upper() if code.endswith(".SH"): return "sh" + code[:-3] if code.endswith(".SZ"): return "sz" + code[:-3] if code.endswith(".BJ"): return "bj" + code[:-3] if code.startswith(("6", "9")): return "sh" + code if code.startswith(("4", "8")): return "bj" + code return "sz" + code def _extract_jsonp(payload: str) -> list[dict[str, Any]]: """从 JSONP 中提取数组：容忍前导注释 / var data=([...]) 包裹 / 尾部杂字符。 直接取首个 '[' 与末个 ']' 之间的内容（行情数组为扁平结构，无嵌套数组）。 """ start = payload.find("[") end = payload.rfind("]") if start == -1 or end <= start: raise DataSourceError("新浪行情返回格式无法解析") return json.loads(payload[start : end + 1]) def _d(value) -> Decimal | None: if value is None: return None try: return Decimal(str(value)) except (ValueError, TypeError): return None def _to_date(value: str) -> date: """兼容 20240831 / 2024-08-31 等格式。""" digits = re.sub(r"\D", "", str(value))[:8] return datetime.strptime(digits, "%Y%m%d").date() class SinaProvider: """新浪财经备用数据源：财务（与 Tushare schema 一致）+ 日线（前复权）。""" name = "sina" def __init__( self, *, timeout: float = 10.0, retries: int = 2, urlopen=urllib.request.urlopen, ) -> None: self._timeout = timeout self._retries = retries self._urlopen = urlopen # ---- HTTP（统一 UA / 重试） ---- def _open(self, url: str) -> bytes: req = urllib.request.Request(url, headers={"User-Agent": _UA}) last_error: Exception | None = None for attempt in range(self._retries): try: with self._urlopen(req, timeout=self._timeout) as resp: return resp.read() except Exception as exc: # noqa: BLE001 last_error = exc if attempt < self._retries - 1: time.sleep(0.5 * (attempt + 1)) raise DataSourceError(f"sina 请求失败: {last_error}") from last_error # ---- 财务（兜底 Tushare fina_indicator） ---- def get_financial( self, symbol: str, start: date | None = None, end: date | None = None, ) -> list[FinancialIndicator]: """新浪关键指标（source=gjzb），含披露日 publish_date → announce_date。 新浪不支持按报告期窗口拉取：忽略 start/end 时返回其全部返回的 报告期；传入窗口则按 report_date 客户端过滤（新浪行 source=sina）。 """ params = { "paperCode": _to_sina_symbol(symbol), "source": "gjzb", "type": "0", "page": "1", "num": "100", } url = f"{_FIN_BASE}?{urllib.parse.urlencode(params)}" payload = json.loads(self._open(url).decode("utf-8", errors="replace")) try: data = payload["result"]["data"] report_dates = [item["date_value"] for item in data["report_date"]] except (KeyError, TypeError) as exc: raise DataSourceError(f"新浪财务返回结构异常（{symbol}）: {exc}") from exc rows: list[FinancialIndicator] = [] for rd in report_dates: entry = data["report_list"].get(rd) if not entry: continue announce = entry.get("publish_date") if not announce: continue # 无披露日不可用于研究（防未来函数） report_day = _to_date(str(rd)) if start is not None and report_day < start: continue if end is not None and report_day > end: continue fields: dict[str, Decimal | None] = { "eps": None, "roe": None, "total_revenue": None, "net_profit": None, "gross_margin": None, } for item in entry.get("data", []): std = _FIN_FIELD_MAP.get(item.get("item_title", "")) if std and fields.get(std) is None: fields[std] = _d(item.get("item_value")) rows.append( FinancialIndicator( symbol=symbol, report_date=report_day, announce_date=_to_date(str(announce)), source="sina", eps=fields["eps"], roe=fields["roe"], total_revenue=fields["total_revenue"], net_profit=fields["net_profit"], gross_margin=fields["gross_margin"], ) ) return rows # ---- 日 K（前复权兜底，标记 adjust=qfq） ---- def get_daily(self, symbol: str, start: date, end: date, datalen: int = 320) -> list[DailyBar]: """拉取前复权日 K（新浪仅支持最近 datalen 个自然日窗口）。""" url = _KLINE_JSONP.format(sina_symbol=_to_sina_symbol(symbol), datalen=datalen) payload = self._open(url).decode("utf-8", errors="replace") bars: list[DailyBar] = [] for rec in _extract_jsonp(payload): day = datetime.strptime(rec["day"], "%Y-%m-%d").date() if day < start or day > end: continue bars.append( DailyBar( symbol=symbol, trade_date=day, source="sina", adjust="qfq", open=_d(rec.get("open")), high=_d(rec.get("high")), low=_d(rec.get("low")), close=_d(rec.get("close")), volume=_d(rec.get("volume")), ) ) return bars # ---- 不支持 ---- def get_stock_basic(self): raise DataSourceNotSupported("新浪不提供股票基础信息列表") def get_index_weight(self, index_code): raise DataSourceNotSupported("新浪不提供指数成分接口") def get_trade_cal(self, start, end): raise DataSourceNotSupported("新浪不提供交易日历") def get_adjust_factor(self, symbol, start, end): raise DataSourceNotSupported("新浪不提供复权因子（日线接口为前复权口径）")