--- name: freestocklineskill description: Free no-API-key A-share ecosystem data query skill for OpenClaw and Hermes. Use when users ask for A股、沪深京股票、指数、ETF/LOF、可转债、行业/概念板块、实时行情、K线、涨跌排行、涨停/跌停/炸板、资金流、财务估值、公告、龙虎榜、大宗交易、融资融券、新闻快讯、筹码等公开数据查询，并且不希望填写 API Key、Token、Cookie 或付费账号。 version: "0.1.1" homepage: https://github.com/Etherstrings/freeStockLIneskill#donate metadata: openclaw: requires: bins: ["python3"] --- # freeStockLIneskill ## Payment / Donation Notice This skill is free to install on ClawHub, but it is donation-supported. If freeStockLIneskill helps you save time, please support ongoing use and maintenance here: - Donate / Sponsor: - Afdian: - ClawHub page: 这是给 OpenClaw / Hermes 用的 A 股免费公开源数据查询 skill。它的设计目标很明确：不需要用户输入任何 apikey/API Key、Token、Cookie 或付费账号；在完全免费信源下尽量覆盖较大范围的 A 股生态数据，包括股票、指数、ETF/LOF、可转债、行业/概念板块、行情/K 线、排行、涨停/跌停/炸板、资金流、财务估值、公告、龙虎榜等。 ## 先执行这个只要用户是自然语言提问，Agent 不要猜接口，直接把用户原话交给 `smart-query`： ```bash python3 freestocklineskill/scripts/stockline_cli.py smart-query --query "贵州茅台最新价" ``` 脚本会自动完成： - 判断意图：行情、K 线、大盘、榜单、涨停、资金流、板块、财务、公告、龙虎榜、可转债等。 - 解析标的：股票名、6 位代码、`600519.SH`、`sh600519`、指数、ETF、可转债、行业/概念。 - 调用免费源：腾讯财经、新浪财经、东方财富、巨潮资讯、AKShare、efinance。 - 输出统一 JSON：成功和失败都包含 `ok`、`intent`、`normalized`、`source_chain`、`data`、`warnings`、`meta`。不要要求用户提供 API Key、Token、Cookie、iFinD、Wind、Choice、Tushare Pro 或任何付费账号。 ## 常用命令 ```bash python3 freestocklineskill/scripts/stockline_cli.py endpoint-list python3 freestocklineskill/scripts/stockline_cli.py search-entity --query "宁德时代" python3 freestocklineskill/scripts/stockline_cli.py quote-realtime --symbol 600519 python3 freestocklineskill/scripts/stockline_cli.py quote-history --symbol 300750 --days 30 --period daily --adjust qfq python3 freestocklineskill/scripts/stockline_cli.py market-snapshot python3 freestocklineskill/scripts/stockline_cli.py rank --kind amount --limit 10 python3 freestocklineskill/scripts/stockline_cli.py limit-pool --kind up --limit 30 python3 freestocklineskill/scripts/stockline_cli.py money-flow --scope market --period instant --limit 20 python3 freestocklineskill/scripts/stockline_cli.py sector --kind industry --action rank python3 freestocklineskill/scripts/stockline_cli.py fundamental --symbol 600519 --pack all python3 freestocklineskill/scripts/stockline_cli.py announcement --symbol 600519 --keyword 年报 python3 freestocklineskill/scripts/stockline_cli.py dragon-tiger --date 2026-04-24 python3 freestocklineskill/scripts/stockline_cli.py news --symbol 300750 --kind news --limit 10 python3 freestocklineskill/scripts/stockline_cli.py news --symbol 300750 --kind research --limit 10 python3 freestocklineskill/scripts/stockline_cli.py chip --symbol 600519 --limit 100 python3 freestocklineskill/scripts/stockline_cli.py block-trade --date 2026-04-24 --limit 50 python3 freestocklineskill/scripts/stockline_cli.py margin-trading --date 2026-04-24 --limit 100 python3 freestocklineskill/scripts/stockline_cli.py bond --action rank --limit 20 ``` ## Agent 规则 1. 高频自然语言先用 `smart-query`。 2. 用户明确说查某类数据时，才使用显式命令。 3. 读 [references/capability-matrix.md](references/capability-matrix.md) 判断稳定支持、best-effort、还是不要假装能做。 4. 读 [references/natural-language-routing.md](references/natural-language-routing.md) 了解关键词如何路由。 5. 读 [references/use-cases.md](references/use-cases.md) 找类似用户问法。 6. 输出给用户时必须带来源、交易日/时间戳、必要 warning。 7. `ok: false` 时不要编造数据，复述 `error.message` 和可尝试的下一步。 ## 能力边界稳定支持： - 个股、指数、ETF、可转债实时行情 - 日/周/月/分钟 K 线 - 大盘指数与市场宽度 - A 股涨跌幅、成交额、成交量、换手率、量比、振幅、市值、PE/PB 排行 - 涨停池、跌停池、炸板池、强势股池 - 全市场主力资金流排行 - 行业/概念板块排行、成分股、个股所属板块 - 基本信息、估值、公开财务摘要、公告 PDF、龙虎榜、可转债排行 - 新闻快讯、公开研报/评级、筹码分布、大宗交易、融资融券 best-effort Best-effort： - 个股资金流、行业/概念资金流 - 筹码分布 - 股东、分红、研报、新闻快讯、大宗交易、融资融券 - 公开源字段会随网站变化，返回中必须保留 `warnings` 不要假装能做： - 实时交易所 Level-2 逐笔、盘口队列 - Wind/Choice/iFinD 专属字段 - 需要登录、Cookie、Token 或付费授权的数据 - 投资建议、买卖点承诺、收益保证 ## Support - Support ongoing development: - OpenClaw / ClawHub skill page: ### Donate Alipay: ![Alipay QR](https://raw.githubusercontent.com/Etherstrings/freeStockLIneskill/main/docs/assets/donate/alipay_clawhub.jpg) WeChat Pay: ![WeChat Pay QR](https://raw.githubusercontent.com/Etherstrings/freeStockLIneskill/main/docs/assets/donate/wechat_clawhub.jpg) FILE:agents/openai.yaml interface: display_name: "freeStockLIneskill" short_description: "无 API Key 的 A 股生态免费数据查询" default_prompt: "Use $freestocklineskill for free no-key A-share ecosystem data queries. For natural-language requests, first call `python3 freestocklineskill/scripts/stockline_cli.py smart-query --query \"<用户原话>\"`, then answer from the returned JSON with source and trade-date warnings." policy: allow_implicit_invocation: true FILE:references/capability-matrix.md # 能力矩阵给 Agent 用，不是营销文档。先判断能力，再调用命令。 ## 稳定支持 | 能力 | 入口 | 免费源 | |---|---|---| | 股票/指数/ETF/可转债实时行情 | `smart-query` / `quote-realtime` | 腾讯财经、AKShare | | 日/周/月/分钟 K 线 | `smart-query` / `quote-history` | 腾讯财经、efinance | | 大盘与市场宽度 | `smart-query` / `market-snapshot` | 腾讯财经、东方财富 | | 涨跌幅/成交额/成交量/换手率/量比/振幅/市值/PE/PB 榜 | `smart-query` / `rank` | 东方财富、新浪 | | 涨停池/跌停池/炸板池/强势股池 | `smart-query` / `limit-pool` | 东方财富 | | 全市场主力资金净流入排行 | `smart-query` / `money-flow` | 新浪、AKShare | | 行业/概念板块排行 | `smart-query` / `sector` | AKShare | | 板块成分股 | `sector --action constituents` | AKShare | | 个股所属板块 | `sector --action belong` | efinance | | 公司基础信息/估值 | `smart-query` / `fundamental` | 腾讯、AKShare | | 公告列表/PDF 链接 | `smart-query` / `announcement` | 巨潮资讯 | | 龙虎榜 | `smart-query` / `dragon-tiger` | AKShare | | 可转债排行/报价/K 线 | `smart-query` / `bond` | AKShare | | 新闻快讯/公开研报/评级 | `smart-query` / `news` | AKShare/东方财富公开源 | | 大宗交易 | `smart-query` / `block-trade` | AKShare/东方财富公开源 | | 融资融券 | `smart-query` / `margin-trading` | AKShare/交易所公开源 | ## Best-effort | 能力 | 入口 | 说明 | |---|---|---| | 个股资金流 | `money-flow --scope stock` | 公开源字段和接口稳定性随网站变化 | | 行业/概念资金流 | `money-flow --scope industry/concept` | 返回时必须保留 warning | | 筹码分布 | `smart-query` / `chip` | 免费源不稳定，必须保留 warning | | 股东/分红/研报/新闻快讯 | `fundamental` / `news` | 按公开源可得字段返回 | ## 不要假装能做 - 需要登录、Cookie、Token、API Key、付费账号的数据。 - Wind、Choice、iFinD、Tushare Pro 专属字段。 - 实时逐笔成交、Level-2 盘口队列。 - 自动生成买卖建议、目标价承诺、收益保证。 - 全市场复杂组合条件选股如果现有 `rank`、`sector`、`fundamental` 无法表达，就返回当前未覆盖。 ## 回答口径 - 稳定能力：直接调用并说明来源。 - Best-effort：调用，返回时说明公开源波动和 warning。 - 不支持：不要编 payload，不要编数据，直接说明当前 skill 不覆盖。 FILE:references/free-sources.md # 免费信源本 skill 不要求用户配置任何 API Key、Token、Cookie 或付费账号。 ## 高频直连源 - 腾讯财经公开接口：实时行情、指数行情、K 线。 - 新浪财经公开接口：A 股排行、主力资金榜、名称 suggest。 - 东方财富公开接口：排行、涨停/跌停/炸板/强势股池、市场宽度。 - 巨潮资讯公开接口：公告列表、PDF 附件链接。 ## 免费 Python 库 - AKShare：财务、资金流、龙虎榜、板块、可转债等广覆盖能力。 - efinance：东方财富封装，适合行情、K 线、ETF、所属板块等兜底。 - pandas：把 DataFrame 统一转成 JSON records。 ## 数据使用注意 - 免费源可能延迟、限流、反爬或字段调整。 - 返回 JSON 中 `source_chain` 表示尝试过哪些源。 - 返回 JSON 中 `warnings` 表示 best-effort 或回退情况。 - 公开源返回的日期可能是最近交易日，不一定等于自然日今天。 FILE:references/natural-language-routing.md # 自然语言路由默认入口： ```bash python3 freestocklineskill/scripts/stockline_cli.py smart-query --query "<用户原话>" ``` ## 路由优先级 1. `可转债`、`转债` -> `bond` 2. `新闻`、`快讯`、`研报`、`评级`、`目标价` -> `news` 3. `筹码`、`筹码分布` -> `chip` 4. `大宗交易`、`大宗成交` -> `block-trade` 5. `融资融券`、`两融`、`融资余额`、`融券` -> `margin-trading` 6. `公告`、`年报`、`季报`、`业绩预告`、`PDF`、`披露` -> `announcement` 7. `龙虎榜` -> `dragon-tiger` 8. `资金流`、`主力资金`、`净流入`、`净流出` -> `money-flow` 9. `涨停`、`跌停`、`炸板`、`连板`、`封板`、`强势股` -> `limit-pool` 10. `排行`、`排名`、`榜`、`top`、`前十`、`最高`、`最低` -> `rank` 11. `板块`、`行业`、`概念` -> `sector` 12. `财务`、`基本面`、`估值`、`ROE`、`毛利率`、`市盈率`、`市净率`、`股东`、`分红` -> `fundamental` 13. `走势`、`历史`、`K线`、`日线`、`周线`、`月线`、`分钟`、`最近` -> `quote-history` 14. `大盘`、`三大指数`、`市场整体`、`涨跌家数`、多指数同问 -> `market-snapshot` 15. 默认 -> `quote-realtime` ## 常见参数解析 - `前十`、`前20`、`top 50` -> `limit` - `近30天`、`最近一个月`、`近半年` -> `days` - `2026-04-24`、`20260424`、`4月24日`、`今天`、`昨天` -> `date` - `前复权` -> `qfq` - `后复权` -> `hfq` - `不复权` -> `none` - `行业` -> `sector --kind industry` - `概念` -> `sector --kind concept` - `属于什么板块` -> `sector --action belong` - `板块成分股` -> `sector --action constituents` ## 标的解析优先顺序： 1. 指数别名：上证指数、深成指、创业板指、沪深300、科创50、北证50。 2. 代码：`600519`、`600519.SH`、`sh600519`。 3. 常见中文名：贵州茅台、宁德时代等内置别名。 4. 腾讯 smartbox / 新浪 suggest 免费接口。无法稳定识别时返回 `ok: false` 和候选提示，不让 Agent 猜。 FILE:references/use-cases.md # 常见 Use Cases 这些问法都应该优先走： ```bash python3 freestocklineskill/scripts/stockline_cli.py smart-query --query "<用户原话>" ``` ## 行情 1. 贵州茅台最新价 2. 宁德时代现在多少钱 3. 查一下 600519 行情 4. 300750 今天涨跌幅 5. 招商银行最新报价 6. 比亚迪实时行情 7. 中国平安今天开盘价 8. 五粮液最高最低价 9. 中芯国际成交额 10. 东方财富量比和换手率 11. 科创50ETF 最新价 12. 510300 现在多少钱 13. 可转债 123xxx 最新报价 14. 上证指数现在多少 15. 沪深300 今天涨跌幅 ## 历史与 K 线 16. 贵州茅台近30天走势 17. 宁德时代最近一个月表现 18. 600519 近一年日线 19. 300750 近半年 K线 20. 招商银行周线走势 21. 比亚迪月线 22. 东方财富近7天走势 23. 510300 近三个月走势 24. 可转债 123xxx 近20天走势 25. 上证指数近5天走势 26. 创业板指最近一周 27. 贵州茅台后复权 K线 28. 宁德时代不复权日线 29. 600519 前复权走势 30. 科创50 近90天 K线 ## 大盘与市场宽度 31. 今天大盘怎么样 32. 三大指数现在如何 33. A股市场整体表现 34. 看一下大盘 35. 今日市场涨跌家数 36. 上证深成创业板快照 37. 沪深300 和科创50 今天怎样 38. 北证50 最新表现 ## 排行 39. A股涨幅榜前十 40. 今日跌幅榜前20 41. 成交额最高的股票 42. A股成交额前十 43. 成交量榜前50 44. 换手率排行 45. 量比榜前十 46. 振幅榜前二十 47. 总市值前十 48. 市盈率排行 49. 市净率最低的股票 50. 今日领涨股票 51. 今日领跌股票 52. top 30 成交额 53. A股 PE 排行 54. A股 PB 排行 ## 涨停与短线情绪 55. 今日涨停 56. 今天的A股涨停数据 57. 涨停板有哪些 58. 连板股有哪些 59. 跌停池 60. 今日跌停股票 61. 炸板股有哪些 62. 强势股池 63. 2026-04-24 涨停池 64. 今天封板时间最早的股票 65. 今日涨停前50 ## 资金流 66. 主力资金净流入前十 67. 今天主力资金流入排行 68. 主力资金净流出前20 69. 宁德时代资金流向 70. 贵州茅台资金流 71. 行业资金流排行 72. 概念资金流前十 73. 5日主力资金流入排行 74. 10日资金净流入前十 75. 20日资金净流出榜 ## 板块 76. 行业板块涨幅排行 77. 概念板块排行 78. 今天领涨行业 79. 今天领跌概念 80. 半导体板块成分股 81. 白酒板块有哪些股票 82. 贵州茅台属于什么行业 83. 宁德时代属于什么概念 84. 新能源车概念成分股 85. AI 概念板块排行 ## 基本面与估值 86. 贵州茅台基本面 87. 宁德时代估值怎么样 88. 600519 市盈率市净率 89. 招商银行 ROE 90. 比亚迪毛利率净利率 91. 东方财富资产负债率 92. 贵州茅台财务摘要 93. 宁德时代利润表 94. 招商银行现金流量表 95. 贵州茅台股东户数 96. 宁德时代十大股东 97. 贵州茅台分红记录 98. 300750 全部基本面 ## 事件与资料 99. 贵州茅台公告 100. 宁德时代年报 PDF 101. 600519 业绩预告 102. 比亚迪回购公告 103. 招商银行分红公告 104. 今日龙虎榜 105. 2026-04-24 龙虎榜 106. 贵州茅台龙虎榜 107. 可转债涨幅榜前十 108. 可转债成交额排行 109. 可转债 123xxx 近30天走势 110. 新闻快讯里有没有宁德时代 111. 宁德时代新闻 112. 贵州茅台研报评级 113. 比亚迪目标价研报 114. 贵州茅台筹码分布 115. 2026-04-24 大宗交易 116. 贵州茅台大宗交易 117. 2026-04-24 融资融券 118. 宁德时代融资余额 119. 上证深成创业板快照 120. 沪深300 和科创50 今天怎样 ## 散户口语和不专业问法 121. 茅台现在咋样 122. 宁王今天啥价 123. 东财现在多少 124. 招行今天咋样 125. 比王今天涨了吗 126. 中芯现在多少钱 127. 大A今天红不红 128. 今天盘面咋样 129. 两市有没有赚钱效应 130. 今天涨的多还是跌的多 131. 今天哪个票最猛 132. 今天谁跌得最惨 133. 成交最多的是谁 134. 谁最活跃 135. 换手最高的是啥 136. 便宜市盈率的票看看 137. 今天有多少涨停 138. 今天炸了哪些板 139. 连板都有哪些 140. 封单最大的票 141. 钱都往哪儿跑了 142. 主力今天买啥了 143. 主力在卖啥 144. 白酒今天强不强 145. 半导体今天咋样 146. 机器人这条线有哪些票 147. AI 这块有哪些股票 148. 茅台属于啥板块 149. 宁王是哪个行业 150. 茅台财报好不好 151. 东财估值贵不贵 152. 招行分不分红 153. 茅台有啥公告 154. 宁王最近公告 155. 有没有回购的公告 156. 今天谁上榜了 157. 茅台上榜没 158. 茅台有啥新闻 159. 宁王有没有研报 160. 茅台筹码松不松 161. 最近有没有大宗 162. 茅台有没有大宗 163. 宁王两融咋样 164. 转债今天谁最强 165. 帮我瞅眼 300750 166. 别整分析，看看大A ## 不要假装能做 - 给我保证明天涨停的股票。 - 现在还能追茅台吗。 - 宁王明天能回本吗。 - 用 Wind 字段查某只股票。 - 查询需要登录账号的逐笔成交。 - 给出确定买点和收益率。 - 不说明来源就回答一个数字。 FILE:scripts/runtime/freestocklineskill_runtime/__init__.py """Runtime package for freeStockLIneskill.""" __all__ = ["__version__"] __version__ = "0.1.0" FILE:scripts/runtime/freestocklineskill_runtime/cli.py from __future__ import annotations import argparse import json import sys from typing import Any, Dict, List, Optional from .endpoint_catalog import ENDPOINTS from .envelope import failure from .envelope import success from .routing import Entity from .routing import RoutePlan from .routing import build_route_plan from .routing import entity_search_candidates from .routing import entity_from_symbol from .routing import normalize_symbol from .routing import resolve_local_entity from .sources import SourceClient def main(argv: Optional[List[str]] = None) -> int: result = run_command(sys.argv[1:] if argv is None else argv) print(json.dumps(result, ensure_ascii=False)) return 0 if result.get("ok") else 1 def run_command(argv: List[str]) -> Dict[str, Any]: parser = build_parser() try: args = parser.parse_args(argv) except SystemExit: return failure(intent="cli", error_type="invalid_request", error_message="invalid arguments") client = SourceClient(timeout=getattr(args, "timeout", 12.0)) try: if args.command == "endpoint-list": return success(intent="endpoint_list", data={"endpoints": ENDPOINTS}) if args.command == "search-entity": return handle_search_entity(client, args.query) if args.command == "smart-query": return handle_smart_query(client, args.query) if args.command == "quote-realtime": entity = resolve_entity_for_arg(client, args.symbol, "quote_realtime") return wrap_result("quote_realtime", args.symbol, entity.to_dict(), client.quote_realtime(entity)) if args.command == "quote-history": entity = resolve_entity_for_arg(client, args.symbol, "quote_history") return wrap_result( "quote_history", args.symbol, entity.to_dict(), client.quote_history(entity, days=args.days, period=args.period, adjust=args.adjust), ) if args.command == "market-snapshot": return wrap_result("market_snapshot", "", {}, client.market_snapshot()) if args.command == "rank": return wrap_result("rank", "", {"kind": args.kind, "order": args.order, "limit": args.limit}, client.rank(args.kind, args.limit, args.order)) if args.command == "limit-pool": return wrap_result( "limit_pool", "", {"kind": args.kind, "date": args.date, "limit": args.limit}, client.limit_pool(args.kind, args.date, args.limit), ) if args.command == "money-flow": entity = resolve_entity_for_arg(client, args.symbol, "money_flow") if args.symbol else None normalized = {"scope": args.scope, "period": args.period, "limit": args.limit} if entity is not None: normalized["entity"] = entity.to_dict() return wrap_result("money_flow", args.symbol or "", normalized, client.money_flow(args.scope, args.period, entity, args.limit)) if args.command == "sector": entity = resolve_entity_for_arg(client, args.symbol, "sector") if args.symbol else None normalized = {"kind": args.kind, "action": args.action, "limit": args.limit} if entity is not None: normalized["entity"] = entity.to_dict() return wrap_result("sector", args.symbol or "", normalized, client.sector(args.kind, args.action, entity, args.query or "", args.limit)) if args.command == "fundamental": entity = resolve_entity_for_arg(client, args.symbol, "fundamental") return wrap_result("fundamental", args.symbol, {"entity": entity.to_dict(), "pack": args.pack}, client.fundamental(entity, args.pack)) if args.command == "announcement": entity = resolve_entity_for_arg(client, args.symbol, "announcement") if args.symbol else None normalized = {"keyword": args.keyword, "limit": args.limit} if entity is not None: normalized["entity"] = entity.to_dict() return wrap_result("announcement", args.symbol or "", normalized, client.announcement(entity, args.keyword, args.limit)) if args.command == "dragon-tiger": entity = resolve_entity_for_arg(client, args.symbol, "dragon_tiger") if args.symbol else None normalized = {"date": args.date, "limit": args.limit} if entity is not None: normalized["entity"] = entity.to_dict() return wrap_result("dragon_tiger", args.symbol or "", normalized, client.dragon_tiger(args.date, entity, args.limit)) if args.command == "news": entity = resolve_entity_for_arg(client, args.symbol, "news") if args.symbol else None normalized = {"kind": args.kind, "keyword": args.keyword, "limit": args.limit} if entity is not None: normalized["entity"] = entity.to_dict() return wrap_result("news", args.symbol or args.keyword or "", normalized, client.news(entity, args.keyword, args.kind, args.limit)) if args.command == "chip": entity = resolve_entity_for_arg(client, args.symbol, "chip") return wrap_result("chip", args.symbol, {"entity": entity.to_dict(), "limit": args.limit}, client.chip(entity, args.limit)) if args.command == "block-trade": entity = resolve_entity_for_arg(client, args.symbol, "block_trade") if args.symbol else None normalized = {"date": args.date, "limit": args.limit} if entity is not None: normalized["entity"] = entity.to_dict() return wrap_result("block_trade", args.symbol or "", normalized, client.block_trade(args.date, entity, args.limit)) if args.command == "margin-trading": entity = resolve_entity_for_arg(client, args.symbol, "margin_trading") if args.symbol else None normalized = {"date": args.date, "limit": args.limit} if entity is not None: normalized["entity"] = entity.to_dict() return wrap_result("margin_trading", args.symbol or "", normalized, client.margin_trading(args.date, entity, args.limit)) if args.command == "bond": entity = resolve_entity_for_arg(client, args.symbol, "bond") if args.symbol else None normalized = {"action": args.action, "limit": args.limit, "days": args.days} if entity is not None: normalized["entity"] = entity.to_dict() return wrap_result("bond", args.symbol or "", normalized, client.bond(args.action, entity, args.limit, args.days)) except Exception as exc: return failure( intent=command_to_intent(args.command), query=getattr(args, "query", "") or getattr(args, "symbol", "") or "", error_type=type(exc).__name__, error_message=str(exc), ) return failure(intent="cli", error_type="invalid_request", error_message="unknown command") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="stockline-cli") parser.add_argument("--timeout", type=float, default=12.0) subparsers = parser.add_subparsers(dest="command", required=True) p = subparsers.add_parser("smart-query") p.add_argument("--query", required=True) p = subparsers.add_parser("search-entity") p.add_argument("--query", required=True) p = subparsers.add_parser("quote-realtime") p.add_argument("--symbol", required=True) p = subparsers.add_parser("quote-history") p.add_argument("--symbol", required=True) p.add_argument("--days", type=int, default=30) p.add_argument("--period", choices=["daily", "weekly", "monthly", "minute"], default="daily") p.add_argument("--adjust", choices=["qfq", "hfq", "none"], default="qfq") subparsers.add_parser("market-snapshot") p = subparsers.add_parser("rank") p.add_argument("--kind", choices=["gainers", "losers", "amount", "volume", "turnover", "volume-ratio", "amplitude", "market-cap", "pe", "pb"], default="gainers") p.add_argument("--order", choices=["auto", "asc", "desc"], default="auto") p.add_argument("--limit", type=int, default=20) p = subparsers.add_parser("limit-pool") p.add_argument("--kind", choices=["up", "down", "broken", "strong"], default="up") p.add_argument("--date") p.add_argument("--limit", type=int, default=50) p = subparsers.add_parser("money-flow") p.add_argument("--scope", choices=["stock", "market", "industry", "concept"], default="market") p.add_argument("--symbol") p.add_argument("--period", choices=["instant", "3d", "5d", "10d", "20d"], default="instant") p.add_argument("--limit", type=int, default=20) p = subparsers.add_parser("sector") p.add_argument("--kind", choices=["industry", "concept"], default="industry") p.add_argument("--action", choices=["rank", "constituents", "belong"], default="rank") p.add_argument("--symbol") p.add_argument("--query") p.add_argument("--limit", type=int, default=20) p = subparsers.add_parser("fundamental") p.add_argument("--symbol", required=True) p.add_argument("--pack", choices=["basic", "valuation", "financials", "holders", "dividend", "all"], default="basic") p = subparsers.add_parser("announcement") p.add_argument("--symbol") p.add_argument("--keyword") p.add_argument("--limit", type=int, default=20) p = subparsers.add_parser("dragon-tiger") p.add_argument("--date") p.add_argument("--symbol") p.add_argument("--limit", type=int, default=50) p = subparsers.add_parser("news") p.add_argument("--symbol") p.add_argument("--keyword") p.add_argument("--kind", choices=["news", "research"], default="news") p.add_argument("--limit", type=int, default=20) p = subparsers.add_parser("chip") p.add_argument("--symbol", required=True) p.add_argument("--limit", type=int, default=200) p = subparsers.add_parser("block-trade") p.add_argument("--date") p.add_argument("--symbol") p.add_argument("--limit", type=int, default=50) p = subparsers.add_parser("margin-trading") p.add_argument("--date") p.add_argument("--symbol") p.add_argument("--limit", type=int, default=100) p = subparsers.add_parser("bond") p.add_argument("--action", choices=["quote", "history", "rank"], default="rank") p.add_argument("--symbol") p.add_argument("--limit", type=int, default=20) p.add_argument("--days", type=int, default=30) subparsers.add_parser("endpoint-list") return parser def handle_search_entity(client: SourceClient, query: str) -> Dict[str, Any]: local = resolve_local_entity(query) if local is not None: return success(intent="search_entity", query=query, normalized={"entity": local.to_dict()}, source_chain=[{"source": "local_alias", "ok": True}], data=local.to_dict()) resolved = search_entity_candidates(client, query) entity = resolved["entity"] return success( intent="search_entity", query=query, normalized={"entity": entity.to_dict()}, source_chain=resolved["source_chain"], data=entity.to_dict(), warnings=resolved["warnings"], ) def handle_smart_query(client: SourceClient, query: str) -> Dict[str, Any]: plan = build_route_plan(query) if plan.command == "unsupported": return failure( intent=plan.intent, query=query, error_type="unsupported_request", error_message="freeStockLIneskill 只做免费公开数据查询，不做预测、荐股或收益保证", normalized=plan.normalized(), data={"hints": ["可以改问：某股票最新价、近一个月走势、公告、资金流、龙虎榜、板块成分股等"]}, ) may_have_entity = bool(entity_search_candidates(query)) should_search_entity = plan.command in { "quote-realtime", "quote-history", "fundamental", "money-flow", "announcement", "dragon-tiger", "news", "chip", "block-trade", "margin-trading", "bond", } if plan.command == "sector" and plan.params.get("action") == "belong": should_search_entity = True if plan.command == "money-flow" and plan.params.get("scope") in {"industry", "concept"}: should_search_entity = False if plan.command == "bond" and plan.params.get("action") == "rank": should_search_entity = False if should_search_entity and plan.entity is None and may_have_entity: try: plan = build_route_plan(query, entity=search_entity_candidates(client, query, asset_hint="bond" if plan.intent == "bond" else None)["entity"]) except Exception: pass required_entity_missing = plan.command in {"quote-realtime", "quote-history", "fundamental"} and plan.entity is None optional_entity_was_requested = may_have_entity and plan.command in {"money-flow", "announcement", "dragon-tiger", "news", "block-trade", "margin-trading", "bond"} and plan.entity is None if plan.command == "money-flow" and plan.params.get("scope") in {"market", "industry", "concept"}: optional_entity_was_requested = False if plan.command == "bond" and plan.params.get("action") == "rank": optional_entity_was_requested = False required_entity_missing = required_entity_missing or (plan.command in {"chip"} and plan.entity is None) if plan.command == "sector" and plan.params.get("action") == "belong" and plan.entity is None: required_entity_missing = True if required_entity_missing or optional_entity_was_requested: return failure( intent=plan.intent, query=query, error_type="entity_not_found", error_message="无法从自然语言中稳定识别股票、指数、ETF 或可转债标的", normalized=plan.normalized(), data={"hints": ["请提供 6 位代码，如 600519", "请提供更完整的中文简称，如贵州茅台"]}, ) try: result = execute_plan(client, plan) return wrap_result(plan.intent, query, plan.normalized(), result) except Exception as exc: return failure( intent=plan.intent, query=query, error_type=type(exc).__name__, error_message=str(exc), normalized=plan.normalized(), ) def search_entity_candidates(client: SourceClient, query: str, asset_hint: Optional[str] = None) -> Dict[str, Any]: warnings: List[str] = [] tried: List[str] = [] for candidate in entity_search_candidates(query): if candidate in tried: continue tried.append(candidate) local = resolve_local_entity(candidate, asset_hint=asset_hint) if local is not None: return { "entity": local, "source_chain": [{"source": "local_alias", "ok": True, "query": candidate}], "warnings": warnings, } normalized = normalize_symbol(candidate, asset_hint=asset_hint) if normalized: return { "entity": entity_from_symbol(candidate, normalized, query=query), "source_chain": [{"source": "symbol_rule", "ok": True, "query": candidate}], "warnings": warnings, } try: resolved = client.search_entity(candidate) for item in resolved.get("source_chain", []): if isinstance(item, dict): item.setdefault("query", candidate) if candidate != query: resolved.setdefault("warnings", []).append("已从自然语言中抽取标的候选：%s" % candidate) return resolved except Exception as exc: warnings.append("候选标的 %s 解析失败: %s" % (candidate, exc)) raise RuntimeError("无法解析标的：%s" % query) def execute_plan(client: SourceClient, plan: RoutePlan) -> Dict[str, Any]: if plan.command == "quote-realtime": return client.quote_realtime(plan.entity) # type: ignore[arg-type] if plan.command == "quote-history": return client.quote_history( plan.entity, # type: ignore[arg-type] days=plan.params.get("days", 30), period=plan.params.get("period", "daily"), adjust=plan.params.get("adjust", "qfq"), ) if plan.command == "market-snapshot": return client.market_snapshot() if plan.command == "rank": return client.rank(plan.params.get("kind", "gainers"), plan.params.get("limit", 20), plan.params.get("order", "desc")) if plan.command == "limit-pool": return client.limit_pool(plan.params.get("kind", "up"), plan.params.get("date"), plan.params.get("limit", 50)) if plan.command == "money-flow": return client.money_flow(plan.params.get("scope", "market"), plan.params.get("period", "instant"), plan.entity, plan.params.get("limit", 20)) if plan.command == "sector": return client.sector(plan.params.get("kind", "industry"), plan.params.get("action", "rank"), plan.entity, plan.query, plan.params.get("limit", 20)) if plan.command == "fundamental": return client.fundamental(plan.entity, plan.params.get("pack", "basic")) # type: ignore[arg-type] if plan.command == "announcement": return client.announcement(plan.entity, plan.params.get("keyword"), plan.params.get("limit", 20)) if plan.command == "dragon-tiger": return client.dragon_tiger(plan.params.get("date"), plan.entity, plan.params.get("limit", 50)) if plan.command == "news": return client.news(plan.entity, plan.params.get("keyword"), plan.params.get("kind", "news"), plan.params.get("limit", 20)) if plan.command == "chip": return client.chip(plan.entity, plan.params.get("limit", 200)) # type: ignore[arg-type] if plan.command == "block-trade": return client.block_trade(plan.params.get("date"), plan.entity, plan.params.get("limit", 50)) if plan.command == "margin-trading": return client.margin_trading(plan.params.get("date"), plan.entity, plan.params.get("limit", 100)) if plan.command == "bond": return client.bond(plan.params.get("action", "rank"), plan.entity, plan.params.get("limit", 20), plan.params.get("days", 30)) raise RuntimeError("unsupported route command: %s" % plan.command) def wrap_result(intent: str, query: str, normalized: Dict[str, Any], result: Dict[str, Any]) -> Dict[str, Any]: return success( intent=intent, query=query, normalized=normalized, source_chain=result.get("source_chain", []), data=result.get("data"), warnings=result.get("warnings", []), trade_date=_extract_trade_date(result.get("data")), source_status=_source_status(result.get("source_chain", [])), ) def resolve_entity_for_arg(client: SourceClient, raw: str, intent: str) -> Entity: local = resolve_local_entity(raw, asset_hint="bond" if intent == "bond" else None) if local is not None: return local normalized = normalize_symbol(raw, asset_hint="bond" if intent == "bond" else None) if normalized: return entity_from_symbol(raw, normalized, query="转债" if intent == "bond" else "") return client.search_entity(raw)["entity"] def command_to_intent(command: str) -> str: return command.replace("-", "_") def _source_status(chain: List[Dict[str, Any]]) -> Dict[str, Any]: return {str(item.get("source")): bool(item.get("ok")) for item in chain} def _extract_trade_date(data: Any) -> Any: if isinstance(data, dict): for key in ("trade_date", "end_date", "date"): if data.get(key): return data[key] if isinstance(data.get("data"), dict): return _extract_trade_date(data["data"]) return None FILE:scripts/runtime/freestocklineskill_runtime/endpoint_catalog.py from __future__ import annotations from typing import Dict, List ENDPOINTS: List[Dict[str, object]] = [ { "name": "smart-query", "description": "自然语言万能入口。Agent 首选，把用户原话放进 --query。", "example": 'python3 freestocklineskill/scripts/stockline_cli.py smart-query --query "贵州茅台最新价"', }, {"name": "search-entity", "description": "解析股票/指数/ETF/可转债名称或代码。"}, {"name": "quote-realtime", "description": "个股、指数、ETF、可转债实时行情。"}, {"name": "quote-history", "description": "日/周/月/分钟 K 线，支持 qfq/hfq/none。"}, {"name": "market-snapshot", "description": "主要指数与市场宽度快照。"}, {"name": "rank", "description": "涨跌幅、成交额、成交量、换手率、量比、振幅、市值、PE/PB 榜。"}, {"name": "limit-pool", "description": "涨停池、跌停池、炸板池、强势股池。"}, {"name": "money-flow", "description": "个股、全市场、行业、概念资金流。"}, {"name": "sector", "description": "行业/概念板块排行、成分股、个股所属板块。"}, {"name": "fundamental", "description": "基本信息、估值、财务报表、股东、分红。"}, {"name": "announcement", "description": "公告列表与 PDF 链接。"}, {"name": "dragon-tiger", "description": "龙虎榜。"}, {"name": "news", "description": "公开新闻、快讯、研报、评级。"}, {"name": "chip", "description": "筹码分布 best-effort。"}, {"name": "block-trade", "description": "大宗交易明细。"}, {"name": "margin-trading", "description": "融资融券明细。"}, {"name": "bond", "description": "可转债报价、K 线、排行。"}, ] FILE:scripts/runtime/freestocklineskill_runtime/envelope.py from __future__ import annotations from datetime import date from datetime import datetime from datetime import timezone from decimal import Decimal import math from typing import Any, Dict, Iterable, List, Optional def utc_now_iso() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") def success( *, intent: str, query: str = "", normalized: Optional[Dict[str, Any]] = None, source_chain: Optional[List[Dict[str, Any]]] = None, data: Any = None, warnings: Optional[Iterable[str]] = None, trade_date: Optional[str] = None, source_status: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: return { "ok": True, "intent": intent, "query": query, "normalized": json_safe(normalized or {}), "source_chain": json_safe(source_chain or []), "data": json_safe(data), "warnings": list(warnings or []), "meta": { "generated_at": utc_now_iso(), "trade_date": json_safe(trade_date), "source_status": json_safe(source_status or {}), }, } def failure( *, intent: str, query: str = "", error_type: str = "runtime_failed", error_message: str, normalized: Optional[Dict[str, Any]] = None, source_chain: Optional[List[Dict[str, Any]]] = None, warnings: Optional[Iterable[str]] = None, source_status: Optional[Dict[str, Any]] = None, data: Any = None, ) -> Dict[str, Any]: return { "ok": False, "intent": intent, "query": query, "normalized": json_safe(normalized or {}), "source_chain": json_safe(source_chain or []), "data": json_safe(data), "warnings": list(warnings or []), "error": { "type": error_type, "message": error_message, }, "meta": { "generated_at": utc_now_iso(), "trade_date": None, "source_status": json_safe(source_status or {}), }, } def json_safe(value: Any) -> Any: if value is None or isinstance(value, (str, bool, int)): return value if isinstance(value, float): return value if math.isfinite(value) else None if isinstance(value, Decimal): return float(value) if value.is_finite() else None if isinstance(value, datetime): return value.isoformat() if isinstance(value, date): return value.isoformat() if isinstance(value, dict): return {str(key): json_safe(item) for key, item in value.items()} if isinstance(value, (list, tuple, set)): return [json_safe(item) for item in value] if hasattr(value, "item"): try: return json_safe(value.item()) except Exception: pass if hasattr(value, "isoformat"): try: return value.isoformat() except Exception: pass return str(value) FILE:scripts/runtime/freestocklineskill_runtime/routing.py from __future__ import annotations from dataclasses import dataclass from datetime import date from datetime import timedelta import re import unicodedata from typing import Any, Dict, List, Optional, Tuple INDEX_ALIASES = { "上证指数": ("上证指数", "000001.SH"), "上证综指": ("上证指数", "000001.SH"), "沪指": ("上证指数", "000001.SH"), "上证": ("上证指数", "000001.SH"), "上证50": ("上证50", "000016.SH"), "深证成指": ("深证成指", "399001.SZ"), "深成指": ("深证成指", "399001.SZ"), "深证": ("深证成指", "399001.SZ"), "创业板指": ("创业板指", "399006.SZ"), "创业板": ("创业板指", "399006.SZ"), "沪深300": ("沪深300", "000300.SH"), "科创50": ("科创50", "000688.SH"), "北证50": ("北证50", "899050.BJ"), "中证500": ("中证500", "000905.SH"), "中证1000": ("中证1000", "000852.SH"), "国证2000": ("国证2000", "399303.SZ"), "中证红利指数": ("中证红利", "000922.SH"), "中证红利": ("中证红利", "000922.SH"), } ETF_ALIASES = { "沪深300ETF": ("沪深300ETF", "510300.SH"), "300ETF": ("沪深300ETF", "510300.SH"), "科创50ETF": ("科创50ETF", "588000.SH"), "科创板50ETF": ("科创板50ETF", "588080.SH"), "科创板ETF": ("科创板ETF", "588080.SH"), "创业板ETF": ("创业板ETF", "159915.SZ"), "创业板50ETF": ("创业板50ETF", "159949.SZ"), "中证500ETF": ("中证500ETF", "510500.SH"), "中证1000ETF": ("中证1000ETF", "512100.SH"), "上证50ETF": ("上证50ETF", "510050.SH"), } COMMON_SYMBOLS = { "美的集团": "000333.SZ", "贵州茅台": "600519.SH", "茅子": "600519.SH", "茅台": "600519.SH", "宁德时代": "300750.SZ", "宁王": "300750.SZ", "宁德": "300750.SZ", "东方财富": "300059.SZ", "东财": "300059.SZ", "中国平安": "601318.SH", "平安银行": "000001.SZ", "招商银行": "600036.SH", "招行": "600036.SH", "五粮液": "000858.SZ", "比亚迪": "002594.SZ", "比王": "002594.SZ", "万科A": "000002.SZ", "万科": "000002.SZ", "中信证券": "600030.SH", "工商银行": "601398.SH", "农业银行": "601288.SH", "中国银行": "601988.SH", "建设银行": "601939.SH", "中芯国际": "688981.SH", "中芯": "688981.SH", "寒武纪": "688256.SH", "药明康德": "603259.SH", "迈瑞医疗": "300760.SZ", "隆基绿能": "601012.SH", "立讯精密": "002475.SZ", "紫金矿业": "601899.SH", "长江电力": "600900.SH", "中国移动": "600941.SH", "中移动": "600941.SH", "海天味业": "603288.SH", "海天": "603288.SH", "三一重工": "600031.SH", "三一": "600031.SH", "牧原股份": "002714.SZ", "牧原": "002714.SZ", "牧原猪肉": "002714.SZ", "韦尔股份": "603501.SH", "韦尔": "603501.SH", "中国船舶": "600150.SH", "船舶": "600150.SH", "赛力斯": "601127.SH", "工业富联": "601138.SH", "阳光电源": "300274.SZ", "爱尔眼科": "300015.SZ", "中国神华": "601088.SH", "长城汽车": "601633.SH", "保利发展": "600048.SH", "中际旭创": "300308.SZ", "新易盛": "300502.SZ", "江淮汽车": "600418.SH", "北方华创": "002371.SZ", "中微公司": "688012.SH", "中航沈飞": "600760.SH", "歌尔股份": "002241.SZ", "海螺水泥": "600585.SH", } BOARD_HINTS = { "白酒": "industry", "半导体": "industry", "券商": "industry", "证券": "industry", "银行": "industry", "猪肉": "concept", "光伏": "concept", "低空经济": "concept", "人形机器人": "concept", "机器人": "concept", "AI": "concept", "CPO": "concept", "算力": "concept", "新能源车": "concept", } CN_NUM = { "零": 0, "一": 1, "二": 2, "两": 2, "俩": 2, "三": 3, "四": 4, "五": 5, "六": 6, "七": 7, "八": 8, "九": 9, "十": 10, "百": 100, "〇": 0, } SYMBOL_RE = re.compile(r"(?i)(? Dict[str, Any]: return { "raw": self.raw, "symbol": self.symbol, "code": self.code, "market": self.market, "name": self.name, "asset_type": self.asset_type, } @dataclass(frozen=True) class RoutePlan: intent: str command: str query: str entity: Optional[Entity] params: Dict[str, Any] def normalized(self) -> Dict[str, Any]: result = { "command": self.command, "params": self.params, } if self.entity is not None: result["entity"] = self.entity.to_dict() return result def normalize_symbol(text: str, prefer_index: bool = False, asset_hint: Optional[str] = None) -> Optional[str]: raw = normalize_query_text(text).strip().upper().replace(" ", "").replace("-", "") if not raw: return None if raw.startswith(("SH.", "SZ.", "BJ.")) and len(raw) == 9 and raw[3:].isdigit(): raw = "%s.%s" % (raw[3:], raw[:2]) if raw.startswith(("SH", "SZ", "BJ")) and len(raw) == 8 and raw[2:].isdigit(): raw = "%s.%s" % (raw[2:], raw[:2]) if "." in raw: code, market = raw.split(".", 1) market = market.upper() if code.isdigit() and len(code) == 6 and market in {"SH", "SZ", "BJ"}: return "%s.%s" % (code, market) return None if not raw.isdigit() or len(raw) != 6: return None market = infer_market(raw, prefer_index=prefer_index, asset_hint=asset_hint) return "%s.%s" % (raw, market) def infer_market(code: str, prefer_index: bool = False, asset_hint: Optional[str] = None) -> str: if prefer_index and code in {"000001", "000300", "000688", "000905"}: return "SH" if prefer_index and code.startswith("399"): return "SZ" if prefer_index and code.startswith("899"): return "BJ" if asset_hint == "bond": if code.startswith(("110", "113", "118")): return "SH" return "SZ" if code.startswith(("50", "51", "52", "56", "58")): return "SH" if code.startswith(("15", "16", "18")): return "SZ" if code.startswith(("4", "8", "92")): return "BJ" if code.startswith(("6", "9")): return "SH" return "SZ" def asset_type_for_symbol(symbol: str, query: str = "") -> str: code, market = symbol.split(".", 1) if "转债" in query or code.startswith(("110", "113", "118", "123", "127", "128")): return "bond" if any(name for name, (_, value) in INDEX_ALIASES.items() if value == symbol): return "index" if code.startswith(("000", "399", "899")) and market in {"SH", "SZ", "BJ"} and "股票" not in query: if symbol in {item[1] for item in INDEX_ALIASES.values()}: return "index" if code.startswith(("50", "51", "52", "56", "58", "15", "16", "18")): return "fund" return "stock" def entity_from_symbol(raw: str, symbol: str, name: Optional[str] = None, query: str = "") -> Entity: code, market = symbol.split(".", 1) return Entity( raw=raw, symbol=symbol, code=code, market=market, name=name, asset_type=asset_type_for_symbol(symbol, query=query), ) def resolve_local_entity(query: str, prefer_index: bool = False, asset_hint: Optional[str] = None) -> Optional[Entity]: text = normalize_query_text(query) compact_text = re.sub(r"\s+", "", text) for alias, (name, symbol) in sorted(ETF_ALIASES.items(), key=lambda item: len(item[0]), reverse=True): if alias.lower() in text.lower() or alias.lower() in compact_text.lower(): return entity_from_symbol(alias, symbol, name=name, query=text) for alias, (name, symbol) in sorted(INDEX_ALIASES.items(), key=lambda item: len(item[0]), reverse=True): if alias.lower() in text.lower() or alias.lower() in compact_text.lower(): return entity_from_symbol(alias, symbol, name=name, query=text) match = SYMBOL_RE.search(re.sub(r"\s+", "", text)) if match: raw = match.group(0) symbol = normalize_symbol(raw, prefer_index=prefer_index, asset_hint=asset_hint) if symbol: return entity_from_symbol(raw, symbol, query=text) for name, symbol in sorted(COMMON_SYMBOLS.items(), key=lambda item: len(item[0]), reverse=True): if name in text or name in compact_text: return entity_from_symbol(name, symbol, name=name, query=text) return None def entity_search_candidates(query: str) -> List[str]: text = normalize_query_text(query).strip() if not text: return [] candidates: List[str] = [] def add(value: str) -> None: cleaned = _clean_entity_candidate(value) if cleaned and cleaned not in candidates and not _looks_like_non_entity_query(cleaned): candidates.append(cleaned) symbol_match = SYMBOL_RE.search(text) if symbol_match: add(symbol_match.group(0)) compact = re.sub(r"[\s,，。！？?！：:；;、（）()【】\[\]\"'“”‘’]+", "", text) stripped = FULL_DATE_RE.sub("", compact) stripped = CHINESE_FULL_DATE_RE.sub("", stripped) stripped = MONTH_DAY_RE.sub("", stripped) stripped = DATE_SPAN_RE.sub("", stripped) for noise in ENTITY_QUERY_NOISE: stripped = stripped.replace(noise, "") stripped = re.sub(r"(?i)(?:pe|pb|roe|top|pdf)", "", stripped) stripped = re.sub(r"\d{1,3}", "", stripped) add(stripped) if stripped == compact: add(text) for segment in re.findall(r"[\u4e00-\u9fffA-Za-z0-9]{2,12}", compact): candidate = segment candidate = FULL_DATE_RE.sub("", candidate) candidate = CHINESE_FULL_DATE_RE.sub("", candidate) candidate = MONTH_DAY_RE.sub("", candidate) candidate = DATE_SPAN_RE.sub("", candidate) for noise in ENTITY_QUERY_NOISE: candidate = candidate.replace(noise, "") candidate = re.sub(r"(?i)(pe|pb|roe|top|pdf)", "", candidate) add(candidate) return candidates[:8] def parse_chinese_number(text: str) -> Optional[int]: cleaned = normalize_query_text(text).strip() if not cleaned: return None if cleaned.isdigit(): return int(cleaned) if cleaned.startswith("零") and len(cleaned) > 1: return parse_chinese_number(cleaned[1:]) if cleaned == "十": return 10 if "百" in cleaned: left, right = cleaned.split("百", 1) hundreds = CN_NUM.get(left, 1) if left else 1 rest = parse_chinese_number(right) if right else 0 if rest is None: return None return hundreds * 100 + rest if "十" in cleaned: left, right = cleaned.split("十", 1) tens = CN_NUM.get(left, 1) if left else 1 units = CN_NUM.get(right, 0) if right else 0 return tens * 10 + units return CN_NUM.get(cleaned) def extract_limit(query: str, default: int = 20) -> int: query = normalize_query_text(query) lowered = query.lower() match = re.search(r"(?:top|前)\s*(\d{1,3})", lowered) if match: return max(1, min(300, int(match.group(1)))) compact = re.sub(r"\s+", "", lowered) match = re.search(r"前\s*([一二三四五六七八九十两俩百]+)", lowered) or re.search(r"前([一二三四五六七八九十两俩百]+)", compact) if match: parsed = parse_chinese_number(match.group(1)) if parsed is not None: return max(1, min(300, parsed)) if "前十" in query or "前十" in compact: return 10 if "前二十" in query or "前二十" in compact: return 20 if "前五十" in query or "前五十" in compact: return 50 return default def extract_date(query: str, today: Optional[date] = None) -> Optional[str]: effective_today = today or date.today() text = normalize_query_text(query) compact_match = re.search(r"(20\d{2})(\d{2})(\d{2})", re.sub(r"\D", "", text)) if compact_match: year, month, day = compact_match.groups() return "%04d-%02d-%02d" % (int(year), int(month), int(day)) match = re.search(r"(20\d{2})[-/.年]?\s*(\d{1,2})[-/.月]?\s*(\d{1,2})日?", text) if match: year, month, day = match.groups() return "%04d-%02d-%02d" % (int(year), int(month), int(day)) chinese = _extract_chinese_date(text) if chinese: return chinese match = re.search(r"(\d{1,2})月(\d{1,2})[日号]?", text) if match: month, day = match.groups() return "%04d-%02d-%02d" % (effective_today.year, int(month), int(day)) if "昨天" in text: return (effective_today - timedelta(days=1)).isoformat() if "今天" in text or "今日" in text: return effective_today.isoformat() return None def _extract_chinese_date(text: str) -> Optional[str]: match = re.search(r"([二〇零一二三四五六七八九]{4})年([一二三四五六七八九十两俩]{1,3})月([一二三四五六七八九十两俩]{1,3})[日号]?", text) if not match: return None raw_year, raw_month, raw_day = match.groups() year_digits = [] for char in raw_year: digit = CN_NUM.get(char) if digit is None or digit >= 10: return None year_digits.append(str(digit)) month = parse_chinese_number(raw_month) day = parse_chinese_number(raw_day) if month is None or day is None: return None return "%04d-%02d-%02d" % (int("".join(year_digits)), month, day) def extract_days(query: str, default: int = 30) -> int: text = normalize_query_text(query) if any(word in text for word in ["近半个月", "最近半个月", "过去半个月", "近半月", "最近半月", "过去半月"]): return 15 match = re.search(r"(?:近|最近|过去)\s*(\d{1,4})\s*(?:天|日|个?交易日)", text) if match: return max(1, min(5000, int(match.group(1)))) match = re.search(r"(?:近|最近|过去)\s*([零〇一二三四五六七八九十两俩百]+)\s*(?:天|日|个?交易日)", text) if match: parsed = parse_chinese_number(match.group(1)) if parsed is not None: return max(1, min(5000, parsed)) if any(word in text for word in ["近一周", "最近一周"]): return 7 match = re.search(r"(?:近|最近|过去)\s*(\d{1,3}|[零〇一二三四五六七八九十两俩百]+)\s*(周|个?星期|个月|月|年)", text) if match: raw_count, unit = match.groups() count = int(raw_count) if raw_count.isdigit() else parse_chinese_number(raw_count) if count is not None: multiplier = 7 if unit in {"周", "星期", "个星期"} else 30 if unit in {"个月", "月"} else 365 return max(1, min(5000, count * multiplier)) if any(word in text for word in ["近一个月", "最近一个月"]): return 30 if any(word in text for word in ["近三个月", "最近三个月"]): return 90 if "近半年" in text: return 180 if "近一年" in text: return 365 if "半年报" not in text and ("半年" in text or "半年度" in text): return 180 if "去年" in text or "一年走势" in text or "一年K线" in text or "一年k线" in text: return 365 return default def history_period(query: str) -> str: lowered = normalize_query_text(query).lower() if any(word in lowered for word in ["分钟", "分时", "minute", "1m", "5m", "15m", "30m", "60m"]): return "minute" if any(word in lowered for word in ["周k", "周线", "weekly", "week"]): return "weekly" if any(word in lowered for word in ["月k", "月线", "monthly"]): return "monthly" return "daily" def adjust_mode(query: str) -> str: lowered = normalize_query_text(query).lower() if "后复权" in lowered or "hfq" in lowered: return "hfq" if "不复权" in lowered or "bfq" in lowered: return "none" return "qfq" def rank_kind(query: str) -> str: lowered = normalize_query_text(query).lower() scan = lowered + re.sub(r"\s+", "", lowered) if any(word in scan for word in ["跌幅", "领跌", "跌得", "最惨", "最狠"]): return "losers" if "成交额" in scan or "金额" in scan or "成交最多" in scan: return "amount" if "成交量" in scan: return "volume" if "换手" in scan or "最活跃" in scan: return "turnover" if "量比" in scan: return "volume-ratio" if "振幅" in scan: return "amplitude" if "市值" in scan: return "market-cap" if "市盈" in scan or "pe" in scan: return "pe" if "市净" in scan or "pb" in scan: return "pb" return "gainers" def rank_order(query: str, kind: str) -> str: lowered = normalize_query_text(query).lower() scan = lowered + re.sub(r"\s+", "", lowered) if kind == "losers": return "asc" if any(word in scan for word in ["最低", "最小", "从低到高", "低到高", "便宜", "别太高", "不太高", "lowest", "asc"]): return "asc" return "desc" def money_period(query: str) -> str: text = normalize_query_text(query) scan = re.sub(r"(?i)top\s*\d{1,3}", "", text) scan = re.sub(r"前\s*(?:\d{1,3}|[一二三四五六七八九十两俩百]+)", "", scan) if re.search(r"20\s*[日天]", scan) or "二十日" in scan or "二十天" in scan: return "20d" if re.search(r"10\s*[日天]", scan) or "十日" in scan or "十天" in scan: return "10d" if re.search(r"5\s*[日天]", scan) or "五日" in scan or "五天" in scan: return "5d" if re.search(r"3\s*[日天]", scan) or "三日" in scan or "三天" in scan: return "3d" return "instant" def generic_market_query(query: str) -> bool: text = normalize_query_text(query) compact = re.sub(r"\s+", "", text) if any(alias in text or alias in compact for alias in INDEX_ALIASES): return False return any( word in text or word in compact for word in [ "大盘", "大A", "股市", "盘面", "两市", "市场整体", "三大指数", "今天市场", "市场怎么样", "市场热不热", "市场涨跌家数", "涨跌家数", "市场宽度", "赚钱效应", "涨的多还是跌的多", "涨多还是跌多", "指数们", ] ) def major_index_mention_count(query: str) -> int: text = normalize_query_text(query) compact = re.sub(r"[\s+＋,，、/]+", "", text) scan = text + compact groups = [ ["上证指数", "上证综指", "沪指", "上证"], ["深证成指", "深成指", "深证"], ["创业板指", "创业板"], ["沪深300"], ["科创50"], ["北证50"], ["中证500"], ["中证1000"], ["国证2000"], ["中证红利"], ] return sum(1 for aliases in groups if any(alias in scan for alias in aliases)) def detect_intent(query: str) -> str: query = normalize_query_text(query) lowered = query.lower() scan = lowered + re.sub(r"\s+", "", lowered) if unsupported_query(query): return "unsupported" if generic_market_query(query) or major_index_mention_count(query) >= 2 or any(word in scan for word in ["指数快照", "大盘快照"]): return "market_snapshot" if any(word in scan for word in ["可转债", "转债"]) or ("债" in scan and SYMBOL_RE.search(re.sub(r"\s+", "", query))): return "bond" if any(word in scan for word in ["筹码", "筹码分布"]): return "chip" if any(word in scan for word in ["研报", "评级", "目标价", "机构怎么看", "机构看法", "机构评级"]): return "news" if any(word in scan for word in ["新闻", "快讯", "消息", "资讯", "利好", "利空", "有雷", "啥雷", "有没有雷", "有没有啥雷", "雷不雷", "雷没雷", "爆雷", "暴雷"]): return "news" if any(word in scan for word in ["大宗交易", "大宗成交", "有没有大宗", "有啥大宗", "大宗"]): return "block_trade" if any(word in scan for word in ["融资融券", "两融", "融资余额", "融资盘", "融券"]): return "margin_trading" if any(word in scan for word in ["公告", "年报", "季报", "业绩预告", "pdf", "披露", "减持", "增持", "回购"]): return "announcement" if "龙虎榜" in scan or "上榜" in scan: return "dragon_tiger" board_money_query = "资金" in scan and board_scope_hint(query) is not None if board_money_query or any(word in scan for word in ["资金流", "主力资金", "净流入", "净流出", "钱都往哪", "钱往哪", "钱流哪", "钱去哪", "资金去哪", "资金往哪", "资金在进还是出", "资金抱团", "资金在买", "吸金", "钱进没", "还有人买", "主力买", "主力卖", "主力在买", "主力在卖", "主力跑路", "跑路"]) or ("主力" in scan and any(word in scan for word in ["买啥", "卖啥", "买什么", "卖什么"])): return "money_flow" if any(word in scan for word in ["涨停", "跌停", "炸板", "炸了哪些板", "连板", "封板", "封单", "强势股", "回封", "地天板", "天地板", "封得", "封死", "开板", "跌停潮"]): return "limit_pool" if any(word in scan for word in ["板块", "行业", "概念", "题材", "这条线", "这块"]) or _looks_like_board_chat(query): return "sector" has_rank_word = any(word in scan for word in ["排行", "排名", "榜", "top", "前十", "前二十", "前50", "领涨", "领跌"]) has_chat_rank_word = any(word in scan for word in ["哪个票最猛", "最猛", "最能打", "跌得最惨", "最惨", "最狠", "杀得最狠", "成交最多", "最活跃", "谁涨", "谁跌", "谁最", "换手最高", "便宜市盈率"]) has_extreme_word = any(word in scan for word in ["最高", "最低", "最大", "最小", "highest", "lowest"]) if not has_rank_word and not has_chat_rank_word and any(word in scan for word in ["开盘", "收盘", "昨收", "最高最低", "最高价", "最低价", "量比"]): return "quote_realtime" has_valuation_rank = ( ("市盈" in scan or "市净" in scan or "pe" in scan or "pb" in scan) and any(word in scan for word in ["最高", "最低", "最大", "最小", "highest", "lowest", "股票"]) ) if has_rank_word or has_chat_rank_word or has_valuation_rank or has_extreme_word: return "rank" if any(word in scan for word in ["财务", "财报", "财务底子", "基本面", "基本资料", "基本情况", "估值", "roe", "毛利率", "净利率", "资产负债", "利润表", "现金流", "股东", "分红", "派息", "每年分", "市盈率", "市净率", "便宜了", "便宜不便宜"]): return "fundamental" if re.search(r"\b(?:pe|pb)\b", lowered): return "fundamental" if DATE_SPAN_RE.search(query) or any(word in scan for word in ["走势", "历史", "k线", "kline", "日线", "周线", "月线", "周k", "月k", "分钟", "分时", "近一个月", "最近", "咋走", "走的", "走得"]): return "quote_history" return "quote_realtime" def unsupported_query(query: str) -> bool: lowered = normalize_query_text(query).lower() compact = re.sub(r"\s+", "", lowered) if any(word in compact for word in ["推荐股票", "推荐一只", "推荐一个", "买哪只", "买什么", "能买", "能不能买", "现在买", "值得买", "该买", "卖不卖", "买不买", "投资建议", "还能追", "能不能追", "能回本", "回本吗", "抄底吗", "能抄底"]): return True if any(word in compact for word in ["保证", "预测", "预判", "一定", "稳赚", "翻倍"]): return True if "明天" in compact and any(word in compact for word in ["涨", "跌", "涨停", "走势", "大盘", "买", "卖", "回本"]): return True if any(word in compact for word in ["会不会涨", "会不会跌", "能不能涨", "能不能跌", "会不会反弹", "能不能反弹", "反弹吗", "未来会涨", "未来会跌", "下周会涨", "下周会跌"]): return True return False def limit_kind(query: str) -> str: lowered = (query or "").lower() scan = lowered + re.sub(r"\s+", "", lowered) if "跌停" in scan: return "down" if "强势" in scan or "回封" in scan or "地天板" in scan: return "strong" if "炸板" in scan or "炸了" in scan or "开板" in scan: return "broken" return "up" def sector_kind(query: str) -> str: text = normalize_query_text(query) compact = re.sub(r"\s+", "", text) if "概念" in text or "概念" in compact: return "concept" for board, kind in BOARD_HINTS.items(): if board.lower() in text.lower() or board.lower() in compact.lower(): return kind return "industry" def board_scope_hint(query: str) -> Optional[str]: text = normalize_query_text(query) compact = re.sub(r"\s+", "", text) for board, kind in BOARD_HINTS.items(): if board.lower() in text.lower() or board.lower() in compact.lower(): return kind return None def sector_action(query: str, has_entity: bool) -> str: query = normalize_query_text(query) scan = query + re.sub(r"\s+", "", query) if has_entity or any(word in scan for word in ["属于", "所属", "哪个行业", "啥板块", "是哪个行业"]): return "belong" if any(word in scan for word in ["成分", "成份", "包含", "有哪些股票", "有哪些", "有哪些票", "有啥票", "有啥股票", "什么票", "哪些股票", "哪些票", "都有谁", "都有啥"]): return "constituents" return "rank" def fundamental_pack(query: str) -> str: lowered = normalize_query_text(query).lower() scan = lowered + re.sub(r"\s+", "", lowered) if any(word in scan for word in ["股东", "十大股东", "户数"]): return "holders" if "分红" in scan or "派息" in scan or "分不分红" in scan or "每年分" in scan: return "dividend" if any(word in scan for word in ["财务", "财务底子", "资产负债", "利润表", "现金流", "财报", "roe", "毛利率", "净利率", "资产负债率"]): return "financials" if any(word in scan for word in ["估值", "贵不贵", "便宜了", "便宜不便宜", "市盈", "市净", "pe", "pb", "市值"]): return "valuation" if "全部" in scan or "完整" in scan: return "all" return "basic" def build_route_plan(query: str, entity: Optional[Entity] = None, today: Optional[date] = None) -> RoutePlan: query = normalize_query_text(query) intent = detect_intent(query) resolved = entity or resolve_local_entity( query, prefer_index=intent in {"market_snapshot", "quote_history"}, asset_hint="bond" if intent == "bond" else None, ) if intent == "quote_realtime": return RoutePlan(intent, "quote-realtime", query, resolved, {}) if intent == "quote_history": return RoutePlan( intent, "quote-history", query, resolved, {"days": extract_days(query), "period": history_period(query), "adjust": adjust_mode(query), "date": extract_date(query, today)}, ) if intent == "market_snapshot": return RoutePlan(intent, "market-snapshot", query, None, {}) if intent == "rank": kind = rank_kind(query) return RoutePlan(intent, "rank", query, None, {"kind": kind, "order": rank_order(query, kind), "limit": extract_limit(query)}) if intent == "limit_pool": return RoutePlan(intent, "limit-pool", query, None, {"kind": limit_kind(query), "date": extract_date(query, today), "limit": extract_limit(query, 50)}) if intent == "money_flow": scope = "stock" if resolved is not None else "market" compact = re.sub(r"\s+", "", query) if "行业" in query or "行业" in compact: scope = "industry" if "概念" in query or "概念" in compact: scope = "concept" board_scope = board_scope_hint(query) if resolved is None and board_scope is not None: scope = board_scope return RoutePlan(intent, "money-flow", query, resolved, {"scope": scope, "period": money_period(query), "limit": extract_limit(query)}) if intent == "sector": return RoutePlan(intent, "sector", query, resolved, {"kind": sector_kind(query), "action": sector_action(query, resolved is not None), "limit": extract_limit(query, 20)}) if intent == "fundamental": return RoutePlan(intent, "fundamental", query, resolved, {"pack": fundamental_pack(query)}) if intent == "announcement": return RoutePlan(intent, "announcement", query, resolved, {"keyword": announcement_keyword(query), "limit": extract_limit(query, 20), "date": extract_date(query, today)}) if intent == "dragon_tiger": return RoutePlan(intent, "dragon-tiger", query, resolved, {"date": extract_date(query, today), "limit": extract_limit(query, 50)}) if intent == "news": return RoutePlan(intent, "news", query, resolved, {"kind": news_kind(query), "keyword": news_keyword(query), "limit": extract_limit(query, 20)}) if intent == "chip": return RoutePlan(intent, "chip", query, resolved, {"limit": extract_limit(query, 200)}) if intent == "block_trade": return RoutePlan(intent, "block-trade", query, resolved, {"date": extract_date(query, today), "limit": extract_limit(query, 50)}) if intent == "margin_trading": return RoutePlan(intent, "margin-trading", query, resolved, {"date": extract_date(query, today), "limit": extract_limit(query, 100)}) if intent == "bond": action = "rank" bond_scan = query.lower() + re.sub(r"\s+", "", query.lower()) if resolved is not None and (DATE_SPAN_RE.search(query) or any(word in bond_scan for word in ["走势", "历史", "k线", "kline", "日线", "最近", "咋走", "走的", "走得"])): action = "history" elif resolved is not None: action = "quote" return RoutePlan(intent, "bond", query, resolved, {"action": action, "limit": extract_limit(query), "days": extract_days(query)}) if intent == "unsupported": return RoutePlan(intent, "unsupported", query, resolved, {}) return RoutePlan("unknown", "smart-query", query, resolved, {}) def announcement_keyword(query: str) -> Optional[str]: query = normalize_query_text(query) scan = query + re.sub(r"\s+", "", query) for word in ["年报", "半年报", "季报", "一季报", "三季报", "业绩预告", "分红", "减持", "增持", "回购", "重大事项"]: if word in scan: return word return None def news_kind(query: str) -> str: lowered = normalize_query_text(query).lower() scan = lowered + re.sub(r"\s+", "", lowered) if any(word in scan for word in ["研报", "评级", "目标价", "机构怎么看", "机构看法", "机构评级"]): return "research" return "news" def news_keyword(query: str) -> Optional[str]: query = normalize_query_text(query) scan = query + re.sub(r"\s+", "", query) if any(word in scan for word in ["机构评级", "机构怎么看", "机构看法"]): return "研报" if any(word in scan for word in ["消息", "资讯", "利好", "利空", "有雷", "爆雷"]): return None for word in ["研报", "评级", "目标价", "新闻", "快讯"]: if word in scan: return word return None def _looks_like_board_chat(query: str) -> bool: text = normalize_query_text(query) compact = re.sub(r"\s+", "", text) if not any(board.lower() in compact.lower() for board in BOARD_HINTS): return False explicit_board = any(word in compact for word in ["板块", "行业", "概念", "题材", "这条线", "这块", "这个方向", "方向"]) if not explicit_board and resolve_local_entity(query) is not None: return False return any(word in compact for word in ["强不强", "最强", "哪个最强", "咋样", "怎么样", "有动静", "动静没", "拉了吗", "趴着", "起来了", "谁在涨", "有哪些", "有哪些票", "有啥票", "什么票", "哪些票", "都有谁", "这条线", "这块", "这个方向", "方向", "题材"]) def _clean_entity_candidate(value: str) -> str: cleaned = normalize_query_text(value).strip() cleaned = re.sub(r"^[的了呢吧啊呀嘛吗]+|[的了呢吧啊呀嘛吗]+$", "", cleaned) cleaned = re.sub(r"[\s,，。！？?！：:；;、（）()【】\[\]\"'“”‘’]+", "", cleaned) cleaned = re.sub(r"^[和与及]+|[和与及]+$", "", cleaned) return cleaned def normalize_query_text(text: str) -> str: normalized = unicodedata.normalize("NFKC", text or "") normalized = normalized.replace("\u200b", "").replace("\ufeff", "") normalized = normalized.replace("﹣", "-").replace("－", "-") return normalized def _looks_like_non_entity_query(value: str) -> bool: if len(value) < 2: return True lowered = value.lower() if lowered in {"pe", "pb", "roe", "top"}: return True if value.isdigit() and len(value) != 6: return True if re.fullmatch(r"前?\d{1,3}", value): return True if re.fullmatch(r"前?[一二三四五六七八九十两俩百]+条?", value): return True if "年" in value and "月" in value and re.fullmatch(r"[二〇零一二三四五六七八九十两俩年月日号]+", value): return True if re.fullmatch(r"\d{0,3}日(?:入|出|资金)?", value): return True if re.fullmatch(r"[一二三四五六七八九十两俩百]+日(?:入|出|资金)?", value): return True generic_words = [ "大盘", "市场", "指数", "涨停", "跌停", "炸板", "排行", "排名", "主力资金", "资金流", "资金", "流入", "流出", "成交额", "成交量", "可转债", "转债", "行业", "概念", "板块", "成分股", "成分", "有哪些股票", "有哪些", "包含", "最低", "最高", "最大", "最小", "回购", "减持", "增持", "重大事项", "新闻", "快讯", "有没有", "里", "研报", "评级", "目标价", "筹码", "筹码分布", "大宗交易", "大宗成交", "大宗", "融资融券", "两融", "融资", "融资余额", "融券", "全市场", "余额", "最火", "现在", "这票", "这个票", "这只票", "这公司", "有啥", "最近", "都往跑", "往跑", "在卖", "记录", "全部", "资产负债率", ] return value in generic_words FILE:scripts/runtime/freestocklineskill_runtime/sources.py from __future__ import annotations from contextlib import redirect_stderr from contextlib import redirect_stdout from datetime import date import io import json import os import re import warnings from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple warnings.filterwarnings("ignore", message="urllib3 v2 only supports OpenSSL.*") import requests from .routing import Entity from .routing import entity_from_symbol from .routing import normalize_symbol USER_AGENT = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36" ) EASTMONEY_UT = "7eea3edcaed734bea9cbfc24409ed989" A_STOCK_FS = "m:0+t:6,m:0+t:80,m:1+t:2,m:1+t:23" class SourceError(RuntimeError): pass class SourceClient: def __init__(self, timeout: float = 12.0) -> None: self.timeout = timeout self.session = requests.Session() self.session.trust_env = False self.session.headers.update( { "User-Agent": USER_AGENT, "Accept": "*/*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Connection": "close", } ) def _get(self, url: str, params: Optional[Dict[str, Any]] = None, referer: Optional[str] = None) -> requests.Response: headers = {"Referer": referer} if referer else None response = self.session.get(url, params=params, headers=headers, timeout=self.timeout) response.raise_for_status() return response def _post(self, url: str, data: Optional[Dict[str, Any]] = None, referer: Optional[str] = None) -> requests.Response: headers = {"Referer": referer} if referer else None response = self.session.post(url, data=data, headers=headers, timeout=self.timeout) response.raise_for_status() return response def _json_get(self, url: str, params: Optional[Dict[str, Any]] = None, referer: Optional[str] = None) -> Dict[str, Any]: response = self._get(url, params=params, referer=referer) payload = response.json() if not isinstance(payload, dict): raise SourceError("response JSON is not an object") return payload def search_entity(self, query: str) -> Dict[str, Any]: chain: List[Dict[str, Any]] = [] warnings: List[str] = [] for name, func in [ ("tencent_smartbox", self._search_tencent), ("sina_suggest", self._search_sina), ]: try: entity = func(query) chain.append({"source": name, "ok": entity is not None}) if entity is not None: return {"entity": entity, "source_chain": chain, "warnings": warnings} except Exception as exc: chain.append({"source": name, "ok": False, "error": str(exc)}) warnings.append("%s 解析失败: %s" % (name, exc)) raise SourceError("无法解析标的：%s" % query) def _search_tencent(self, query: str) -> Optional[Entity]: response = self._get("https://smartbox.gtimg.cn/s3/", {"q": query, "t": "all"}, referer="https://gu.qq.com/") text = _decode_response(response) match = re.search(r'v_hint="(.*)"', text, flags=re.S) if not match: return None for row in match.group(1).split("^"): parts = row.split("~") if len(parts) < 5: continue market, code, name, _, kind = parts[:5] symbol = normalize_symbol("%s.%s" % (code, market)) if symbol: return entity_from_symbol(query, symbol, name=_json_unescape(name), query=query if kind != "ZS" else "指数") return None def _search_sina(self, query: str) -> Optional[Entity]: response = self._get( "https://suggest3.sinajs.cn/suggest/type=11,12,13,14,15", {"key": query, "name": "suggestdata"}, referer="https://finance.sina.com.cn/", ) text = _decode_response(response) match = re.search(r'"(.*)"', text, flags=re.S) if not match: return None for row in match.group(1).split(";"): parts = row.split(",") if len(parts) < 4: continue name, code, provider_symbol = parts[0].strip(), parts[2].strip(), parts[3].strip().lower() symbol = normalize_symbol(provider_symbol) if symbol: return entity_from_symbol(query, symbol, name=name, query=query) if code: symbol = normalize_symbol(code) if symbol: return entity_from_symbol(query, symbol, name=name, query=query) return None def quote_realtime(self, entity: Entity) -> Dict[str, Any]: chain: List[Dict[str, Any]] = [] try: data = self._quote_tencent([entity.symbol]) chain.append({"source": "tencent_finance", "ok": True}) return {"data": data[0], "source_chain": chain, "warnings": []} except Exception as exc: chain.append({"source": "tencent_finance", "ok": False, "error": str(exc)}) try: rows = self._akshare_call("stock_zh_a_spot_em") item = _find_row(rows, entity.code) if item: chain.append({"source": "akshare.stock_zh_a_spot_em", "ok": True}) return {"data": _normalize_akshare_quote(item, entity), "source_chain": chain, "warnings": []} raise SourceError("akshare did not return target code") except Exception as exc: chain.append({"source": "akshare.stock_zh_a_spot_em", "ok": False, "error": str(exc)}) raise SourceError("实时行情所有免费源失败") def _quote_tencent(self, symbols: Iterable[str]) -> List[Dict[str, Any]]: provider_symbols = [to_tencent_symbol(symbol) for symbol in symbols] response = self._get("https://qt.gtimg.cn/q=" + ",".join(provider_symbols), referer="https://gu.qq.com/") text = _decode_response(response) quotes: List[Dict[str, Any]] = [] for match in re.finditer(r"v_([a-z]{2}\d{6})=\"([^\"]*)\";", text): provider_symbol = match.group(1) fields = match.group(2).split("~") if len(fields) < 34 or not fields[1]: continue quotes.append(_parse_tencent_quote(provider_symbol, fields)) if not quotes: raise SourceError("腾讯行情返回为空") return quotes def quote_history(self, entity: Entity, *, days: int, period: str, adjust: str) -> Dict[str, Any]: chain: List[Dict[str, Any]] = [] try: data = self._history_tencent(entity.symbol, days=days, period=period, adjust=adjust) chain.append({"source": "tencent_fqkline", "ok": True}) return {"data": data, "source_chain": chain, "warnings": []} except Exception as exc: chain.append({"source": "tencent_fqkline", "ok": False, "error": str(exc)}) try: data = self._history_efinance(entity, days=days) chain.append({"source": "efinance.get_quote_history", "ok": True}) return {"data": data, "source_chain": chain, "warnings": ["已从腾讯 K 线回退到 efinance"]} except Exception as exc: chain.append({"source": "efinance.get_quote_history", "ok": False, "error": str(exc)}) if period == "minute": return { "data": {"symbol": entity.symbol, "period": period, "adjust": adjust, "count": 0, "candles": []}, "source_chain": chain, "warnings": ["分钟 K 线公开源暂时不可用，已返回空列表和失败详情"], } raise SourceError("历史行情所有免费源失败") def _history_tencent(self, symbol: str, *, days: int, period: str, adjust: str) -> Dict[str, Any]: provider_symbol = to_tencent_symbol(symbol) period_key = {"daily": "day", "weekly": "week", "monthly": "month", "minute": "mline"}.get(period, "day") adjust_key = {"qfq": "qfq", "hfq": "hfq", "none": ""}.get(adjust, "qfq") if period_key == "mline": param = "%s,m1,,,%d" % (provider_symbol, max(1, min(days, 240))) else: param = "%s,%s,,,%d,%s" % (provider_symbol, period_key, max(1, min(days, 5000)), adjust_key) payload = self._json_get( "https://web.ifzq.gtimg.cn/appstock/app/fqkline/get", {"param": param}, referer="https://gu.qq.com/", ) node = payload.get("data", {}).get(provider_symbol, {}) rows = node.get("%s%s" % (adjust_key, period_key)) or node.get(period_key) or node.get("data") or [] candles = [_parse_kline_row(row) for row in rows if isinstance(row, list) and len(row) >= 6] if not candles: raise SourceError("腾讯 K 线为空") qt = node.get("qt", {}).get(provider_symbol) quote = _parse_tencent_quote(provider_symbol, qt) if isinstance(qt, list) else None return { "symbol": symbol, "period": period, "adjust": adjust, "count": len(candles), "start_date": candles[0].get("date"), "end_date": candles[-1].get("date"), "latest_quote": quote, "candles": candles, } def _history_efinance(self, entity: Entity, *, days: int) -> Dict[str, Any]: import efinance as ef df = _quiet_call(ef.stock.get_quote_history, stock_codes=entity.code) rows = _df_to_records(df)[-days:] candles = [ { "date": _pick(row, ["日期", "date"]), "open": _to_float(_pick(row, ["开盘", "open"])), "close": _to_float(_pick(row, ["收盘", "close"])), "high": _to_float(_pick(row, ["最高", "high"])), "low": _to_float(_pick(row, ["最低", "low"])), "volume": _to_float(_pick(row, ["成交量", "volume"])), "amount": _to_float(_pick(row, ["成交额", "amount"])), } for row in rows ] return {"symbol": entity.symbol, "period": "daily", "adjust": "provider_default", "count": len(candles), "candles": candles} def market_snapshot(self) -> Dict[str, Any]: symbols = ["000001.SH", "399001.SZ", "399006.SZ", "000300.SH", "000688.SH", "899050.BJ"] chain: List[Dict[str, Any]] = [] data = {"indices": [], "breadth": None} warnings: List[str] = [] try: data["indices"] = self._quote_tencent(symbols) chain.append({"source": "tencent_finance", "ok": True}) except Exception as exc: chain.append({"source": "tencent_finance", "ok": False, "error": str(exc)}) warnings.append("主要指数获取失败: %s" % exc) try: data["breadth"] = self._market_breadth_eastmoney() chain.append({"source": "eastmoney_market_breadth", "ok": True}) except Exception as exc: chain.append({"source": "eastmoney_market_breadth", "ok": False, "error": str(exc)}) warnings.append("市场宽度获取失败: %s" % exc) if not data["indices"] and not data["breadth"]: raise SourceError("大盘快照所有免费源失败") return {"data": data, "source_chain": chain, "warnings": warnings} def _market_breadth_eastmoney(self) -> Dict[str, Any]: payload = self._json_get( "https://push2.eastmoney.com/api/qt/ulist.np/get", { "fltt": 2, "fields": "f104,f105,f106,f3,f6", "secids": "1.000001,0.399001,0.399006", }, referer="https://quote.eastmoney.com/", ) rows = payload.get("data", {}).get("diff") or [] up = sum(_to_int(row.get("f104")) or 0 for row in rows if isinstance(row, dict)) down = sum(_to_int(row.get("f105")) or 0 for row in rows if isinstance(row, dict)) flat = sum(_to_int(row.get("f106")) or 0 for row in rows if isinstance(row, dict)) return {"up_count": up, "down_count": down, "flat_count": flat} def rank(self, kind: str, limit: int, order: str = "desc") -> Dict[str, Any]: if order == "auto": order = "asc" if kind == "losers" else "desc" chain: List[Dict[str, Any]] = [] try: data = self._rank_eastmoney(kind, limit, order) chain.append({"source": "eastmoney_clist", "ok": True}) return {"data": data, "source_chain": chain, "warnings": []} except Exception as exc: chain.append({"source": "eastmoney_clist", "ok": False, "error": str(exc)}) try: data = self._rank_sina(kind, limit, order) chain.append({"source": "sina_market_center", "ok": True}) return {"data": data, "source_chain": chain, "warnings": ["已从东方财富排行回退到新浪公开接口"]} except Exception as exc: chain.append({"source": "sina_market_center", "ok": False, "error": str(exc)}) raise SourceError("榜单所有免费源失败") def _rank_eastmoney(self, kind: str, limit: int, order: str = "desc") -> Dict[str, Any]: config = { "gainers": "f3", "losers": "f3", "amount": "f6", "volume": "f5", "turnover": "f8", "volume-ratio": "f10", "amplitude": "f7", "market-cap": "f20", "pe": "f9", "pb": "f23", }.get(kind) if config is None: raise SourceError("unsupported rank kind: %s" % kind) fid = config po = 0 if order == "asc" else 1 params = { "pn": 1, "pz": max(1, min(limit, 200)), "po": po, "np": 1, "ut": EASTMONEY_UT, "fltt": 2, "invt": 2, "fid": fid, "fs": A_STOCK_FS, "fields": "f12,f13,f14,f2,f3,f4,f5,f6,f7,f8,f9,f10,f15,f16,f17,f18,f20,f21,f23", } payload = None last_error: Optional[Exception] = None for url in [ "https://push2.eastmoney.com/api/qt/clist/get", "https://push2delay.eastmoney.com/api/qt/clist/get", ]: try: payload = self._json_get(url, params, referer="https://quote.eastmoney.com/") break except Exception as exc: last_error = exc if payload is None: raise SourceError("东方财富排行失败: %s" % last_error) rows = payload.get("data", {}).get("diff") or [] items = [_parse_eastmoney_rank_row(row, index + 1) for index, row in enumerate(rows) if isinstance(row, dict)] if not items: raise SourceError("东方财富排行为空") return {"kind": kind, "order": order, "items": items, "returned_count": len(items), "total_count": payload.get("data", {}).get("total")} def _rank_sina(self, kind: str, limit: int, order: str = "desc") -> Dict[str, Any]: mapping = { "gainers": "changepercent", "losers": "changepercent", "amount": "amount", "volume": "volume", "turnover": "turnoverratio", "market-cap": "mktcap", "pe": "per", "pb": "pb", } if kind not in mapping: raise SourceError("sina does not support rank kind: %s" % kind) sort = mapping[kind] asc = 1 if order == "asc" else 0 response = self._get( "https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/Market_Center.getHQNodeData", {"page": 1, "num": max(1, min(limit, 100)), "sort": sort, "asc": asc, "node": "hs_a", "symbol": "", "_s_r_a": "page"}, referer="https://finance.sina.com.cn/", ) rows = json.loads(_decode_response(response)) items = [_parse_sina_rank_row(row, index + 1) for index, row in enumerate(rows) if isinstance(row, dict)] if not items: raise SourceError("新浪排行为空") return {"kind": kind, "order": order, "items": items, "returned_count": len(items)} def limit_pool(self, kind: str, query_date: Optional[str], limit: int) -> Dict[str, Any]: endpoint = {"up": "getTopicZTPool", "down": "getTopicDTPool", "broken": "getTopicZBPool", "strong": "getTopicQSPool"}.get(kind, "getTopicZTPool") sort = "fbt:asc" if kind in {"up", "down", "broken"} else "zdp:desc" date_text = (query_date or date.today().isoformat()).replace("-", "") payload = self._json_get( "https://push2ex.eastmoney.com/%s" % endpoint, {"ut": EASTMONEY_UT, "dpt": "wz.ztzt", "Pageindex": 0, "pagesize": max(1, min(limit, 500)), "sort": sort, "date": date_text}, referer="https://quote.eastmoney.com/ztb/detail", ) data = payload.get("data") or {} pool = data.get("pool") or [] items = [_parse_pool_row(row, index + 1) for index, row in enumerate(pool) if isinstance(row, dict)] return { "data": {"kind": kind, "query_date": date_text, "trade_date": str(data.get("qdate") or ""), "total_count": data.get("tc"), "items": items}, "source_chain": [{"source": "eastmoney_%s" % endpoint, "ok": True}], "warnings": [] if items else ["公开源返回空池，可能是非交易日或接口短暂不可用"], } def money_flow(self, scope: str, period: str, entity: Optional[Entity], limit: int) -> Dict[str, Any]: chain: List[Dict[str, Any]] = [] if scope == "market": try: data = self._money_flow_sina_market(limit) chain.append({"source": "sina_moneyflow", "ok": True}) return {"data": data, "source_chain": chain, "warnings": []} except Exception as exc: chain.append({"source": "sina_moneyflow", "ok": False, "error": str(exc)}) if scope == "stock": try: if entity is None: raise SourceError("个股资金流需要 symbol") data = self._money_flow_eastmoney_stock(entity, period, limit) chain.append({"source": "eastmoney_stock_moneyflow", "ok": True}) return {"data": data, "source_chain": chain, "warnings": []} except Exception as exc: chain.append({"source": "eastmoney_stock_moneyflow", "ok": False, "error": str(exc)}) try: data = self._money_flow_akshare(scope, period, entity, limit) chain.append({"source": "akshare_moneyflow", "ok": True}) return {"data": data, "source_chain": chain, "warnings": ["资金流为公开源 best-effort 数据"]} except Exception as exc: chain.append({"source": "akshare_moneyflow", "ok": False, "error": str(exc)}) try: if scope == "stock": data = self._money_flow_ths(scope, period, limit, entity) else: data = self._money_flow_ths(scope, period, limit) chain.append({"source": "akshare_ths_moneyflow", "ok": True}) return {"data": data, "source_chain": chain, "warnings": ["已从东方财富资金流回退到同花顺公开源"]} except Exception as exc: chain.append({"source": "akshare_ths_moneyflow", "ok": False, "error": str(exc)}) if scope == "stock": return { "data": {"scope": scope, "symbol": entity.symbol if entity else None, "period": period, "items": [], "returned_count": 0}, "source_chain": chain, "warnings": ["个股资金流公开源暂时不可用，已返回空列表和失败详情"], } raise SourceError("资金流公开源失败") def _money_flow_sina_market(self, limit: int) -> Dict[str, Any]: response = self._get( "https://vip.stock.finance.sina.com.cn/quotes_service/api/json_v2.php/MoneyFlow.ssl_bkzj_ssggzj", {"page": 1, "num": max(1, min(limit, 100)), "sort": "netamount", "asc": 0}, referer="https://finance.sina.com.cn/", ) rows = json.loads(_decode_response(response)) return {"scope": "market", "period": "instant", "items": [_parse_sina_money_row(row, i + 1) for i, row in enumerate(rows) if isinstance(row, dict)]} def _money_flow_eastmoney_stock(self, entity: Entity, period: str, limit: int) -> Dict[str, Any]: secid = "%s.%s" % (1 if entity.market == "SH" else 0, entity.code) try: payload = self._json_get( "https://push2his.eastmoney.com/api/qt/stock/fflow/kline/get", { "lmt": 0, "klt": 101, "secid": secid, "fields1": "f1,f2,f3,f7", "fields2": "f51,f52,f53,f54,f55,f56", "ut": "b2884a393a59ad64002292a3e90d46a5", }, referer="https://quote.eastmoney.com/", ) source_data = payload.get("data") or {} rows = [_parse_eastmoney_stock_money_flow_line(line) for line in source_data.get("klines") or []] rows = [row for row in rows if row] if not rows: raise SourceError("东方财富个股资金流为空") lookback = {"instant": 1, "3d": 3, "5d": 5, "10d": 10, "20d": 20}.get(period, max(1, min(limit, 120))) selected = rows[-lookback:] summary = { "main_net_inflow": sum(_to_float(row.get("main_net_inflow")) or 0 for row in selected), "small_net_inflow": sum(_to_float(row.get("small_net_inflow")) or 0 for row in selected), "medium_net_inflow": sum(_to_float(row.get("medium_net_inflow")) or 0 for row in selected), "large_net_inflow": sum(_to_float(row.get("large_net_inflow")) or 0 for row in selected), "super_large_net_inflow": sum(_to_float(row.get("super_large_net_inflow")) or 0 for row in selected), } return { "scope": "stock", "symbol": entity.symbol, "name": source_data.get("name") or entity.name, "period": period, "items": selected, "returned_count": len(selected), "summary": summary, } except Exception as exc: if period == "20d": raise return self._money_flow_eastmoney_stock_rank(entity, period, limit, str(exc)) def _money_flow_eastmoney_stock_rank(self, entity: Entity, period: str, limit: int, fallback_reason: str) -> Dict[str, Any]: config = { "instant": ("f62", "f12,f14,f2,f3,f62,f184,f66,f69,f72,f75,f78,f81,f84,f87,f204,f205,f124", "今日"), "3d": ("f267", "f12,f14,f2,f127,f267,f268,f269,f270,f271,f272,f273,f274,f275,f276,f257,f258,f124", "3日"), "5d": ("f164", "f12,f14,f2,f109,f164,f165,f166,f167,f168,f169,f170,f171,f172,f173,f257,f258,f124", "5日"), "10d": ("f174", "f12,f14,f2,f160,f174,f175,f176,f177,f178,f179,f180,f181,f182,f183,f260,f261,f124", "10日"), }.get(period) if config is None: raise SourceError("东方财富排行式个股资金流不支持周期 %s: %s" % (period, fallback_reason)) fid, fields, label = config base_params = { "fid": fid, "po": "1", "pz": "100", "pn": "1", "np": "1", "fltt": "2", "invt": "2", "ut": "b2884a393a59ad64002292a3e90d46a5", "fs": "m:0+t:6+f:!2,m:0+t:13+f:!2,m:0+t:80+f:!2,m:1+t:2+f:!2,m:1+t:23+f:!2,m:0+t:7+f:!2,m:1+t:3+f:!2", "fields": fields, } total = 6000 matched: List[Dict[str, Any]] = [] for page in range(1, 61): params = dict(base_params) params["pn"] = str(page) payload = self._json_get("https://push2delay.eastmoney.com/api/qt/clist/get", params, referer="https://data.eastmoney.com/") data = payload.get("data") or {} rows = data.get("diff") or [] total = _to_int(data.get("total")) or total for row in rows: if isinstance(row, dict) and str(row.get("f12")) == entity.code: matched.append(_parse_eastmoney_stock_money_flow_rank_row(row, period, label)) if matched or page * 100 >= total or not rows: break if not matched: raise SourceError("东方财富排行式个股资金流未匹配到 %s: %s" % (entity.code, fallback_reason)) return { "scope": "stock", "symbol": entity.symbol, "name": matched[0].get("name") or entity.name, "period": period, "items": matched[:limit], "returned_count": len(matched[:limit]), "source_note": "东方财富资金流排行公开接口；历史明细接口失败后按排行页匹配个股", } def _money_flow_akshare(self, scope: str, period: str, entity: Optional[Entity], limit: int) -> Dict[str, Any]: import akshare as ak if scope == "stock": if entity is None: raise SourceError("个股资金流需要 symbol") df = _quiet_call(ak.stock_individual_fund_flow, stock=entity.code, market=entity.market.lower()) return {"scope": scope, "symbol": entity.symbol, "period": period, "items": _df_to_records(df)[-limit:]} indicator = {"instant": "今日", "3d": "3日", "5d": "5日", "10d": "10日", "20d": "20日"}.get(period, "今日") sector_type = "行业资金流" if scope == "industry" else "概念资金流" df = _quiet_call(ak.stock_sector_fund_flow_rank, indicator=indicator, sector_type=sector_type) return {"scope": scope, "period": period, "items": _df_to_records(df)[:limit]} def _money_flow_ths(self, scope: str, period: str, limit: int, entity: Optional[Entity] = None) -> Dict[str, Any]: import akshare as ak symbol = {"instant": "即时", "3d": "3日排行", "5d": "5日排行", "10d": "10日排行", "20d": "20日排行"}.get(period, "即时") if scope == "stock": if entity is None: raise SourceError("个股资金流需要 symbol") try: matched = self._money_flow_ths_stock_by_code(period, entity, limit) except Exception: rows = _df_to_records(_quiet_call(ak.stock_fund_flow_individual, symbol=symbol)) matched = [row for row in rows if _row_matches_code(row, entity.code)] if not matched: raise SourceError("同花顺个股资金流未匹配到 %s" % entity.code) return { "scope": scope, "symbol": entity.symbol, "period": period, "items": matched[:limit], "returned_count": len(matched[:limit]), "source_note": "同花顺资金流公开页", } if scope not in {"industry", "concept"}: raise SourceError("同花顺资金流兜底仅支持个股/行业/概念") func = ak.stock_fund_flow_industry if scope == "industry" else ak.stock_fund_flow_concept rows = _df_to_records(_quiet_call(func, symbol=symbol)) return {"scope": scope, "period": period, "items": rows[:limit], "source_note": "同花顺资金流公开页"} def _money_flow_ths_stock_by_code(self, period: str, entity: Entity, limit: int) -> List[Dict[str, Any]]: import pandas as pd import py_mini_racer from akshare.stock_feature.stock_fund_flow import _get_file_content_ths board = {"3d": "3", "5d": "5", "10d": "10", "20d": "20"}.get(period) path = "board/%s/field/code/order/asc/page/{}/ajax/1/free/1/" % board if board else "field/code/order/asc/page/{}/ajax/1/free/1/" url_template = "http://data.10jqka.com.cn/funds/ggzjl/%s" % path js_code = py_mini_racer.MiniRacer() js_code.eval(_get_file_content_ths("ths.js")) headers = { "Accept": "text/html, */*; q=0.01", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Cache-Control": "no-cache", "hexin-v": js_code.call("v"), "Host": "data.10jqka.com.cn", "Referer": "http://data.10jqka.com.cn/funds/hyzjl/", "User-Agent": USER_AGENT, "X-Requested-With": "XMLHttpRequest", } cache: Dict[int, Tuple[List[Dict[str, Any]], str]] = {} def fetch_page(page: int) -> Tuple[List[Dict[str, Any]], str]: if page in cache: return cache[page] response = self.session.get(url_template.format(page), headers=headers, timeout=self.timeout) response.raise_for_status() text = _decode_response(response) tables = pd.read_html(io.StringIO(text)) rows = _df_to_records(tables[0]) if tables else [] cache[page] = (rows, text) return cache[page] rows, text = fetch_page(1) total_pages = _extract_ths_total_pages(text) or 120 target = int(entity.code) low, high = 1, total_pages while low <= high: page = (low + high) // 2 rows, _ = fetch_page(page) codes = sorted(_row_code_as_int(row) for row in rows if _row_code_as_int(row) is not None) if not codes: break matched = [row for row in rows if _row_matches_code(row, entity.code)] if matched: return matched[:limit] if target < codes[0]: high = page - 1 elif target > codes[-1]: low = page + 1 else: break return [] def sector(self, kind: str, action: str, entity: Optional[Entity], query: str, limit: int) -> Dict[str, Any]: chain: List[Dict[str, Any]] = [] try: data = self._sector_akshare(kind, action, entity, query, limit) chain.append({"source": "akshare_sector", "ok": True}) return {"data": data, "source_chain": chain, "warnings": []} except Exception as exc: chain.append({"source": "akshare_sector", "ok": False, "error": str(exc)}) try: data = self._sector_ths(kind, action, entity, query, limit) chain.append({"source": "akshare_ths_or_sina_sector", "ok": True}) return {"data": data, "source_chain": chain, "warnings": ["已从东方财富板块源回退到同花顺/新浪公开源"]} except Exception as exc: chain.append({"source": "akshare_ths_or_sina_sector", "ok": False, "error": str(exc)}) if action == "constituents": return { "data": {"kind": kind, "action": action, "board": _extract_board_name(query), "items": [], "returned_count": 0}, "source_chain": chain, "warnings": ["板块成分股公开源暂时不可用或未匹配到板块，已返回空列表和失败详情"], } raise SourceError("板块公开源失败: %s" % exc) def _sector_akshare(self, kind: str, action: str, entity: Optional[Entity], query: str, limit: int) -> Dict[str, Any]: import akshare as ak if action == "rank": func = ak.stock_board_concept_name_em if kind == "concept" else ak.stock_board_industry_name_em return {"kind": kind, "action": action, "items": _df_to_records(_quiet_call(func))[:limit]} if action == "constituents": name = _extract_board_name(query) if not name: raise SourceError("缺少板块名称") func = ak.stock_board_concept_cons_em if kind == "concept" else ak.stock_board_industry_cons_em return {"kind": kind, "action": action, "board": name, "items": _df_to_records(_quiet_call(func, symbol=name))[:limit]} if action == "belong": if entity is None: raise SourceError("所属板块需要股票代码") import efinance as ef df = _quiet_call(ef.stock.get_belong_board, entity.code) return {"kind": kind, "action": action, "symbol": entity.symbol, "items": _df_to_records(df)[:limit]} raise SourceError("unknown sector action") def _sector_ths(self, kind: str, action: str, entity: Optional[Entity], query: str, limit: int) -> Dict[str, Any]: import akshare as ak import pandas as pd if action == "rank": if kind == "industry": df = _quiet_call(ak.stock_sector_spot, indicator="新浪行业") rows = _df_to_records(df) rows = sorted(rows, key=lambda row: _to_float(row.get("涨跌幅")) or -999999, reverse=True) return {"kind": kind, "action": action, "items": rows[:limit]} df = _quiet_call(ak.stock_board_concept_name_ths) rows = _df_to_records(df) return {"kind": kind, "action": action, "items": rows[:limit], "note": "同花顺概念板块兜底仅返回名称和代码"} if action == "constituents": name = _extract_board_name(query) if not name: raise SourceError("缺少板块名称") listing_func = ak.stock_board_concept_name_ths if kind == "concept" else ak.stock_board_industry_name_ths listing = _df_to_records(_quiet_call(listing_func)) matched = _match_board_row(listing, name) if not matched: raise SourceError("同花顺未找到板块：%s" % name) code = str(matched.get("code") or matched.get("代码") or "").strip() path = "gn" if kind == "concept" else "thshy" response = self._get( "https://q.10jqka.com.cn/%s/detail/code/%s/" % (path, code), referer="https://q.10jqka.com.cn/", ) tables = _quiet_call(pd.read_html, io.StringIO(_decode_response(response))) if not tables: raise SourceError("同花顺板块成分股表格为空") selected = tables[0] for table in tables: columns = {str(column) for column in getattr(table, "columns", [])} lowered_columns = {column.lower() for column in columns} if {"代码", "名称"}.issubset(columns) or {"code", "name"}.issubset(lowered_columns): selected = table break return {"kind": kind, "action": action, "board": matched.get("name") or matched.get("名称") or name, "items": _df_to_records(selected)[:limit]} if action == "belong": raise SourceError("同花顺兜底不支持个股所属板块") raise SourceError("unknown sector action") def fundamental(self, entity: Entity, pack: str) -> Dict[str, Any]: chain: List[Dict[str, Any]] = [] data: Dict[str, Any] = {"symbol": entity.symbol, "pack": pack} warnings: List[str] = [] try: data["quote_valuation"] = self.quote_realtime(entity)["data"] chain.append({"source": "tencent_quote_valuation", "ok": True}) except Exception as exc: chain.append({"source": "tencent_quote_valuation", "ok": False, "error": str(exc)}) warnings.append("估值快照失败: %s" % exc) try: data.update(self._fundamental_akshare(entity, pack)) chain.append({"source": "akshare_fundamental", "ok": True}) except Exception as exc: chain.append({"source": "akshare_fundamental", "ok": False, "error": str(exc)}) warnings.append("akshare 基本面 best-effort 失败: %s" % exc) if len(data) <= 2: raise SourceError("基本面公开源失败") return {"data": data, "source_chain": chain, "warnings": warnings} def _fundamental_akshare(self, entity: Entity, pack: str) -> Dict[str, Any]: import akshare as ak result: Dict[str, Any] = {} if pack in {"basic", "all"}: result["basic"] = _df_to_records(_quiet_call(ak.stock_individual_info_em, symbol=entity.code)) if pack in {"financials", "all"}: result["financial_abstract"] = _df_to_records(_quiet_call(ak.stock_financial_abstract, symbol=entity.code)) result["financial_indicator"] = _df_to_records(_quiet_call(ak.stock_financial_analysis_indicator, symbol=entity.code)) if pack in {"holders", "all"}: holder_count_history = _optional_ak_call(ak, "stock_zh_a_gdhs_detail_em", symbol=entity.code) main_holders = _optional_ak_call(ak, "stock_main_stock_holder", stock=entity.code) result["holder_count_history"] = _tail_records(holder_count_history, 40) result["main_holders"] = _latest_records(main_holders, ["截至日期", "报告期"], 20) result["holders_scope"] = { "holder_count_history": "latest_40_records", "main_holders": "latest_report_period_top_20", } if pack in {"dividend", "all"}: result["dividend"] = _optional_ak_call(ak, "stock_dividend_cninfo", symbol=entity.code) return result def announcement(self, entity: Optional[Entity], keyword: Optional[str], limit: int) -> Dict[str, Any]: chain: List[Dict[str, Any]] = [] try: data = self._announcement_cninfo(entity, keyword, limit) chain.append({"source": "cninfo_public", "ok": True}) return {"data": data, "source_chain": chain, "warnings": []} except Exception as exc: chain.append({"source": "cninfo_public", "ok": False, "error": str(exc)}) raise SourceError("公告公开源失败") def _announcement_cninfo(self, entity: Optional[Entity], keyword: Optional[str], limit: int) -> Dict[str, Any]: data = { "pageNum": 1, "pageSize": max(1, min(limit, 50)), "column": "szse", "tabName": "fulltext", "sortName": "", "sortType": "", "isHLtitle": "true", } if entity is not None: search_name = _canonical_entity_search_name(entity) search_parts = [search_name or entity.code] if keyword: search_parts.append(_announcement_search_keyword(keyword)) data["searchkey"] = " ".join(search_parts) elif keyword: data["searchkey"] = _announcement_search_keyword(keyword) payload = self._post("https://www.cninfo.com.cn/new/hisAnnouncement/query", data=data, referer="https://www.cninfo.com.cn/").json() rows = payload.get("announcements") or [] items = [] for row in rows: if not isinstance(row, dict): continue adjunct_url = row.get("adjunctUrl") items.append( { "title": _strip_html(row.get("announcementTitle")), "code": row.get("secCode"), "name": row.get("secName"), "announcement_time": row.get("announcementTime"), "announcement_id": row.get("announcementId"), "pdf_url": "https://static.cninfo.com.cn/%s" % adjunct_url if adjunct_url else None, } ) return {"symbol": entity.symbol if entity else None, "keyword": keyword, "items": items, "returned_count": len(items)} def dragon_tiger(self, query_date: Optional[str], entity: Optional[Entity], limit: int) -> Dict[str, Any]: import akshare as ak date_text = (query_date or date.today().isoformat()).replace("-", "") try: df = _quiet_call(ak.stock_lhb_detail_em, start_date=date_text, end_date=date_text) rows = _df_to_records(df) chain = [{"source": "akshare.stock_lhb_detail_em", "ok": True}] warnings_list = ["龙虎榜为公开源 best-effort 数据"] except Exception as exc: rows = [] chain = [{"source": "akshare.stock_lhb_detail_em", "ok": False, "error": str(exc)}] warnings_list = ["龙虎榜公开源暂时不可用，已返回空列表和失败详情"] if entity is not None: rows = [row for row in rows if entity.code in json.dumps(row, ensure_ascii=False, default=str)] return {"data": {"date": date_text, "symbol": entity.symbol if entity else None, "items": rows[:limit], "returned_count": min(len(rows), limit)}, "source_chain": chain, "warnings": warnings_list} def news(self, entity: Optional[Entity], keyword: Optional[str], kind: str, limit: int) -> Dict[str, Any]: import akshare as ak source = "akshare.stock_research_report_em" if kind == "research" else "akshare.stock_news_em" if entity is not None else "akshare.stock_news_main_cx" try: if kind == "research": if entity is None: raise SourceError("研报/评级需要股票代码或名称") df = _quiet_call(ak.stock_research_report_em, symbol=entity.code) elif entity is not None: df = _quiet_call(ak.stock_news_em, symbol=entity.code) else: df = _quiet_call(ak.stock_news_main_cx) rows = _df_to_records(df) chain = [{"source": source, "ok": True}] warnings_list = ["新闻/研报为公开源 best-effort 数据"] except Exception as exc: rows = [] chain = [{"source": source, "ok": False, "error": str(exc)}] warnings_list = ["新闻/研报公开源暂时不可用，已返回空列表和失败详情"] non_filter_keywords = {"新闻", "快讯", "消息", "资讯", "利好", "利空", "有雷", "爆雷", "研报", "评级", "目标价", "机构评级", "机构怎么看", "机构看法"} if keyword and keyword not in non_filter_keywords: rows = [row for row in rows if keyword in json.dumps(row, ensure_ascii=False, default=str)] return { "data": {"kind": kind, "symbol": entity.symbol if entity else None, "keyword": keyword, "items": rows[:limit], "returned_count": min(len(rows), limit)}, "source_chain": chain, "warnings": warnings_list, } def chip(self, entity: Entity, limit: int) -> Dict[str, Any]: import akshare as ak try: df = _quiet_call(ak.stock_cyq_em, symbol=entity.code) rows = _df_to_records(df) chain = [{"source": "akshare.stock_cyq_em", "ok": True}] warnings_list = ["筹码分布为公开源 best-effort 数据，字段和复权口径可能随公开源变化"] except Exception as exc: rows = [] chain = [{"source": "akshare.stock_cyq_em", "ok": False, "error": str(exc)}] warnings_list = ["筹码分布公开源暂时不可用，已返回空列表和失败详情"] return { "data": {"symbol": entity.symbol, "items": rows[-limit:], "returned_count": min(len(rows), limit)}, "source_chain": chain, "warnings": warnings_list, } def block_trade(self, query_date: Optional[str], entity: Optional[Entity], limit: int) -> Dict[str, Any]: import akshare as ak date_text = (query_date or date.today().isoformat()).replace("-", "") try: df = _quiet_call(ak.stock_dzjy_mrmx, symbol="A股", start_date=date_text, end_date=date_text) rows = _df_to_records(df) chain = [{"source": "akshare.stock_dzjy_mrmx", "ok": True}] warnings_list = ["大宗交易为公开源 best-effort 数据，非交易日可能返回空列表"] except Exception as exc: rows = [] chain = [{"source": "akshare.stock_dzjy_mrmx", "ok": False, "error": str(exc)}] warnings_list = ["大宗交易公开源暂时不可用，已返回空列表和失败详情"] if entity is not None: rows = [row for row in rows if entity.code in json.dumps(row, ensure_ascii=False, default=str)] return { "data": {"date": date_text, "symbol": entity.symbol if entity else None, "items": rows[:limit], "returned_count": min(len(rows), limit)}, "source_chain": chain, "warnings": warnings_list, } def margin_trading(self, query_date: Optional[str], entity: Optional[Entity], limit: int) -> Dict[str, Any]: import akshare as ak date_text = (query_date or date.today().isoformat()).replace("-", "") chain: List[Dict[str, Any]] = [] rows: List[Dict[str, Any]] = [] for source, func, kwargs in [ ("akshare.stock_margin_detail_sse", ak.stock_margin_detail_sse, {"date": date_text}), ("akshare.stock_margin_detail_szse", ak.stock_margin_detail_szse, {"date": date_text}), ]: try: part = _df_to_records(_quiet_call(func, **kwargs)) rows.extend(part) chain.append({"source": source, "ok": True}) except Exception as exc: chain.append({"source": source, "ok": False, "error": str(exc)}) if entity is not None: rows = [row for row in rows if entity.code in json.dumps(row, ensure_ascii=False, default=str)] if not rows and not any(item.get("ok") for item in chain): raise SourceError("融资融券公开源失败") return { "data": {"date": date_text, "symbol": entity.symbol if entity else None, "items": rows[:limit], "returned_count": min(len(rows), limit)}, "source_chain": chain, "warnings": ["融资融券为交易所公开源 best-effort 数据，深沪字段可能不完全一致"], } def bond(self, action: str, entity: Optional[Entity], limit: int, days: int) -> Dict[str, Any]: import akshare as ak if action in {"rank", "quote"}: df = _quiet_call(ak.bond_zh_hs_cov_spot) rows = _df_to_records(df) if entity is not None: rows = [row for row in rows if entity.code in json.dumps(row, ensure_ascii=False, default=str)] return {"data": {"action": action, "items": rows[:limit]}, "source_chain": [{"source": "akshare.bond_zh_hs_cov_spot", "ok": True}], "warnings": []} if action == "history": if entity is None: raise SourceError("可转债历史需要 symbol") chain: List[Dict[str, Any]] = [] warnings_list: List[str] = [] try: data = self._history_tencent(entity.symbol, days=days, period="daily", adjust="qfq") data["action"] = action data["asset_type"] = "bond" chain.append({"source": "tencent_fqkline", "ok": True}) return {"data": data, "source_chain": chain, "warnings": warnings_list} except Exception as exc: chain.append({"source": "tencent_fqkline", "ok": False, "error": str(exc)}) try: df = _quiet_call(ak.bond_zh_hs_cov_daily, symbol=entity.code) rows = _df_to_records(df)[-days:] chain.append({"source": "akshare.bond_zh_hs_cov_daily", "ok": True}) except Exception as exc: rows = [] chain.append({"source": "akshare.bond_zh_hs_cov_daily", "ok": False, "error": str(exc)}) warnings_list = ["可转债历史公开源暂时不可用，已返回空列表和失败详情"] return {"data": {"action": action, "symbol": entity.symbol, "items": rows, "returned_count": len(rows)}, "source_chain": chain, "warnings": warnings_list} raise SourceError("unsupported bond action") def _akshare_call(self, name: str, *args: Any, **kwargs: Any) -> List[Dict[str, Any]]: import akshare as ak func = getattr(ak, name) return _df_to_records(_quiet_call(func, *args, **kwargs)) def to_tencent_symbol(symbol: str) -> str: normalized = normalize_symbol(symbol) if not normalized: return symbol.lower() code, market = normalized.split(".", 1) return market.lower() + code def _decode_response(response: requests.Response) -> str: content = response.content for encoding in ("utf-8", "gb18030", "gbk"): try: return content.decode(encoding) except UnicodeDecodeError: continue return response.text def _json_unescape(text: str) -> str: if "\\u" not in text: return text try: return json.loads('"%s"' % text) except Exception: return text def _to_float(value: Any) -> Optional[float]: if value is None: return None if isinstance(value, (int, float)): return float(value) text = str(value).strip().replace(",", "") if text in {"", "-", "--", "None", "false"}: return None try: return float(text) except ValueError: return None def _to_int(value: Any) -> Optional[int]: number = _to_float(value) if number is None: return None return int(number) def _parse_tencent_quote(provider_symbol: str, fields: List[Any]) -> Dict[str, Any]: def field(index: int) -> str: return str(fields[index]).strip() if index < len(fields) else "" code = field(2) symbol = normalize_symbol("%s.%s" % (code, provider_symbol[:2])) or provider_symbol return { "symbol": symbol, "provider_symbol": provider_symbol, "code": code, "name": field(1), "latest": _to_float(field(3)), "previous_close": _to_float(field(4)), "open": _to_float(field(5)), "volume": _to_float(field(6)), "market_time": field(30), "change": _to_float(field(31)), "change_ratio": _to_float(field(32)), "high": _to_float(field(33)), "low": _to_float(field(34)), "amount": _to_float(field(37)), "turnover_ratio": _to_float(field(38)), "pe": _to_float(field(39)), "amplitude": _to_float(field(43)), "total_market_cap": _to_float(field(44)), "float_market_cap": _to_float(field(45)), "pb": _to_float(field(46)), "volume_ratio": _to_float(field(49)), } def _parse_kline_row(row: List[Any]) -> Dict[str, Any]: return { "date": row[0], "open": _to_float(row[1]), "close": _to_float(row[2]), "high": _to_float(row[3]), "low": _to_float(row[4]), "volume": _to_float(row[5]), "amount": _to_float(row[6]) if len(row) > 6 else None, } def _parse_eastmoney_rank_row(row: Dict[str, Any], rank: int) -> Dict[str, Any]: code = str(row.get("f12", "")).strip() symbol = normalize_symbol(code) or code return { "rank": rank, "symbol": symbol, "code": code, "name": row.get("f14"), "latest": _to_float(row.get("f2")), "change_ratio": _to_float(row.get("f3")), "change": _to_float(row.get("f4")), "volume": _to_float(row.get("f5")), "amount": _to_float(row.get("f6")), "amplitude": _to_float(row.get("f7")), "turnover_ratio": _to_float(row.get("f8")), "pe": _to_float(row.get("f9")), "volume_ratio": _to_float(row.get("f10")), "high": _to_float(row.get("f15")), "low": _to_float(row.get("f16")), "open": _to_float(row.get("f17")), "previous_close": _to_float(row.get("f18")), "total_market_cap": _to_float(row.get("f20")), "float_market_cap": _to_float(row.get("f21")), "pb": _to_float(row.get("f23")), } def _parse_sina_rank_row(row: Dict[str, Any], rank: int) -> Dict[str, Any]: return { "rank": rank, "symbol": row.get("symbol"), "code": row.get("code"), "name": row.get("name"), "latest": _to_float(row.get("trade")), "change": _to_float(row.get("pricechange")), "change_ratio": _to_float(row.get("changepercent")), "volume": _to_float(row.get("volume")), "amount": _to_float(row.get("amount")), "turnover_ratio": _to_float(row.get("turnoverratio")), "pe": _to_float(row.get("per")), "pb": _to_float(row.get("pb")), "tick_time": row.get("ticktime"), } def _parse_sina_money_row(row: Dict[str, Any], rank: int) -> Dict[str, Any]: symbol = row.get("symbol") return { "rank": rank, "symbol": symbol, "code": str(symbol)[-6:] if symbol else None, "name": row.get("name") or "", "latest": _to_float(row.get("trade")), "change_ratio": _to_float(row.get("changeratio")), "amount": _to_float(row.get("amount")), "net_amount": _to_float(row.get("netamount")), "main_net_amount": _to_float(row.get("r0_net")), "turnover": _to_float(row.get("turnover")), } def _parse_eastmoney_stock_money_flow_line(line: Any) -> Dict[str, Any]: parts = str(line).split(",") if len(parts) < 6: return {} return { "date": parts[0], "main_net_inflow": _to_float(parts[1]), "small_net_inflow": _to_float(parts[2]), "medium_net_inflow": _to_float(parts[3]), "large_net_inflow": _to_float(parts[4]), "super_large_net_inflow": _to_float(parts[5]), } def _parse_eastmoney_stock_money_flow_rank_row(row: Dict[str, Any], period: str, label: str) -> Dict[str, Any]: if period == "instant": return { "code": str(row.get("f12") or ""), "name": row.get("f14"), "latest": _to_float(row.get("f2")), "change_ratio": _to_float(row.get("f3")), "period": label, "main_net_inflow": _to_float(row.get("f62")), "main_net_inflow_ratio": _to_float(row.get("f184")), "super_large_net_inflow": _to_float(row.get("f66")), "super_large_net_inflow_ratio": _to_float(row.get("f69")), "large_net_inflow": _to_float(row.get("f72")), "large_net_inflow_ratio": _to_float(row.get("f75")), "medium_net_inflow": _to_float(row.get("f78")), "medium_net_inflow_ratio": _to_float(row.get("f81")), "small_net_inflow": _to_float(row.get("f84")), "small_net_inflow_ratio": _to_float(row.get("f87")), } field_map = { "3d": ("f127", "f267", "f268", "f269", "f270", "f271", "f272", "f273", "f274", "f275", "f276"), "5d": ("f109", "f164", "f165", "f166", "f167", "f168", "f169", "f170", "f171", "f172", "f173"), "10d": ("f160", "f174", "f175", "f176", "f177", "f178", "f179", "f180", "f181", "f182", "f183"), } change_key, main_key, main_ratio_key, super_key, super_ratio_key, large_key, large_ratio_key, medium_key, medium_ratio_key, small_key, small_ratio_key = field_map[period] return { "code": str(row.get("f12") or ""), "name": row.get("f14"), "latest": _to_float(row.get("f2")), "change_ratio": _to_float(row.get(change_key)), "period": label, "main_net_inflow": _to_float(row.get(main_key)), "main_net_inflow_ratio": _to_float(row.get(main_ratio_key)), "super_large_net_inflow": _to_float(row.get(super_key)), "super_large_net_inflow_ratio": _to_float(row.get(super_ratio_key)), "large_net_inflow": _to_float(row.get(large_key)), "large_net_inflow_ratio": _to_float(row.get(large_ratio_key)), "medium_net_inflow": _to_float(row.get(medium_key)), "medium_net_inflow_ratio": _to_float(row.get(medium_ratio_key)), "small_net_inflow": _to_float(row.get(small_key)), "small_net_inflow_ratio": _to_float(row.get(small_ratio_key)), } def _parse_pool_row(row: Dict[str, Any], rank: int) -> Dict[str, Any]: code = str(row.get("c", "")).strip() symbol = normalize_symbol(code) or code price_raw = _to_float(row.get("p")) stat = row.get("zttj") if isinstance(row.get("zttj"), dict) else {} return { "rank": rank, "symbol": symbol, "code": code, "name": row.get("n"), "latest": round(price_raw / 1000, 3) if price_raw is not None else None, "change_ratio": _to_float(row.get("zdp")), "amount": _to_float(row.get("amount")), "turnover_ratio": _to_float(row.get("hs")), "board_count": _to_int(row.get("lbc")), "first_limit_time": _format_hhmmss(row.get("fbt")), "last_limit_time": _format_hhmmss(row.get("lbt")), "sealed_fund": _to_float(row.get("fund")), "break_count": _to_int(row.get("zbc")), "sector": row.get("hybk"), "limit_stat_days": stat.get("days"), "limit_stat_count": stat.get("ct"), } def _format_hhmmss(value: Any) -> Optional[str]: number = _to_int(value) if number is None: return None text = "%06d" % number return "%s:%s:%s" % (text[0:2], text[2:4], text[4:6]) def _df_to_records(df: Any) -> List[Dict[str, Any]]: if df is None: return [] if isinstance(df, list): return df if hasattr(df, "to_dict"): return df.to_dict(orient="records") return [] def _find_row(rows: List[Dict[str, Any]], code: str) -> Optional[Dict[str, Any]]: for row in rows: if str(row.get("代码") or row.get("code") or row.get("股票代码") or "") == code: return row return None def _row_matches_code(row: Dict[str, Any], code: str) -> bool: for key in ["代码", "code", "股票代码", "证券代码"]: value = row.get(key) if value is None: continue text = str(value).strip() if text.endswith(".0"): text = text[:-2] if text.zfill(6) == code: return True return False def _row_code_as_int(row: Dict[str, Any]) -> Optional[int]: for key in ["代码", "code", "股票代码", "证券代码"]: value = row.get(key) if value is None: continue text = str(value).strip() if text.endswith(".0"): text = text[:-2] if text.isdigit(): return int(text) return None def _extract_ths_total_pages(text: str) -> Optional[int]: match = re.search(r'class=["\']page_info["\'][^>]*>\s*\d+\s*/\s*(\d+)', text) if match: return _to_int(match.group(1)) match = re.search(r">\s*\d+\s*/\s*(\d+)\s*<", text) if match: return _to_int(match.group(1)) return None def _normalize_akshare_quote(row: Dict[str, Any], entity: Entity) -> Dict[str, Any]: return { "symbol": entity.symbol, "code": entity.code, "name": _pick(row, ["名称", "股票名称", "name"]), "latest": _to_float(_pick(row, ["最新价", "最新", "price"])), "change_ratio": _to_float(_pick(row, ["涨跌幅", "change_ratio"])), "change": _to_float(_pick(row, ["涨跌额", "change"])), "volume": _to_float(_pick(row, ["成交量", "volume"])), "amount": _to_float(_pick(row, ["成交额", "amount"])), "turnover_ratio": _to_float(_pick(row, ["换手率", "turnover"])), "volume_ratio": _to_float(_pick(row, ["量比", "volume_ratio"])), "pe": _to_float(_pick(row, ["市盈率-动态", "市盈率", "pe"])), "pb": _to_float(_pick(row, ["市净率", "pb"])), } def _pick(row: Dict[str, Any], keys: Iterable[str]) -> Any: for key in keys: if key in row: return row[key] return None def _optional_ak_call(module: Any, name: str, **kwargs: Any) -> List[Dict[str, Any]]: func = getattr(module, name, None) if func is None: return [] try: return _df_to_records(_quiet_call(func, **kwargs)) except TypeError: return _df_to_records(_quiet_call(func)) def _tail_records(records: List[Dict[str, Any]], limit: int) -> List[Dict[str, Any]]: if limit <= 0: return [] return records[-limit:] def _latest_records(records: List[Dict[str, Any]], date_keys: Iterable[str], limit: int) -> List[Dict[str, Any]]: if not records or limit <= 0: return [] latest_value: Optional[str] = None for row in records: for key in date_keys: value = row.get(key) if value is None: continue text = str(value) if latest_value is None or text > latest_value: latest_value = text break if latest_value is None: return records[:limit] latest_rows = [ row for row in records if any(row.get(key) is not None and str(row.get(key)) == latest_value for key in date_keys) ] return latest_rows[:limit] def _strip_html(value: Any) -> str: text = str(value or "") return re.sub(r"<[^>]+>", "", text).strip() def _canonical_entity_search_name(entity: Entity) -> Optional[str]: canonical_by_code = { "600519": "贵州茅台", "300750": "宁德时代", "300059": "东方财富", "600036": "招商银行", "002594": "比亚迪", "688981": "中芯国际", "603288": "海天味业", "600031": "三一重工", "002714": "牧原股份", "603501": "韦尔股份", "600150": "中国船舶", "601127": "赛力斯", "601138": "工业富联", "300274": "阳光电源", "300308": "中际旭创", "601088": "中国神华", "002371": "北方华创", "688012": "中微公司", } return canonical_by_code.get(entity.code) or entity.name def _announcement_search_keyword(keyword: str) -> str: mapping = { "年报": "年度报告", "半年报": "半年度报告", "一季报": "第一季度报告", "三季报": "第三季度报告", } return mapping.get(keyword, keyword) def _extract_board_name(query: str) -> Optional[str]: text = re.sub(r"\s+", "", query or "") aliases = [ "低空经济", "人形机器人", "机器人", "CPO", "算力", "AI", "光伏", "猪肉", "券商", "证券", "银行", "白酒", "半导体", "新能源车", ] for alias in aliases: if alias.lower() in text.lower(): return alias for noise in [ "有哪些股票", "哪些股票", "有什么股票", "成分股", "板块", "行业", "概念", "包含", "排行", "排名", "有哪些", "哪些", "有啥票", "有啥股票", "都有谁", "都有啥", "方向", "这块", "这条线", "那条线", "谁在涨", "今天", "强不强", "咋样", "怎么样", "股票", ]: text = text.replace(noise, " ") text = re.sub(r"\s+", "", text) return text or None def _match_board_row(rows: List[Dict[str, Any]], name: str) -> Optional[Dict[str, Any]]: for row in rows: row_name = str(row.get("name") or row.get("名称") or row.get("板块名称") or "") if row_name == name: return row for row in rows: row_name = str(row.get("name") or row.get("名称") or row.get("板块名称") or "") if name in row_name or row_name in name: return row return None def _quiet_call(func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: sink = io.StringIO() proxy_names = ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy", "NO_PROXY", "no_proxy"] env = os.environ saved = {name: env.get(name) for name in proxy_names} for name in proxy_names: if name.lower() == "no_proxy": env[name] = "*" else: env.pop(name, None) try: with redirect_stdout(sink), redirect_stderr(sink): return func(*args, **kwargs) finally: for name, value in saved.items(): if value is None: env.pop(name, None) else: env[name] = value FILE:scripts/stockline_cli.py #!/usr/bin/env python3 # -*- coding: utf-8 -*- from __future__ import annotations from pathlib import Path import sys import warnings warnings.filterwarnings("ignore", message="urllib3 v2 only supports OpenSSL.*") RUNTIME_DIR = Path(__file__).resolve().parent / "runtime" if RUNTIME_DIR.is_dir(): runtime_path = str(RUNTIME_DIR) if runtime_path not in sys.path: sys.path.insert(0, runtime_path) from freestocklineskill_runtime.cli import main if __name__ == "__main__": raise SystemExit(main())

freeStockLIneskill

Comments (0)