feat(sina): 新增新浪财报三表 — 三层接入(API/CLI/Web),独立数据源

This commit is contained in:
Justin Gu
2026-06-15 00:31:13 +08:00
parent a5ce783487
commit b49cfd66f8
10 changed files with 910 additions and 10 deletions
+55 -3
View File
@@ -744,10 +744,14 @@ with MacClient.from_best_host() as c:
### 财务
```bash
easy-tdx f10 SH 600519 # F10 公司信息
easy-tdx fund-flow SH 600519 # 历史资金流向
easy-tdx f10 600519 # 茅台利润表,最近 8 期(默认 lrb)
easy-tdx f10 600519 --type fzb --num 4 # 资产负债表,最近 4 期
easy-tdx f10 000001 --type llb --table # 平安现金流量表,表格输出
```
> 新浪财经数据源,``--type`` 支持 ``lrb``(利润表)/``fzb``(资产负债表)/``llb``(现金流量表)。
> 独立于 TDX 行情服务器,``item_value`` 已转 float 可直接数值计算,同比附 ``{科目}_同比`` 列。
### 扩展市场(港股/美股/期货)
```bash
@@ -875,6 +879,11 @@ curl "http://localhost:8000/api/v1/announcements?code=688017&count=30&page=1"
# 返回每条含 url(4 参数可直点打开)和 pdf_url(PDF 直链):
# {"data": [{"title":"...","type":"...","date":"...","url":".../detail?stockCode=...","pdf_url":"http://static.cninfo.com.cn/.../xxx.PDF",...}], "count": 30}
# ── 财报三表(新浪财经,独立数据源)──
# 利润表(type: lrb/fzb/llb
curl "http://localhost:8000/api/v1/sina/financial-report?code=600519&type=lrb&num=8"
# 返回每行一期(最新在前),列为科目名(float)+ {科目}_同比(如有):
# ── 排行 / 竞价 / 异动 ──
# 全 A 涨幅排行前 20
curl "http://localhost:8000/api/v1/mac/quote-list?category=A&count=20&sort_type=CHANGE_PCT"
@@ -972,7 +981,7 @@ uvicorn.run(app, host="0.0.0.0", port=8000)
| `screen scan` | 策略选股扫描(纯离线,全市场信号扫描) |
| `screen rank` | 扫描结果回测排名(按夏普/回撤等指标排序) |
| `serve` | 启动 Web API 服务器(REST + WebSocket,需 `easy-tdx[web]` |
| `f10` | F10 公司信息 |
| `f10` | 财报三表(新浪:利润表/资产负债表/现金流量表) |
| `fund-flow` | 历史资金流向 |
| `ex kline` | 扩展市场 K 线 |
| `ex quote` | 扩展市场报价 |
@@ -1377,6 +1386,34 @@ for _, row in df.iterrows():
print(f"跳过(无附件或失败): {e}")
```
### 财报三表(新浪财经)
独立数据源(新浪财经),无需连接 TDX 行情服务器即可获取利润表/资产负债表/现金流量表。
标准库 urllib 实现,零额外依赖。
```python
from easy_tdx.sina import SinaClient
client = SinaClient()
# 利润表(默认 8 期,最新在前)
df = client.get_financial_report("600519", report_type="lrb")
# → DataFrame,每行一期,列 = [报告期, 营业总收入, 营业总收入_同比, ...]
# 资产负债表 / 现金流量表(report_type 也接受中文别名:利润表/资产负债表/现金流量表)
df = client.get_financial_report("600519", report_type="fzb", num=4)
df = client.get_financial_report("600519", report_type="llb", num=4)
# 返回示例(item_value 已转 float,可直接数值计算):
# 报告期 营业总收入 营业总收入_同比 营业收入 营业收入_同比
# 0 2026-03-31 54702912385.23 0.06336 53909252220.51 0.06538
# 1 2025-12-31 174000000000.00 0.10000 NaN NaN
```
> - ``item_value`` 是字符串(新浪原始格式),本实现转 float;空/非数值转 None
> - 有同比的科目附加 ``{科目}_同比`` 列(float 比例,如 0.06336 = +6.3%
> - 大类标题行(如 ``流动资产``,原 ``item_value=""``)保留为 None,反映报表结构
## 枚举参考
### PeriodK 线周期)
@@ -1581,6 +1618,21 @@ ruff format --check src/ tests/ # format check
## Changelog
### 1.14.0 (2026-06-15)
**新增新浪财报三表** — 三层接入(编程 API / CLI / Web API),独立数据源,无需连接 TDX 行情服务器。
- 新模块 `easy_tdx.sina``SinaClient().get_financial_report(code, report_type=, num=)` 返回 `DataFrame`(每行一期,列为科目名 + `{科目}_同比`
- 三表:`lrb`(利润表)/ `fzb`(资产负债表)/ `llb`(现金流量表),report_type 支持中英文别名
- CLI`easy-tdx f10 600519 [--type lrb|fzb|llb] [--num N]`(接管原 f10 占位符)
- Web`GET /api/v1/sina/financial-report?code=&type=&num=`
- 标准库 urllib 实现,零新依赖
- 修复参考脚本 bug`item_value` 字符串转 float(原 object 列无法数值计算)
- 大类标题行(如「流动资产」)保留为 None,完整反映报表结构
- `SinaError` 继承 `TdxError`,保证全局 `except TdxError` 覆盖
测试:`tests/unit/test_sina.py` 27 个离线用例(mock HTTP,零网络),覆盖三表解析、数值转换、报告期格式化、同比键、paperCode 推导、错误转换。
### 1.13.1 (2026-06-15)
**cninfo 公告检索 Bug 修复 + PDF 下载**(实测 `easy-tdx announcement 601088` 暴露的 3 个 Bug + 新增 PDF 下载功能)。
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "easy-tdx"
version = "1.13.1"
version = "1.14.0"
description = "通达信 TCP 协议行情数据客户端,支持在线行情、离线数据读取与写入同步"
readme = "README.md"
requires-python = ">=3.10"
+36 -6
View File
@@ -1,4 +1,4 @@
"""财务数据命令(暂未实现)"""
"""财务数据命令。"""
from __future__ import annotations
@@ -6,16 +6,46 @@ import click
@click.command("f10")
@click.argument("market")
@click.argument("code")
def f10(market: str, code: str) -> None:
"""获取 F10 财务数据(暂未实现)。
@click.option(
"--type",
"report_type",
type=click.Choice(["lrb", "fzb", "llb"], case_sensitive=False),
default="lrb",
help="报表类型: lrb(利润表) / fzb(资产负债表) / llb(现金流量表)",
)
@click.option("--num", default=8, type=int, help="取最近 N 期(默认 8")
@click.option("--table", "use_table", is_flag=True, help="表格输出")
@click.option("--output", "output_fmt", type=click.Choice(["json", "table", "csv"]), default="json")
def f10(code: str, report_type: str, num: int, use_table: bool, output_fmt: str) -> None:
"""获取财报三表(新浪数据源,独立于 TDX 行情服务器)。
\b
报表类型:
lrb 利润表(默认)
fzb 资产负债表
llb 现金流量表
\b
示例:
easy-tdx f10 SZ 000001
easy-tdx f10 600519 # 茅台利润表,最近 8 期
easy-tdx f10 600519 --type fzb --num 4 # 资产负债表,最近 4 期
easy-tdx f10 000001 --type llb --table # 平安现金流量表,表格输出
"""
raise click.UsageError("f10 命令暂未实现,请使用 TdxClient.get_finance_info() API")
from ..sina import SinaClient, SinaError
from .output import print_error, print_output
fmt = "table" if use_table else output_fmt
client = SinaClient()
try:
df = client.get_financial_report(code, report_type=report_type, num=num)
except SinaError as e:
print_error(str(e))
raise SystemExit(1) from e
print_output(df, fmt)
@click.command("fund-flow")
+31
View File
@@ -0,0 +1,31 @@
"""新浪财经财报三表 —— 独立于 TDX 协议的 HTTP 数据源。
零额外依赖(标准库 urllib),无需连接 TDX 服务器即可使用。
支持三表:
- ``lrb`` 利润表
- ``fzb`` 资产负债表
- ``llb`` 现金流量表
用法::
from easy_tdx.sina import SinaClient
client = SinaClient()
# 利润表(默认 8 期)
df = client.get_financial_report("600519", report_type="lrb")
# → DataFrame,每行一期,列 = [报告期, 营业总收入, 营业总收入_同比, ...]
"""
from __future__ import annotations
from .client import SinaClient
from .models import ReportType, SinaError, normalize_report_type, report_type_name
__all__ = [
"SinaClient",
"ReportType",
"SinaError",
"normalize_report_type",
"report_type_name",
]
+165
View File
@@ -0,0 +1,165 @@
"""新浪财经财报三表客户端。
独立于 TDX 协议的 HTTP 数据源(标准库 urllib,零额外依赖)。
公开方法返回 ``pd.DataFrame``,遵循项目 ``get_*`` 约定。
新浪财报接口 ``CompanyFinanceService.getFinanceReport2022`` 返回结构::
result.data.report_list = { "20260331": { "data": [行项...] }, ... }
↑ 报告期(YYYYMMDD) 为键,倒序排列
每行项含 ``item_title``(科目名)/ ``item_value``(字符串数值)/ ``item_tongbi``
(同比比例,如 0.06336 = +6.3%/ ``item_display``(大类/小类)。
参考脚本的 bug``item_value`` 是字符串(如 "54702912385.230000"),
直接存入 DataFrame 导致列是 object 类型无法数值计算。本实现转 float,
空字符串/非数值转 None(保留行,因为大类标题行有价值)。
"""
from __future__ import annotations
import json
import logging
from typing import Any
from urllib import parse
from urllib import request as urlrequest
import pandas as pd
from .models import ReportType, SinaError, normalize_report_type
logger = logging.getLogger(__name__)
_UA = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
)
_API_URL = "https://quotes.sina.cn/cn/api/openapi.php/CompanyFinanceService.getFinanceReport2022"
def _http_get_json(url: str, params: dict[str, str], timeout: float = 15.0) -> Any:
"""GET JSON,自动 urlencode query 参数(stdlib urllibmonkeypatch 点)。"""
full = f"{url}?{parse.urlencode(params)}" if params else url
req = urlrequest.Request(full, headers={"User-Agent": _UA})
with urlrequest.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read().decode("utf-8"))
def _to_float(v: Any) -> float | None:
"""item_value 字符串转 float,空/非数值返回 None。"""
if v is None or v == "":
return None
try:
return float(v)
except (TypeError, ValueError):
return None
def _format_period(period: str) -> str:
"""``20260331`` → ``2026-03-31``。"""
if len(period) == 8 and period.isdigit():
return f"{period[:4]}-{period[4:6]}-{period[6:8]}"
return period
class SinaClient:
"""新浪财经财报检索客户端(无状态 HTTP,无需 connect/close)。
用法::
from easy_tdx.sina import SinaClient
client = SinaClient()
# 利润表(默认 8 期,最新在前)
df = client.get_financial_report("600519", report_type="lrb")
# → DataFrame,每行一期,列 = [报告期, 营业总收入, 营业总收入_同比, ...]
"""
def __init__(self, *, timeout: float = 15.0) -> None:
self.timeout = timeout
def _build_paper_code(self, code: str) -> str:
"""6 位代码 → 新浪 paperCodesh/sz 前缀)。"""
prefix = "sh" if code.startswith("6") else "sz"
return f"{prefix}{code}"
def get_financial_report(
self,
code: str,
report_type: ReportType | str = "lrb",
*,
num: int = 8,
) -> pd.DataFrame:
"""获取财报三表数据。
Args:
code: 6 位股票代码(不含市场前缀),如 ``600519``。
report_type: 报表类型,标准值 ``lrb``(利润表)/ ``fzb``(资产负债表)/
``llb``(现金流量表),也接受中文/英文别名(如 ``利润表``/``income``)。
num: 取最近 N 期(默认 8)。
Returns:
``DataFrame``,每行一期报告(最新在前)。
- 第一列 ``报告期````YYYY-MM-DD`` 格式)
- 其余列为科目名(如 ``营业总收入``),值为 float
- 有同比数据的科目附加 ``{科目}_同比`` 列(float 比例值,如 0.06336 = +6.3%
大类标题行(如 ``流动资产``)的 ``item_value`` 为 None(保留行以反映报表结构)。
无结果时返回空 DataFrame(含 ``报告期`` 列名)。
"""
rt = normalize_report_type(str(report_type))
rows = self._query(code, rt, num=num)
if not rows:
return pd.DataFrame(columns=["报告期"])
return pd.DataFrame(rows)
def _query(self, code: str, report_type: str, *, num: int) -> list[dict[str, Any]]:
"""调用新浪 API,解析为「按报告期」的行列表。
整个 HTTP + 解析过程统一捕获异常并转为 ``SinaError``。
"""
paper_code = self._build_paper_code(code)
params = {
"paperCode": paper_code,
"source": report_type,
"type": "0",
"page": "1",
"num": str(num),
}
try:
d = _http_get_json(_API_URL, params, timeout=self.timeout)
report_list = (
d.get("result", {}).get("data", {}).get("report_list", {})
if isinstance(d, dict)
else {}
)
if not report_list:
return []
rows: list[dict[str, Any]] = []
# 按报告期倒序,取最近 num 期
for period in sorted(report_list.keys(), reverse=True)[:num]:
obj = report_list[period]
if not isinstance(obj, dict):
continue
rec: dict[str, Any] = {"报告期": _format_period(period)}
for it in obj.get("data", []) or []:
if not isinstance(it, dict):
continue
title = it.get("item_title", "")
if not title:
continue
# item_value 字符串转 float(空/非数值 → None,保留行)
rec[title] = _to_float(it.get("item_value"))
tongbi = it.get("item_tongbi")
if tongbi not in (None, ""):
tb = _to_float(tongbi)
if tb is not None:
rec[f"{title}_同比"] = tb
rows.append(rec)
return rows
except SinaError:
raise
except Exception as e: # noqa: BLE001 — HTTP/JSON/解析统一转领域异常
raise SinaError(f"新浪财报查询失败: {e}") from e
+53
View File
@@ -0,0 +1,53 @@
"""新浪财经数据模型。"""
from __future__ import annotations
from typing import Literal
from easy_tdx.exceptions import TdxError
# 财报三表类型(新浪 API 的 source 参数值)
ReportType = Literal["lrb", "fzb", "llb"]
# 中文别名 → API source 值(CLI/Web 层方便用户)
_REPORT_TYPE_ALIASES: dict[str, str] = {
"lrb": "lrb",
"利润表": "lrb",
"income": "lrb",
"fzb": "fzb",
"资产负债表": "fzb",
"balance": "fzb",
"llb": "llb",
"现金流量表": "llb",
"cashflow": "llb",
}
_REPORT_TYPE_NAMES: dict[str, str] = {
"lrb": "利润表",
"fzb": "资产负债表",
"llb": "现金流量表",
}
def normalize_report_type(s: str) -> str:
"""归一化 report_type 输入:接受 lrb/fzb/llb 及中文/英文别名,返回标准三值之一。
Raises:
ValueError: 无法识别的输入。
"""
key = s.strip().lower() if s.isascii() else s.strip()
if key in _REPORT_TYPE_ALIASES:
return _REPORT_TYPE_ALIASES[key]
raise ValueError(
f"无法识别的报表类型: {s!r}"
"(支持 lrb/利润表/income、fzb/资产负债表/balance、llb/现金流量表/cashflow"
)
def report_type_name(report_type: str) -> str:
"""报表类型的中文名(用于展示)。"""
return _REPORT_TYPE_NAMES.get(report_type, report_type)
class SinaError(TdxError):
"""新浪财经数据请求或解析失败。"""
+3
View File
@@ -152,6 +152,7 @@ def _create_app(
from easy_tdx.web.routers.mac_quotes import router as mac_quotes_router
from easy_tdx.web.routers.market import router as market_router
from easy_tdx.web.routers.realtime import router as realtime_router
from easy_tdx.web.routers.sina import router as sina_router
app.include_router(market_router, prefix="/api/v1")
app.include_router(bars_router, prefix="/api/v1")
@@ -169,5 +170,7 @@ def _create_app(
app.include_router(indicator_router, prefix="/api/v1")
# 公告检索路由(巨潮资讯网,独立数据源)
app.include_router(announcement_router, prefix="/api/v1")
# 新浪财报三表路由(独立数据源)
app.include_router(sina_router, prefix="/api/v1")
return app
+39
View File
@@ -0,0 +1,39 @@
"""新浪财报三表路由(独立数据源,不依赖 TDX 服务器)。"""
from __future__ import annotations
import asyncio
from fastapi import APIRouter, HTTPException, Query
from easy_tdx.web.schemas import DataFrameResponse
router = APIRouter(tags=["sina"])
@router.get("/sina/financial-report", response_model=DataFrameResponse)
async def financial_report(
code: str = Query(..., min_length=6, max_length=6, description="6位股票代码"),
type: str = Query(
"lrb",
pattern=r"^(lrb|fzb|llb)$",
description="报表类型: lrb(利润表)/fzb(资产负债表)/llb(现金流量表)",
),
num: int = Query(8, ge=1, le=40, description="取最近 N 期"),
) -> DataFrameResponse:
"""获取财报三表(新浪数据源,独立于 TDX 行情服务器)。
返回每行一期报告(最新在前),列为科目名 + ``{科目}_同比``(如有同比)。
"""
from easy_tdx.sina import SinaClient, SinaError
client = SinaClient()
def _fetch() -> DataFrameResponse:
df = client.get_financial_report(code, report_type=type, num=num)
return DataFrameResponse.from_dataframe(df)
try:
return await asyncio.to_thread(_fetch)
except SinaError as e:
raise HTTPException(status_code=503, detail=str(e)) from e
+526
View File
@@ -0,0 +1,526 @@
"""新浪财经财报模块离线测试 —— mock HTTP,零网络依赖。
覆盖:响应解析(三表)、数值转换(字符串→float)、报告期格式化、
同比键、paperCode 推导、错误转换、模块导出。
"""
from __future__ import annotations
import json
from typing import Any
import pandas as pd
import pytest
# ---------------------------------------------------------------------------
# 导出
# ---------------------------------------------------------------------------
def test_public_exports() -> None:
"""模块应导出 SinaClient / ReportType / SinaError。"""
from easy_tdx import sina
assert hasattr(sina, "SinaClient")
assert hasattr(sina, "ReportType")
assert hasattr(sina, "SinaError")
def test_sina_error_subclasses_tdx_error() -> None:
"""SinaError 必须继承 TdxError,保证全局 except TdxError 覆盖。"""
from easy_tdx.exceptions import TdxError
from easy_tdx.sina import SinaError
assert issubclass(SinaError, TdxError)
assert issubclass(SinaError, Exception)
# ---------------------------------------------------------------------------
# report_type 归一化
# ---------------------------------------------------------------------------
def test_normalize_report_type_standard() -> None:
from easy_tdx.sina import normalize_report_type
assert normalize_report_type("lrb") == "lrb"
assert normalize_report_type("fzb") == "fzb"
assert normalize_report_type("llb") == "llb"
def test_normalize_report_type_uppercase() -> None:
from easy_tdx.sina import normalize_report_type
assert normalize_report_type("LRB") == "lrb"
assert normalize_report_type("FZB") == "fzb"
def test_normalize_report_type_chinese_alias() -> None:
from easy_tdx.sina import normalize_report_type
assert normalize_report_type("利润表") == "lrb"
assert normalize_report_type("资产负债表") == "fzb"
assert normalize_report_type("现金流量表") == "llb"
def test_normalize_report_type_english_alias() -> None:
from easy_tdx.sina import normalize_report_type
assert normalize_report_type("income") == "lrb"
assert normalize_report_type("balance") == "fzb"
assert normalize_report_type("cashflow") == "llb"
def test_normalize_report_type_invalid_raises() -> None:
from easy_tdx.sina import normalize_report_type
with pytest.raises(ValueError, match="无法识别"):
normalize_report_type("xyz")
# ---------------------------------------------------------------------------
# _to_float / _format_period
# ---------------------------------------------------------------------------
def test_to_float_numeric_string() -> None:
from easy_tdx.sina.client import _to_float
assert _to_float("54702912385.230000") == 54702912385.23
assert _to_float("0") == 0.0
assert _to_float("-123.45") == -123.45
def test_to_float_empty_and_none() -> None:
from easy_tdx.sina.client import _to_float
assert _to_float("") is None
assert _to_float(None) is None
def test_to_float_non_numeric() -> None:
from easy_tdx.sina.client import _to_float
assert _to_float("N/A") is None
assert _to_float("--") is None
assert _to_float("abc") is None
def test_format_period() -> None:
from easy_tdx.sina.client import _format_period
assert _format_period("20260331") == "2026-03-31"
assert _format_period("20251231") == "2025-12-31"
# 非 8 位数字原样返回
assert _format_period("2026Q1") == "2026Q1"
# ---------------------------------------------------------------------------
# paperCode 推导
# ---------------------------------------------------------------------------
def test_build_paper_code_sh() -> None:
"""6 开头 → sh 前缀。"""
from easy_tdx.sina import SinaClient
assert SinaClient()._build_paper_code("600519") == "sh600519"
assert SinaClient()._build_paper_code("601318") == "sh601318"
def test_build_paper_code_sz() -> None:
"""非 6 开头 → sz 前缀(含北交所 8/4)。"""
from easy_tdx.sina import SinaClient
assert SinaClient()._build_paper_code("000001") == "sz000001"
assert SinaClient()._build_paper_code("002594") == "sz002594"
assert SinaClient()._build_paper_code("300750") == "sz300750"
assert SinaClient()._build_paper_code("830799") == "sz830799"
# ---------------------------------------------------------------------------
# 财报查询与解析(mock HTTP)
# ---------------------------------------------------------------------------
def _make_response(periods: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
"""构造新浪 API 响应。periods = {period: [行项...]}"""
report_list = {
period: {
"data": items,
"publish_date": "2026-04-30",
"rType": "lrb",
}
for period, items in periods.items()
}
return {
"result": {
"status": {"code": 0},
"data": {
"report_count": len(periods),
"report_date": sorted(periods.keys(), reverse=True),
"report_list": report_list,
},
}
}
# 典型利润表行项
_LRB_ITEMS_2026Q1 = [
{
"item_field": "BIZTOTINCO",
"item_title": "营业总收入",
"item_value": "54702912385.230000",
"item_display": "大类",
"item_tongbi": 0.06336,
},
{
"item_field": "BIZINCO",
"item_title": "营业收入",
"item_value": "53909252220.510000",
"item_display": "小类",
"item_tongbi": 0.06538,
},
]
_LRB_ITEMS_2025 = [
{
"item_field": "BIZTOTINCO",
"item_title": "营业总收入",
"item_value": "174000000000.000000",
"item_display": "大类",
"item_tongbi": 0.10,
},
]
# 资产负债表(含大类标题行,item_value 为空)
_FZB_ITEMS = [
{
"item_field": "",
"item_title": "流动资产",
"item_value": "",
"item_display": "大类",
"item_tongbi": "",
},
{
"item_field": "CURFDS",
"item_title": "货币资金",
"item_value": "48786691397.550000",
"item_display": "小类",
"item_tongbi": -0.06538,
},
]
def _patch_http(monkeypatch: pytest.MonkeyPatch, response: dict[str, Any]) -> None:
"""让 _http_get_json 返回固定响应(不触网)。"""
monkeypatch.setattr(
"easy_tdx.sina.client._http_get_json",
lambda url, params, timeout=15.0: response,
)
def test_get_financial_report_returns_dataframe(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""应返回 DataFrame,第一列是报告期,其余是科目(float)。"""
resp = _make_response({"20260331": _LRB_ITEMS_2026Q1, "20251231": _LRB_ITEMS_2025})
_patch_http(monkeypatch, resp)
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519", report_type="lrb")
assert isinstance(df, pd.DataFrame)
# 第一列报告期
assert df.columns[0] == "报告期"
# 最新期在前
assert df.iloc[0]["报告期"] == "2026-03-31"
assert df.iloc[1]["报告期"] == "2025-12-31"
# 科目名作为列
assert "营业总收入" in df.columns
assert "营业收入" in df.columns
# 值已转 float(非 object 字符串)
assert df.iloc[0]["营业总收入"] == 54702912385.23
assert isinstance(df.iloc[0]["营业总收入"], float)
def test_get_financial_report_tongbi_columns(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""有同比的科目应附加 {科目}_同比 列。"""
resp = _make_response({"20260331": _LRB_ITEMS_2026Q1})
_patch_http(monkeypatch, resp)
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519", report_type="lrb")
assert "营业总收入_同比" in df.columns
assert df.iloc[0]["营业总收入_同比"] == pytest.approx(0.06336)
def test_get_financial_report_no_tongbi_omits_column(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""tongbi 为 None/空时不应生成 _同比 列。"""
items = [
{
"item_field": "NETPROFIT",
"item_title": "净利润",
"item_value": "100000000.00",
"item_tongbi": None,
}
]
resp = _make_response({"20260331": items})
_patch_http(monkeypatch, resp)
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519")
assert "净利润" in df.columns
assert "净利润_同比" not in df.columns
def test_get_financial_report_dalei_header_row_preserved(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""大类标题行(item_value="")应保留,值为 None(不跳过)。"""
resp = _make_response({"20260331": _FZB_ITEMS})
_patch_http(monkeypatch, resp)
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519", report_type="fzb")
assert "流动资产" in df.columns
# 大类标题行的值为 None(空字符串转换)
assert df.iloc[0]["流动资产"] is None or pd.isna(df.iloc[0]["流动资产"])
# 小类行有值
assert df.iloc[0]["货币资金"] == 48786691397.55
def test_get_financial_report_dalei_no_tongbi_column(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""大类标题行 tongbi="" 不生成 _同比 列。"""
resp = _make_response({"20260331": _FZB_ITEMS})
_patch_http(monkeypatch, resp)
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519", report_type="fzb")
assert "流动资产_同比" not in df.columns
def test_get_financial_report_non_numeric_value_to_none(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""非数值 item_value(如 'N/A')应转 None,不抛错。"""
items = [
{
"item_field": "X",
"item_title": "某科目",
"item_value": "N/A",
"item_tongbi": None,
}
]
resp = _make_response({"20260331": items})
_patch_http(monkeypatch, resp)
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519")
assert "某科目" in df.columns
assert df.iloc[0]["某科目"] is None or pd.isna(df.iloc[0]["某科目"])
def test_get_financial_report_skips_no_title(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""item_title 为空应跳过该行项。"""
items = [
{"item_field": "", "item_title": "", "item_value": "123.45"},
{"item_field": "X", "item_title": "有效科目", "item_value": "1.00"},
]
resp = _make_response({"20260331": items})
_patch_http(monkeypatch, resp)
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519")
assert "有效科目" in df.columns
# 空 title 的行不应产生列(pandas 会把无名列丢进一个奇怪的列名,确认它不是有效科目)
def test_get_financial_report_empty(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""空 report_list → 带「报告期」列名的空 DataFrame。"""
resp = {"result": {"status": {"code": 0}, "data": {"report_list": {}}}}
_patch_http(monkeypatch, resp)
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519")
assert isinstance(df, pd.DataFrame)
assert df.empty
assert list(df.columns) == ["报告期"]
def test_get_financial_report_missing_keys(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""响应缺少 result/data/report_list 链应视为空结果,不抛错。"""
_patch_http(monkeypatch, {"unexpected": True})
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519")
assert df.empty
def test_get_financial_report_request_failure_raises_sina_error(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""HTTP 异常应转为 SinaError。"""
def _boom(url: str, params: dict[str, str], timeout: float = 15.0) -> Any:
raise OSError("connection refused")
monkeypatch.setattr("easy_tdx.sina.client._http_get_json", _boom)
from easy_tdx.sina import SinaClient, SinaError
with pytest.raises(SinaError):
SinaClient().get_financial_report("600519")
def test_get_financial_report_num_limit(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""num 应限制返回期数(sorted reverse 后取前 num)。"""
resp = _make_response(
{
"20260331": _LRB_ITEMS_2026Q1,
"20251231": _LRB_ITEMS_2025,
"20250930": _LRB_ITEMS_2025,
"20250630": _LRB_ITEMS_2025,
}
)
_patch_http(monkeypatch, resp)
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519", num=2)
assert len(df) == 2
assert df.iloc[0]["报告期"] == "2026-03-31"
assert df.iloc[1]["报告期"] == "2025-12-31"
def test_get_financial_report_params_passed(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""paperCode/source/num 应正确传入 params。"""
captured: dict[str, Any] = {}
def _capture(url: str, params: dict[str, str], timeout: float = 15.0) -> Any:
captured.update(params)
return {"result": {"data": {"report_list": {}}}}
monkeypatch.setattr("easy_tdx.sina.client._http_get_json", _capture)
from easy_tdx.sina import SinaClient
SinaClient().get_financial_report("600519", report_type="fzb", num=4)
assert captured["paperCode"] == "sh600519"
assert captured["source"] == "fzb"
assert captured["num"] == "4"
assert captured["type"] == "0"
def test_get_financial_report_report_type_alias(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""中文别名(利润表)应归一化为 lrb 传入 source。"""
captured: dict[str, Any] = {}
def _capture(url: str, params: dict[str, str], timeout: float = 15.0) -> Any:
captured.update(params)
return {"result": {"data": {"report_list": {}}}}
monkeypatch.setattr("easy_tdx.sina.client._http_get_json", _capture)
from easy_tdx.sina import SinaClient
SinaClient().get_financial_report("000001", report_type="现金流量表")
assert captured["source"] == "llb"
assert captured["paperCode"] == "sz000001"
def test_get_financial_report_invalid_report_type_raises() -> None:
"""无法识别的 report_type 应抛 ValueErrornormalize 阶段,不触网)。"""
from easy_tdx.sina import SinaClient
with pytest.raises(ValueError, match="无法识别"):
SinaClient().get_financial_report("600519", report_type="xyz")
def test_get_financial_report_skips_non_dict_period(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""report_list 中某期值非 dict(如 list)应跳过该期。"""
resp = {
"result": {
"data": {
"report_list": {
"20260331": {"data": _LRB_ITEMS_2026Q1}, # 正常(含 data 键)
"20251231": ["not", "a", "dict"], # 异常,跳过
}
}
}
}
_patch_http(monkeypatch, resp)
from easy_tdx.sina import SinaClient
df = SinaClient().get_financial_report("600519")
assert len(df) == 1
assert df.iloc[0]["报告期"] == "2026-03-31"
# ---------------------------------------------------------------------------
# urllib helper 烟雾测试(不触网,验证 query 拼接 + headers
# ---------------------------------------------------------------------------
def test_http_get_json_builds_query_string(monkeypatch: pytest.MonkeyPatch) -> None:
"""_http_get_json 应把 params urlencode 到 URL。"""
import easy_tdx.sina.client as mod
captured: dict[str, Any] = {}
class _FakeResp:
def read(self) -> bytes:
return json.dumps({"ok": 1}).encode("utf-8")
def __enter__(self) -> _FakeResp:
return self
def __exit__(self, *args: Any) -> None:
pass
def _fake_urlopen(req: Any, timeout: float = 15.0) -> _FakeResp:
captured["url"] = req.full_url
captured["headers"] = {k.lower(): v for k, v in req.header_items()}
return _FakeResp()
monkeypatch.setattr(mod.urlrequest, "urlopen", _fake_urlopen)
result = mod._http_get_json(
"https://example.com/api", {"paperCode": "sh600519", "source": "lrb"}
)
assert result == {"ok": 1}
assert "paperCode=sh600519" in captured["url"]
assert "source=lrb" in captured["url"]
assert "user-agent" in captured["headers"]
def test_http_get_json_no_params(monkeypatch: pytest.MonkeyPatch) -> None:
"""params 为空时 URL 不附加 ?。"""
import easy_tdx.sina.client as mod
class _FakeResp:
def read(self) -> bytes:
return b'{"ok": 1}'
def __enter__(self) -> _FakeResp:
return self
def __exit__(self, *args: Any) -> None:
pass
monkeypatch.setattr(mod.urlrequest, "urlopen", lambda req, timeout=15.0: _FakeResp())
assert mod._http_get_json("https://example.com/x", {}) == {"ok": 1}
+1
View File
@@ -269,6 +269,7 @@ def test_full_app_routes_registered():
"/api/v1/block",
"/api/v1/chanlun",
"/api/v1/announcements",
"/api/v1/sina/financial-report",
"/ws/realtime",
]
for prefix in expected_prefixes: