1. 项目概述从零构建理杏仁数据获取工具做可转债投资的朋友都知道数据是决策的基石。无论是分析转股溢价率、纯债溢价率还是监控强赎、下修条款没有及时、准确的数据一切分析都是空中楼阁。市面上虽然有一些付费的数据终端但对于个人投资者或者小团队来说成本不菲而且API调用往往有诸多限制。今天要聊的就是如何利用Python自己动手搭建一个稳定、高效的理杏仁数据获取工具。这不仅仅是写几行爬虫代码那么简单它涉及到对理杏仁网站结构的逆向工程、应对反爬策略、设计合理的数据存储结构以及构建一个可持续维护的代码框架。我把自己在实际操作中踩过的坑、总结的技巧以及完整的代码实现思路分享出来希望能帮你省下大量摸索的时间。2. 核心思路与技术选型解析2.1 为什么选择理杏仁作为数据源在动手之前首先要明确数据源的选择。理杏仁在可转债数据方面有几个突出的优势数据维度相对齐全更新较为及时并且其免费版提供了足够个人投资者使用的数据深度。与一些仅提供基础行情的数据源不同理杏仁包含了诸如转股价、修正条款触发情况、剩余规模、信用评级等关键字段这些都是可转债量化分析不可或缺的。当然它的网页结构并不简单数据也并非通过公开的API直接提供这恰恰是我们需要攻克的技术点。选择自己抓取而非依赖第三方封装库最大的好处是可控性——你可以完全自定义需要的数据字段、更新频率和存储格式后续进行数据清洗和分析的灵活性极高。2.2 整体技术架构设计这个项目的目标不是写一个一次性的脚本而是构建一个健壮的数据管道。我的设计核心思路是“模块化”和“容错性”。整个流程可以拆解为以下几个核心模块请求与会话管理模块负责模拟浏览器行为维持登录状态处理Cookie和Token并应对网站可能存在的反爬机制如请求频率限制、验证码等。数据定位与解析模块这是核心中的核心。需要精准地分析理杏仁数据页面的HTML结构或异步加载的JSON数据接口找到目标数据的“藏身之处”并用稳定的方法将其提取出来。数据清洗与格式化模块从网页上抓取下来的数据往往是文本格式夹杂着百分号、逗号等字符。这个模块负责将字符串转换为标准的数值、日期等Python数据类型并处理缺失值或异常值。持久化存储模块将处理好的数据保存下来。根据数据量和使用场景可以选择CSV文件、SQLite数据库或者更专业的时序数据库如InfluxDB。我倾向于使用SQLite它轻量、无需单独服务非常适合个人项目。调度与日志模块实现定时自动运行并记录每次运行的成功与否、抓取了哪些数据、遇到了什么错误便于后期监控和排查问题。基于这个架构技术选型就很明确了Requests库处理HTTP请求BeautifulSoup4或lxml解析HTMLPandas进行数据清洗和转换SQLAlchemy或直接使用sqlite3进行数据库操作schedule或APScheduler实现定时任务标准库的logging记录日志。注意直接对商业网站进行高频抓取可能违反其服务条款。在实际操作中务必在请求头中设置合理的User-Agent并在请求间添加随机延时例如time.sleep(random.uniform(1, 3))模拟人类操作行为避免对目标服务器造成压力这也是基本的网络礼仪。3. 关键环节实现与代码拆解3.1 逆向工程定位数据接口这是最具挑战性的一步。理杏仁的数据页面通常不是简单的静态HTML很多数据是通过JavaScript异步加载的。直接解析初始HTML往往找不到数据。方法一分析网络请求推荐打开Chrome开发者工具F12切换到“Network”网络选项卡然后刷新或打开理杏仁上某个可转债的详情页例如“湖广转债”。在所有的网络请求中仔细寻找类型为XHR或Fetch的请求。这些请求通常返回JSON格式的数据。通过查看请求的Preview预览和Headers标头我们可以找到真实的数据接口API Endpoint和请求所需的参数如查询参数、请求体、特定的Headers如Authorization等。例如你可能会发现一个类似https://www.lixinger.com/api/.../bond/...的请求其响应体就是一个结构清晰的JSON包含了所有我们需要的数据。如果能找到这样的接口那么数据获取就变成了一个标准的API调用问题比解析HTML要稳定和高效得多。方法二解析动态渲染的HTML如果找不到清晰的JSON接口或者接口参数过于复杂那么可能需要解析最终渲染的HTML。此时BeautifulSoup仍然是好帮手但你需要确保你获取的HTML是JavaScript执行完毕后的“最终状态”。对于简单的页面requests库获取的HTML可能就足够了。但对于复杂页面你可能需要借助Selenium或Playwright这类浏览器自动化工具来模拟用户操作等待数据加载完成后再获取页面源码。这种方法资源消耗大、速度慢但能应对绝大多数复杂场景。假设我们通过方法一找到了一个返回债券基本信息的JSON接口。核心的请求代码可能如下所示import requests import pandas as pd import time import random def fetch_bond_data(bond_code): 根据债券代码获取理杏仁债券数据 # 1. 配置请求头模拟浏览器 headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36, Accept: application/json, text/plain, */*, Accept-Language: zh-CN,zh;q0.9,en;q0.8, # 注意如果接口需要认证可能需要添加Referer、Authorization或其他自定义Header # Referer: https://www.lixinger.com/analytics/stock/bond/, } # 2. 构建请求URL和参数这里需要根据实际找到的接口调整 # 这是一个假设的URL实际需要替换为真实的接口地址和参数 url fhttps://www.lixinger.com/api/analytics/stock/bond/{bond_code}/overview params { timestamp: int(time.time() * 1000), # 常见的防缓存参数 } try: # 3. 发送GET请求 response requests.get(url, headersheaders, paramsparams, timeout10) response.raise_for_status() # 如果状态码不是200抛出HTTPError异常 # 4. 解析JSON响应 data_json response.json() # 假设返回的数据在 data 字段下 raw_data data_json.get(data, {}) # 5. 数据提取与转换 # 这里需要根据实际返回的JSON结构来映射字段 bond_info { 代码: bond_code, 名称: raw_data.get(name), 现价: float(raw_data.get(price, 0)), # 转换为浮点数 转股价: float(raw_data.get(conversionPrice, 0)), 转股价值: float(raw_data.get(conversionValue, 0)), 溢价率: float(raw_data.get(premiumRate, 0).rstrip(%)) if raw_data.get(premiumRate) else None, # 去除百分号 更新日期: pd.to_datetime(raw_data.get(updateTime), unitms) if raw_data.get(updateTime) else None, # 处理时间戳 } return bond_info except requests.exceptions.RequestException as e: print(f请求失败 for {bond_code}: {e}) return None except (KeyError, ValueError) as e: print(f数据解析失败 for {bond_code}: {e}) return None # 使用示例 if __name__ __main__: # 假设我们要获取123456示例代码这只可转债的数据 result fetch_bond_data(123456) if result: print(pd.DataFrame([result])) # 重要请求间添加延时避免被封IP time.sleep(random.uniform(2, 5))3.2 数据清洗与规整从接口获取的原始数据往往不能直接使用。例如溢价率字段可能是“23.45%”这样的字符串我们需要将其转换为浮点数0.2345市值可能是“125.6亿”需要识别“亿”、“万”等单位并转换为统一的数值如元。日期字段也可能有多种格式时间戳、ISO格式字符串等。我通常会专门写一个data_cleaner.py模块里面包含一系列清洗函数import re def clean_percentage(value): 清洗百分比字符串如 15.6% - 0.156 if not value or pd.isna(value): return None if isinstance(value, str): # 移除逗号、空格和百分号并转换为浮点数 num_str value.replace(,, ).replace( , ).rstrip(%) try: return float(num_str) / 100 except ValueError: return None return float(value) / 100 if isinstance(value, (int, float)) else None def clean_market_cap(value): 清洗市值字符串如 125.6亿 - 12560000000 (元) if not value or pd.isna(value): return None if isinstance(value, str): match re.search(r([\d\.])([万亿]?), value) if match: num, unit match.groups() num float(num) multiplier {: 1, 万: 1e4, 亿: 1e8, 万亿: 1e12} return int(num * multiplier.get(unit, 1)) return value # 在主流程中应用清洗 bond_info[溢价率_清洗后] clean_percentage(bond_info.get(溢价率)) bond_info[市值_清洗后] clean_market_cap(bond_info.get(市值))3.3 数据存储策略对于可转债这类时序数据存储设计要考虑历史追溯和增量更新。我的方案是使用SQLite为每只转债创建一个单独的表或者用一个大的表存储所有转债用“债券代码”和“日期”作为联合主键。import sqlite3 from datetime import datetime def init_database(db_pathconvertible_bonds.db): conn sqlite3.connect(db_path) cursor conn.cursor() # 创建数据表 cursor.execute( CREATE TABLE IF NOT EXISTS bond_daily_data ( bond_code TEXT, date DATE, price REAL, conversion_price REAL, conversion_value REAL, premium_rate REAL, volume INTEGER, PRIMARY KEY (bond_code, date) ) ) conn.commit() conn.close() def save_to_database(data_dict, db_pathconvertible_bonds.db): 将单条数据字典存入数据库 conn sqlite3.connect(db_path) cursor conn.cursor() # 假设data_dict包含所有需要的字段 # 使用INSERT OR REPLACE来更新当日数据 cursor.execute( INSERT OR REPLACE INTO bond_daily_data (bond_code, date, price, conversion_price, conversion_value, premium_rate, volume) VALUES (?, ?, ?, ?, ?, ?, ?) , ( data_dict[代码], datetime.now().date().isoformat(), # 使用当天日期 data_dict[现价], data_dict[转股价], data_dict[转股价值], data_dict[溢价率_清洗后], data_dict.get(成交量, 0) )) conn.commit() conn.close()4. 完整工作流与自动化调度将上述模块组合起来就形成了一个完整的工作流。我通常会创建一个main.py作为入口一个config.py存放配置如数据库路径、需要监控的债券代码列表、请求间隔等。config.py示例# 需要监控的可转债代码列表 BOND_CODES [113050, 123456, 128034, ...] # 替换为真实的代码 # 数据库路径 DB_PATH data/convertible_bonds.db # 请求配置 REQUEST_HEADERS {...} REQUEST_DELAY (1, 5) # 随机延时范围单位秒main.py工作流示例import time import random from datetime import datetime import logging from config import BOND_CODES, DB_PATH, REQUEST_DELAY from data_fetcher import fetch_bond_data # 假设将抓取函数放在此模块 from data_cleaner import clean_percentage, clean_market_cap from database import save_to_database # 配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(data_fetch.log), logging.StreamHandler() ] ) def daily_fetch_task(): 每日数据抓取任务 logging.info(开始每日数据抓取任务...) successful 0 failed 0 for code in BOND_CODES: try: logging.info(f正在处理债券代码: {code}) raw_data fetch_bond_data(code) if raw_data: # 数据清洗 raw_data[溢价率_清洗后] clean_percentage(raw_data.get(溢价率)) # ... 其他字段清洗 # 数据存储 save_to_database(raw_data, DB_PATH) successful 1 logging.info(f债券 {code} 数据抓取并保存成功。) else: failed 1 logging.warning(f债券 {code} 数据抓取失败。) except Exception as e: failed 1 logging.error(f处理债券 {code} 时发生未知错误: {e}, exc_infoTrue) # 关键请求间隔避免被封 delay random.uniform(*REQUEST_DELAY) time.sleep(delay) logging.info(f每日任务完成。成功: {successful}, 失败: {failed}) if __name__ __main__: # 手动执行一次 # daily_fetch_task() # 或者使用schedule库定时执行例如每个交易日15:30后运行 import schedule schedule.every().day.at(15:35).do(daily_fetch_task) logging.info(定时任务调度器已启动...) while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次5. 实战中遇到的典型问题与解决方案在实际运行这套系统的过程中我遇到了不少问题这里把几个典型的坑和解决办法记录下来。5.1 接口变更与稳定性维护理杏仁的网页结构和接口并非一成不变。这是自建数据抓取工具最大的维护成本。问题某天突然发现脚本报错返回的数据为空或结构不对。排查首先检查网络连接和基础请求是否正常状态码是否为200。如果正常立刻打开开发者工具重新检查之前找到的那个数据接口是否还在请求参数和响应格式是否发生了变化。很可能接口URL、必需的请求头如新增了Token或返回的JSON结构已经更新。解决方案防御性编程在解析JSON时大量使用.get()方法而不是直接键值访问并为关键字段设置默认值。监控与告警在日志中记录每次抓取的数据条数或关键字段是否存在。可以设置一个简单的监控如果连续多次抓取到的数据量异常比如为0就发送邮件或钉钉消息告警。代码版本管理将解析网页结构的核心代码如CSS选择器、XPath或JSON字段映射关系集中放在一个配置模块里。一旦需要修改只需更新这个模块而不必到处找代码。5.2 反爬机制与请求限制即使我们礼貌地添加了延时仍然可能触发网站的反爬。问题脚本运行一段时间后返回403错误或收到包含验证码的页面。现象IP地址可能被暂时限制访问。解决方案降低请求频率这是最重要的。将REQUEST_DELAY调得更大比如(3, 10)秒。对于可转债数据日级更新完全不需要高频抓取。模拟更真实的浏览器完善headers包括Accept-Language、Referer通常设置为目标网站的上一级页面甚至模拟完整的Cookie链如果网站需要登录。使用代理IP池高级如果数据量需求极大可以考虑使用付费的代理IP服务在代码中轮换使用不同的IP。但这对于个人投资者获取日级数据来说通常杀鸡用牛刀了。尊重robots.txt检查目标网站的robots.txt文件通常在网站根目录如https://www.lixinger.com/robots.txt了解哪些路径是允许爬虫访问的。5.3 数据质量校验抓取到的数据可能有错漏直接入库会影响后续分析。问题价格数据为负数转股价为0或者日期字段是乱码。解决方案在数据清洗模块之后增加一个数据校验步骤。def validate_bond_data(data): 简单的数据校验 errors [] if data.get(现价, 0) 0: errors.append(f价格异常: {data.get(现价)}) if data.get(转股价, 0) 0: errors.append(f转股价异常: {data.get(转股价)}) # 溢价率通常在-50%到200%之间超出范围需警惕 premium data.get(溢价率_清洗后) if premium is not None and (premium -0.5 or premium 2.0): errors.append(f溢价率异常: {premium}) return errors # 在保存前校验 validation_errors validate_bond_data(cleaned_data) if validation_errors: logging.warning(f数据校验未通过 {bond_code}: {validation_errors}) # 可以选择丢弃、标记或人工复核 else: save_to_database(cleaned_data)5.4 网络异常与重试机制网络请求总有可能失败。问题偶发的网络超时、连接断开。解决方案为请求函数添加重试机制。可以使用tenacity库也可以自己简单实现。import requests from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def fetch_bond_data_with_retry(bond_code): 带重试的数据抓取函数 return fetch_bond_data(bond_code) # 调用之前定义的函数这个装饰器会在函数抛出异常时重试最多3次每次重试的等待时间指数级增加4秒8秒...避免在服务器临时故障时雪上加霜。6. 项目扩展与进阶思考当基础的数据获取管道稳定运行后你可以在此基础上做很多扩展让这个工具发挥更大价值。1. 数据可视化与监控面板使用Dash、Streamlit或Grafana将数据库中的数据可视化。创建一个简单的Web面板实时展示所有关注转债的关键指标溢价率分布、价格走势、双低排名等一目了然。2. 策略信号提醒在数据抓取和清洗后加入策略判断逻辑。例如计算“双低值”价格溢价率*100对排名前10的转债进行标记或者监控即将满足强赎条件的转债。当满足特定条件时通过邮件、钉钉机器人或Telegram Bot发送提醒。3. 历史数据回补如果中途才开始运行这个脚本会缺少历史数据。可以尝试分析理杏仁是否提供历史K线或历史指标数据的接口通常参数中会包含开始日期和结束日期编写一个回补脚本按天或按周获取历史数据填充数据库。4. 容器化与云部署为了让脚本在后台稳定、长期运行可以将其打包成Docker镜像部署到家里的树莓派、旧电脑或者云服务器如腾讯云轻量应用服务器上。使用systemd或supervisor来管理进程确保脚本在意外退出后能自动重启。5. 代码优化与性能当监控的转债数量很多时比如全市场几百只串行抓取会非常慢。可以考虑使用asyncioaiohttp进行异步并发请求或者使用concurrent.futures的线程池大幅缩短数据获取的总时间。但切记并发请求会极大增加对目标服务器的压力必须更加谨慎地控制并发数和请求间隔否则极易被封IP。构建这样一个数据获取工具的过程本身就是一个极佳的学习项目。它迫使你去理解HTTP协议、解析HTML/JSON、设计数据库、处理异常、编写可维护的代码最后还能得到一个对自己投资有实际帮助的成果。我的体会是初期把基础打牢模块之间耦合度尽量低后面无论是扩展功能还是修复问题都会轻松很多。最后一个小建议所有代码和配置一定要用Git管理起来每一次重要的修改都做好记录这是保证项目能长期维护下去的好习惯。