| name | a-stock-data |
| description | A股全栈数据工具包 — 覆盖行情(mootdx+腾讯+百度K线)、研报(东财+同花顺+iwencai)、信号(同花顺热点+北向+龙虎榜+解禁+行业)、资金面(融资融券+大宗交易+股东户数+分红+资金流分钟级+资金流120日)、新闻(东财个股+全球资讯)、基础数据(mootdx财务/F10+东财+新浪三表)、公告(巨潮)、打板(涨停池/连板梯队/炸板率/跌停)、ETF期权(T型报价/希腊字母/IV)、舆情互动(互动易问答/热榜/人气榜)十层数据源，内嵌全部调用代码，自包含零依赖外部文件。优先用通达信(mootdx)/腾讯(不封IP)，东财接口已内置限流防封。适用于个股估值、研报检索、题材归因、龙虎榜跟踪、解禁预警、行业轮动、融资融券跟踪、筹码分析、产业链调研、批量筛选、打板情绪跟踪、ETF期权策略、投资者互动问答、市场热度选题等场景。 |
📦 项目主页：https://github.com/simonlin1212/a-stock-data — 更新、反馈、支持作者
作者：Simon 林 · 抖音「Simon林」· 公众号「硅基世纪」
A股全栈数据工具包 V3.3.0
十层数据架构，40 个端点实测可用（2026-06 验证；财联社快讯已下线，详见 §5.2），覆盖主板/中小板/科创板/ST。
V3.2.3（行业研报新增）：
- §2.1 东财行业研报
eastmoney_industry_reports()：研报层补上行业研报端点（此前只有个股研报）。与个股研报同端点 reportapi.eastmoney.com/report/list，仅 qType=1；industry_code="*" 拉全行业、传东财行业码（如 1238=IT服务Ⅱ）精确过滤，PDF 复用 download_pdf()，走 em_get 限流。端点数 27 → 28。
- 实测（2026-06-20）：全行业
hits=47928、按行业码 1238 过滤 hits=1863，首篇 PDF H3_{infoCode}_1.pdf 下载成功（2.5MB，%PDF 头）；行业码表端点（bxpa 等）404 不存在，用 "*" 拉取后从结果反查行业码。
V3.2.2（失效接口替换 + 隐藏 Bug 修复）：
- §3.3 概念板块归属（#18）：百度 PAE
getrelatedblock 失效（ResultCode 10003 + 空数组）→ 改用东财 slist（spt=3）eastmoney_concept_blocks()，一次请求拿全个股所属板块（行业/概念/地域 + BK码 + 涨跌幅 + 龙头股），零鉴权走 em_get 限流。
- §7.1 巨潮公告 orgId（#19）：硬编码
gssx0{code} 致大量 601xxx 股票 totalAnnouncement=0 → 新增 _cninfo_orgid() 动态查官方映射表 szse_stock.json（6198 只股，模块级缓存），硬编码降为 fallback。
- 综合示例修复：示例仍调用 v3.1 已删的
baidu_fund_flow_history → 改 eastmoney_fund_flow_minute。
- §4.5/§5.1 风控说明：部分大陆住宅 IP 被东财间歇风控（
HTTP 000/空）非代码 Bug，加重试/换网络提示。
- 新代码原样 exec smoke test 实测：板块归属 茅台27/五粮液28/绿的谐波21；公告 平安601318/工行601398 原失效股恢复。
V3.2.1（Bug 修复）： 修复两个内嵌函数的解析逻辑（预先存在，非 V3.2 引入）——
- §5.1 东财个股新闻：东财实际返回里
result.cmsArticleWebOld 直接就是文章列表，旧写法对 list 调 .get("list") 触发 AttributeError / 返回空 → 改为遍历 cmsArticleWebOld 列表本身。
- §6.4 新浪财报三表：新浪实际结构是
result.data.report_list（按报告期为键的 dict，每期 data 才是行项列表），旧写法取 result.data.{lrb} 永久返回空 → 改为遍历 report_list 期次、从每期 data 按 item_title 提取。
- 两函数均用茅台 600519 公开 API（零 key）实测返回非空、字段正确。
V3.2（防封 + 失效修复）：
- 数据源优先级 + 东财防封：明确「通达信(mootdx)/腾讯不封IP 优先用，东财仅用于其独有数据」原则；新增统一节流入口
em_get()，所有东财接口内置串行限流（间隔≥1s+随机抖动）+ 会话复用，AI 抄代码即自带防封。详见「数据源优先级 & 东财防封」章节。
- 财联社快讯下线（#14）：
cls.cn 旧 API 全面 404，标注弃用并改用东财全球资讯。
V3.1 修复： 替换 4 个失效接口（百度 PAE 资金流→东财 push2、大宗交易 RPT 报表名更新、机构席位改用 BUY/SELL 明细筛选）+ 修复东财全球资讯 req_trace 参数 + 修复巨潮公告 orgId 格式。
V3.0 Breaking Change：彻底移除 akshare 依赖，所有数据源改为直连 HTTP API（零第三方数据依赖，仅 mootdx 保留 TCP）。
使用方式： 本 skill 已作为专家包内置 skill 使用，WorkBuddy/CodeBuddy 会自动加载，无需手动安装。
行情层（实时，不封IP）
├── mootdx → K线 + 五档盘口 + 逐笔成交 (TCP 7709)
├── 腾讯财经 API → PE/PB/市值/换手率/涨跌停/指数/ETF (HTTP)
└── 百度股市通 → K线带MA5/10/20 (V3.0 新增，HTTP)
研报层
├── 东财 reportapi → 个股研报 + 行业研报 + PDF下载 + 评级 + 三年EPS
├── 同花顺 THS → 一致预期EPS (直连 basic.10jqka.com.cn)
└── iwencai → NL语义搜索研报 (唯一能力，需X-Claw)
信号层
├── 同花顺热点 → 当日强势股 + 题材归因 reason tags (零鉴权 73ms)
├── 同花顺北向 → hgt/sgt 分钟资金流向 + 本地自缓存历史
├── 东财 slist → 个股所属板块/概念归属 (V3.2.2 替换百度PAE)
├── 东财 push2 → 个股资金流向 分钟级 (V3.1 替换百度PAE)
├── 龙虎榜席位 → 上榜记录 + 买卖席位 TOP5 + 机构动向 (datacenter-web)
├── 全市场龙虎榜 → 每日全市场上榜股票 + 净买额排名 (datacenter-web)
├── 限售解禁日历 → 历史解禁 + 未来90天待解禁 (datacenter-web)
└── 行业板块排名 → 东财行业涨跌/上涨下跌家数 (V3.0 替换同花顺)
资金面 / 筹码层
├── 融资融券明细 → 日级融资余额/买入/偿还 + 融券 (datacenter-web)
├── 大宗交易 → 成交价/量 + 买卖方营业部 (datacenter-web)
├── 股东户数变化 → 季度股东户数 + 环比变化 (datacenter-web)
├── 分红送转 → 历史每股派息/送股/转增 (datacenter-web)
└── 个股资金流120日 → 主力/大单/中单/小单 日级净流入 (push2his)
新闻层
├── 东财个股新闻 → 个股相关新闻 (search-api-web JSONP)
├── 财联社快讯 → ⚠️ 已下线 (cls.cn 迁 Next.js，旧API 404)
└── 东财全球资讯 → 7×24 财经快讯 (np-weblist，财联社替代)
基础数据层
├── mootdx finance → 季报快照 (37字段, EPS/ROE/净利)
├── mootdx F10 → 公司资料 (9大类文本)
├── 东财个股信息 → 行业/总股本/流通股/市值/上市日期 (push2)
└── 新浪财报三表 → 资产负债表/利润表/现金流量表 (quotes.sina.cn)
公告层
├── 巨潮 cninfo → 公告全文检索+下载 (cninfo.com.cn)
└── mootdx F10 → 最新公告摘要
打板层 (V3.3 新增)
├── 东财涨停池 → 连板数/几天几板/封板资金/炸板次数/行业 (push2ex)
├── 东财炸板池 → 涨停后开板 + 振幅/涨速 (push2ex)
├── 东财跌停池 → 封单资金/连续跌停/开板次数 (push2ex)
├── 东财昨涨停池 → 昨涨停今表现，算晋级率/赚钱效应 (push2ex)
└── 同花顺涨停揭秘 → 涨停原因题材/封板成功率/板型 (10jqka)
ETF期权层 (V3.3 新增)
├── 合约清单 → 50ETF/300ETF/科创50/500ETF 各月认购认沽 (新浪)
├── T型报价 → 买卖五档/持仓量/行权价/最新价 (新浪)
└── 希腊字母+IV → Delta/Gamma/Theta/Vega/隐含波动率 (新浪)
舆情互动层 (V3.3 新增)
├── 互动易问答 → 投资者提问+公司回复 (巨潮，AI问答独家)
├── 同花顺热榜 → 人气值/概念标签/排名变化 (10jqka)
├── 东财人气榜 → 排名+排名变化+名称价格 (emappdata)
└── 东财概念命中 → 个股被归到哪些概念在炒+热度 (emappdata)
数据源优先级 & 东财防封（重要，先读）
优先级原则：能用通达信/腾讯，就别用东财
| 优先级 | 数据源 | 协议 | 封 IP 风险 | 覆盖 |
|---|
| 1（首选） | mootdx（通达信） | TCP 7709 二进制 | 不封 IP | K线、五档盘口、逐笔成交、财务快照、F10 |
| 2 | 腾讯财经 | HTTP GBK | 不封 IP | 实时价、PE/PB/市值/换手率/涨跌停、指数、ETF |
| 3 | 新浪 / 巨潮 / 同花顺 | HTTP | 低 | 财报三表、公告、一致预期/热点 |
| 4（仅独有数据才用） | 东财 eastmoney | HTTP | 有风控，会封 IP | 见下 |
凡是行情 / K线 / 实时价 / 市值 / 财务三表能从 mootdx 或腾讯拿到的，一律走它们——TCP 协议和腾讯接口实测不封 IP，可放心高频调用。
东财只用于它「独有、别处拿不到」的数据
下列数据只有东财有，通达信/腾讯/新浪都没有，必须用东财（但要限流）：
龙虎榜席位 · 全市场龙虎榜 · 限售解禁日历 · 融资融券 · 大宗交易 · 股东户数 · 分红送转 · 个股资金流向（分钟/日级）· 行业板块排名 · 研报列表/PDF · 个股新闻 · 全球资讯
东财风控阈值（社区实测，2026-05）
| 行为 | 触发封禁的阈值 | 风险 |
|---|
| 每秒请求数 | > 5 次/秒 | 高 |
| 单 IP 并发连接 | ≥ 10 | 高 |
| 1 分钟请求总数 | ≥ 200 次 | 中高 |
| 5 分钟请求总数 | ≥ 300 次 | 触发封禁 |
| User-Agent | 空 UA / 无浏览器特征 | 中 |
被封表现：连续请求后 403 / 429 / 连接超时 / 返回空数据。临时封禁通常几分钟到几小时。
防封铁律（调用东财时必须遵守）
- 串行，不并发——绝不对东财开多线程/协程并发请求
- 每次间隔 ≥ 1 秒 + 随机抖动（QPS ≤ 2），批量筛选时调大到 1.5~2 秒
- 复用 HTTP 会话（Keep-Alive），不要每次新建连接
- 带正常 UA + Referer（本 SKILL 各端点已配好）
- 批量场景每只股票之间 sleep——AI 跑批量循环（如筛选 100 只股逐个拉龙虎榜/资金流）是被封的头号元凶
已内置限流：所有东财请求走 em_get()
本 SKILL 提供统一的节流入口 em_get()（定义见下方「东财数据中心统一查询（共用 helper）」），它自动做到：串行限流（最小间隔 EM_MIN_INTERVAL=1.0s + 随机抖动）+ 复用 EM_SESSION（Keep-Alive）+ 默认 UA。所有 eastmoney.com 端点的代码块都已改用 em_get 而非裸 requests.get，AI 直接抄代码即自带防封。批量任务把 EM_MIN_INTERVAL 调大即可进一步降速。
注：em_get / EM_SESSION / EM_MIN_INTERVAL 是所有东财代码块共用的前置定义，使用任一东财端点前需先执行「共用 helper」代码块。
When to Activate
- 用户要查 A 股个股估值（一致预期 / PE / PEG / PE消化）
- 用户要拉实时行情（价格 / 五档盘口 / K线 / 涨跌停价）
- 用户要搜研报（按主题 / 按标的 / 按行业 / 下载PDF）
- 用户要看当日强势股 / 题材归因 / 概念热点
- 用户要看北向资金动向（沪股通/深股通分钟流向）
- 用户要看概念板块归属（行业/概念/地域）
- 用户要看个股资金流向（主力/散户/超大单/大单分钟级）
- 用户要看龙虎榜席位（营业部 + 机构买卖）
- 用户要看全市场龙虎榜（当日所有上榜股票 + 净买额排名）
- 用户要看限售解禁日历（历史解禁 + 未来待解禁）
- 用户要做行业横向对比（涨跌排名 / 资金流入 / 领涨股）
- 用户要看融资融券 / 两融数据（融资余额 + 融券余额）
- 用户要看大宗交易（成交价/量 + 买卖方营业部）
- 用户要看股东户数变化（筹码集中度）
- 用户要看分红送转历史（每股派息 + 送股 + 转增）
- 用户要看指数/ETF行情（上证指数 / 沪深300 / 创业板指 / ETF）
- 用户要看涨停 / 打板情绪（涨停池 / 连板梯队 / 炸板率 / 跌停 / 涨停原因题材）
- 用户要看ETF 期权（T型报价 / 希腊字母 Delta·Gamma·Theta·Vega / 隐含波动率 IV）
- 用户要看投资者互动问答（公司如何回应某传闻/利好 · 互动易）
- 用户要看市场热度 / 人气榜（同花顺热榜 / 东财人气榜 / 个股概念命中）
- 用户要看新闻资讯（个股新闻 / 财联社快讯 / 全球资讯）
- 用户要查公告（巨潮公告全文）
- 用户要做产业链调研 / 批量横向对比
- 关键词：估值、一致预期、机构预测、市盈率、PEG、市值、研报、产业链、行业研究、K线、盘口、公告、新闻、强势股、题材、热点、概念归因、北向资金、沪股通、深股通、概念板块、资金流向、主力、龙虎榜、席位、营业部、全市场龙虎榜、净买入、解禁、限售、行业对比、行业轮动、融资融券、两融、大宗交易、股东户数、筹码集中、分红、派息、送股、指数、ETF、涨停、打板、连板、炸板、跌停、涨停原因、封板、晋级率、ETF期权、希腊字母、隐含波动率、互动易、投资者关系、热榜、人气榜、市场热度
Prerequisites
pip install mootdx requests pandas stockstats
| 依赖 | 版本要求 | 用途 |
|---|
| mootdx | >= 0.10 | TCP行情+财务+F10（唯一非HTTP依赖）；0.11.x 用 tdx_client() 规避 BESTIP bug，见上节 |
| requests | any | 所有HTTP API直连 |
| pandas | any | 数据处理+HTML表格解析 |
| stockstats | any | 技术指标计算（RSI/MACD/BOLL等） |
V3.0 架构： 除 mootdx（TCP 二进制协议）外，所有数据源均为直连 HTTP API，零第三方数据封装依赖。每个端点的底层 URL/参数完全暴露，方便调试和定制。
iwencai API Key（仅语义搜索需要）
# 环境变量方式
export IWENCAI_API_KEY="your_key_here"
export IWENCAI_BASE_URL="https://openapi.iwencai.com"
# 申请地址: https://www.iwencai.com/skillhub
# 注册后安装 SkillHub CLI，再安装 report-search 技能即可获得 Key
其他数据源（mootdx / 腾讯 / 东财 / 同花顺 / 百度股市通 / 新浪 / 巨潮）全部免费，无需 key。
mootdx 客户端（必读，规避 0.11.x BESTIP 空串 bug）
已知 bug（mootdx 0.11.x）： 全新安装后 Quotes.factory(market='std') 裸调用可能抛 ValueError: not enough values to unpack (expected 2, got 0)。
根因：~/.mootdx/config.json 的 BESTIP.HQ 初始是空字符串 ""（不是缺失键），mootdx 用 dict.get(key, default) 取不到 default，拆包失败。老用户（config 曾填充过 IP）不会触发，所以容易漏测。
不要靠锁版本解决： 锁 mootdx==0.10.12 在部分环境（如干净的 Python 3.9）下 import mootdx 会因 numpy/pandas 二进制不兼容直接崩。正确做法是用下面的 tdx_client()——显式传 server 绕过 BESTIP，对 0.10 / 0.11 都适用。
统一用以下 helper 创建客户端（所有 mootdx 调用都走它）：
import socket
from mootdx.quotes import Quotes
# 实测可用的备选服务器（按延迟排序，2026-06 验证）
_TDX_SERVERS = [
('119.97.185.59', 7709), ('124.70.133.119', 7709), ('116.205.183.150', 7709),
('123.60.73.44', 7709), ('116.205.163.254', 7709), ('121.36.225.169', 7709),
('123.60.70.228', 7709), ('124.71.9.153', 7709), ('110.41.147.114', 7709),
('124.71.187.122', 7709),
]
def _probe(ip, port, timeout=2.0):
"""TCP 握手探测，判断服务器是否可达"""
try:
with socket.create_connection((ip, port), timeout=timeout):
return True
except Exception:
return False
def tdx_client(market='std'):
"""
创建 mootdx 客户端，规避 0.11.x BESTIP.HQ 空串 bug。
顺序兜底，保证 IP 列表老化/换网时仍能工作：
1) 顺序探测 _TDX_SERVERS，用第一个 TCP 可达的显式 server；
2) 全部不可达 → 回退 mootdx 自带 bestip 测速选优；
3) 再不行 → 回退裸 factory（老用户 config 已有可用 BESTIP 时成立）；
4) 仍失败 → 抛 RuntimeError，明确报错而非死等。
"""
for ip, port in _TDX_SERVERS:
if _probe(ip, port):
return Quotes.factory(market=market, server=(ip, port))
try:
return Quotes.factory(market=market, bestip=True) # fallback 1
except Exception:
pass
try:
return Quotes.factory(market=market) # fallback 2
except Exception as e:
raise RuntimeError(
"所有 mootdx 服务器均不可达。海外网络通常全部超时（TCP 7709），"
"请走国内代理或更新 _TDX_SERVERS 列表。原始错误：%s" % e
)
# 用法：client = tdx_client() # 替代所有 Quotes.factory(market='std')
海外 IP 用户： mootdx 走通达信 TCP 7709，海外环境通常全部超时。tdx_client() 会快速失败给出明确报错，而非死等。
市场前缀规则（全局通用）
def get_prefix(code: str) -> str:
"""6位代码 → 市场前缀"""
if code.startswith(("6", "9")):
return "sh"
elif code.startswith("8"):
return "bj"
else:
return "sz"
Ticker 格式归一化
所有接口统一支持多种输入格式，内部归一化为纯 6 位数字：
| 输入 | 归一化结果 |
|---|
688017 | 688017 |
SH688017 / sh688017 | 688017 |
688017.SH / 688017.sh | 688017 |
SZ000001 | 000001 |
BJ832000 | 832000 |
东财数据中心统一查询（共用 helper）
龙虎榜/解禁/融资融券/大宗交易/股东户数/分红 共用同一 base URL：
import time
import random
import requests
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"
# ── 东财防封：全局节流 + 会话复用 ────────────────────────────────────
# 东财系 HTTP 接口（push2 / datacenter / reportapi / search / np-weblist）有风控：
# 每秒 >5 次 / 单 IP 并发 ≥10 / 1 分钟 ≥200 次 → 临时封 IP。
# 所有 eastmoney.com 请求一律走 em_get()：串行限流（最小间隔 + 随机抖动）+ 复用
# Keep-Alive 会话，批量调用时自动降速，避免被封。详见「数据源优先级 & 东财防封」章节。
EM_SESSION = requests.Session()
EM_SESSION.headers.update({"User-Agent": UA})
# 连接级自动重试：瞬态连接错误 / 429 / 5xx 指数退避重试（住宅IP偶发风控更稳）。
# 注意：403 不重试（东财风控信号，重试无益反而加重；按下方 EM_MIN_INTERVAL 降频应对）。
try:
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
_em_adapter = HTTPAdapter(max_retries=Retry(
total=3, connect=3, backoff_factor=0.6,
status_forcelist=[429, 500, 502, 503, 504], allowed_methods=["GET"]))
EM_SESSION.mount("https://", _em_adapter)
EM_SESSION.mount("http://", _em_adapter)
except Exception:
pass # 老版本 urllib3 缺 allowed_methods 等参数时降级为无重试，不影响主流程
EM_MIN_INTERVAL = 1.0 # 两次东财请求最小间隔(秒)；批量筛选建议调大到 1.5~2
_em_last_call = [0.0] # 模块级上次请求时间戳
def em_get(url: str, params: dict | None = None, headers: dict | None = None,
timeout: int = 15, **kwargs):
"""东财统一请求入口：自动节流 + 复用 session + 默认 UA。
所有 eastmoney.com 接口都应通过它请求，避免高频被封 IP。"""
wait = EM_MIN_INTERVAL - (time.time() - _em_last_call[0])
if wait > 0:
time.sleep(wait + random.uniform(0.1, 0.5))
try:
return EM_SESSION.get(url, params=params, headers=headers, timeout=timeout, **kwargs)
finally:
_em_last_call[0] = time.time()
def eastmoney_datacenter(report_name: str, columns: str = "ALL",
filter_str: str = "", page_size: int = 50,
sort_columns: str = "", sort_types: str = "-1") -> list[dict]:
"""东财数据中心统一查询 — 龙虎榜/解禁/融资融券/大宗交易/股东户数/分红 共用（已内置限流）"""
params = {
"reportName": report_name, "columns": columns,
"filter": filter_str, "pageNumber": "1", "pageSize": str(page_size),
"sortColumns": sort_columns, "sortTypes": sort_types,
"source": "WEB", "client": "WEB",
}
r = em_get(DATACENTER_URL, params=params, timeout=15)
d = r.json()
if d.get("result") and d["result"].get("data"):
return d["result"]["data"]
return []
Layer 1: 行情层（实时，不封IP）
1.1 mootdx — K线 + 五档盘口 + 逐笔成交
TCP 二进制协议，连通达信服务器(7709)，无需注册，不封IP。
from mootdx.quotes import Quotes
client = tdx_client() # 见 Prerequisites 的 tdx_client() helper（规避 0.11.x BESTIP bug；等价 Quotes.factory(market='std')）
# === K线数据 ===
# ⚠️ 参数名是 frequency（不是 category！传 category 会被 **kwargs 静默吞掉，
# 永远退化成默认 frequency=9 日线，拿不到分钟数据）。
# mootdx 0.11.7 实测频率值表：
# 0=5分钟 1=15分钟 2=30分钟 3=60分钟(1小时) 4=日线 5=周线 6=月线
# 8=1分钟 9=日线(默认) 10=季线 11=年线 （7=1分钟除权口径,少用）
klines = client.bars(symbol='688017', frequency=9, offset=10) # 日线
min1 = client.bars(symbol='688017', frequency=8, offset=240) # 1分钟（一个交易日≈240根）
min5 = client.bars(symbol='688017', frequency=0, offset=48) # 5分钟
# 返回: open, close, high, low, vol, amount, datetime
# ⚠️ 复权：bars 返回【不复权】原始价（通达信原始数据，无 adjust 参数）。
# 跨除权除息日做估值/回测前需自行复权，或改用带前复权的日K数据源（腾讯财经）。
# === 实时报价 ===
quotes = client.quotes(symbol=['688017', '300476'])
# 返回 46 个字段:
# price(现价), open, high, low, last_close(昨收)
# bid1~bid5, ask1~ask5, bid_vol1~bid_vol5, ask_vol1~ask_vol5
# vol(成交量), amount(成交额), servertime
# === 逐笔成交（非交易时间返回空）===
trades = client.transaction(symbol='688017', date='20260502')
# 返回: time, price, vol, num, buyorsell(0买/1卖/2中性)
mootdx 不提供 PE / PB / 市值 / 换手率 / 涨跌停价 — 这些走腾讯财经。
1.2 腾讯财经 API — PE/PB/市值/换手率/涨跌停/指数/ETF
HTTP GET，GBK 编码，~ 分隔 88 个字段，不封IP。
import urllib.request
def tencent_quote(codes: list[str]) -> dict[str, dict]:
"""
批量拉取腾讯财经实时行情。
codes: ["688017", "300476", "002463"]
也支持指数: ["000001", "000300", "399006"]
也支持ETF: ["510050", "510300"]
返回: {code: {name, price, pe_ttm, pb, mcap, ...}}
"""
prefixed = []
for c in codes:
if c.startswith(("6", "9")):
prefixed.append(f"sh{c}")
elif c.startswith("8"):
prefixed.append(f"bj{c}")
else:
prefixed.append(f"sz{c}")
url = "https://qt.gtimg.cn/q=" + ",".join(prefixed)
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")
result = {}
for line in data.strip().split(";"):
if not line.strip() or "=" not in line or '"' not in line:
continue
key = line.split("=")[0].split("_")[-1]
vals = line.split('"')[1].split("~")
if len(vals) < 53:
continue
code = key[2:]
result[code] = {
"name": vals[1],
"price": float(vals[3]) if vals[3] else 0,
"last_close": float(vals[4]) if vals[4] else 0,
"open": float(vals[5]) if vals[5] else 0,
"change_amt": float(vals[31]) if vals[31] else 0,
"change_pct": float(vals[32]) if vals[32] else 0,
"high": float(vals[33]) if vals[33] else 0,
"low": float(vals[34]) if vals[34] else 0,
"amount_wan": float(vals[37]) if vals[37] else 0,
"turnover_pct": float(vals[38]) if vals[38] else 0,
"pe_ttm": float(vals[39]) if vals[39] else 0,
"amplitude_pct":float(vals[43]) if vals[43] else 0,
"mcap_yi": float(vals[44]) if vals[44] else 0,
"float_mcap_yi":float(vals[45]) if vals[45] else 0,
"pb": float(vals[46]) if vals[46] else 0,
"limit_up": float(vals[47]) if vals[47] else 0,
"limit_down": float(vals[48]) if vals[48] else 0,
"vol_ratio": float(vals[49]) if vals[49] else 0,
"pe_static": float(vals[52]) if vals[52] else 0,
}
return result
# 用法: 个股
quotes = tencent_quote(["688017", "300476", "002463"])
for code, q in quotes.items():
print(f"{q['name']}({code}): {q['price']}元 PE={q['pe_ttm']} PB={q['pb']} 市值={q['mcap_yi']}亿")
# 用法: 指数 — sh000001=上证指数, sh000300=沪深300, sz399006=创业板指
index_quotes = tencent_quote(["000001", "000300", "399006"])
# 用法: ETF — sh510050=上证50ETF, sh510300=沪深300ETF
etf_quotes = tencent_quote(["510050", "510300"])
腾讯财经字段索引速查（实测校准 2026-05-03）
| 索引 | 含义 | 示例 |
|---|
| 1 | 名称 | 绿的谐波 |
| 3 | 当前价 | 224.12 |
| 4 | 昨收 | 215.01 |
| 5 | 今开 | 214.10 |
| 9-18 | 买一~买五(价+量) | |
| 19-28 | 卖一~卖五(价+量) | |
| 31 | 涨跌额 | 9.11 |
| 32 | 涨跌幅% | 4.24 |
| 33 | 最高 | 229.62 |
| 34 | 最低 | 214.10 |
| 37 | 成交额(万) | 187040 |
| 38 | 换手率% | 4.55 |
| 39 | PE(TTM) | 300.45 |
| 43 | 振幅%（不是PB！） | 7.22 |
| 44 | 总市值(亿) | 410.88 |
| 45 | 流通市值(亿) | 410.88 |
| 46 | PB(市净率) | 11.51 |
| 47 | 涨停价 | 258.01 |
| 48 | 跌停价 | 172.01 |
| 49 | 量比 | 1.20 |
| 52 | PE(静) | 314.76 |
踩坑提醒： 网上很多教程把索引 43 写成 PB，实测是振幅%。PB 在索引 46。
1.3 百度股市通 K线 — 带MA5/MA10/MA20（V3.0 新增）
核心价值： 返回时自带均线数据，无需本地计算。
import requests
def baidu_kline_with_ma(code: str, start_time: str = "") -> dict:
"""百度股市通K线 — 独有能力: 返回时自带 ma5/ma10/ma20 均价"""
url = "https://finance.pae.baidu.com/selfselect/getstockquotation"
params = {
"all": "1", "isIndex": "false", "isBk": "false", "isBlock": "false",
"isFutures": "false", "isStock": "true", "newFormat": "1",
"group": "quotation_kline_ab", "finClientType": "pc",
"code": code, "start_time": start_time, "ktype": "1",
}
headers = {
"User-Agent": "Mozilla/5.0",
"Accept": "application/vnd.finance-web.v1+json",
"Origin": "https://gushitong.baidu.com",
"Referer": "https://gushitong.baidu.com/",
}
r = requests.get(url, params=params, headers=headers, timeout=10)
d = r.json()
result = d.get("Result", {})
md = result.get("newMarketData", {})
keys = md.get("keys", []) # includes: ma5avgprice, ma10avgprice, ma20avgprice
rows = md.get("marketData", "").split(";")
return {"keys": keys, "rows": rows}
# 用法
data = baidu_kline_with_ma("600519")
print("字段:", data["keys"][:10])
print("最近5根K线:", data["rows"][-5:])
# keys 包含: time, open, close, high, low, volume, amount, ma5avgprice, ma10avgprice, ma20avgprice 等
Layer 2: 研报层
2.1 东财研报 API — 研报列表 + PDF下载（主力）
A级接口（公开JSON API），reportapi.eastmoney.com，免费无key。
import requests
import re
import time
from pathlib import Path
REPORT_API = "https://reportapi.eastmoney.com/report/list"
PDF_TPL = "https://pdf.dfcfw.com/pdf/H3_{info_code}_1.pdf"
UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
def eastmoney_reports(code: str, max_pages: int = 5) -> list[dict]:
"""拉取指定股票的研报列表"""
all_records = []
for page in range(1, max_pages + 1):
params = {
"industryCode": "*", "pageSize": "100", "industry": "*",
"rating": "*", "ratingChange": "*",
"beginTime": "2000-01-01", "endTime": "2030-01-01",
"pageNo": str(page), "fields": "", "qType": "0",
"orgCode": "", "code": code, "rcode": "",
"p": str(page), "pageNum": str(page), "pageNumber": str(page),
}
r = em_get(REPORT_API, params=params,
headers={"Referer": "https://data.eastmoney.com/"}, timeout=30) # 已内置限流
d = r.json()
rows = d.get("data") or []
if not rows:
break
all_records.extend(rows)
if page >= (d.get("TotalPage", 1) or 1):
break
return all_records
def download_pdf(record: dict, target_dir: str = "./reports") -> str | None:
"""下载单份研报PDF，返回保存路径或None"""
info_code = record.get("infoCode", "")
if not info_code:
return None
date = (record.get("publishDate") or "")[:10]
org = re.sub(r'[\\/:*?"<>|]', "_", record.get("orgSName") or "未知")[:40]
title = re.sub(r'[\\/:*?"<>|]', "_", record.get("title", ""))[:80]
fname = f"{date}_{org}_{title}.pdf"
target = Path(target_dir) / fname
if target.exists():
return str(target)
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:
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(r.content)
return str(target)
return None
# 用法
reports = eastmoney_reports("688017")
print(f"共 {len(reports)} 篇研报")
for r in reports[:5]:
print(f" {r.get('publishDate','')[:10]} | {r.get('orgSName')} | {r.get('title','')[:60]}")
研报 record 关键字段
| 字段 | 含义 |
|---|
| title | 研报标题 |
| publishDate | 发布日期 |
| orgSName | 机构简称 |
| infoCode | 用于拼 PDF URL |
| predictThisYearEps | 今年EPS预测 |
| predictNextYearEps | 明年EPS预测 |
| predictNextTwoYearEps | 后年EPS预测 |
| emRatingName | 评级(买入/增持/...) |
| indvInduName | 行业分类 |
行业研报列表（qType=1）
与个股研报同一端点（reportapi.eastmoney.com/report/list），仅 qType 不同：qType=0 个股研报，qType=1 行业研报。返回 record 可直接喂给上面的 download_pdf()（PDF 模板通用）。
def eastmoney_industry_reports(industry_code: str = "*", max_pages: int = 5,
begin: str = "2024-01-01") -> list[dict]:
"""拉取行业研报列表（qType=1）。
industry_code="*" = 全行业；传东财行业码（如 "1238"=IT服务Ⅱ）= 单行业。
行业名 / 行业码在每条 record 的 industryName / industryCode 字段。"""
all_records = []
for page in range(1, max_pages + 1):
params = {
"industryCode": industry_code, "pageSize": "100", "industry": "*",
"rating": "*", "ratingChange": "*",
"beginTime": begin, "endTime": "2030-01-01",
"pageNo": str(page), "fields": "", "qType": "1",
}
r = em_get(REPORT_API, params=params,
headers={"Referer": "https://data.eastmoney.com/"}, timeout=30) # 已内置限流
d = r.json()
rows = d.get("data") or []
if not rows:
break
all_records.extend(rows)
if page >= (d.get("TotalPage", 1) or 1):
break
return all_records
# 用法
# 1) 全行业最新研报
reports = eastmoney_industry_reports("*", max_pages=2)
print(f"共 {len(reports)} 篇行业研报")
for r in reports[:5]:
print(f" {r.get('publishDate','')[:10]} | {r.get('industryName')} | {r.get('orgSName')} | {r.get('title','')[:50]}")
# 2) 单行业（IT服务Ⅱ，行业码 1238）+ 下载首篇 PDF（复用 2.1 的 download_pdf）
it = eastmoney_industry_reports("1238", max_pages=1)
if it:
download_pdf(it[0])
行业研报特有/常用字段（其余字段同 2.1 个股研报）：
| 字段 | 含义 |
|---|
| industryName | 行业名称（如 IT服务Ⅱ、风电设备、光伏设备） |
| industryCode | 东财行业代码（用于 industry_code 精确过滤） |
| emRatingName | 行业评级（买入/增持/中性/...） |
| reportType | 报告类型 |
| attachPages / attachSize | PDF 页数 / 大小(KB) |
| infoCode | 喂给 download_pdf() 拼 PDF URL |
行业码怎么拿： 东财行业码不是通用记忆码，没有公开的码表端点（bxpa 等已 404）。常用做法：先用 industry_code="*" 拉一批，从结果的 industryName/industryCode 找到目标行业的码，再用该码精确过滤。
2.2 同花顺一致预期EPS（直连 basic.10jqka.com.cn）
import requests
import pandas as pd
from io import StringIO
def ths_eps_forecast(code: str) -> pd.DataFrame:
"""
同花顺机构一致预期EPS。
直连 basic.10jqka.com.cn，解析HTML表格。
返回 DataFrame: 年度, 预测机构数, 最小值, 均值, 最大值
"均值" = 机构一致预期EPS
"""
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/",
}
r = requests.get(url, headers=headers, timeout=15)
r.encoding = "gbk"
dfs = pd.read_html(StringIO(r.text))
# 找含"每股收益"的表格
for df in dfs:
cols = [str(c) for c in df.columns]
if any("每股收益" in c or "均值" in c for c in cols):
return df
# fallback: 返回第一个表
return dfs[0] if dfs else pd.DataFrame()
# 用法
df = ths_eps_forecast("688017")
print(df)
# "预测机构数" < 3 的要谨慎
2.3 iwencai — NL语义搜索研报（唯一能力）
需要 API Key + X-Claw Headers（SkillHub 2.0 强制要求）。
import os
import json
import secrets
import requests
IWENCAI_BASE = os.environ.get("IWENCAI_BASE_URL", "https://openapi.iwencai.com")
IWENCAI_KEY = os.environ.get("IWENCAI_API_KEY", "")
def _claw_headers(call_type: str = "normal") -> dict:
"""SkillHub 2.0 必须的 X-Claw 鉴权头"""
return {
"X-Claw-Call-Type": call_type,
"X-Claw-Skill-Id": "report-search",
"X-Claw-Skill-Version": "2.0.0",
"X-Claw-Plugin-Id": "none",
"X-Claw-Plugin-Version": "none",
"X-Claw-Trace-Id": secrets.token_hex(32),
}
def iwencai_search(query: str, channel: str = "report", size: int = 50) -> list[dict]:
"""
iwencai 语义搜索。
channel: "report"(研报) / "announcement"(公告) / "news"(新闻)
size: 默认10, 实测可调到50（隐藏参数）
"""
headers = {
"Authorization": f"Bearer {IWENCAI_KEY}",
"Content-Type": "application/json",
**_claw_headers(),
}
payload = {
"channels": [channel],
"app_id": "AIME_SKILL",
"query": query,
"size": size,
}
r = requests.post(
f"{IWENCAI_BASE}/v1/comprehensive/search",
json=payload, headers=headers, timeout=30,
)
if r.status_code != 200:
raise RuntimeError(f"iwencai HTTP {r.status_code}: {r.text[:200]}")
data = r.json()
if data.get("status_code", 0) != 0:
raise RuntimeError(f"iwencai error: {data.get('status_msg', '')}")
return data.get("data") or []
def iwencai_query(query: str, page: int = 1, limit: int = 50) -> list[dict]:
"""
iwencai NL数据查询（结构化字段）。
例: "贵州茅台 ROE" → DataFrame-like rows
"""
headers = {
"Authorization": f"Bearer {IWENCAI_KEY}",
"Content-Type": "application/json",
**_claw_headers(),
}
payload = {
"query": query,
"page": str(page),
"limit": str(limit),
"is_cache": "1",
"expand_index": "true",
}
r = requests.post(
f"{IWENCAI_BASE}/v1/query2data",
json=payload, headers=headers, timeout=30,
)
if r.status_code != 200:
raise RuntimeError(f"iwencai HTTP {r.status_code}: {r.text[:200]}")
data = r.json()
if data.get("status_code", 0) != 0:
raise RuntimeError(f"iwencai error: {data.get('status_msg', '')}")
return data.get("datas") or []
def dedup_articles(articles: list[dict]) -> list[dict]:
"""同一uid仅保留score最高的段落"""
best = {}
for a in articles:
uid = a.get("uid", "") or f"{a.get('title','')}|{a.get('publish_date','')}"
score = float(a.get("score", 0))
if uid not in best or score > float(best[uid].get("score", 0)):
best[uid] = a
return sorted(best.values(), key=lambda x: x.get("publish_date", ""), reverse=True)
# 用法: NL语义搜索研报
articles = iwencai_search("人形机器人 行星滚柱丝杠 2026", channel="report", size=50)
articles = dedup_articles(articles)
for a in articles[:5]:
extra = a.get("extra") or {}
if isinstance(extra, str):
extra = json.loads(extra)
print(f"{a.get('publish_date','')[:10]} | {extra.get('organization','')} | {a.get('title','')[:60]}")
iwencai 的唯一价值： NL 主题搜索。"人形机器人 行星滚柱丝杠" 这种跨主题检索只有 iwencai 能做。按标的搜研报走东财 reportapi 更稳定。
Layer 3: 信号层
3.1 同花顺热点 — 当日强势股 + 题材归因 reason tags（独家）
核心价值： 不只告诉你"哪些走强"，还告诉你**"为什么走强"** —— 同花顺编辑部人工运营的题材标签。
import requests
import pandas as pd
def ths_hot_reason(date: str = None) -> pd.DataFrame:
"""
同花顺当日强势股归因。
date: 'YYYY-MM-DD' 格式，None=今天
返回 DataFrame，含每只股票的题材标签 (reason)。
实测: 73ms 拿到 ~125 只 + 完整字段
"""
from datetime import date as _date
if date is None:
date = _date.today().strftime("%Y-%m-%d")
url = (
f"http://zx.10jqka.com.cn/event/api/getharden/"
f"date/{date}/orderby/date/orderway/desc/charset/GBK/"
)
headers = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"Chrome/117.0.0.0 Safari/537.36"
)
}
r = requests.get(url, headers=headers, timeout=10)
data = r.json()
if data.get("errocode", 0) != 0:
raise RuntimeError(f"同花顺热点错误: {data.get('errormsg', '')}")
rows = data.get("data") or []
df = pd.DataFrame(rows)
if df.empty:
return df
# 字段重命名（中文友好）
rename_map = {
"name": "名称", "code": "代码", "reason": "题材归因",
"close": "收盘价", "zhangdie": "涨跌额", "zhangfu": "涨幅%",
"huanshou": "换手率%", "chengjiaoe": "成交额",
"chengjiaoliang": "成交量", "ddejingliang": "大单净量",
"market": "市场",
}
df = df.rename(columns=rename_map)
return df
# 用法
df = ths_hot_reason("2026-05-09")
print(f"当日强势股: {len(df)} 只")
print(df[["代码", "名称", "涨幅%", "题材归因"]].head(10))
同花顺热点字段速查
| 原字段 | 中文 | 说明 |
|---|
| code | 代码 | 6 位股票代码 |
| name | 名称 | 简称 |
| reason | 题材归因 | 核心字段，人工运营 tags，如"算力租赁+Token工厂+AI政务" |
| zhangfu | 涨幅% | 当日涨幅 |
| huanshou | 换手率% | 当日换手 |
| chengjiaoe | 成交额 | 元 |
| chengjiaoliang | 成交量 | 股 |
| ddejingliang | 大单净量 | 主力净流入指标 |
| close | 收盘价 | 元 |
| zhangdie | 涨跌额 | 元 |
| market | 市场 | 沪/深/北 |
3.2 同花顺北向资金 — hsgtApi 实时分钟流向 + 本地自缓存历史
已知行业性问题： eastmoney 全系北向数据自 2024-08 后净买额字段返回 NaN/0，属上游断供。已改为本地 CSV 自缓存模式——每次拉实时数据后自动写入本地 CSV，历史越跑越丰富。
import requests
import pandas as pd
from pathlib import Path
HSGT_HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"Chrome/117.0.0.0 Safari/537.36"
),
"Host": "data.hexin.cn",
"Referer": "https://data.hexin.cn/",
}
def hsgt_realtime() -> pd.DataFrame:
"""
沪深股通当日实时分钟流向（含集合竞价 09:10–15:00，262 个时间点）。
返回字段: time, hgt(沪股通累计净买入), sgt(深股通累计净买入)
单位: 亿元
"""
url = "https://data.hexin.cn/market/hsgtApi/method/dayChart/"
r = requests.get(url, headers=HSGT_HEADERS, timeout=10)
d = r.json()
times = d.get("time", [])
hgt = d.get("hgt", [])
sgt = d.get("sgt", [])
n = len(times)
return pd.DataFrame({
"time": times,
"hgt_yi": hgt[:n] + [None] * (n - len(hgt)),
"sgt_yi": sgt[:n] + [None] * (n - len(sgt)),
})
# === 自缓存辅助函数 ===
def _northbound_cache_path() -> Path:
"""北向资金本地 CSV 缓存路径"""
p = Path.home() / ".tradingagents" / "cache" / "northbound_daily.csv"
p.parent.mkdir(parents=True, exist_ok=True)
return p
def _save_northbound_snapshot(date: str, hgt: float, sgt: float):
"""写入/更新当天北向收盘数据到 CSV"""
path = _northbound_cache_path()
rows = {}
if path.exists():
for line in path.read_text().strip().split("\n")[1:]:
parts = line.split(",")
if len(parts) == 3:
rows[parts[0]] = line
rows[date] = f"{date},{hgt},{sgt}"
with open(path, "w") as f:
f.write("date,hgt,sgt\n")
for d in sorted(rows.keys()):
f.write(rows[d] + "\n")
def _load_northbound_history(n: int = 20) -> pd.DataFrame:
"""读取最近 N 天北向历史"""
path = _northbound_cache_path()
if not path.exists():
return pd.DataFrame()
df = pd.read_csv(path)
return df.tail(n)
# 用法 1: 实时分钟流向
df = hsgt_realtime()
print(f"分钟点数: {len(df)}")
print(df.tail(5))
# 用法 2: 自动缓存今日收盘数据
if not df.empty:
last = df.dropna().iloc[-1]
_save_northbound_snapshot("2026-05-17", last["hgt_yi"], last["sgt_yi"])
# 用法 3: 读取历史
hist = _load_northbound_history(20)
print(hist)
3.3 东财 slist — 个股所属板块/概念归属（V3.2.2 替换百度）
核心价值： 一次调用拿到个股所属的全部板块（行业 + 概念 + 地域混合），含板块代码（BK码）、当日涨跌幅、板块龙头股。题材归因、板块联动分析必备。
V3.2.2 替换说明： 百度 PAE getrelatedblock 接口已失效（实测返回 ResultCode 10003 + 空数组，#18），改用东财 slist 个股所属板块接口（spt=3，一次请求拿全，零鉴权）。东财把行业/概念/地域混在一个列表里返回，板块名本身已自解释（如「食品饮料」是行业、「贵州板块」是地域、「酿酒概念」是概念），AI 直接用板块名做题材归因即可。
def eastmoney_concept_blocks(code: str) -> dict:
"""
个股所属板块/概念归属（东财 slist，一次请求拿全，已内置限流）。
返回: {total, boards: [{name, code(BK码), change_pct, lead_stock}], concept_tags: [板块名...]}
boards 混合 行业/概念/地域，板块名自解释；concept_tags 是所有板块名的便捷列表。
"""
market_code = 1 if code.startswith("6") else 0
params = {
"fltt": "2", "invt": "2",
"secid": f"{market_code}.{code}",
"spt": "3", "pi": "0", "pz": "200", "po": "1",
"fields": "f12,f14,f3,f128",
}
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)
d = r.json()
except Exception as e:
print(f"[WARN] 东财板块归属请求失败: {e}")
return {"total": 0, "boards": [], "concept_tags": []}
diff = (d.get("data") or {}).get("diff") or {}
items = diff.values() if isinstance(diff, dict) else diff
boards = []
for it in items:
boards.append({
"name": it.get("f14", ""), # 板块名
"code": it.get("f12", ""), # BK 板块代码
"change_pct": it.get("f3", ""), # 板块当日涨跌幅
"lead_stock": it.get("f128", ""), # 板块龙头股
})
return {
"total": len(boards),
"boards": boards,
"concept_tags": [b["name"] for b in boards],
}
# 用法
blocks = eastmoney_concept_blocks("600519")
print(f"共 {blocks['total']} 个板块")
print("板块归属:", blocks["concept_tags"])
# → ['食品饮料', '白酒Ⅲ', '白酒Ⅱ', '贵州板块', '酿酒概念', 'HS300_', ...]
注意： 东财不区分行业/概念/地域类型（混在一个列表返回）。如需精确分类可按板块名判断，或另查全市场板块清单（clist + m:90+t:1/2/3）——但后者每次需多发请求、大页易触发风控，不推荐在批量场景用。
3.4 东财 push2 — 个股资金流向（分钟级）
盘中实时分钟级资金流（主力/大单/中单/小单/超大单净流入）。
V3.1 替换说明： 百度 PAE fundflow 和 fundsortlist 接口已于 2026-05 下线（返回 null），改用东财 push2 资金流 API。日级资金流见 Layer 4.5 stock_fund_flow_120d()。
import requests
def eastmoney_fund_flow_minute(code: str) -> list[dict]:
"""
个股资金流向（分钟级，当日盘中）。
code: 6位股票代码
返回: [{time, main_net, small_net, mid_net, large_net, super_net}, ...]
单位: 元
"""
secid = f"1.{code}" if code.startswith("6") else f"0.{code}"
url = "https://push2.eastmoney.com/api/qt/stock/fflow/kline/get"
params = {
"secid": secid, "klt": 1,
"fields1": "f1,f2,f3,f7",
"fields2": "f51,f52,f53,f54,f55,f56,f57",
}
headers = {
"User-Agent": UA,
"Referer": "https://quote.eastmoney.com/",
"Origin": "https://quote.eastmoney.com",
}
try:
r = em_get(url, params=params, headers=headers, timeout=10)
d = r.json()
except Exception as e:
print(f"[WARN] push2 资金流请求失败: {e}")
return []
rows = []
for line in d.get("data", {}).get("klines", []):
parts = line.split(",")
if len(parts) >= 6:
rows.append({
"time": parts[0],
"main_net": float(parts[1]),
"small_net": float(parts[2]),
"mid_net": float(parts[3]),
"large_net": float(parts[4]),
"super_net": float(parts[5]),
})
return rows
# 用法: 分钟级实时资金流
realtime = eastmoney_fund_flow_minute("000858")
if realtime:
last = realtime[-1]
signal = "bullish" if last["main_net"] > 0 else "bearish"
print(f"主力净流入: {last['main_net']:.0f}元 → {signal}")
# 统计全天主力净流入
total = sum(r["main_net"] for r in realtime)
print(f"全天主力累计: {total/1e4:.0f}万元")
注意： push2 资金流金额单位是元（非万元），使用时注意换算。klt=1 分钟级，klt=101 日级。
3.5 龙虎榜席位 — 个股上榜记录 + 买卖席位 TOP5 + 机构动向
直连东财 datacenter API，不依赖第三方封装。
import requests
from datetime import datetime, timedelta
def dragon_tiger_board(code: str, trade_date: str, look_back: int = 30) -> dict:
"""
龙虎榜数据聚合。
trade_date: YYYY-MM-DD
look_back: 回看天数
返回: {records: [...], seats: {buy: [...], sell: [...]}, institution: {...}}
"""
start = datetime.strptime(trade_date, "%Y-%m-%d") - timedelta(days=look_back)
start_str = start.strftime("%Y-%m-%d")
# 1. 上榜记录
records = []
data = eastmoney_datacenter(
"RPT_DAILYBILLBOARD_DETAILSNEW",
filter_str=f"(TRADE_DATE>='{start_str}')(TRADE_DATE<='{trade_date}')(SECURITY_CODE=\"{code}\")",
page_size=50,
sort_columns="TRADE_DATE", sort_types="-1",
)
for row in data:
records.append({
"date": str(row.get("TRADE_DATE", ""))[:10],
"reason": row.get("EXPLANATION", ""),
"net_buy": round((row.get("BILLBOARD_NET_AMT") or 0) / 10000, 1),
"turnover": round(float(row.get("TURNOVERRATE") or 0), 2),
})
# 2. 最近上榜的买卖席位
seats = {"buy": [], "sell": []}
if records:
latest_date = records[0]["date"]
# 买入席位
buy_data = eastmoney_datacenter(
"RPT_BILLBOARD_DAILYDETAILSBUY",
filter_str=f"(TRADE_DATE='{latest_date}')(SECURITY_CODE=\"{code}\")",
page_size=10,
sort_columns="BUY", sort_types="-1",
)
for row in buy_data[:5]:
seats["buy"].append({
"name": row.get("OPERATEDEPT_NAME", ""),
"buy_amt": round((row.get("BUY") or 0) / 10000, 1),
"sell_amt": round((row.get("SELL") or 0) / 10000, 1),
"net": round((row.get("NET") or 0) / 10000, 1),
})
# 卖出席位
sell_data = eastmoney_datacenter(
"RPT_BILLBOARD_DAILYDETAILSSELL",
filter_str=f"(TRADE_DATE='{latest_date}')(SECURITY_CODE=\"{code}\")",
page_size=10,
sort_columns="SELL", sort_types="-1",
)
for row in sell_data[:5]:
seats["sell"].append({
"name": row.get("OPERATEDEPT_NAME", ""),
"buy_amt": round((row.get("BUY") or 0) / 10000, 1),
"sell_amt": round((row.get("SELL") or 0) / 10000, 1),
"net": round((row.get("NET") or 0) / 10000, 1),
})
# 3. 机构买卖统计（从买卖席位明细中筛选 OPERATEDEPT_CODE="0" 即机构专用席位）
institution = {"buy_amt": 0, "sell_amt": 0, "net_amt": 0}
for detail_data, side in [(buy_data, "buy"), (sell_data, "sell")]:
for row in detail_data:
if str(row.get("OPERATEDEPT_CODE", "")) == "0":
amt = (row.get("BUY") or 0) if side == "buy" else (row.get("SELL") or 0)
if side == "buy":
institution["buy_amt"] += amt
else:
institution["sell_amt"] += amt
institution["buy_amt"] = round(institution["buy_amt"] / 10000, 1)
institution["sell_amt"] = round(institution["sell_amt"] / 10000, 1)
institution["net_amt"] = round(institution["buy_amt"] - institution["sell_amt"], 1)
return {"records": records, "seats": seats, "institution": institution}
# 用法
data = dragon_tiger_board("002475", "2026-05-17")
print(f"近30日上榜 {len(data['records'])} 次")
for r in data["records"]:
print(f" {r['date']}: {r['reason']}")
if data["seats"]["buy"]:
print("买入席位 TOP5:")
for s in data["seats"]["buy"]:
print(f" {s['name']}: 买{s['buy_amt']}万 卖{s['sell_amt']}万 净{s['net']}万")
ST 股注意： 5% 涨跌停更容易触发龙虎榜（"连续三日偏离值累计达12%"），科创板 20% 涨跌停则较少触发。
3.6 限售解禁日历 — 历史解禁 + 未来 90 天待解禁
from datetime import datetime, timedelta
def lockup_expiry(code: str, trade_date: str, forward_days: int = 90) -> dict:
"""
限售解禁日历。
返回: {history: [...], upcoming: [...]}
"""
# 1. 历史解禁记录
history_data = eastmoney_datacenter(
"RPT_LIFT_STAGE",
filter_str=f"(SECURITY_CODE=\"{code}\")",
page_size=15,
sort_columns="FREE_DATE", sort_types="-1",
)
history = []
for row in history_data:
history.append({
"date": str(row.get("FREE_DATE", ""))[:10],
"type": row.get("LIMITED_STOCK_TYPE", ""),
"shares": row.get("FREE_SHARES_NUM", 0),
"ratio": row.get("FREE_RATIO", 0),
})
# 2. 未来待解禁
end_date = datetime.strptime(trade_date, "%Y-%m-%d") + timedelta(days=forward_days)
end_str = end_date.strftime("%Y-%m-%d")
upcoming_data = eastmoney_datacenter(
"RPT_LIFT_STAGE",
filter_str=f"(SECURITY_CODE=\"{code}\")(FREE_DATE>='{trade_date}')(FREE_DATE<='{end_str}')",
page_size=20,
sort_columns="FREE_DATE", sort_types="1",
)
upcoming = []
for row in upcoming_data:
upcoming.append({
"date": str(row.get("FREE_DATE", ""))[:10],
"type": row.get("LIMITED_STOCK_TYPE", ""),
"shares": row.get("FREE_SHARES_NUM", 0),
"ratio": row.get("FREE_RATIO", 0),
})
return {"history": history, "upcoming": upcoming}
# 用法
data = lockup_expiry("002475", "2026-05-17")
print(f"历史解禁 {len(data['history'])} 批")
for h in data["history"][:5]:
print(f" {h['date']}: {h['type']} 数量={h['shares']}")
if data["upcoming"]:
print(f"未来90天待解禁 {len(data['upcoming'])} 批")
else:
print("未来90天无待解禁")
限售股类型参考：
- 首发原股东限售股份（IPO 后 1-3 年）
- 首发机构配售股份（IPO 战略配售）
- 定向增发机构配售股份（6-18 个月）
- 股权激励限售股份
3.7 行业板块排名（V3.0 改用东财 — 同花顺加了反爬401）
东财行业板块涨跌幅排名，一次调用看全市场行业轮动。
import requests
def industry_comparison(top_n: int = 20) -> dict:
"""
全行业涨跌幅排名（东财行业板块，~100 个行业）。
返回: {top: [...], bottom: [...], total: int}
"""
url = "https://push2.eastmoney.com/api/qt/clist/get"
params = {
"pn": "1", "pz": "100", "po": "1", "np": "1",
"fltt": "2", "invt": "2",
"fs": "m:90+t:2",
"fields": "f2,f3,f4,f12,f13,f14,f104,f105,f128,f136,f140,f141,f207",
}
headers = {"User-Agent": UA}
r = em_get(url, params=params, headers=headers, timeout=15)
d = r.json()
items = d.get("data", {}).get("diff", [])
if not items:
return {"top": [], "bottom": [], "total": 0}
rows = []
for i, item in enumerate(items):
rows.append({
"rank": i + 1,
"name": item.get("f14", ""),
"change_pct": item.get("f3", 0),
"code": item.get("f12", ""),
"up_count": item.get("f104", 0),
"down_count": item.get("f105", 0),
"leader": item.get("f140", ""),
"leader_change": item.get("f136", 0),
})
return {
"top": rows[:top_n],
"bottom": rows[-top_n:],
"total": len(rows),
}
# 用法
data = industry_comparison(20)
print(f"共 {data['total']} 个行业")
print("\nTOP 10 涨幅:")
for r in data["top"][:10]:
print(f" {r['rank']}. {r['name']}: {r['change_pct']}% 涨{r['up_count']}跌{r['down_count']} 领涨{r['leader']}")
print("\nBOTTOM 5 跌幅:")
for r in data["bottom"][-5:]:
print(f" {r['rank']}. {r['name']}: {r['change_pct']}%")
3.8 全市场龙虎榜
每日全市场龙虎榜汇总——当日所有触发龙虎榜的股票 + 上榜原因 + 买卖净额 + 换手率。
from datetime import datetime
def daily_dragon_tiger(trade_date: str = None, min_net_buy: float = None) -> dict:
"""
全市场龙虎榜。
trade_date: YYYY-MM-DD（默认当日）
min_net_buy: 净买入下限（万元），None 不过滤
返回: {date, total_records, stocks: [{code, name, reason, close, change_pct,
net_buy_wan, buy_wan, sell_wan, turnover_pct}]}
"""
if trade_date is None:
trade_date = datetime.now().strftime("%Y-%m-%d")
data = eastmoney_datacenter(
"RPT_DAILYBILLBOARD_DETAILSNEW",
filter_str=f"(TRADE_DATE>='{trade_date}')(TRADE_DATE<='{trade_date}')",
page_size=500,
sort_columns="BILLBOARD_NET_AMT", sort_types="-1",
)
if not data:
return {"date": trade_date, "total_records": 0, "stocks": [],
"note": "无数据（非交易日或盘后未更新）"}
actual_date = str(data[0].get("TRADE_DATE", ""))[:10] if data else trade_date
stocks = []
for row in data:
net_buy = (row.get("BILLBOARD_NET_AMT") or 0) / 10000
if min_net_buy is not None and net_buy < min_net_buy:
continue
stocks.append({
"code": row.get("SECURITY_CODE", ""),
"name": row.get("SECURITY_NAME_ABBR", ""),
"reason": row.get("EXPLANATION", ""),
"close": row.get("CLOSE_PRICE") or 0,
"change_pct": round(float(row.get("CHANGE_RATE") or 0), 2),
"net_buy_wan": round(net_buy, 1),
"buy_wan": round((row.get("BILLBOARD_BUY_AMT") or 0) / 10000, 1),
"sell_wan": round((row.get("BILLBOARD_SELL_AMT") or 0) / 10000, 1),
"turnover_pct": round(float(row.get("TURNOVERRATE") or 0), 2),
})
return {"date": actual_date, "total_records": len(stocks), "stocks": stocks}
# 用法
data = daily_dragon_tiger("2026-05-16")
print(f"{data['date']} 龙虎榜共 {data['total_records']} 条记录")
for s in data["stocks"][:10]:
print(f" {s['code']} {s['name']}: {s['reason']} | 净买{s['net_buy_wan']}万 涨跌{s['change_pct']}%")
# 只看净买入 > 5000 万的
data = daily_dragon_tiger("2026-05-16", min_net_buy=5000)
print(f"\n净买入 > 5000万: {data['total_records']} 条")
3.9 信号层组合用法：题材热度 + 资金验证
# 拉当日强势股 reason
df_hot = ths_hot_reason()
# 词频统计 reason 列里的题材关键词
from collections import Counter
all_tags = []
for r in df_hot["题材归因"].dropna():
tags = [t.strip() for t in str(r).split("+") if t.strip()]
all_tags.extend(tags)
cnt = Counter(all_tags)
print("当日 TOP 10 题材热度:")
for tag, n in cnt.most_common(10):
print(f" {tag}: {n} 只")
# 同时拉北向当日流向，看资金流方向是否对应题材
df_north = hsgt_realtime()
hgt_close = df_north["hgt_yi"].dropna().iloc[-1] if not df_north.empty else 0
sgt_close = df_north["sgt_yi"].dropna().iloc[-1] if not df_north.empty else 0
print(f"\n北向收盘累计: 沪股通 {hgt_close} 亿 / 深股通 {sgt_close} 亿")
# V3.0: 叠加行业对比，看哪些行业资金在流入
comp = industry_comparison(10)
print("\n行业涨幅 TOP 5:")
for r in comp["top"][:5]:
print(f" {r['name']}: {r['change_pct']}% 涨{r['up_count']}跌{r['down_count']}")
Layer 4: 资金面 / 筹码层（V3.0 新增）
4.1 融资融券明细
def margin_trading(code: str, page_size: int = 30) -> list[dict]:
"""
融资融券明细（日级）。
返回: [{date, rzye(融资余额), rzmre(融资买入), rqye(融券余额), ...}]
"""
data = eastmoney_datacenter(
"RPTA_WEB_RZRQ_GGMX",
filter_str=f'(SCODE="{code}")',
page_size=page_size,
sort_columns="DATE", sort_types="-1",
)
rows = []
for row in data:
rows.append({
"date": str(row.get("DATE", ""))[:10],
"rzye": row.get("RZYE", 0), # 融资余额(元)
"rzmre": row.get("RZMRE", 0), # 融资买入额
"rzche": row.get("RZCHE", 0), # 融资偿还额
"rqye": row.get("RQYE", 0), # 融券余额(元)
"rqmcl": row.get("RQMCL", 0), # 融券卖出量
"rqchl": row.get("RQCHL", 0), # 融券偿还量
"rzrqye": row.get("RZRQYE", 0), # 融资融券余额合计
})
return rows
# 用法
data = margin_trading("600519")
for d in data[:5]:
print(f"{d['date']}: 融资余额={d['rzye']/1e8:.2f}亿 融券余额={d['rqye']/1e8:.2f}亿")
4.2 大宗交易
def block_trade(code: str, page_size: int = 20) -> list[dict]:
"""
大宗交易记录。
返回: [{date, price, vol, amount, buyer, seller, premium_pct}]
"""
data = eastmoney_datacenter(
"RPT_DATA_BLOCKTRADE",
filter_str=f'(SECURITY_CODE="{code}")',
page_size=page_size,
sort_columns="TRADE_DATE", sort_types="-1",
)
rows = []
for row in data:
close = row.get("CLOSE_PRICE") or 0
deal_price = row.get("DEAL_PRICE") or 0
premium = ((deal_price / close - 1) * 100) if close else 0
rows.append({
"date": str(row.get("TRADE_DATE", ""))[:10],
"price": deal_price,
"close": close,
"premium_pct": round(premium, 2),
"vol": row.get("DEAL_VOLUME", 0),
"amount": row.get("DEAL_AMT", 0),
"buyer": row.get("BUYER_NAME", ""),
"seller": row.get("SELLER_NAME", ""),
})
return rows
# 用法
data = block_trade("600519")
for d in data[:5]:
print(f"{d['date']}: 价格={d['price']} 溢价={d['premium_pct']}% 买方={d['buyer']}")
4.3 股东户数变化
def holder_num_change(code: str, page_size: int = 10) -> list[dict]:
"""
股东户数变化（季度级）。
返回: [{date, holder_num, change_num, change_ratio, avg_shares}]
"""
data = eastmoney_datacenter(
"RPT_HOLDERNUMLATEST",
filter_str=f'(SECURITY_CODE="{code}")',
page_size=page_size,
sort_columns="END_DATE", sort_types="-1",
)
rows = []
for row in data:
rows.append({
"date": str(row.get("END_DATE", ""))[:10],
"holder_num": row.get("HOLDER_NUM", 0),
"change_num": row.get("HOLDER_NUM_CHANGE", 0),
"change_ratio": row.get("HOLDER_NUM_RATIO", 0), # 环比%
"avg_shares": row.get("AVG_FREE_SHARES", 0), # 户均持股
})
return rows
# 用法
data = holder_num_change("600519")
for d in data[:5]:
print(f"{d['date']}: 股东数={d['holder_num']} 变化={d['change_ratio']}% 户均={d['avg_shares']}")
# 股东户数持续减少 = 筹码集中 = 主力吸筹信号
4.4 分红送转历史
def dividend_history(code: str, page_size: int = 20) -> list[dict]:
"""
分红送转历史。
返回: [{date, bonus_rmb(每股派息), transfer_ratio(转增比例), bonus_ratio(送股比例)}]
"""
data = eastmoney_datacenter(
"RPT_SHAREBONUS_DET",
filter_str=f'(SECURITY_CODE="{code}")',
page_size=page_size,
sort_columns="EX_DIVIDEND_DATE", sort_types="-1",
)
rows = []
for row in data:
rows.append({
"date": str(row.get("EX_DIVIDEND_DATE", ""))[:10],
"bonus_rmb": row.get("PRETAX_BONUS_RMB", 0), # 每股派息(税前)
"transfer_ratio": row.get("TRANSFER_RATIO", 0), # 每10股转增
"bonus_ratio": row.get("BONUS_RATIO", 0), # 每10股送股
"plan": row.get("ASSIGN_PROGRESS", ""), # 进度
})
return rows
# 用法
data = dividend_history("600519")
for d in data[:5]:
print(f"{d['date']}: 每股派息={d['bonus_rmb']}元 转增={d['transfer_ratio']} 送={d['bonus_ratio']}")
4.5 个股资金流（120日，日级）
import requests
def stock_fund_flow_120d(code: str) -> list[dict]:
"""
个股资金流（日级，最近120个交易日）。
返回: [{date, main_net(主力净流入), small_net, mid_net, large_net, super_net}]
单位: 元
"""
market_code = 1 if code.startswith("6") else 0
url = "https://push2his.eastmoney.com/api/qt/stock/fflow/daykline/get"
params = {
"secid": f"{market_code}.{code}",
"fields1": "f1,f2,f3,f7",
"fields2": "f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63,f64,f65",
"lmt": "120",
}
headers = {
"User-Agent": UA,
"Referer": "https://quote.eastmoney.com/",
"Origin": "https://quote.eastmoney.com",
}
try:
r = em_get(url, params=params, headers=headers, timeout=15)
d = r.json()
except Exception as e:
print(f"[WARN] push2 资金流请求失败: {e}")
return []
klines = d.get("data", {}).get("klines", [])
rows = []
for line in klines:
parts = line.split(",")
if len(parts) >= 7:
rows.append({
"date": parts[0],
"main_net": float(parts[1]) if parts[1] != "-" else 0,
"small_net": float(parts[2]) if parts[2] != "-" else 0,
"mid_net": float(parts[3]) if parts[3] != "-" else 0,
"large_net": float(parts[4]) if parts[4] != "-" else 0,
"super_net": float(parts[5]) if parts[5] != "-" else 0,
})
return rows
# 用法
data = stock_fund_flow_120d("600519")
for d in data[-5:]:
print(f"{d['date']}: 主力净流入={d['main_net']/1e4:.0f}万 超大单={d['super_net']/1e4:.0f}万")
# 统计近20日主力净流入
recent_20 = data[-20:]
total_main = sum(d["main_net"] for d in recent_20)
print(f"\n近20日主力累计净流入: {total_main/1e8:.2f}亿")
⚠️ 大陆住宅 IP 间歇封锁（#18）： push2/push2his 系列对部分大陆住宅宽带 IP 有连接级风控，表现为偶发 HTTP 000（连接被拒/超时）或返回空——这不是代码问题（同一代码在其他网络/时段实测正常）。遇到时：① 隔几分钟重试；② 换网络环境（如手机热点）；③ 降低请求频率（调大 EM_MIN_INTERVAL）。日级资金流务实替代：仍可用 mootdx 算量价，或换时段重试。
Layer 5: 新闻层
5.1 东财个股新闻（直连 search-api-web）
import requests
import re
import json
def eastmoney_stock_news(code: str, page_size: int = 20) -> list[dict]:
"""
东财个股新闻（JSONP 接口）。
返回: [{title, content, time, source, url}]
"""
# 构造 JSONP 参数
cb = "jQuery_news"
url = "https://search-api-web.eastmoney.com/search/jsonp"
inner_params = json.dumps({
"uid": "",
"keyword": code,
"type": ["cmsArticleWebOld"],
"client": "web",
"clientType": "web",
"clientVersion": "curr",
"param": {"cmsArticleWebOld": {"searchScope": "default", "sort": "default",
"pageIndex": 1, "pageSize": page_size, "preTag": "", "postTag": ""}},
}, separators=(',', ':'))
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)
# 解析 JSONP
text = r.text
json_str = text[text.index("(") + 1 : text.rindex(")")]
d = json.loads(json_str)
rows = []
# 东财实际返回里 result.cmsArticleWebOld 直接就是文章列表（非 {list:[...]} 嵌套）
articles = d.get("result", {}).get("cmsArticleWebOld", []) or []
for a in articles:
rows.append({
"title": re.sub(r'<[^>]+>', '', a.get("title", "")),
"content": re.sub(r'<[^>]+>', '', a.get("content", ""))[:200],
"time": a.get("date", ""),
"source": a.get("mediaName", ""),
"url": a.get("url", ""),
})
return rows
# 用法
news = eastmoney_stock_news("688017")
for n in news[:5]:
print(f" {n['time']} | {n['source']} | {n['title']}")
⚠️ 间歇性返回空（#18）： 部分大陆住宅 IP 调本接口会只拿到 passportWeb（股民资料）而无 cmsArticleWebOld（文章列表）——这是东财对该 IP 的间歇风控，非代码问题。代码已对空结果安全返回 []；遇到时隔几分钟或换网络重试即可。
5.2 财联社快讯（直连 cls.cn）— ⚠️ 已下线，改用 §5.3
⚠️ 2026-05 已失效（#14）： 财联社网站迁移到 Next.js 架构，旧版公开接口
cls.cn/nodeapi/telegraphList 全面下线（返回 404），新版 API 需签名认证，无法
公开 HTTP 调用。全市场实时快讯请改用 §5.3「东财全球资讯」（7×24 滚动，免费无 key）。
下面代码仅作历史参考，已不可用。
import requests
def cls_telegraph(page_size: int = 50) -> list[dict]:
"""
财联社电报（全市场实时快讯）。
返回: [{title, content, time}]
"""
url = "https://www.cls.cn/nodeapi/telegraphList"
params = {"rn": str(page_size), "page": "1"}
headers = {"User-Agent": UA, "Referer": "https://www.cls.cn/"}
r = requests.get(url, params=params, headers=headers, timeout=10)
d = r.json()
rows = []
for item in d.get("data", {}).get("roll_data", []):
rows.append({
"title": item.get("title", "") or item.get("brief", ""),
"content": item.get("content", "") or item.get("brief", ""),
"time": item.get("ctime", ""),
})
return rows
# 用法
news = cls_telegraph()
for n in news[:10]:
print(f" {n['time']} | {n['title'][:60]}")
5.3 东财全球资讯（7x24）
import requests
import uuid
def eastmoney_global_news(page_size: int = 50) -> list[dict]:
"""
东方财富全球财经资讯（7x24 滚动）。
返回: [{title, summary, time}]
"""
url = "https://np-weblist.eastmoney.com/comm/web/getFastNewsList"
params = {
"client": "web", "biz": "web_724",
"fastColumn": "102", "sortEnd": "",
"pageSize": str(page_size),
"req_trace": str(uuid.uuid4()),
}
headers = {"User-Agent": UA, "Referer": "https://kuaixun.eastmoney.com/"}
r = em_get(url, params=params, headers=headers, timeout=10)
d = r.json()
rows = []
for item in d.get("data", {}).get("fastNewsList", []):
rows.append({
"title": item.get("title", ""),
"summary": item.get("summary", "")[:200],
"time": item.get("showTime", ""),
})
return rows
# 用法
news = eastmoney_global_news()
for n in news[:10]:
print(f" {n['time']} | {n['title']}")
Layer 6: 基础数据层
6.1 mootdx 财务快照（37字段季报数据）
from mootdx.quotes import Quotes
client = tdx_client() # 见 Prerequisites 的 tdx_client() helper（规避 0.11.x BESTIP bug；等价 Quotes.factory(market='std')）
# market: 0=深圳, 1=上海
fin = client.finance(symbol='688017')
# 返回 37 个字段的季报快照:
# liutongguben(流通股本), zongguben(总股本)
# eps(每股收益), bvps(每股净资产), roe(净资产收益率%)
# profit(净利润), income(主营收入)
# meigujingzichan(每股净资产), meigugongjijin(每股公积金)
# meiguweifeipeili(每股未分配利润)
# 等37个季报财务字段
6.2 mootdx F10（公司文本资料）
from mootdx.quotes import Quotes
client = tdx_client() # 见 Prerequisites 的 tdx_client() helper（规避 0.11.x BESTIP bug；等价 Quotes.factory(market='std')）
# 9 大类文本数据:
categories = [
"最新提示", "公司概况", "财务分析",
"股东研究", "股本结构", "资本运作",
"业内点评", "行业分析", "公司大事",
]
for cat in categories:
text = client.F10(symbol='688017', name=cat)
print(f"=== {cat} ===")
print(text[:200] if text else "(空)")
优化提示： "股东研究" 中的【4.股东变化】章节含大量历史十大股东列表，实测 16000+ chars。建议只保留最新一期（-70% token）。
6.3 东财个股基本面（直连 push2 API）
import requests
def eastmoney_stock_info(code: str) -> dict:
"""
东财个股基本面信息。
返回: {code, name, industry, total_shares, float_shares, mcap, float_mcap, list_date}
"""
market_code = 1 if code.startswith("6") else 0
url = "https://push2.eastmoney.com/api/qt/stock/get"
params = {
"fltt": "2", "invt": "2",
"fields": "f57,f58,f84,f85,f127,f116,f117,f189,f43",
"secid": f"{market_code}.{code}",
}
headers = {"User-Agent": UA}
r = em_get(url, params=params, headers=headers, timeout=10)
d = r.json().get("data", {})
return {
"code": d.get("f57", ""),
"name": d.get("f58", ""),
"industry": d.get("f127", ""),
"total_shares": d.get("f84", 0), # 总股本(股)
"float_shares": d.get("f85", 0), # 流通股(股)
"mcap": d.get("f116", 0), # 总市值(元)
"float_mcap": d.get("f117", 0), # 流通市值(元)
"list_date": str(d.get("f189", "")), # 上市日期 YYYYMMDD
"price": d.get("f43", 0),
}
# 用法
info = eastmoney_stock_info("688017")
print(f"{info['name']}({info['code']}): 行业={info['industry']} 总市值={info['mcap']/1e8:.0f}亿 上市={info['list_date']}")
6.4 新浪财报三表（资产负债表/利润表/现金流量表）
import requests
def sina_financial_report(code: str, report_type: str = "lrb", num: int = 8) -> list[dict]:
"""
新浪财报三表。
code: 6位代码
report_type: "fzb"(资产负债表) / "lrb"(利润表) / "llb"(现金流量表)
num: 取最近 N 期（默认 8 期）
返回: 按报告期倒序的记录列表，每期一条 dict：
{"报告期": "2026-03-31", "<科目>": "<值>", "<科目>_同比": <同比>, ...}
（item_value 为新浪原始字符串数值，仅在有同比时附 "_同比" 键）
"""
prefix = "sh" if code.startswith("6") else "sz"
paper_code = f"{prefix}{code}"
url = "https://quotes.sina.cn/cn/api/openapi.php/CompanyFinanceService.getFinanceReport2022"
params = {
"paperCode": paper_code,
"source": report_type,
"type": "0",
"page": "1",
"num": str(num),
}
headers = {"User-Agent": UA}
r = requests.get(url, params=params, headers=headers, timeout=15)
# 新浪实际结构: result.data.report_list 是「按报告期(如 '20260331')为键」的 dict,
# 每期对象的 data 字段才是行项列表 [{item_title, item_value, item_tongbi}]。
report_list = r.json().get("result", {}).get("data", {}).get("report_list", {}) or {}
rows = []
for period in sorted(report_list.keys(), reverse=True)[:num]:
obj = report_list[period]
rec = {"报告期": f"{period[:4]}-{period[4:6]}-{period[6:8]}"}
for it in obj.get("data", []) or []:
title = it.get("item_title", "")
if not title or it.get("item_value") is None:
continue
rec[title] = it.get("item_value")
tongbi = it.get("item_tongbi")
if tongbi not in (None, ""):
rec[title + "_同比"] = tongbi
rows.append(rec)
return rows
# 用法: 利润表
lrb = sina_financial_report("600519", "lrb")
for item in lrb[:3]:
print(f"报告期: {item.get('报告期', '')} 净利润: {item.get('净利润', '')}")
# 用法: 资产负债表
fzb = sina_financial_report("600519", "fzb")
# 用法: 现金流量表
llb = sina_financial_report("600519", "llb")
Layer 7: 公告层
7.1 巨潮公告（直连 cninfo.com.cn）
import requests
from datetime import datetime
def _cninfo_ts_to_date(ts):
"""巨潮 announcementTime 返回 Unix 毫秒整数，需转换为日期字符串。"""
if isinstance(ts, (int, float)):
return datetime.fromtimestamp(ts / 1000).strftime("%Y-%m-%d")
return str(ts)[:10] if ts else ""
# 巨潮 股票→orgId 映射（模块级缓存，首次调用时拉取一次，全程复用）
_CNINFO_ORGID_MAP = {}
def _cninfo_orgid(code: str) -> str:
"""查股票真实 orgId。巨潮 orgId 并非统一 `gssx0{code}` 格式（如 601318→9900002221、
601398→jjxt0000019、688017→9900041602），硬编码会导致大量股票（尤其 601xxx 段）
返回 totalAnnouncement=0、查不到公告（#19）。优先动态查官方映射表，查不到再回退硬编码。"""
global _CNINFO_ORGID_MAP
if not _CNINFO_ORGID_MAP:
try:
r = requests.get("http://www.cninfo.com.cn/new/data/szse_stock.json",
headers={"User-Agent": UA}, timeout=15)
_CNINFO_ORGID_MAP = {s["code"]: s["orgId"]
for s in r.json().get("stockList", [])}
except Exception as e:
print(f"[WARN] 巨潮 orgId 映射表拉取失败，回退硬编码规则: {e}")
org = _CNINFO_ORGID_MAP.get(code)
if org:
return org
# fallback：老格式（仅部分老股票如 600519/600036 适用）
if code.startswith("6"):
return f"gssh0{code}"
elif code.startswith("8") or code.startswith("4"):
return f"gsbj0{code}"
return f"gssz0{code}"
def cninfo_announcements(code: str, page_size: int = 30) -> list[dict]:
"""
巨潮公告全文检索。
返回: [{title, type, date, url}]
"""
url = "https://www.cninfo.com.cn/new/hisAnnouncement/query"
org_id = _cninfo_orgid(code) # 动态查真实 orgId（#19 修复，自带硬编码 fallback）
payload = {
"stock": f"{code},{org_id}",
"tabName": "fulltext",
"pageSize": str(page_size),
"pageNum": "1",
"column": "",
"category": "",
"plate": "",
"seDate": "",
"searchkey": "",
"secid": "",
"sortName": "",
"sortType": "",
"isHLtitle": "true",
}
headers = {
"User-Agent": UA,
"Content-Type": "application/x-www-form-urlencoded",
"Referer": "https://www.cninfo.com.cn/new/disclosure",
"Origin": "https://www.cninfo.com.cn",
}
r = requests.post(url, data=payload, headers=headers, timeout=15)
d = r.json()
rows = []
for item in d.get("announcements", []) or []:
rows.append({
"title": item.get("announcementTitle", ""),
"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', '')}",
})
return rows
# 用法
anns = cninfo_announcements("688017")
for a in anns[:10]:
print(f" {a['date']} | {a['type']} | {a['title']}")
7.2 mootdx F10 公告摘要
from mootdx.quotes import Quotes
client = tdx_client() # 见 Prerequisites 的 tdx_client() helper（规避 0.11.x BESTIP bug；等价 Quotes.factory(market='std')）
text = client.F10(symbol='688017', name='最新提示')
# 包含最近的公告/分红/股东大会决议等摘要
Layer 8: 打板层（涨停 / 炸板 / 跌停 / 题材情绪，V3.3.0 新增）
连板梯队、炸板率、晋级率、涨停原因题材——打板与题材跟踪的高频需求（#23 / #15）。东财四池走 push2ex.eastmoney.com（与现有 push2 同源，已纳入 em_get() 限流）；涨停原因题材增强用同花顺。全部免登录、零鉴权。
8.1 东财涨停板池 — 涨停 / 炸板 / 跌停 / 昨日涨停
import requests
ZTB_UT = "7eea3edcaed734bea9cbfc24409ed989"
def _fmt_zt_time(t) -> str:
"""涨停板时间整数 → HH:MM:SS（92500 → 09:25:00）。"""
s = str(t).zfill(6)
return f"{s[0:2]}:{s[2:4]}:{s[4:6]}"
def _em_zt_api(endpoint: str, sort: str, date: str) -> list[dict]:
"""东财涨停板行情中心通用请求（push2ex，走 em_get 限流）。
endpoint: getTopicZTPool / getTopicZBPool / getTopicDTPool / getYesterdayZTPool
返回 data.pool 原始列表（data 为 null = 非交易日 / 参数错）。"""
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/"}
try:
r = em_get(url, params=params, headers=headers, timeout=10)
return (r.json().get("data") or {}).get("pool") or []
except Exception as e:
print(f"[WARN] 涨停板池 {endpoint} 请求失败: {e}")
return []
def em_zt_pool(date: str) -> list[dict]:
"""涨停池。date=YYYYMMDD（交易日）。
返回每只: code/name/price/pct/amount/float_cap/turnover/limit_days(连板数)/
first_seal/last_seal(封板时间)/seal_fund(封板资金,元)/break_times(炸板次数)/
industry/zt_stat(N天M板)"""
out = []
for p in _em_zt_api("getTopicZTPool", "fbt:asc", date):
out.append({"code": p["c"], "name": p["n"], "price": p["p"] / 1000,
"pct": round(p["zdp"], 2), "amount": p["amount"], "float_cap": p["ltsz"],
"turnover": round(p["hs"], 2), "limit_days": p["lbc"],
"first_seal": _fmt_zt_time(p["fbt"]), "last_seal": _fmt_zt_time(p["lbt"]),
"seal_fund": p["fund"], "break_times": p["zbc"], "industry": p.get("hybk", ""),
"zt_stat": f'{(p.get("zttj") or {}).get("days","?")}天{(p.get("zttj") or {}).get("ct","?")}板'})
return out
def em_zb_pool(date: str) -> list[dict]:
"""炸板池（涨停后开板）。返回 code/name/price/limit_price(涨停价)/pct/turnover/
first_seal/break_times/amplitude(振幅)/speed(涨速)/industry/zt_stat"""
out = []
for p in _em_zt_api("getTopicZBPool", "fbt:asc", date):
out.append({"code": p["c"], "name": p["n"], "price": p["p"] / 1000,
"limit_price": p["ztp"] / 1000, "pct": round(p["zdp"], 2),
"turnover": round(p["hs"], 2), "first_seal": _fmt_zt_time(p["fbt"]),
"break_times": p["zbc"], "amplitude": round(p["zf"], 2),
"speed": round(p["zs"], 2), "industry": p.get("hybk", ""),
"zt_stat": f'{(p.get("zttj") or {}).get("days","?")}天{(p.get("zttj") or {}).get("ct","?")}板'})
return out
def em_dt_pool(date: str) -> list[dict]:
"""跌停池。返回 code/name/price/pct/turnover/pe/seal_fund(封单资金)/last_seal/
board_amount(板上成交额)/dt_days(连续跌停)/open_times(开板次数)/industry"""
out = []
for p in _em_zt_api("getTopicDTPool", "fund:asc", date):
out.append({"code": p["c"], "name": p["n"], "price": p["p"] / 1000,
"pct": round(p["zdp"], 2), "turnover": round(p["hs"], 2), "pe": p.get("pe"),
"seal_fund": p["fund"], "last_seal": _fmt_zt_time(p["lbt"]),
"board_amount": p.get("fba"), "dt_days": p.get("days"),
"open_times": p.get("oc"), "industry": p.get("hybk", "")})
return out
def em_yzt_pool(date: str) -> list[dict]:
"""昨日涨停池（昨涨停今表现，算晋级率/赚钱效应）。返回 code/name/price/
pct(今日涨幅)/turnover/amplitude/speed/y_first_seal(昨封板时间)/
y_limit_days(昨连板)/industry/zt_stat"""
out = []
for p in _em_zt_api("getYesterdayZTPool", "zs:desc", date):
out.append({"code": p["c"], "name": p["n"], "price": p["p"] / 1000,
"pct": round(p["zdp"], 2), "turnover": round(p["hs"], 2),
"amplitude": round(p["zf"], 2), "speed": round(p["zs"], 2),
"y_first_seal": _fmt_zt_time(p["yfbt"]), "y_limit_days": p["ylbc"],
"industry": p.get("hybk", ""), "zt_stat": f'{(p.get("zttj") or {}).get("days","?")}天{(p.get("zttj") or {}).get("ct","?")}板'})
return out
# 用法
zt = em_zt_pool("20260626")
print(f"今日涨停 {len(zt)} 只")
for s in zt[:3]:
print(f" {s['name']} {s['zt_stat']} 封板{s['seal_fund']/1e8:.2f}亿 {s['industry']}")
坑： ① 价格字段 price/limit_price 已 ÷1000（原始值是 ×1000 整数）。② 四池只有 sort 不同（涨停/炸板=fbt:asc、跌停=fund:asc、昨涨停=zs:desc），dpt 都是 wz.ztzt。③ date 必须传交易日，非交易日 data 返回 null。④ 金额单位均为元。
8.2 同花顺涨停揭秘 — 涨停原因题材 + 封板成功率 + 板型
from datetime import datetime
def ths_limit_up_pool(date: str) -> list[dict]:
"""同花顺涨停揭秘（涨停原因 + 封板质量增强源）。date=YYYYMMDD。
返回每只: code/name/price/pct/reason(涨停原因题材)/board_type(换手板/一字板/T字板)/
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"
params = {"page": 1, "limit": 200,
"field": "199112,10,9001,330323,330324,330325,9002,330329,133971,133970,1968584,3475914,9003,9004",
"filter": "HS,GEM2STAR", "order_field": "330324", "order_type": "0", "date": date}
try:
r = requests.get(url, params=params, headers={"User-Agent": UA}, timeout=10)
info = (r.json().get("data") or {}).get("info", [])
except Exception as e:
print(f"[WARN] 同花顺涨停揭秘请求失败: {e}")
return []
out = []
for it in info:
ft = it.get("first_limit_up_time")
out.append({"code": it.get("code"), "name": it.get("name"),
"price": it.get("latest"), "pct": it.get("change_rate"),
"reason": it.get("reason_type", ""), "board_type": it.get("limit_up_type", ""),
"seal_rate": it.get("limit_up_suc_rate"), "break_times": it.get("open_num") or 0,
"seal_amount": it.get("order_amount"), "high_days": it.get("high_days", ""),
"first_time": datetime.fromtimestamp(int(ft)).strftime("%H:%M:%S") if ft else "",
"is_again": it.get("is_again_limit")})
return out
# 用法: 涨停原因题材归因
for s in ths_limit_up_pool("20260626")[:5]:
print(f" {s['name']} {s['high_days']} | {s['reason']} | 封板率{s['seal_rate']}")
坑： first_limit_up_time 是 Unix 秒时间戳（要 datetime.fromtimestamp），不是 HHMMSS。field 那串是同花顺内部字段 ID，照抄即可。filter=HS,GEM2STAR 控制板块范围（沪深主板 + 创业板 + 科创板）。
8.3 打板情绪速算 — 炸板率 / 连板高度 / 连板梯队
def limit_up_sentiment(date: str) -> dict:
"""打板情绪温度计：连板梯队 + 炸板率 + 涨跌停对比。"""
zt, zb, dt = em_zt_pool(date), em_zb_pool(date), em_dt_pool(date)
ladder = {}
for s in zt:
ladder[s["limit_days"]] = ladder.get(s["limit_days"], 0) + 1
zt_n, zb_n = len(zt), len(zb)
return {"date": date, "zt_count": zt_n, "zb_count": zb_n, "dt_count": len(dt),
"break_rate": round(zb_n / (zt_n + zb_n) * 100, 1) if (zt_n + zb_n) else 0, # 炸板率%
"max_height": max((s["limit_days"] for s in zt), default=0), # 最高连板
"ladder": dict(sorted(ladder.items()))} # 连板梯队 {板数:家数}
# 用法
s = limit_up_sentiment("20260626")
print(f"涨停{s['zt_count']} 炸板{s['zb_count']}(炸板率{s['break_rate']}%) "
f"跌停{s['dt_count']} 最高{s['max_height']}连板")
print(f"连板梯队: {s['ladder']}")
晋级率（昨涨停今仍涨停 / 昨涨停总数）可用 em_yzt_pool() 的 pct >= 9.8 计数除以总数自算。
Layer 9: ETF 期权层（T型报价 + 希腊字母 + IV，V3.3.0 新增）
50ETF / 300ETF / 科创50ETF / 500ETF 期权（#13）。走新浪源——T型报价、希腊字母、隐含波动率均由交易所/新浪预先算好，无需本地算 BSM。免费直连，唯一注意带 Referer。
9.1 合约清单 + T型报价 + 希腊字母
import requests
SINA_OPT_HDR = {"Referer": "https://stock.finance.sina.com.cn/", "User-Agent": UA}
def _opt_f(x):
try: return float(x)
except Exception: return x
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)
r.encoding = "gbk"
t = r.text
return t.split('"')[1].split(",") if '"' in t else []
def sina_option_codes(underlying: str = "510050", call: bool = True) -> dict:
"""ETF期权合约清单。underlying: 510050/510300/588000/510500。call=True认购/False认沽。
返回 {月份YYMM: [合约代码,...]}，第一个 key 即近月。"""
cate = {"510050": "50ETF", "510300": "300ETF",
"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}")
try:
months = requests.get(url, headers=SINA_OPT_HDR, timeout=10).json()["result"]["data"]["contractMonth"]
except Exception as e:
print(f"[WARN] 期权月份获取失败: {e}")
return {}
months = [m.replace("-", "")[2:] for m in months[1:]] # 丢首个，转 YYMM
flag = "OP_UP_" if call else "OP_DOWN_"
out = {}
for m in months:
codes = [c.replace("CON_OP_", "") for c in _sina_opt_list(f"{flag}{underlying}{m}")
if c.startswith("CON_OP_")]
if codes:
out[m] = codes
return out
def sina_option_tquote(code: str) -> dict:
"""期权T型报价。返回 bid_vol/bid/last/ask/ask_vol/open_interest(持仓量)/pct/
strike(行权价)/prev_close/open/limit_up/limit_down/name/amplitude/high/low/volume/amount。"""
v = _sina_opt_list(f"CON_OP_{code}")
if len(v) < 43:
return {}
return {"bid_vol": _opt_f(v[0]), "bid": _opt_f(v[1]), "last": _opt_f(v[2]),
"ask": _opt_f(v[3]), "ask_vol": _opt_f(v[4]), "open_interest": _opt_f(v[5]),
"pct": _opt_f(v[6]), "strike": _opt_f(v[7]), "prev_close": _opt_f(v[8]),
"open": _opt_f(v[9]), "limit_up": _opt_f(v[10]), "limit_down": _opt_f(v[11]),
"name": v[37], "amplitude": _opt_f(v[38]), "high": _opt_f(v[39]),
"low": _opt_f(v[40]), "volume": _opt_f(v[41]), "amount": _opt_f(v[42])}
def sina_option_greeks(code: str) -> dict:
"""期权希腊字母 + 隐含波动率。返回 name/volume/delta/gamma/theta/vega/
iv(隐含波动率,小数)/high/low/trade_code/strike/last/theory(理论价值)。"""
raw = _sina_opt_list(f"CON_SO_{code}")
if len(raw) < 16:
return {}
v = [raw[0]] + raw[4:] # ⚠️ raw[1:4] 是 3 个空串，必须跳过否则字段错位
return {"name": v[0], "volume": _opt_f(v[1]), "delta": _opt_f(v[2]),
"gamma": _opt_f(v[3]), "theta": _opt_f(v[4]), "vega": _opt_f(v[5]),
"iv": _opt_f(v[6]), "high": _opt_f(v[7]), "low": _opt_f(v[8]),
"trade_code": v[9], "strike": _opt_f(v[10]), "last": _opt_f(v[11]), "theory": _opt_f(v[12])}
# 用法: 取 50ETF 近月平值附近一档的 T型报价 + 希腊字母
codes = sina_option_codes("510050", call=True)
near = list(codes)[0] # 近月
c = codes[near][len(codes[near]) // 2] # 中间档≈平值附近
q, g = sina_option_tquote(c), sina_option_greeks(c)
print(f"{q['name']} 行权价{q['strike']} 最新{q['last']} 持仓{q['open_interest']:.0f}")
print(f" Delta={g['delta']} Gamma={g['gamma']} Theta={g['theta']} Vega={g['vega']} IV={g['iv']:.2%}")
坑： ① 新浪源 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)、科创50ETF(588000) 同理，换 underlying 即可。
Layer 10: 舆情互动层（互动易问答 + 热榜 + 人气榜，V3.3.0 新增）
投资者互动问答 + 市场热度——AI 问答与选题的独家信源。互动易（巨潮）能答"公司怎么回应某传闻/利好"，别处拿不到；同花顺热榜 / 东财人气榜给"当下最热个股 + 被归到什么概念在炒"。全部免登录、零鉴权。
10.1 互动易问答（巨潮 — 投资者提问 + 公司回复）
import requests
from datetime import datetime
def cninfo_irm(code: str, page_size: int = 30, page_num: int = 1) -> list[dict]:
"""互动易问答（深沪统一走巨潮）。code: 6位代码。
返回每条: code/company/question(投资者提问)/answer(公司回复,None=未回复)/
answerer(回答方)/ask_time。"""
try:
r1 = requests.post("https://irm.cninfo.com.cn/newircs/index/queryKeyboardInfo",
data={"keyWord": code}, headers={"User-Agent": UA}, timeout=10)
d1 = r1.json().get("data") or []
if not d1:
return []
org_id = d1[0].get("secid")
# ⚠️ 第二步参数必须放 query string（POST 但 body 空），否则 HTTP 400
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",
params=params, headers={"User-Agent": UA}, timeout=10)
rows = r2.json().get("rows") or []
except Exception as e:
print(f"[WARN] 互动易请求失败: {e}")
return []
out = []
for it in rows:
pd = it.get("pubDate")
out.append({"code": it.get("stockCode"), "company": it.get("companyShortName"),
"question": it.get("mainContent"), "answer": it.get("attachedContent"),
"answerer": it.get("attachedAuthor"),
"ask_time": datetime.fromtimestamp(pd / 1000).strftime("%Y-%m-%d %H:%M") if pd else ""})
return out
# 用法: 看公司怎么回应投资者关切
for q in cninfo_irm("002594", page_size=30):
if q["answer"]:
print(f" Q: {q['question'][:30]}\n A[{q['answerer']}]: {q['answer'][:50]}")
坑： ① 第二步参数放 query string（不是 body），否则 400。② orgId 取自第一步的 secid（即便前缀是 gshk，靠 stockcode 过滤照样拿 A 股问答）。③ 最新提问常未回复（answer=None），回复率因公司而异（实测立讯精密 002475 回复多、京东方 000725 几乎不回）。④ 时间是毫秒时间戳。
10.2 同花顺热榜 + 东财人气榜（市场热度 + 概念命中）
EM_HOT_BODY = {"appId": "appId01", "globalId": "786e4c21-70dc-435a-93bb-38"}
def ths_hot_list(period: str = "hour") -> list[dict]:
"""同花顺热榜（单接口拿名称+人气+概念标签+排名变化）。period: hour/day。
返回每只: 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",
params={"stock_type": "a", "type": period, "list_type": "normal"},
headers={"User-Agent": UA}, timeout=10)
lst = (r.json().get("data") or {}).get("stock_list") or []
except Exception as e:
print(f"[WARN] 同花顺热榜失败: {e}")
return []
out = []
for it in lst:
tag = it.get("tag") or {}
out.append({"rank": it.get("order"), "code": it.get("code"), "name": it.get("name"),
"heat": it.get("rate"), "pct": it.get("rise_and_fall"), "rank_chg": it.get("hot_rank_chg"),
"concepts": tag.get("concept_tag") or [], "tag": tag.get("popularity_tag", "")})
return out
def em_hot_rank(top: int = 50) -> list[dict]:
"""东财人气榜（排名 + 排名变化 + 名称/价格）。返回 rank/code/name/price/pct/rank_chg。"""
try:
r = requests.post("https://emappdata.eastmoney.com/stockrank/getAllCurrentList",
json={**EM_HOT_BODY, "marketType": "", "pageNo": 1, "pageSize": top},
headers={"User-Agent": UA}, timeout=10)
data = r.json().get("data") or []
if not data:
return []
# 人气榜只给带前缀代码，用 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",
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=10)
diff = (u.json().get("data") or {}).get("diff") or []
if isinstance(diff, dict): # push2 的 diff 有时是 dict
diff = list(diff.values())
nm = {x["f12"]: (x.get("f14"), x.get("f2"), x.get("f3")) for x in diff}
except Exception as e:
print(f"[WARN] 东财人气榜失败: {e}")
return []
out = []
for it in data:
code = it["sc"][2:]
name, price, pct = nm.get(code, ("", None, None))
out.append({"rank": it["rk"], "code": code, "name": name,
"price": price, "pct": pct, "rank_chg": it.get("hisRc")})
return out
def em_hot_concept(code: str) -> list[dict]:
"""东财个股热门概念命中（这只票当下被市场归到哪些概念在炒）。
返回 [{concept, bk, hit(命中热度)}, ...]，按热度降序。"""
try:
prefix = "SH" if code.startswith("6") else "SZ"
r = requests.post("https://emappdata.eastmoney.com/stockrank/getHotStockRankList",
json={**EM_HOT_BODY, "srcSecurityCode": prefix + code},
headers={"User-Agent": UA}, timeout=10)
data = r.json().get("data") or []
except Exception as e:
print(f"[WARN] 东财个股概念失败: {e}")
return []
return [{"concept": x.get("conceptName"), "bk": x.get("conceptId"),
"hit": x.get("hitCount")} for x in data]
# 用法
for s in ths_hot_list()[:5]:
print(f" #{s['rank']} {s['name']} 热度{s['heat']} {s['concepts']} {s['tag']}")
hot = em_hot_rank(10) # 东财人气榜 TOP10
print("人气第一:", hot[0]["name"], "概念命中:", em_hot_concept(hot[0]["code"])[:3])
坑： ① 东财人气榜 getAllCurrentList 只返回带前缀代码（SZ/SH），名称要再走 ulist.np 补（SZ→0.、SH→1.）。② ulist.np 的 diff 偶尔是 dict（按序号为键），已做 list(values()) 归一化。③ 同花顺热榜 type 可选 hour/day。
估值计算公式
前向PE
