qlib/backend/app/infrastructure/data_sources/sina.py

"""新浪财经 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, list_status: str = "L"):
        raise DataSourceNotSupported("新浪不提供股票基础信息列表")

    def get_name_changes(self, start, end):
        raise DataSourceNotSupported("新浪不提供股票名称变更历史（namechange）")

    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("新浪不提供复权因子（日线接口为前复权口径）")

    def get_daily_basic(self, trade_date):
        """新浪无估值/股息率接口。

        禁止返回空列表冒充成功 —— 否则 daily_basic 同步会把「接口不支持」
        误记成「当日无数据」，导致缺口被静默固化（AGENT.md §7）。
        """
        raise DataSourceNotSupported("新浪不提供每日指标（估值/股息率/市值）接口")