pytdx获取f10信息

以下是重构后的专业级代码，包含清晰的函数划分、完整的字段说明文档，以及健壮的数据获取逻辑，适用于量化研究、基本面分析等场景。

✅ 一、完整可运行代码（模块化设计）

-- coding: utf-8 --
"""
通达信 F10 公司信息获取工具
支持：查询目录、分段读取、全文拼接、保存文件
作者：miniqmt
日期：2025-12-04
"""

from pytdx.hq import TdxHq_API
from pytdx.params import TDXParams
import os
import time
import traceback
默认行情服务器（深圳双子星）
DEFAULT_HOST = '218.6.170.87'
DEFAULT_PORT = 7709
MAX_CHUNK_SIZE = 60000 # 单次安全读取上限（字节）

def connect_api(host=DEFAULT_HOST, port=DEFAULT_PORT):
"""
创建并连接 TDX 行情 API 实例

Returns:
TdxHq_API: 已连接的 API 对象（需用 with 管理）
"""
api = TdxHq_API()
api.connect(host, port)
return api

def get_company_info_category(api, market, stock_code):
"""
查询公司信息目录（F10 分段列表）

Args:
api (TdxHq_API): 已连接的 API 实例
market (int): 市场代码（TDXParams.MARKET_SZ / MARKET_SH）
stock_code (str): 股票代码，如 '000001'

Returns:
list[dict]: 分段信息列表，每个元素包含：
'filename' (str): 文件名（通常为 {code}.txt）
'start' (int): 内容起始偏移（字节）
'length' (int): 段长度（字节）
'file_type' (int, optional): 文件类型（常缺失）
'info' (str, optional): 附加信息（常为空）
"""
return api.get_company_info_category(market, stock_code)

def read_company_info_segment(api, market, stock_code, filename, start, length):
"""
安全读取一段公司信息内容（自动分块 + 编码处理）

Args:
api (TdxHq_API): 已连接的 API 实例
market (int): 市场代码
stock_code (str): 股票代码
filename (str): 文件名（如 '000001.txt'）
start (int): 起始偏移
length (int): 要读取的总长度

Returns:
str: 该段的完整文本内容（UTF-8 字符串），若失败则返回空字符串
"""
if length <= 0:
return ""

parts = []
offset = 0
while offset < length:
read_size = min(MAX_CHUNK_SIZE, length - offset)
raw = api.get_company_info_content(market, stock_code, filename, start + offset, read_size)

if raw is None:
# 可选：加短暂延迟后重试（此处直接跳过）
offset += read_size
continue

# 统一转为字符串
if isinstance(raw, bytes):
try:
text = raw.decode('gbk')
except UnicodeDecodeError:
text = raw.decode('utf-8', errors='replace')
else:
text = str(raw)

parts.append(text)
offset += read_size
# 可选：time.sleep(0.01) # 防止请求过快

return ''.join(parts)

def fetch_full_f10_text(api, market, stock_code):
"""
获取完整 F10 公司信息全文

Args:
api (TdxHq_API): 已连接的 API 实例
market (int): 市场代码
stock_code (str): 股票代码

Returns:
tuple[str, int]: (完整文本, 总字符数)
"""
categories = get_company_info_category(api, market, stock_code)
if not categories:
return "", 0

full_parts = []
for i, seg in enumerate(categories):
print(f" ➤ 读取分段 [{i+1}/{len(categories)}]: 起始={seg['start']}, 长度={seg['length']}")
text = read_company_info_segment(
api, market, stock_code,
seg['filename'], seg['start'], seg['length']
)
full_parts.append(text)

full_text = ''.join(full_parts)
return full_text, len(full_text)

def save_text_to_file(text, filepath, encoding='utf-8'):
"""
将文本保存到文件

Args:
text (str): 要保存的文本
filepath (str): 目标文件路径
encoding (str): 文件编码，默认 UTF-8
"""
os.makedirs(os.path.dirname(filepath), exist_ok=True)
with open(filepath, 'w', encoding=encoding) as f:
f.write(text)

def get_company_f10(stock_code, market=TDXParams.MARKET_SZ, output_dir="./f10_data"):
"""
主函数：获取指定股票的完整 F10 信息并保存

Args:
stock_code (str): 股票代码，如 '000001'
market (int): 市场（默认深圳）
output_dir (str): 输出目录

Returns:
str: 保存的文件绝对路径
"""
print(f"🔍 开始获取 {stock_code} 的 F10 公司信息...")
with connect_api() as api:
# 1. 获取目录
categories = get_company_info_category(api, market, stock_code)
if not categories:
raise ValueError(f"未找到 {stock_code} 的公司信息（仅深圳股票支持）")

print(f"✅ 发现 {len(categories)} 个数据分段")

# 2. 获取完整文本
full_text, char_count = fetch_full_f10_text(api, market, stock_code)
print(f"✅ 全文获取完成，共 {char_count:,} 字符")

# 3. 保存文件
filepath = os.path.join(output_dir, f"{stock_code}_F10_full.txt")
save_text_to_file(full_text, filepath)
print(f"📁 已保存至: {os.path.abspath(filepath)}")

return os.path.abspath(filepath)
========================
示例与测试入口
========================
if __name__ == '__main__':
try:
# 获取平安银行 F10
file_path = get_company_f10('000001')

# 可选：打印前 500 字符预览
with open(file_path, 'r', encoding='utf-8') as f:
preview = f.read(500)
print("\n📄 预览开头：")
print(preview + ("..." if len(preview) == 500 else ""))

except Exception as e:
print("❌ 执行失败:")
traceback.print_exc()

📚 二、关键数据结构与字段说明

1. get_company_info_category() 返回结构

字段名 类型 说明 示例

filename str 逻辑文件名（所有段相同） “000001.txt”
start int 该段在文件中的起始字节偏移 0, 36941
length int 该段的字节长度 36941
file_type int (可选) 文件类型（通常缺失或为 1） 1
info str (可选) 附加描述（通常为空） “”
🔔 注意：
所有分段属于同一个逻辑文件（如 000001.txt）
必须按 start 顺序拼接才能得到完整 F10 文本
仅深圳市场（MARKET_SZ）有效，上海股票通常返回空列表

1. F10 文本典型内容结构（通达信格式）

F10 是纯文本，包含以下标准模块（顺序固定）：

【1. 最新提示】
【2. 互动问答】
【3. 最新公告】
【4. 最新报道】
【5. 基本资料】 ← 公司全称、注册地址、法人代表、联系电话等
【6. 主营业务】 ← 业务范围、产品分类
【7. 财务分析】 ← 利润表、资产负债表摘要（含每股收益、净资产等）
【8. 股东研究】 ← 十大股东、股本结构、限售解禁
【9. 资本运作】 ← 增发、并购、分红送股记录
【10. 行业分析】
【11. 风险提示】
【12. 重大事项】
💡 关键字段示例（可用于正则提取）：
每股收益(元) → 1.8700
每股净资产(元) → 15.2300
实际控制人 → 中国平安保险(集团)股份有限公司
所属行业 → 银行

🛠 三、使用说明
安装依赖
bash
pip install pytdx
基本用法
python
获取单只股票 F10
file_path = get_company_f10(‘000001’) # 平安银行
自定义输出目录
get_company_f10(‘000858’, output_dir=’./my_f10’)
批量获取（示例）
python
stocks = [‘000001’, ‘000858’, ‘002415’]
for code in stocks:
try:
get_company_f10(code)
time.sleep(0.5) # 避免请求过快
except Exception as e:
print(f”跳过 {code}: {e}”)

⚠️ 四、注意事项

1. 仅支持深圳股票（000/002/003/300 开头）
2. 服务器限制：
   单次读取 ≤ 64KB，否则返回 None
   高频请求可能被临时拒绝
3. 编码问题：
   优先使用 GBK 解码（通达信标准）
   失败时回退到 UTF-8
4. 内容非结构化：
   F10 是纯文本，需自行解析（如用正则表达式）