simonlin1212/a-stock-data

**#46 东财资金流将沪市 ETF 510300 错误路由为深市**（感谢 [@zmcmcc](https://github.com/zmcmcc)）— **已修**

**#47 datacenter filter 的 `>` `<` 未 URL 编码**（[@alicessr](https://github.com/alicessr)）— **不适用于本项目，不改**

issue 的前提是「`requests.get(url)` 不会自动编码 `>` `<`」。但本项目的 `eastmoney_datacenter()`

感谢 [@tongflau-dongzhu](https://github.com/tongflau-dongzhu) 的报告——根因、复现和

- **备用源速查 K线行补腾讯分钟 K 线（#43）**：同花顺 K 线备胎只有 30/60 分，mootdx 一挂就无 5 分钟源。补腾讯 `ifzq.gtimg.cn/appstock/app/kline/mkline`（`m1/m5/m15/m30/m60`，≤320 根，零鉴权不封 IP，需 `Referer: https://gu.qq.com/`）。⚠<U+FE0F> 文档标注字段坑：返回数组第 7 个字段是**换手率基点**不是成交额（当成交额读会小三个数量级），成交额需自

Apache License                            Version 2.0, January 2004                         http://www.apache.org/licenses/

http://www.apache.org/licenses/

http://www.apache.org/licenses/LICENSE-2.0

<p align="center">   <a href="LICENSE"><img src="https://img.shields.io/badge/License-Apache_2.0-blue.svg" alt="License"></a>

<a href="LICENSE"><img src="https://img.shields.io/badge/License-Apache_2.0-blue.svg" alt="License"></a>

<img src="https://img.shields.io/badge/Python-3.9+-3776AB?logo=python&logoColor=white" alt="Python">

<a href="https://github.com/simonlin1212/a-stock-data/stargazers"><img src="https://img.shields.io/github/stars/simonlin1212/a-stock-data?style=social" alt="Stars"></a>

<br>   <img src="https://img.shields.io/badge/layers-15-2ea44f.svg" alt="Layers">   <img src="https://img.shields.io/badge/endpoints-87-2ea44f.svg" alt="Endpoints">

<img src="https://img.shields.io/badge/layers-15-2ea44f.svg" alt="Layers">

<img src="https://img.shields.io/badge/endpoints-87-2ea44f.svg" alt="Endpoints">

<img src="https://img.shields.io/badge/sources-34-2ea44f.svg" alt="Sources">

<img src="https://img.shields.io/badge/auth-zero-success.svg" alt="Zero Auth">

> 兼容 [Claude Code](https://github.com/anthropics/claude-code) · [Codex](https://github.com/openai/codex) · [OpenClaw](https://github.com/openclaw/openclaw) · [WorkBuddy](https://www.workbuddy.cn/) · [Muse](https://muse.ai/) · [Grok Bot](htt

# 2. 把 SKILL.md 放进去 curl -o ~/.claude/skills/a-stock-data/SKILL.md \

curl -o ~/.claude/skills/a-stock-data/SKILL.md \

https://raw.githubusercontent.com/simonlin1212/a-stock-data/main/SKILL.md

除 iwencai 外，其余已接入数据源无需用户注册或 API Key；V3.8 的中证、国证和北交所，以及 V3.9 新增的 12 个来源（通达信官网、华尔街见闻、央视网、上证e互动、中债、中国货币网、五家期货交易所、上金所）也不需要用户凭据（北交所使用官网匿名 Cookie）。仅 iwencai 语义搜索需要 API Key（[申请地址](https://www.iwencai.com/skillhub)）。候选增强源不计入上述能力。

**作者：** Simon 林 · X [@linsizhen](https://x.com/linsizhen) · 邮箱：[simonlin0423@gmail.com](mailto:simonlin0423@gmail.com)

<p align="center">   <a href="LICENSE"><img src="https://img.shields.io/badge/License-Apache_2.0-blue.svg" alt="License"></a>

<a href="LICENSE"><img src="https://img.shields.io/badge/License-Apache_2.0-blue.svg" alt="License"></a>

<img src="https://img.shields.io/badge/Python-3.9+-3776AB?logo=python&logoColor=white" alt="Python">

<a href="https://github.com/simonlin1212/a-stock-data/stargazers"><img src="https://img.shields.io/github/stars/simonlin1212/a-stock-data?style=social" alt="Stars"></a>

<br>   <img src="https://img.shields.io/badge/layers-15-2ea44f.svg" alt="Layers">   <img src="https://img.shields.io/badge/endpoints-87-2ea44f.svg" alt="Endpoints">

<img src="https://img.shields.io/badge/layers-15-2ea44f.svg" alt="Layers">

<img src="https://img.shields.io/badge/endpoints-87-2ea44f.svg" alt="Endpoints">

<img src="https://img.shields.io/badge/sources-34-2ea44f.svg" alt="Sources">

<img src="https://img.shields.io/badge/auth-zero-success.svg" alt="Zero Auth">

> Compatible with [Claude Code](https://github.com/anthropics/claude-code) · [Codex](https://github.com/openai/codex) · [OpenClaw](https://github.com/openclaw/openclaw) · [WorkBuddy](https://www.workbuddy.cn/) · [Muse](https://muse.ai/) · [

# 2. Download SKILL.md curl -o ~/.claude/skills/a-stock-data/SKILL.md \

curl -o ~/.claude/skills/a-stock-data/SKILL.md \

https://raw.githubusercontent.com/simonlin1212/a-stock-data/main/SKILL.md

All integrated sources except iwencai require no user registration or API key. CSI, CNI and BSE (V3.8) and the 12 sources added in V3.9 (TDX official site, Wallstreetcn, CCTV, SSE e-Interaction, ChinaBond, China Money, the five futures exch

**Author:** Simon Lin · X [@linsizhen](https://x.com/linsizhen) · Email: [simonlin0423@gmail.com](mailto:simonlin0423@gmail.com)

> 📦 项目主页：https://github.com/simonlin1212/a-stock-data — 更新、反馈、支持作者

> 作者：Simon 林 · X [@linsizhen](https://x.com/linsizhen) · 邮箱：simonlin0423@gmail.com

### ⚠<U+FE0F> 实测封禁案例（2026-06-30，一手数据，感谢 [@luodada99](https://github.com/luodada99) issue #36）

[Credential-related source omitted]

[Credential-related source omitted]

export IWENCAI_BASE_URL="https://openapi.iwencai.com"

# 申请地址: https://www.iwencai.com/skillhub

"""TCP 握手探测（快速粗筛）。注意：握手成功 ≠ 能取数，必须再经 _validate 验活。"""     try:         with socket.create_connection((ip, port), timeout=timeout):

try:         with socket.create_connection((ip, port), timeout=timeout):             return True

with socket.create_connection((ip, port), timeout=timeout):

UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36" DATACENTER_URL = "https://datacenter-web.eastmoney.com/api/data/v1/get"

DATACENTER_URL = "https://datacenter-web.eastmoney.com/api/data/v1/get"

# 所有 eastmoney.com 请求一律走 em_get()：串行限流（最小间隔 + 随机抖动）+ 复用 # Keep-Alive 会话，批量调用时自动降速，避免被封。详见「数据源优先级 & 东财防封」章节。 EM_SESSION = requests.Session()

# Keep-Alive 会话，批量调用时自动降速，避免被封。详见「数据源优先级 & 东财防封」章节。 EM_SESSION = requests.Session() EM_SESSION.headers.update({"User-Agent": UA})

EM_SESSION = requests.Session()

# 连接失败、握手或等响应头超时、429 / 5xx 指数退避重试（写法同 EM_SESSION）；读响应体中途超时不重试， # 也不按服务端给的 Retry-After 等待。 V39_RETRY_SESSION = requests.Session()

# 也不按服务端给的 Retry-After 等待。 V39_RETRY_SESSION = requests.Session() try:

V39_RETRY_SESSION = requests.Session()

url = "https://qt.gtimg.cn/q=" + ",".join(prefixed)     req = urllib.request.Request(url)

req = urllib.request.Request(url)

req.add_header("User-Agent", "Mozilla/5.0")     resp = urllib.request.urlopen(req, timeout=10)     data = resp.read().decode("gbk")

resp = urllib.request.urlopen(req, timeout=10)

# 三个入口是同一后端，但限流各自独立（#52 实测单域名约 600 次后返回空 JSON）。 TENCENT_KLINE_HOSTS = ["https://web.ifzq.gtimg.cn",

TENCENT_KLINE_HOSTS = ["https://web.ifzq.gtimg.cn",

"https://proxy.finance.qq.com/ifzqgtimg",

"https://ifzq.gtimg.cn"]

try:             response = _v39_http(host + path, params={"param": param},                                  headers={"Referer": "https://gu.qq.com/"}, timeout=(8, 20))

response = _v39_http(host + path, params={"param": param},                                  headers={"Referer": "https://gu.qq.com/"}, timeout=(8, 20))             payload = _v39_json(response) if response.text.strip() else {}

headers={"Referer": "https://gu.qq.com/"}, timeout=(8, 20))

通达信官网每天发布的增量数据包（HTTP 下载约 2.7MB），一次拿到**沪深北全部证券**当日的昨收、开高低收、成交量（股）、 成交额（元）。它走 HTTP，和 §1.7 失效的 TCP 行情命令是两条路。适合收盘后全市场截面筛选、每日落库。 二进制布局参考 [jing2uo/tdx2db](https://github.com/jing2uo/tdx2db)（MIT），已与腾讯收盘价对拍。

成交额（元）。它走 HTTP，和 §1.7 失效的 TCP 行情命令是两条路。适合收盘后全市场截面筛选、每日落库。 二进制布局参考 [jing2uo/tdx2db](https://github.com/jing2uo/tdx2db)（MIT），已与腾讯收盘价对拍。 非交易日、或当天包还没发布（通常收盘后数小时）抛 `ValueError`，不返回空表。

二进制布局参考 [jing2uo/tdx2db](https://github.com/jing2uo/tdx2db)（MIT），已与腾讯收盘价对拍。

TDX_PACKAGE_URL = "https://www.tdx.com.cn/products/data/data/g4day/{ymd}.zip"

TENCENT_TICK_URL = "https://stock.gtimg.cn/data/index.php"

TENCENT_QT_URL = "https://qt.gtimg.cn/q="

def baidu_kline_with_ma(code: str, start_time: str = "") -> dict:     """百度股市通K线 — 独有能力: 返回时自带 ma5/ma10/ma20 均价"""     url = "https://finance.pae.baidu.com/selfselect/getstockquotation"

"""百度股市通K线 — 独有能力: 返回时自带 ma5/ma10/ma20 均价"""     url = "https://finance.pae.baidu.com/selfselect/getstockquotation"     params = {

url = "https://finance.pae.baidu.com/selfselect/getstockquotation"

"User-Agent": "Mozilla/5.0",         "Accept": "application/vnd.finance-web.v1+json",         "Origin": "https://gushitong.baidu.com",

"Accept": "application/vnd.finance-web.v1+json",         "Origin": "https://gushitong.baidu.com",         "Referer": "https://gushitong.baidu.com/",

"Origin": "https://gushitong.baidu.com",

"Referer": "https://gushitong.baidu.com/",     }     r = requests.get(url, params=params, headers=headers, timeout=10)

}     r = requests.get(url, params=params, headers=headers, timeout=10)     d = r.json()

r = requests.get(url, params=params, headers=headers, timeout=10)

prefix = {"xshg": "sh", "xshe": "sz"}.get(m.group(1).lower(), m.group(1).lower()) if m else get_prefix(digits)     symbol = f"{prefix}{digits}"     url = f"https://finance.sina.com.cn/realstock/company/{symbol}/{kind}.js"

symbol = f"{prefix}{digits}"     url = f"https://finance.sina.com.cn/realstock/company/{symbol}/{kind}.js"     r = requests.get(url, headers={"User-Agent": "Mozilla/5.0",

url = f"https://finance.sina.com.cn/realstock/company/{symbol}/{kind}.js"     r = requests.get(url, headers={"User-Agent": "Mozilla/5.0",                                    "Referer": "https://finance.sina.com.cn/"}, timeout=10)

r = requests.get(url, headers={"User-Agent": "Mozilla/5.0",

"Referer": "https://finance.sina.com.cn/"}, timeout=10)

REPORT_API = "https://reportapi.eastmoney.com/report/list"

PDF_TPL = "https://pdf.dfcfw.com/pdf/H3_{info_code}_1.pdf"

}         r = em_get(REPORT_API, params=params,                    headers={"Referer": "https://data.eastmoney.com/"}, timeout=30)  # 已内置限流

r = em_get(REPORT_API, params=params,                    headers={"Referer": "https://data.eastmoney.com/"}, timeout=30)  # 已内置限流         d = r.json()

headers={"Referer": "https://data.eastmoney.com/"}, timeout=30)  # 已内置限流

return str(target)     url = PDF_TPL.format(info_code=info_code)     r = em_get(url, headers={"Referer": "https://data.eastmoney.com/"}, timeout=60)

url = PDF_TPL.format(info_code=info_code)     r = em_get(url, headers={"Referer": "https://data.eastmoney.com/"}, timeout=60)     if r.status_code == 200 and len(r.content) >= 1024:

r = em_get(url, headers={"Referer": "https://data.eastmoney.com/"}, timeout=60)

}         r = em_get(REPORT_API, params=params,                    headers={"Referer": "https://data.eastmoney.com/"}, timeout=30)  # 已内置限流

r = em_get(REPORT_API, params=params,                    headers={"Referer": "https://data.eastmoney.com/"}, timeout=30)  # 已内置限流         d = r.json()

headers={"Referer": "https://data.eastmoney.com/"}, timeout=30)  # 已内置限流

"""     code = norm_ticker(code, stock_only=True)   # URL 路径只认纯 6 位，带前缀会 404 到空表     url = f"https://basic.10jqka.com.cn/new/{code}/worth.html"

code = norm_ticker(code, stock_only=True)   # URL 路径只认纯 6 位，带前缀会 404 到空表     url = f"https://basic.10jqka.com.cn/new/{code}/worth.html"     headers = {

url = f"https://basic.10jqka.com.cn/new/{code}/worth.html"

headers = {         "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",         "Referer": "https://basic.10jqka.com.cn/",

"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",         "Referer": "https://basic.10jqka.com.cn/",     }

"Referer": "https://basic.10jqka.com.cn/",     }     r = requests.get(url, headers=headers, timeout=15)

}     r = requests.get(url, headers=headers, timeout=15)     r.encoding = "gbk"

r = requests.get(url, headers=headers, timeout=15)

[Credential-related source omitted]

[Credential-related source omitted]

[Credential-related source omitted]

"size": size,     }     r = requests.post(

}     r = requests.post(         f"{IWENCAI_BASE}/v1/comprehensive/search",

r = requests.post(

"expand_index": "true",     }     r = requests.post(

}     r = requests.post(         f"{IWENCAI_BASE}/v1/query2data",

r = requests.post(

SINA_REPORT_URL = "https://vip.stock.finance.sina.com.cn/q/go.php/vReport_List/kind/{kind}/index.phtml"

try:             response = _v39_http(url, params=params,                                  headers={"Referer": "https://finance.sina.com.cn/"})

response = _v39_http(url, params=params,                                  headers={"Referer": "https://finance.sina.com.cn/"})         finally:

headers={"Referer": "https://finance.sina.com.cn/"})

url = (         f"http://zx.10jqka.com.cn/event/api/getharden/"

f"http://zx.10jqka.com.cn/event/api/getharden/"

)     }     r = requests.get(url, headers=headers, timeout=10)

}     r = requests.get(url, headers=headers, timeout=10)     data = r.json()

r = requests.get(url, headers=headers, timeout=10)

),     "Host": "data.hexin.cn",     "Referer": "https://data.hexin.cn/",

"Host": "data.hexin.cn",     "Referer": "https://data.hexin.cn/", }

"Referer": "https://data.hexin.cn/",

单位: 亿元     """     url = "https://data.hexin.cn/market/hsgtApi/method/dayChart/"

"""     url = "https://data.hexin.cn/market/hsgtApi/method/dayChart/"     r = requests.get(url, headers=HSGT_HEADERS, timeout=10)

url = "https://data.hexin.cn/market/hsgtApi/method/dayChart/"     r = requests.get(url, headers=HSGT_HEADERS, timeout=10)     d = r.json()

r = requests.get(url, headers=HSGT_HEADERS, timeout=10)

"fields": "f12,f14,f3,f128",     }     headers = {"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"}

}     headers = {"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"}     try:

headers = {"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"}

try:         r = em_get("https://push2.eastmoney.com/api/qt/slist/get",                    params=params, headers=headers, timeout=15)

r = em_get("https://push2.eastmoney.com/api/qt/slist/get",

#    东财这个**个股**资金流接口本身不覆盖 ETF。ETF 资金流请另找端点。     secid = em_secid(code)     url = "https://push2.eastmoney.com/api/qt/stock/fflow/kline/get"

secid = em_secid(code)     url = "https://push2.eastmoney.com/api/qt/stock/fflow/kline/get"     params = {

url = "https://push2.eastmoney.com/api/qt/stock/fflow/kline/get"

headers = {         "User-Agent": UA,         "Referer": "https://quote.eastmoney.com/",

"User-Agent": UA,         "Referer": "https://quote.eastmoney.com/",         "Origin": "https://quote.eastmoney.com",

"Referer": "https://quote.eastmoney.com/",

"Origin": "https://quote.eastmoney.com",

返回: {top: [...], bottom: [...], total: int}     """     url = "https://push2.eastmoney.com/api/qt/clist/get"

"""     url = "https://push2.eastmoney.com/api/qt/clist/get"     params = {

url = "https://push2.eastmoney.com/api/qt/clist/get"

url = "https://push2.eastmoney.com/api/qt/clist/get"

"""     market_code = em_market_code(code)      # #46     url = "https://push2his.eastmoney.com/api/qt/stock/fflow/daykline/get"

market_code = em_market_code(code)      # #46     url = "https://push2his.eastmoney.com/api/qt/stock/fflow/daykline/get"     params = {

url = "https://push2his.eastmoney.com/api/qt/stock/fflow/daykline/get"

headers = {         "User-Agent": UA,         "Referer": "https://quote.eastmoney.com/",

"User-Agent": UA,         "Referer": "https://quote.eastmoney.com/",         "Origin": "https://quote.eastmoney.com",

"Referer": "https://quote.eastmoney.com/",

"Origin": "https://quote.eastmoney.com",

SSE_ETF_SHARES_URL = "https://query.sse.com.cn/commonQuery.do"

SZSE_FUND_LIST_URL = "https://fund.szse.cn/api/report/ShowReport/data"

"pageHelp.beginPage": 1, "pageHelp.cacheSize": 1, "pageHelp.endPage": 1}     response = _v39_http(SSE_ETF_SHARES_URL, params=params,                          headers={"Referer": "https://www.sse.com.cn/"})

response = _v39_http(SSE_ETF_SHARES_URL, params=params,                          headers={"Referer": "https://www.sse.com.cn/"})     payload = _v39_json(response)

headers={"Referer": "https://www.sse.com.cn/"})

params={"SHOWTYPE": "JSON", "CATALOGID": "1000_lf", "TABKEY": "tab1",                                      "selectJjlb": "ETF", "PAGENO": page},                              headers={"Referer": "https://fund.szse.cn/"})

"selectJjlb": "ETF", "PAGENO": page},                              headers={"Referer": "https://fund.szse.cn/"})         payload = _v39_json(response)

headers={"Referer": "https://fund.szse.cn/"})

# 构造 JSONP 参数     cb = "jQuery_news"     url = "https://search-api-web.eastmoney.com/search/jsonp"

cb = "jQuery_news"     url = "https://search-api-web.eastmoney.com/search/jsonp"     inner_params = json.dumps({

url = "https://search-api-web.eastmoney.com/search/jsonp"

}, separators=(',', ':'))     params = {"cb": cb, "param": inner_params}     headers = {"User-Agent": UA, "Referer": "https://so.eastmoney.com/"}

params = {"cb": cb, "param": inner_params}     headers = {"User-Agent": UA, "Referer": "https://so.eastmoney.com/"}     r = em_get(url, params=params, headers=headers, timeout=15)

headers = {"User-Agent": UA, "Referer": "https://so.eastmoney.com/"}

qs = "&".join(f"{k}={params[k]}" for k in sorted(params))     sign = hashlib.md5(hashlib.sha1(qs.encode()).hexdigest().encode()).hexdigest()     url = f"https://www.cls.cn/v1/roll/get_roll_list?{qs}&sign={sign}"

sign = hashlib.md5(hashlib.sha1(qs.encode()).hexdigest().encode()).hexdigest()     url = f"https://www.cls.cn/v1/roll/get_roll_list?{qs}&sign={sign}"     headers = {"User-Agent": UA, "Referer": "https://www.cls.cn/"}

url = f"https://www.cls.cn/v1/roll/get_roll_list?{qs}&sign={sign}"     headers = {"User-Agent": UA, "Referer": "https://www.cls.cn/"}     r = requests.get(url, headers=headers, timeout=10)

headers = {"User-Agent": UA, "Referer": "https://www.cls.cn/"}     r = requests.get(url, headers=headers, timeout=10)     d = r.json()

r = requests.get(url, headers=headers, timeout=10)

返回: [{title, summary, time}]     """     url = "https://np-weblist.eastmoney.com/comm/web/getFastNewsList"

"""     url = "https://np-weblist.eastmoney.com/comm/web/getFastNewsList"     params = {

url = "https://np-weblist.eastmoney.com/comm/web/getFastNewsList"

"req_trace": str(uuid.uuid4()),     }     headers = {"User-Agent": UA, "Referer": "https://kuaixun.eastmoney.com/"}

}     headers = {"User-Agent": UA, "Referer": "https://kuaixun.eastmoney.com/"}     r = em_get(url, params=params, headers=headers, timeout=10)

headers = {"User-Agent": UA, "Referer": "https://kuaixun.eastmoney.com/"}

WSCN_LIVES_URL = "https://api-one-wscn.awtmt.com/apiv1/content/lives"

CCTV_DAY_URL = "https://tv.cctv.com/lm/xwlb/day/{ymd}.shtml"

"""     market_code = em_market_code(code)      # #46     url = "https://push2.eastmoney.com/api/qt/stock/get"

market_code = em_market_code(code)      # #46     url = "https://push2.eastmoney.com/api/qt/stock/get"     params = {

url = "https://push2.eastmoney.com/api/qt/stock/get"

prefix = get_prefix(code)               # #46：51x/588x/900x 也是沪市     paper_code = f"{prefix}{code}"     url = "https://quotes.sina.cn/cn/api/openapi.php/CompanyFinanceService.getFinanceReport2022"

paper_code = f"{prefix}{code}"     url = "https://quotes.sina.cn/cn/api/openapi.php/CompanyFinanceService.getFinanceReport2022"     params = {

url = "https://quotes.sina.cn/cn/api/openapi.php/CompanyFinanceService.getFinanceReport2022"

}     headers = {"User-Agent": UA}     r = requests.get(url, params=params, headers=headers, timeout=15)

headers = {"User-Agent": UA}     r = requests.get(url, params=params, headers=headers, timeout=15)     # 新浪实际结构: result.data.report_list 是「按报告期(如 '20260331')为键」的 dict,

r = requests.get(url, params=params, headers=headers, timeout=15)

SW_URL = "https://www.swsresearch.com/swindex/pdf/SwClass2021/StockClassifyUse_stock.xls"

"""申万行业归属变迁史 — 每只股票每次行业调整一行"""     try:         r = requests.get(SW_URL, headers={"User-Agent": "Mozilla/5.0"}, timeout=60)

try:         r = requests.get(SW_URL, headers={"User-Agent": "Mozilla/5.0"}, timeout=60)         r.raise_for_status()

r = requests.get(SW_URL, headers={"User-Agent": "Mozilla/5.0"}, timeout=60)

EM_CLIST_HOSTS = ["https://push2.eastmoney.com", "https://push2delay.eastmoney.com"]

raise RuntimeError(f"东财 clist 翻页后 {len(rows)} 条，与 total={total} 不符")         return rows, url     raise requests.ConnectionError("东财 push2 / push2delay 均不可达: " + "; ".join(errors))

raise requests.ConnectionError("东财 push2 / push2delay 均不可达: " + "; ".join(errors))

if not _CNINFO_ORGID_MAP:         try:             r = requests.get("http://www.cninfo.com.cn/new/data/szse_stock.json",

try:             r = requests.get("http://www.cninfo.com.cn/new/data/szse_stock.json",                              headers={"User-Agent": UA}, timeout=15)

r = requests.get("http://www.cninfo.com.cn/new/data/szse_stock.json",

返回: [{title, type, date, url}]     """     url = "https://www.cninfo.com.cn/new/hisAnnouncement/query"

"""     url = "https://www.cninfo.com.cn/new/hisAnnouncement/query"     org_id = _cninfo_orgid(code)   # 动态查真实 orgId（#19 修复，自带硬编码 fallback）

url = "https://www.cninfo.com.cn/new/hisAnnouncement/query"

"User-Agent": UA,         "Content-Type": "application/x-www-form-urlencoded",         "Referer": "https://www.cninfo.com.cn/new/disclosure",

"Content-Type": "application/x-www-form-urlencoded",         "Referer": "https://www.cninfo.com.cn/new/disclosure",         "Origin": "https://www.cninfo.com.cn",

"Referer": "https://www.cninfo.com.cn/new/disclosure",

"Origin": "https://www.cninfo.com.cn",     }     r = requests.post(url, data=payload, headers=headers, timeout=15)

}     r = requests.post(url, data=payload, headers=headers, timeout=15)     d = r.json()

r = requests.post(url, data=payload, headers=headers, timeout=15)

"type": item.get("announcementTypeName", ""),             "date": _cninfo_ts_to_date(item.get("announcementTime")),             "url": f"https://www.cninfo.com.cn/new/disclosure/detail?annoId={item.get('announcementId', '')}",

"date": _cninfo_ts_to_date(item.get("announcementTime")),             "url": f"https://www.cninfo.com.cn/new/disclosure/detail?annoId={item.get('announcementId', '')}",         })

"url": f"https://www.cninfo.com.cn/new/disclosure/detail?annoId={item.get('announcementId', '')}",

endpoint: getTopicZTPool / getTopicZBPool / getTopicDTPool / getYesterdayZTPool     返回 data.pool 原始列表（data 为 null = 非交易日 / 参数错）。"""     url = f"https://push2ex.eastmoney.com/{endpoint}"

返回 data.pool 原始列表（data 为 null = 非交易日 / 参数错）。"""     url = f"https://push2ex.eastmoney.com/{endpoint}"     params = {"ut": ZTB_UT, "dpt": "wz.ztzt", "Pageindex": 0,

url = f"https://push2ex.eastmoney.com/{endpoint}"

params = {"ut": ZTB_UT, "dpt": "wz.ztzt", "Pageindex": 0,               "pagesize": 10000, "sort": sort, "date": date}     headers = {"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"}

"pagesize": 10000, "sort": sort, "date": date}     headers = {"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"}     try:

headers = {"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"}

seal_rate(封板成功率,0~1)/break_times(炸板次数)/seal_amount(封单额,元)/     high_days(几天几板)/first_time(首次涨停时间)/is_again(是否回封 0/1)"""     url = "https://data.10jqka.com.cn/dataapi/limit_up/limit_up_pool"

high_days(几天几板)/first_time(首次涨停时间)/is_again(是否回封 0/1)"""     url = "https://data.10jqka.com.cn/dataapi/limit_up/limit_up_pool"     params = {"page": 1, "limit": 200,

url = "https://data.10jqka.com.cn/dataapi/limit_up/limit_up_pool"

"filter": "HS,GEM2STAR", "order_field": "330324", "order_type": "0", "date": date}     try:         r = requests.get(url, params=params, headers={"User-Agent": UA}, timeout=10)

try:         r = requests.get(url, params=params, headers={"User-Agent": UA}, timeout=10)         info = (r.json().get("data") or {}).get("info", [])

r = requests.get(url, params=params, headers={"User-Agent": UA}, timeout=10)

MONITOR_URL = "https://mobappconfig.securities.eastmoney.com/emcfg/stock_monitor.json"

返回: [{code, name, market, start, end, link}]     """     r = em_get(MONITOR_URL, headers={"Referer": "https://vipmoney.eastmoney.com/"}, timeout=20)

"""     r = em_get(MONITOR_URL, headers={"Referer": "https://vipmoney.eastmoney.com/"}, timeout=20)     rows = r.json() or []

r = em_get(MONITOR_URL, headers={"Referer": "https://vipmoney.eastmoney.com/"}, timeout=20)

ANOMALY_BASE = "https://dycalchis.eastmoney.com/price-anomaly"

params = {**HQ_PARAMS, "pageSize": str(page_size), "pageNo": str(page_no), **extra}     r = em_get(f"{ANOMALY_BASE}/{path}", params=params,                headers={"Referer": "https://vipmoney.eastmoney.com/"}, timeout=20)

r = em_get(f"{ANOMALY_BASE}/{path}", params=params,                headers={"Referer": "https://vipmoney.eastmoney.com/"}, timeout=20)     d = r.json()

headers={"Referer": "https://vipmoney.eastmoney.com/"}, timeout=20)

SINA_OPT_HDR = {"Referer": "https://stock.finance.sina.com.cn/", "User-Agent": UA}

def _sina_opt_list(param: str) -> list:     """新浪 hq.sinajs.cn 取值（GBK，逗号分隔，去 var hq_str_XXX="..." 壳）。"""     r = requests.get(f"https://hq.sinajs.cn/list={param}", headers=SINA_OPT_HDR, timeout=10)

"""新浪 hq.sinajs.cn 取值（GBK，逗号分隔，去 var hq_str_XXX="..." 壳）。"""     r = requests.get(f"https://hq.sinajs.cn/list={param}", headers=SINA_OPT_HDR, timeout=10)     r.encoding = "gbk"

r = requests.get(f"https://hq.sinajs.cn/list={param}", headers=SINA_OPT_HDR, timeout=10)

cate = {"510050": "50ETF", "510300": "300ETF",             "588000": "科创50ETF", "510500": "500ETF"}.get(underlying, "50ETF")     url = ("https://stock.finance.sina.com.cn/futures/api/openapi.php/"

"588000": "科创50ETF", "510500": "500ETF"}.get(underlying, "50ETF")     url = ("https://stock.finance.sina.com.cn/futures/api/openapi.php/"            f"StockOptionService.getStockName?exchange=null&cate={cate}")

url = ("https://stock.finance.sina.com.cn/futures/api/openapi.php/"

f"StockOptionService.getStockName?exchange=null&cate={cate}")     try:         months = requests.get(url, headers=SINA_OPT_HDR, timeout=10).json()["result"]["data"]["contractMonth"]

try:         months = requests.get(url, headers=SINA_OPT_HDR, timeout=10).json()["result"]["data"]["contractMonth"]     except Exception as e:

months = requests.get(url, headers=SINA_OPT_HDR, timeout=10).json()["result"]["data"]["contractMonth"]

> **坑：** ① 新浪源 **GBK 编码**、**逗号分隔**、需去 `var hq_str_XXX="..."` 壳。② 必带 `Referer: https://stock.finance.sina.com.cn/`，否则 403。③ 希腊字母解析 **`[raw[0]] + raw[4:]`**——`raw[1:4]` 是 3 个空串，不跳过则 Delta/IV 全错位。④ `iv` 是小数（0.1735 = 17.35%）。⑤ 300ETF(510300)、科创

answerer(回答方)/ask_time。"""     try:         r1 = requests.post("https://irm.cninfo.com.cn/newircs/index/queryKeyboardInfo",

try:         r1 = requests.post("https://irm.cninfo.com.cn/newircs/index/queryKeyboardInfo",             data={"keyWord": code}, headers={"User-Agent": UA}, timeout=10)

r1 = requests.post("https://irm.cninfo.com.cn/newircs/index/queryKeyboardInfo",

params = {"_t": 1, "stockcode": code, "orgId": org_id, "pageSize": page_size,                   "pageNum": page_num, "keyWord": "", "startDay": "", "endDay": ""}         r2 = requests.post("https://irm.cninfo.com.cn/newircs/company/question

"pageNum": page_num, "keyWord": "", "startDay": "", "endDay": ""}         r2 = requests.post("https://irm.cninfo.com.cn/newircs/company/question",             params=params, headers={"User-Agent": UA}, timeout=10)

r2 = requests.post("https://irm.cninfo.com.cn/newircs/company/question",

返回每只: rank/code/name/heat(人气值)/pct/rank_chg(排名变化)/concepts(概念标签)/tag。"""     try:         r = requests.get("https://dq.10jqka.com.cn/fuyao/hot_list_data/out/hot_list/v1/stock",

try:         r = requests.get("https://dq.10jqka.com.cn/fuyao/hot_list_data/out/hot_list/v1/stock",             params={"stock_type": "a", "type": period, "list_type": "normal"},

r = requests.get("https://dq.10jqka.com.cn/fuyao/hot_list_data/out/hot_list/v1/stock",

"""东财人气榜（排名 + 排名变化 + 名称/价格）。返回 rank/code/name/price/pct/rank_chg。"""     try:         r = requests.post("https://emappdata.eastmoney.com/stockrank/getAllCurrentList",

try:         r = requests.post("https://emappdata.eastmoney.com/stockrank/getAllCurrentList",             json={**EM_HOT_BODY, "marketType": "", "pageNo": 1, "pageSize": top},

r = requests.post("https://emappdata.eastmoney.com/stockrank/getAllCurrentList",

# 人气榜只给带前缀代码，用 push2 ulist.np 批量补名称/价格         secids = [("0." if it["sc"].startswith("SZ") else "1.") + it["sc"][2:] for it in data]         u = requests.get("https://push2.eastmoney.com/api/qt/ulist.np/get",

secids = [("0." if it["sc"].startswith("SZ") else "1.") + it["sc"][2:] for it in data]         u = requests.get("https://push2.eastmoney.com/api/qt/ulist.np/get",             params={"ut": "f057cbcbce2a86e2866ab8877db1d059", "fltt": 2, "inv

u = requests.get("https://push2.eastmoney.com/api/qt/ulist.np/get",

params={"ut": "f057cbcbce2a86e2866ab8877db1d059", "fltt": 2, "invt": 2,                     "fields": "f14,f3,f12,f2", "secids": ",".join(secids)},             headers={"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"}, timeout=1

"fields": "f14,f3,f12,f2", "secids": ",".join(secids)},             headers={"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"}, timeout=10)         diff = (u.json().get("data") or {}).get("diff") or []

headers={"User-Agent": UA, "Referer": "https://quote.eastmoney.com/"}, timeout=10)

try:         prefix = get_prefix(code).upper()   # #46；emappdata 用大写 SH/SZ/BJ         r = requests.post("https://emappdata.eastmoney.com/stockrank/getHotStockRankList",

prefix = get_prefix(code).upper()   # #46；emappdata 用大写 SH/SZ/BJ         r = requests.post("https://emappdata.eastmoney.com/stockrank/getHotStockRankList",             json={**EM_HOT_BODY, "srcSecurityCode": prefix + code},

r = requests.post("https://emappdata.eastmoney.com/stockrank/getHotStockRankList",

SSE_E_BASE = "https://sns.sseinfo.com"

_UA = {"User-Agent": "Mozilla/5.0"} PBC_BASE = "https://www.pbc.gov.cn"

PBC_BASE = "https://www.pbc.gov.cn"

def _macro_get(url: str, timeout: int = 30) -> str:     r = requests.get(url, headers=_UA, timeout=timeout)

r = requests.get(url, headers=_UA, timeout=timeout)

content = requests.get(_abs_pbc(books[0]), headers=_UA, timeout=60).content

NBS_INDEX = "https://www.stats.gov.cn/sj/zxfb/"

"""与 §11.1 同名同实现 —— 本块按「端点路由速查」单独取用时也能独立跑，     不必先执行 §11.1。两处同时执行时后定义覆盖前者，行为一致，无副作用。"""     r = requests.get(url, headers=_UA, timeout=timeout)

不必先执行 §11.1。两处同时执行时后定义覆盖前者，行为一致，无副作用。"""     r = requests.get(url, headers=_UA, timeout=timeout)     r.raise_for_status()

r = requests.get(url, headers=_UA, timeout=timeout)

CHINABOND_HISTORY_URL = "https://yield.chinabond.com.cn/cbweb-pbc-web/pbc/historyQuery"

CHINAMONEY_FIXING_URL = "https://www.chinamoney.com.cn/r/cms/www/chinamoney/data/currency/{name}-chrt.csv"

raise ValueError("kind 只能是 'FR' 或 'FDR'")     url = CHINAMONEY_FIXING_URL.format(name=names[kind])     response = _v39_http(url, headers={"Referer": "https://www.chinamoney.com.cn/chinese/bkfrr/"})

url = CHINAMONEY_FIXING_URL.format(name=names[kind])     response = _v39_http(url, headers={"Referer": "https://www.chinamoney.com.cn/chinese/bkfrr/"})     rows = []

response = _v39_http(url, headers={"Referer": "https://www.chinamoney.com.cn/chinese/bkfrr/"})

WSCN_MACRO_URL = "https://api-one-wscn.awtmt.com/apiv1/finance/macrodatas"

def _official_get(url, params=None, referer=None):     response = requests.get(

response = requests.get(

if provider == "csi":         kind = "closeweight" if weights else "cons"         url = ("https://oss-ch.csindex.com.cn/static/html/csindex/public/uploads/file/"

kind = "closeweight" if weights else "cons"         url = ("https://oss-ch.csindex.com.cn/static/html/csindex/public/uploads/file/"                f"autofile/{kind}/{index_code}{kind}.xls")

url = ("https://oss-ch.csindex.com.cn/static/html/csindex/public/uploads/file/"

cols.append("权重(%)weight")     else:         url = "https://www.cnindex.com.cn/sample-detail/download-history"

else:         url = "https://www.cnindex.com.cn/sample-detail/download-history"         response = _official_get(url, {"indexcode": index_code})

url = "https://www.cnindex.com.cn/sample-detail/download-history"

"""中证近期 PE/股息率文件；不含 PB，两种股本口径不混用。"""     index_code = _official_code(index_code)     url = ("https://oss-ch.csindex.com.cn/static/html/csindex/public/uploads/file/"

index_code = _official_code(index_code)     url = ("https://oss-ch.csindex.com.cn/static/html/csindex/public/uploads/file/"            f"autofile/indicator/{index_code}indicator.xls")

url = ("https://oss-ch.csindex.com.cn/static/html/csindex/public/uploads/file/"

last = calendar.monthrange(year, month)[1]     expected = {datetime(year, month, day).date().isoformat() for day in range(1, last + 1)}     url = "https://www.szse.cn/api/report/exchange/onepersistenthour/monthList"

expected = {datetime(year, month, day).date().isoformat() for day in range(1, last + 1)}     url = "https://www.szse.cn/api/report/exchange/onepersistenthour/monthList"     response = _official_get(url, {"month": f"{year}-{month}"})

url = "https://www.szse.cn/api/report/exchange/onepersistenthour/monthList"

原始端点的发现与交叉核对参考： [AKShare 中证成分](https://github.com/akfamily/akshare/blob/main/akshare/index/index_cons.py)、

[AKShare 中证成分](https://github.com/akfamily/akshare/blob/main/akshare/index/index_cons.py)、

[AKShare 中证估值](https://github.com/akfamily/akshare/blob/main/akshare/index/index_stock_zh_csindex.py)、

[AKShare 国证](https://github.com/akfamily/akshare/blob/main/akshare/index/index_cni.py)、

[Qlib 交易日历](https://github.com/microsoft/qlib/blob/main/scripts/data_collector/utils.py)。

FUTURES_EXCHANGES = ("SHFE", "INE", "CZCE", "CFFEX", "GFEX") _SHFE_HOSTS = {"SHFE": "https://www.shfe.com.cn", "INE": "https://www.ine.cn"}

_SHFE_HOSTS = {"SHFE": "https://www.shfe.com.cn", "INE": "https://www.ine.cn"}

CZCE_FILE_URL = "https://www.czce.com.cn/cn/DFSStaticFiles/{kind}/{year}/{ymd}/{name}.txt"

CZCE_FIRST_DAY = "20150921"      # 郑商所现行文件路径的第一天；更早是另一套无表头 CSV，未接入 CFFEX_DAILY_URL = "http://www.cffex.com.cn/fzjy/mrhq/{ym}/{dd}/{ymd}_1.csv" CFFEX_DAILY_XML = "http://www.cffex.com.cn/fzjy/mrhq/{ym}/{dd}/index.xml"

CFFEX_DAILY_URL = "http://www.cffex.com.cn/fzjy/mrhq/{ym}/{dd}/{ymd}_1.csv"

CFFEX_DAILY_XML = "http://www.cffex.com.cn/fzjy/mrhq/{ym}/{dd}/index.xml"

CFFEX_RANK_URL = "http://www.cffex.com.cn/sj/ccpm/{ym}/{dd}/{product}_1.csv"

CFFEX_RANK_FIRST_DAY = {"IF": "20100416", "IH": "20150416", "IC": "20150416", "IM": "20220722",                         "TS": "20180817", "TF": "20130906", "T": "20150320", "TL": "20230421"} GFEX_DAILY_URL = "http://www.gfex.com.cn/u/interf

"TS": "20180817", "TF": "20130906", "T": "20150320", "TL": "20230421"} GFEX_DAILY_URL = "http://www.gfex.com.cn/u/interfacesWebTiDayQuotes/loadList" SINA_HQ_URL = "https://hq.sinajs.cn/list="

GFEX_DAILY_URL = "http://www.gfex.com.cn/u/interfacesWebTiDayQuotes/loadList"

SINA_HQ_URL = "https://hq.sinajs.cn/list="

SINA_FUT_KLINE_URL = ("https://stock2.finance.sina.com.cn/futures/api/jsonp.php/var%20_{code}="

"/InnerFuturesNewService.getDailyKLine") SGE_DAILY_URL = "https://www.sge.com.cn/graph/Dailyhq" _CFFEX_SINA_PRODUCTS = ("IF", "IH", "IC", "IM", "TS", "TF", "T", "TL")

SGE_DAILY_URL = "https://www.sge.com.cn/graph/Dailyhq"

def _gfex_rows(ymd, trade_type):     response = _v39_http(GFEX_DAILY_URL, method="POST", data={"trade_date": ymd, "trade_type": trade_type},                          headers={"Referer": "http://www.gfex.com.cn/gfex/rihq/hqsj_tjsj.shtml"})

response = _v39_http(GFEX_DAILY_URL, method="POST", data={"trade_date": ymd, "trade_type": trade_type},                          headers={"Referer": "http://www.gfex.com.cn/gfex/rihq/hqsj_tjsj.shtml"})     payload = _v39_json(response)

headers={"Referer": "http://www.gfex.com.cn/gfex/rihq/hqsj_tjsj.shtml"})

def _sina_hq(codes):     response = _v39_http(SINA_HQ_URL + ",".join(codes), headers={"Referer": "https://finance.sina.com.cn/"})

response = _v39_http(SINA_HQ_URL + ",".join(codes), headers={"Referer": "https://finance.sina.com.cn/"})

raise ValueError(f"start {lo} 晚于 end {hi}")     response = _v39_http(SINA_FUT_KLINE_URL.format(code=code), params={"symbol": code}, timeout=(5, 20),                          headers={"Referer": "https://finance.sina.com.cn/"}, session=V39_R

response = _v39_http(SINA_FUT_KLINE_URL.format(code=code), params={"symbol": code}, timeout=(5, 20),                          headers={"Referer": "https://finance.sina.com.cn/"}, session=V39_RETRY_SESSION)     text = response.content.decode

headers={"Referer": "https://finance.sina.com.cn/"}, session=V39_RETRY_SESSION)

"""     response = _v39_http(SGE_DAILY_URL, method="POST", data={"instid": instrument},                          headers={"Referer": "https://www.sge.com.cn/"})

response = _v39_http(SGE_DAILY_URL, method="POST", data={"instid": instrument},                          headers={"Referer": "https://www.sge.com.cn/"})     payload = _v39_json(response)

headers={"Referer": "https://www.sge.com.cn/"})

prefix = ("bj" if code.startswith(("92", "8"))               else "sh" if code.startswith(("6", "9")) else "sz")     url = f"https://qt.gtimg.cn/q={prefix}{code}"

else "sh" if code.startswith(("6", "9")) else "sz")     url = f"https://qt.gtimg.cn/q={prefix}{code}"     req = urllib.request.Request(url)

url = f"https://qt.gtimg.cn/q={prefix}{code}"     req = urllib.request.Request(url)     req.add_header("User-Agent", "Mozilla/5.0")

req = urllib.request.Request(url)

req.add_header("User-Agent", "Mozilla/5.0")     resp = urllib.request.urlopen(req, timeout=10)     data = resp.read().decode("gbk")

resp = urllib.request.urlopen(req, timeout=10)

code = _official_margin_code(code, exchange)     if exchange == "SH":         url = "https://query.sse.com.cn/marketdata/tradedata/queryMargin.do"

if exchange == "SH":         url = "https://query.sse.com.cn/marketdata/tradedata/queryMargin.do"         response = _official_get(url, {

url = "https://query.sse.com.cn/marketdata/tradedata/queryMargin.do"

"pageHelp.pageSize": 5000, "pageHelp.pageNo": 1, "pageHelp.beginPage": 1,             "pageHelp.cacheSize": 1, "pageHelp.endPage": 1,         }, "https://www.sse.com.cn/")

"pageHelp.cacheSize": 1, "pageHelp.endPage": 1,         }, "https://www.sse.com.cn/")         page = response.json().get("pageHelp") or {}

}, "https://www.sse.com.cn/")

for src, dest in fields.items()}})     else:         url = "https://www.szse.cn/api/report/ShowReport"

else:         url = "https://www.szse.cn/api/report/ShowReport"         response = _official_get(url, {"SHOWTYPE": "xlsx", "CATALOGID": "1837_xxpl",

url = "https://www.szse.cn/api/report/ShowReport"

response = _official_get(url, {"SHOWTYPE": "xlsx", "CATALOGID": "1837_xxpl",                                       "TABKEY": "tab2", "txtDate": trade_date}, "https://www.szse.cn/")         data = _official_excel(response)

"TABKEY": "tab2", "txtDate": trade_date}, "https://www.szse.cn/")

if not code.startswith(("4", "8", "92")):             raise ValueError("请输入北交所代码（4/8/92 开头）")     page_url = "https://www.bse.cn/nq/quotation.html"

raise ValueError("请输入北交所代码（4/8/92 开头）")     page_url = "https://www.bse.cn/nq/quotation.html"     url = "https://www.bse.cn/nqhqController/nqhq_en.do"

page_url = "https://www.bse.cn/nq/quotation.html"

url = "https://www.bse.cn/nqhqController/nqhq_en.do"

raw_rows = []     total = None     with requests.Session() as session:

total = None     with requests.Session() as session:         session.headers.update({"User-Agent": "Mozilla/5.0", "Referer": page_url,

with requests.Session() as session:

端点与字段交叉参考 [CNEquity 两融适配](https://github.com/rootSunc/CNEquity/blob/main/src/cnequity/adapters/exchange/margin_trading.py)

及 [北交所适配](https://github.com/rootSunc/CNEquity/blob/main/src/cnequity/adapters/bse/daily_quotes.py)。

"""龙虎榜官方备用源（东财被封时用）：上交所+深交所官方，零鉴权权威一手，含营业部席位。"""     out = {"date": trade_date, "sse_raw": "", "szse": []}     su = (f"https://www.szse.cn/api/report/ShowReport/data?SHOWTYPE=JSON"

out = {"date": trade_date, "sse_raw": "", "szse": []}     su = (f"https://www.szse.cn/api/report/ShowReport/data?SHOWTYPE=JSON"           f"&CATALOGID=1842_xxpl&TABKEY=tab1&txtStart={trade_date}&txtEnd={trade_date}&random=0.9")

su = (f"https://www.szse.cn/api/report/ShowReport/data?SHOWTYPE=JSON"           f"&CATALOGID=1842_xxpl&TABKEY=tab1&txtStart={trade_date}&txtEnd={trade_date}&random=0.9")     req = urllib.request.Request(su, headers={"User-Agent": UA,

f"&CATALOGID=1842_xxpl&TABKEY=tab1&txtStart={trade_date}&txtEnd={trade_date}&random=0.9")     req = urllib.request.Request(su, headers={"User-Agent": UA,           "Referer": "https://www.szse.cn/disclosure/supervision/dealinfo/index.html"}

req = urllib.request.Request(su, headers={"User-Agent": UA,

"Referer": "https://www.szse.cn/disclosure/supervision/dealinfo/index.html"})     with urllib.request.urlopen(req, timeout=15, context=_ctx) as r:         d = json.loads(r.read())

with urllib.request.urlopen(req, timeout=15, context=_ctx) as r:

out["szse"].append({"code": row.get("zqdm"), "name": row.get("zqjc"),                             "amount": row.get("cjje"), "reason": row.get("plyy")})     eu = (f"https://query.sse.com.cn/infodisplay/showTradePublicFile.do?"

"amount": row.get("cjje"), "reason": row.get("plyy")})     eu = (f"https://query.sse.com.cn/infodisplay/showTradePublicFile.do?"           f"jsonCallBack=cb&isPagination=false&dateTx={trade_date}")

eu = (f"https://query.sse.com.cn/infodisplay/showTradePublicFile.do?"           f"jsonCallBack=cb&isPagination=false&dateTx={trade_date}")     req = urllib.request.Request(eu, headers={"User-Agent": UA,

f"jsonCallBack=cb&isPagination=false&dateTx={trade_date}")     req = urllib.request.Request(eu, headers={"User-Agent": UA,           "Referer": "https://www.sse.com.cn/disclosure/diclosure/public/"})

req = urllib.request.Request(eu, headers={"User-Agent": UA,

"Referer": "https://www.sse.com.cn/disclosure/diclosure/public/"})     with urllib.request.urlopen(req, timeout=15) as r:         t = r.read().decode("utf-8", "ignore")

with urllib.request.urlopen(req, timeout=15) as r:

pre = ("bj" if code.startswith(("92", "8"))            else "sh" if code.startswith(("6", "9")) else "sz") + code     u = (f"https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/"

else "sh" if code.startswith(("6", "9")) else "sz") + code     u = (f"https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/"          f"MoneyFlow.ssl_qsfx_zjlrqs?page=1&num={days}&sort=opendate&asc=0&daima={pre}")

u = (f"https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/"          f"MoneyFlow.ssl_qsfx_zjlrqs?page=1&num={days}&sort=opendate&asc=0&daima={pre}")     req = urllib.request.Request(u, headers={"User-Agent": UA, "Referer":

f"MoneyFlow.ssl_qsfx_zjlrqs?page=1&num={days}&sort=opendate&asc=0&daima={pre}")     req = urllib.request.Request(u, headers={"User-Agent": UA, "Referer": "https://finance.sina.com.cn/"})     with urllib.request.urlopen(req, timeout=15) as r

req = urllib.request.Request(u, headers={"User-Agent": UA, "Referer": "https://finance.sina.com.cn/"})

with urllib.request.urlopen(req, timeout=15) as r:

body = json.dumps({"channelCode": ["listedNotice_disc"], "pageSize": page_size,                            "pageNum": 1, "stock": [code]}).encode()         req = urllib.request.Request("https://www.szse.cn/api/disc/announcement/annList", da

"pageNum": 1, "stock": [code]}).encode()         req = urllib.request.Request("https://www.szse.cn/api/disc/announcement/annList", data=body,               headers={"User-Agent": UA, "Content-Type": "application/json",

req = urllib.request.Request("https://www.szse.cn/api/disc/announcement/annList", data=body,

headers={"User-Agent": UA, "Content-Type": "application/json",                        "Referer": "https://www.szse.cn/disclosure/listed/notice/index.html"})         with urllib.request.urlopen(req, timeout=15, context=_ctx) as r:

"Referer": "https://www.szse.cn/disclosure/listed/notice/index.html"})         with urllib.request.urlopen(req, timeout=15, context=_ctx) as r:             d = json.loads(r.read())

with urllib.request.urlopen(req, timeout=15, context=_ctx) as r:

d = json.loads(r.read())         return [{"title": a.get("title"), "time": a.get("publishTime", "")[:10],                  "pdf": "https://disc.static.szse.cn/download" + a.get("attachPath", "")}

return [{"title": a.get("title"), "time": a.get("publishTime", "")[:10],                  "pdf": "https://disc.static.szse.cn/download" + a.get("attachPath", "")}                 for a in d.get("data", [])]

"pdf": "https://disc.static.szse.cn/download" + a.get("attachPath", "")}

for a in d.get("data", [])]     u = (f"https://np-anotice-stock.eastmoney.com/api/security/ann?sr=-1&page_size={page_size}"          f"&page_index=1&ann_type=A&client_source=web&stock_list={code}&f_node=0&s_node=0")

u = (f"https://np-anotice-stock.eastmoney.com/api/security/ann?sr=-1&page_size={page_size}"          f"&page_index=1&ann_type=A&client_source=web&stock_list={code}&f_node=0&s_node=0")     req = urllib.request.Request(u, headers={"User-Agent

f"&page_index=1&ann_type=A&client_source=web&stock_list={code}&f_node=0&s_node=0")     req = urllib.request.Request(u, headers={"User-Agent": UA})     with urllib.request.urlopen(req, timeout=15) as r:

req = urllib.request.Request(u, headers={"User-Agent": UA})

with urllib.request.urlopen(req, timeout=15) as r:

d = json.loads(r.read())     return [{"title": a.get("title"), "time": a.get("notice_date", "")[:10],              "pdf": f"https://pdf.dfcfw.com/pdf/H2_{a.get('art_code','')}_1.pdf"}

return [{"title": a.get("title"), "time": a.get("notice_date", "")[:10],              "pdf": f"https://pdf.dfcfw.com/pdf/H2_{a.get('art_code','')}_1.pdf"}             for a in (d.get("data") or {}).get("list") or []]     # #46 同因

"pdf": f"https://pdf.dfcfw.com/pdf/H2_{a.get('art_code','')}_1.pdf"}

### Q: 东财 PDF 下载 403 A: 必须带 `Referer: https://data.eastmoney.com/` header。

A: 必须带 `Referer: https://data.eastmoney.com/` header。

[Credential-related source omitted]

> 📦 https://github.com/simonlin1212/a-stock-data — Star ⭐ 是最好的支持

- [AKShare：中证成分/权重](https://github.com/akfamily/akshare/blob/main/akshare/index/index_cons.py)

- [AKShare：中证估值](https://github.com/akfamily/akshare/blob/main/akshare/index/index_stock_zh_csindex.py)

- [AKShare：国证指数](https://github.com/akfamily/akshare/blob/main/akshare/index/index_cni.py)

- [Qlib：深交所交易日历](https://github.com/microsoft/qlib/blob/main/scripts/data_collector/utils.py)

- [CNEquity：沪深官方两融](https://github.com/rootSunc/CNEquity/blob/main/src/cnequity/adapters/exchange/margin_trading.py)

- [CNEquity：北交所行情](https://github.com/rootSunc/CNEquity/blob/main/src/cnequity/adapters/bse/daily_quotes.py)

| 候选 | 已核实价值 | 决定及剩余条件 | |---|---|---| | [easy_tdx](https://github.com/handsomejustin/easy_tdx) | 通达信竞价过程；2026-09-04平安银行实测63点；MIT | 同一通达信数据来源的实现备选。要求Python≥3.10，先验证可选依赖与竞价契约，不改变主包Python3.9兼容 |

|---|---|---| | [easy_tdx](https://github.com/handsomejustin/easy_tdx) | 通达信竞价过程；2026-09-04平安银行实测63点；MIT | 同一通达信数据来源的实现备选。要求Python≥3.10，先验证可选依赖与竞价契约，不改变主包Python3.9兼容 | | [eltdx](https://github.com/electkismet/eltdx) | 当日竞价82点及历史09:25快照可用；历史

| [easy_tdx](https://github.com/handsomejustin/easy_tdx) | 通达信竞价过程；2026-09-04平安银行实测63点；MIT | 同一通达信数据来源的实现备选。要求Python≥3.10，先验证可选依赖与竞价契约，不改变主包Python3.9兼容 |

| [eltdx](https://github.com/electkismet/eltdx) | 当日竞价82点及历史09:25快照可用；历史完整过程未复现 | 当前 LICENSE 为 Research-Only，限制商业/生产用途；不纳入本Apache-2.0项目，不复制实现 |

| [同花顺官方 Financial-API](https://github.com/HiThink-Tech/Financial-API) | 文档含竞价、基金、批量数据等增强能力 | 需用户Key，未做带Key数据验收；不增加强制鉴权，不计入主源数量 |

| [TickFlow](https://github.com/tickflow-org/tickflow) | 免费日K实测沪市和北交所成功 | 观察型独立服务备选；本次未确认数据来源、再分发条件和长期服务保障 |

| [CNEquity](https://github.com/rootSunc/CNEquity) | 官方端点和日期/分页防错参考，含国证历史调样适配 | 继续作为实现参考；历史调样另需时点验证，不把当前月末文件称为历史成分 |

| [AKShare](https://github.com/akfamily/akshare)、[adata](https://github.com/1nchaos/adata)、[efinance](https://github.com/Micro-sheep/efinance)、[Ashare](https://github.com/mpquant/Ashare) | 可检索上游端点与解析经验 | 多数为已有来源的包装；不作为新独立来源，不恢复运行时依赖 |

| [free-stockdb](https://github.com/hello245m/free-stockdb) | 本地存储和同步设计 | 上游数据出处不够清楚，暂不接入 |

- [AKShare](https://github.com/akfamily/akshare)：期货交易所日行情、持仓排名、上金所、中债、货币网等端点格式

- [jing2uo/tdx2db](https://github.com/jing2uo/tdx2db)：通达信官网盘后包地址与 `.cod` / `.md1` 布局

- [Xiaoyuan-Liu/national-team-position](https://github.com/Xiaoyuan-Liu/national-team-position)：上交所 ETF 份额归档接口

- 新浪研报每页序号都从 1 编起；翻过末页、以及没有研报的股票，页面都写「没有找到相关内容」。上证e互动问答为空时，公司维度写「近1个月暂无回复 / 暂无提问」，   全市场翻过末页写「暂时没有问答内容」。华尔街见闻宏观日历 31 周 5976 条没有重复 id。 - 中金所、广期所的数据地址只能走 HTTP：2026-09-20 用 curl 8.7.1 与 Python requests 实测，中金所 443 握手即断（SSL_ERROR_SYSCALL），

全市场翻过末页写「暂时没有问答内容」。华尔街见闻宏观日历 31 周 5976 条没有重复 id。 - 中金所、广期所的数据地址只能走 HTTP：2026-09-20 用 curl 8.7.1 与 Python requests 实测，中金所 443 握手即断（SSL_ERROR_SYSCALL），   广期所 443 返回的是 `*.unionpayintl.com` 的证书。交易日靠来源自己核对：中金所日行情 / 期权用同目录 `index.xml`，持仓排名用行内交易日，

- 中金所、广期所的数据地址只能走 HTTP：2026-09-20 用 curl 8.7.1 与 Python requests 实测，中金所 443 握手即断（SSL_ERROR_SYSCALL），

Executable file

Executable file

Executable file

Executable file

var sh600519hfq={"total":33,"data":[{"d":"2026-06-26", "f":"8.8825126710204000"},{"d":"2025-12-19", "f":"8.6771454751862000"},{"d":"2025-06-26", "f":"8.5318775687229000"},{"d":"2024-12-20", "f":"8.3678784177632990"},{"d":"2024-06-19", "f":"

Executable file

var sh600519qfq={"total":33,"data":[{"d":"2026-06-26", "f":"1.0000000000000000"},{"d":"2025-12-19", "f":"1.0236675985693000"},{"d":"2025-06-26", "f":"1.0410970621031000"},{"d":"2024-12-20", "f":"1.0615011628473000"},{"d":"2024-06-19", "f":"

Executable file

Executable file

Executable file

Executable file

Executable file

Executable file

exec(compile(block, f"SKILL.md:{name}", "exec"), namespace)

self.data = data         self.status_code = status         self.url = "https://official.example/data"

self.status_code = status         self.url = "https://official.example/data"         self.text = json.dumps(data) if data is not None else ""

self.url = "https://official.example/data"

def raise_for_status(self):         if self.status_code >= 400:             raise requests.HTTPError(str(self.status_code))

raise requests.HTTPError(str(self.status_code))

[Credential-related source omitted]

[Credential-related source omitted]

Executable file

exec(compile(_marker_code("v310-tencent-ticks"), "SKILL.md:v310-tencent-ticks", "exec"), ns)

RequestException=REAL_REQUESTS.RequestException)         with patch.dict(self.ns, {"requests": no_new_connection}):             response = self.ns["_v39_http"]("https://example.test/x", params={"a": 1}, session=session)

with patch.dict(self.ns, {"requests": no_new_connection}):             response = self.ns["_v39_http"]("https://example.test/x", params={"a": 1}, session=session)         self.assertEqual(response.content, b"ok")

response = self.ns["_v39_http"]("https://example.test/x", params={"a": 1}, session=session)

self.assertEqual(response.content, b"ok")         self.assertEqual(session.request.call_args[0], ("GET", "https://example.test/x"))         self.assertEqual(session.request.call_args[1]["params"], {"a": 1})

self.assertEqual(session.request.call_args[0], ("GET", "https://example.test/x"))

session.request.side_effect = REAL_REQUESTS.ConnectionError("reset")         with self.assertRaises(RuntimeError):                    # 重试用完仍失败：照旧是「来源坏了」             self.ns["_v39_http"]("https://example.test/x", session=session)

self.ns["_v39_http"]("https://example.test/x", session=session)

if tuple(int(p) for p in urllib3.__version__.split(".")[:2]) < (1, 26):             self.skipTest("urllib3 < 1.26 没有重试会话，见下一条测试")         for url in ("https://stock.gtimg.cn/data/index.php", "https://qt.gtimg.cn/q=sz000001",

self.skipTest("urllib3 < 1.26 没有重试会话，见下一条测试")         for url in ("https://stock.gtimg.cn/data/index.php", "https://qt.gtimg.cn/q=sz000001",                     "https://stock2.finance.sina.com.cn/futures/api/jsonp.php/"):

for url in ("https://stock.gtimg.cn/data/index.php", "https://qt.gtimg.cn/q=sz000001",

"https://stock2.finance.sina.com.cn/futures/api/jsonp.php/"):

[Credential-related source omitted]

("futures_kline M0", lambda: ns["futures_kline"]("M0", start="2026-01-01"))):             with self.subTest(name=name):                 frame = fetch()

with self.subTest(name=name):                 frame = fetch()                 self.assertGreater(len(frame), 0)

frame = fetch()

Executable file

exec(_defs_only(block), ns)

def factors(self, code, kind="qfq", status=200):         suffix = ".html" if status == 404 else ".js"         response = requests.Response()

suffix = ".html" if status == 404 else ".js"         response = requests.Response()         response.status_code = status

response = requests.Response()

response.status_code = status         response.url = "https://example.test/" + code + "/" + kind + ".js"         response.encoding = "utf-8"

response.url = "https://example.test/" + code + "/" + kind + ".js"

for kind in ("qfq", "hfq"):             for row, expected in cases:                 response = requests.Response()

for row, expected in cases:                 response = requests.Response()                 response.status_code = 200

response = requests.Response()

def test_malformed_fund_fields_are_runtime_errors(self):         for fields in ('"s":"bad","u":"0"', '"s":"1"', '"u":"0"'):             response = requests.Response()

for fields in ('"s":"bad","u":"0"', '"s":"1"', '"u":"0"'):             response = requests.Response()             response.status_code = 200

response = requests.Response()

Executable file

exec(code, namespace)

self.text = text if text is not None else content.decode("latin-1")         self._payload = payload         self.url = "https://example.test/response"

self.url = "https://example.test/response"

RequestException=REAL_REQUESTS.RequestException)         with patch.dict(self.ns, {"requests": fake}):             return self.ns["_v39_http"]("https://example.test/x", allow_status=(404,))

return self.ns["_v39_http"]("https://example.test/x", allow_status=(404,))

def fetch(self, response, day="2026-09-18"):

def test_parses_all_three_markets(self):         out = self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"])))

out = self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"])))

def test_early_2022_package_without_bse_files(self):         out = self.fetch(Response(content=tdx_zip("20220104", ["sh", "sz"])), "2022-01-04")

out = self.fetch(Response(content=tdx_zip("20220104", ["sh", "sz"])), "2022-01-04")

with self.assertRaises(RuntimeError):             self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz"])))

self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz"])))

out = self.fetch(Response(content=tdx_zip("20220505", ["sh", "sz"])), "2022-05-05")

def test_missing_shanghai_or_shenzhen_is_an_error(self):         with self.assertRaises(RuntimeError):             self.fetch(Response(content=tdx_zip("20260918", ["sh", "bj"])))

self.fetch(Response(content=tdx_zip("20260918", ["sh", "bj"])))

with self.assertRaises(RuntimeError):             self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"], empty=("bj",))))

self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"], empty=("bj",))))

cod[40:44] = b"\xff\xfe\xff\xfe"

for tweak in (bad_bytes, blank):             with self.subTest(tweak=tweak.__name__), self.assertRaises(RuntimeError):                 self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"], tweak=tweak)))

self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"], tweak=tweak)))

for market in ("sh", "sz", "bj"):             with self.subTest(market=market), self.assertRaises(RuntimeError):                 self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"],

with self.subTest(market=market), self.assertRaises(RuntimeError):                 self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"],                                                     counts={market: TDX_FLOOR[market] - 1

self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"],

for tweak in (dup_code, dup_seq, blank_code, extra_block):             with self.subTest(tweak=tweak.__name__), self.assertRaises(RuntimeError):                 self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"],

with self.subTest(tweak=tweak.__name__), self.assertRaises(RuntimeError):                 self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"],                                                     counts={"sh": 10001, "sz": 300

self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"],

for tweak in (bad_byte, letter, nan_close, inf_amount, nan_in_unpriced_block):             with self.subTest(tweak=tweak.__name__), self.assertRaises(RuntimeError):                 self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz"

with self.subTest(tweak=tweak.__name__), self.assertRaises(RuntimeError):                 self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"],                                                     counts={"sh": 10001, "sz": 300

self.fetch(Response(content=tdx_zip("20260918", ["sh", "sz", "bj"],

def test_404_is_value_error_and_error_page_is_runtime_error(self):         with self.assertRaises(ValueError):             self.fetch(Response(status=404))

with self.assertRaises(ValueError):             self.fetch(Response(status=404))         with self.assertRaises(RuntimeError):

self.fetch(Response(status=404))

self.fetch(Response(content=b"<html>error</html>"))

for content in (b"PK\x03\x04broken", good[:len(good) // 2]):             with self.subTest(size=len(content)), self.assertRaises(RuntimeError):                 self.fetch(Response(content=content))

self.fetch(Response(content=content))

ns = load_shipped_code()         page = MagicMock(return_value=(MagicMock(url="https://sina.example"),

page = MagicMock(return_value=(MagicMock(url="https://sina.example"),

def reports(self, body):         ns = load_shipped_code()         page = MagicMock(return_value=(MagicMock(url="https://sina.example"),

ns = load_shipped_code()         page = MagicMock(return_value=(MagicMock(url="https://sina.example"),                                        f"<table class=tb_01><tr><th>研究员</th></tr>{body}</table>"))

page = MagicMock(return_value=(MagicMock(url="https://sina.example"),

class EtfSharesSseTests(unittest.TestCase):     def fetch(self, result, total, day="2026-09-18"):

def fetch(self, result, total, day="2026-09-18"):

rec = {"STAT_DATE": "2026-09-18", "SEC_CODE": "510300", "SEC_NAME": "样本ETF", "TOT_VOL": "1,234.5"}         rows, _ = self.fetch([rec, dict(rec, SEC_CODE="510500")], 2)

rows, _ = self.fetch([rec, dict(rec, SEC_CODE="510500")], 2)

self.assertEqual([r["code"] for r in rows], ["510300", "510500"])         with self.assertRaises(ValueError):                     # 非交易日：result 空、total 0             self.fetch([], 0, "2026-09-19")

with self.assertRaises(ValueError):                     # 非交易日：result 空、total 0             self.fetch([], 0, "2026-09-19")         for result, total in (([rec], 912), ([rec], True), ([rec], None), ([], 3), ([rec, "x"], 2),

self.fetch([], 0, "2026-09-19")

([dict(rec, TOT_VOL="")], 1), ([dict(rec, TOT_VOL="-")], 1)):             with self.subTest(total=total, n=len(result)), self.assertRaises(RuntimeError):                 self.fetch(result, total)

self.fetch(result, total)

def fetch(self, *pages):

def test_complete_pages_match_record_count(self):         rows, _ = self.fetch(self.page(1), self.page(1))

rows, _ = self.fetch(self.page(1), self.page(1))

broken = Response(text="x", payload=[{"metadata": meta, "data": [rec]}])         with self.assertRaises(RuntimeError):             self.fetch(broken, self.page(1))

self.fetch(broken, self.page(1))

def test_empty_page_or_count_mismatch_raises(self):         with self.assertRaises(RuntimeError):             self.fetch(self.page(1), self.page(0))            # 第 2 页空：只返回第 1 页就是部分快照

with self.assertRaises(RuntimeError):             self.fetch(self.page(1), self.page(0))            # 第 2 页空：只返回第 1 页就是部分快照         with self.assertRaises(RuntimeError):                 # 总数也跟着少报时，只有空页检查拦得住

self.fetch(self.page(1), self.page(0))            # 第 2 页空：只返回第 1 页就是部分快照

with self.assertRaises(RuntimeError):                 # 总数也跟着少报时，只有空页检查拦得住             self.fetch(self.page(1, recordcount=1), self.page(0, recordcount=1))         with self.assertRaises(RuntimeError):

self.fetch(self.page(1, recordcount=1), self.page(0, recordcount=1))

self.fetch(self.page(1, recordcount=3), self.page(1, recordcount=3))

self.page(50, pagecount=2, recordcount=150, subname="2026-09-21")):             with self.subTest(meta=second._payload[0]["metadata"]), self.assertRaises(RuntimeError):                 self.fetch(first, second)

with self.subTest(meta=second._payload[0]["metadata"]), self.assertRaises(RuntimeError):                 self.fetch(first, second)         with self.assertRaises(ValueError):                   # 第 1 页日期不符才是「深市只有当前快照」

self.fetch(first, second)

self.fetch(self.page(1, subname="2026-09-17"))

for subname in (None, "", "2026/09/18", 20260918.0, True):             with self.subTest(subname=subname), self.assertRaises(RuntimeError):                 self.fetch(self.page(1, pagecount=1, recordcount=1, subname=subname))

with self.subTest(subname=subname), self.assertRaises(RuntimeError):                 self.fetch(self.page(1, pagecount=1, recordcount=1, subname=subname))         rows, _ = self.fetch(self.page(1, pagecount=1, recordcount=1, subname="202609

self.fetch(self.page(1, pagecount=1, recordcount=1, subname=subname))

rows, _ = self.fetch(self.page(1, pagecount=1, recordcount=1, subname="20260918"))

{"pagecount": 1, "recordcount": "-1"}, {"pagecount": 1.0, "recordcount": 1}):             with self.subTest(kwargs=kwargs), self.assertRaises(RuntimeError):                 self.fetch(self.page(1, **kwargs))

with self.subTest(kwargs=kwargs), self.assertRaises(RuntimeError):                 self.fetch(self.page(1, **kwargs))         rows, _ = self.fetch(self.page(1, pagecount="1", recordcount="1"))

self.fetch(self.page(1, **kwargs))

rows, _ = self.fetch(self.page(1, pagecount="1", recordcount="1"))

for payload in ([], {}, [None], [{"data": []}], [{"metadata": {}, "data": None}], [{"metadata": [], "data": []}]):             with self.subTest(payload=payload), self.assertRaises(RuntimeError):                 self.fetch(Response(text="x"

with self.subTest(payload=payload), self.assertRaises(RuntimeError):                 self.fetch(Response(text="x", payload=payload))         rec_missing = self.page(1, pagecount=1, recordcount=1)

self.fetch(Response(text="x", payload=payload))

rec_missing._payload[0]["data"][0].pop("dqgm")         with self.assertRaises(RuntimeError):             self.fetch(rec_missing)

self.fetch(rec_missing)

shsz = [{"f12": "600001", "f13": 1, "f14": "*ST样本", "f2": 1.0, "f3": 0.5}]         bj = [{"f12": "920001", "f13": 0, "f14": "ST北交", "f2": 2.0, "f3": 0.1}]         with patch.dict(ns, {"_em_clist_all": MagicMock(side_effect=[(shsz, "https://

bj = [{"f12": "920001", "f13": 0, "f14": "ST北交", "f2": 2.0, "f3": 0.1}]         with patch.dict(ns, {"_em_clist_all": MagicMock(side_effect=[(shsz, "https://a"), ([], "https://a")])}):             with self.assertRaises(RuntimeError):

with patch.dict(ns, {"_em_clist_all": MagicMock(side_effect=[(shsz, "https://a"), ([], "https://a")])}):

with self.assertRaises(RuntimeError):                 ns["st_stock_list"]()         with patch.dict(ns, {"_em_clist_all": MagicMock(side_effect=[(shsz, "https://a"), (bj, "https://b")])}):

ns["st_stock_list"]()         with patch.dict(ns, {"_em_clist_all": MagicMock(side_effect=[(shsz, "https://a"), (bj, "https://b")])}):             out = ns["st_stock_list"]()

with patch.dict(ns, {"_em_clist_all": MagicMock(side_effect=[(shsz, "https://a"), (bj, "https://b")])}):

out = ns["st_stock_list"]()         self.assertEqual(sorted(out.code), ["600001", "920001"])         self.assertEqual(set(out.source_url), {"https://a | https://b"})

self.assertEqual(set(out.source_url), {"https://a | https://b"})

rec = dict(good_sh, **bad)             with self.subTest(bad=bad), patch.dict(ns, {"_em_clist_all": MagicMock(                     side_effect=[([rec], "https://a"), ([good_bj], "https://a")])}):

with self.subTest(bad=bad), patch.dict(ns, {"_em_clist_all": MagicMock(                     side_effect=[([rec], "https://a"), ([good_bj], "https://a")])}):                 if bad == {"f13": 0}:               # 真的是深市：照常出表，market 跟着变

side_effect=[([rec], "https://a"), ([good_bj], "https://a")])}):

rec = {k: v for k, v in good_sh.items() if k != missing}             with self.subTest(missing=missing), patch.dict(ns, {"_em_clist_all": MagicMock(                     side_effect=[([rec], "https://a"), ([good_bj], "https://a")])}), \

with self.subTest(missing=missing), patch.dict(ns, {"_em_clist_all": MagicMock(                     side_effect=[([rec], "https://a"), ([good_bj], "https://a")])}), \                     self.assertRaises(RuntimeError):

side_effect=[([rec], "https://a"), ([good_bj], "https://a")])}), \

shfe_json = MagicMock(return_value=(None, "https://www.ine.cn/x"))

for sigma in ({}, "", 0, {"a": 1}, [1], [{"SIGMA": 0.2}]):             payload = dict(option, o_cursigma=sigma)             stub = MagicMock(return_value=(payload, "https://x"))

payload = dict(option, o_cursigma=sigma)             stub = MagicMock(return_value=(payload, "https://x"))             with self.subTest(sigma=sigma), patch.dict(self.ns, {"_shfe_json": stub}), \

stub = MagicMock(return_value=(payload, "https://x"))

def call(sigma_rows):             stub = MagicMock(return_value=({"o_curinstrument": [inst], "o_cursigma": sigma_rows}, "https://x"))

stub = MagicMock(return_value=({"o_curinstrument": [inst], "o_cursigma": sigma_rows}, "https://x"))

head = "品种：{}　日期：2026-09-18"         body = "\n名次|会员|成交量|增减|会员|持买单量|增减|会员|持卖单量|增减\n1|华泰|100|1|银河|200|2|国泰|300|3\n"         with patch.dict(self.ns, {"_czce_text": MagicMock(return_value=(head.format("SR") + body, "https://x"))}):

body = "\n名次|会员|成交量|增减|会员|持买单量|增减|会员|持卖单量|增减\n1|华泰|100|1|银河|200|2|国泰|300|3\n"         with patch.dict(self.ns, {"_czce_text": MagicMock(return_value=(head.format("SR") + body, "https://x"))}):             out = self.ns["futures_position_ran

with patch.dict(self.ns, {"_czce_text": MagicMock(return_value=(head.format("SR") + body, "https://x"))}):

out = self.ns["futures_position_rank"]("2026-09-18", "CZCE")         self.assertEqual((out.symbol[0], out["rank"][0], out.long_member[0]), ("SR", 1, "银河"))         with patch.dict(self.ns, {"_czce_text": MagicMock(return_value=(head.format(

self.assertEqual((out.symbol[0], out["rank"][0], out.long_member[0]), ("SR", 1, "银河"))         with patch.dict(self.ns, {"_czce_text": MagicMock(return_value=(head.format("2026") + body, "https://x"))}):             with self.assertRaises(R

with patch.dict(self.ns, {"_czce_text": MagicMock(return_value=(head.format("2026") + body, "https://x"))}):

load(body)         sigma = {"o_curinstrument": [], "o_cursigma": [["cu2611", 0.2]], "report_date": "20260918"}         with patch.dict(self.ns, {"_shfe_json": MagicMock(return_value=(sigma, "https://www.ine.cn/x"))}):

sigma = {"o_curinstrument": [], "o_cursigma": [["cu2611", 0.2]], "report_date": "20260918"}         with patch.dict(self.ns, {"_shfe_json": MagicMock(return_value=(sigma, "https://www.ine.cn/x"))}):             with self.assertRaises(Runtim

with patch.dict(self.ns, {"_shfe_json": MagicMock(return_value=(sigma, "https://www.ine.cn/x"))}):

def test_shfe_rank_without_rank_field_is_an_error(self):         payload = {"o_cursor": [{"INSTRUMENTID": "sc2610", "RANK": ""}]}         with patch.dict(self.ns, {"_shfe_json": MagicMock(return_value=(payload, "https://www.ine.cn/x"))}):

payload = {"o_cursor": [{"INSTRUMENTID": "sc2610", "RANK": ""}]}         with patch.dict(self.ns, {"_shfe_json": MagicMock(return_value=(payload, "https://www.ine.cn/x"))}):             with self.assertRaises(RuntimeError):

with patch.dict(self.ns, {"_shfe_json": MagicMock(return_value=(payload, "https://www.ine.cn/x"))}):

[Credential-related source omitted]

for name, fetch in calls:             with self.subTest(name=name):                 frame = fetch()

with self.subTest(name=name):                 frame = fetch()                 self.assertGreater(len(frame), 0)

frame = fetch()

Executable file

exec(compile(ast.Module(body=functions, type_ignores=[]), "SKILL.md:valuation", "exec"), namespace)