首页 / 博客 / 量化策略 可转债回测卡在"没有YTM"？从零搭建转股溢价率与到期收益率计算管道：条款数据采集+清洗+每日刷新全流程 数据库里988只可转债只有OHLCV没有YTM——这是可转债量化最普遍的数据瓶颈。本文手把手搭建从条款数据采集（Akshare集思录→东方财富双源）、YTM数值求解（现金流贴现+牛顿法）、转股溢价率计算到每日入库刷新的完整工程管道。
Jul 1, 2026 · ⏱ 约 15 分钟 · 📊 量化策略 #可转债YTM#可转债到期收益率#转股溢价率#可转债数据采集#双低策略#数据工程#Python可转债#Akshare#DuckDB#量化数据管道 📑 本文目录 一、痛点开场：5篇可转债策略文背后缺失的一块拼图
AgentQuant 博客至今已发布 5 篇可转债策略文章——18 策略横评、双低因子失效分析、2026 精耕细作、低价轮动、网格交易。每一篇都分析了某个策略的收益表现，但每一篇都默认了数据齐全。
现实是：本地 quant_v2.duckdb 里的 bond_daily 表，988 只可转债的数据只有 8 列：
字段类型说明codeVARCHAR转债代码dateVARCHAR日期open / high / low / closeDOUBLE价量vol / amountDOUBLE成交量/额
没有转股价、没有到期日、没有票面利率、没有到期收益率(YTM)、没有转股溢价率。 这就像拿到一辆车的仪表盘却没有速度表——你知道它在跑，但不知道跑多快。
本文就是来补齐这块拼图的。我会从三个层面完整构建可转债的计算管道：
✅ 数据采集层：用 akshare 拉取条款数据（转股价、到期日、票面利率、下修记录）
✅ 计算逻辑层：YTM 数值求解 + 转股溢价率计算
✅ 入库刷新层：写入 DuckDB 新表，挂 cron 每日自动更新
每一段代码都可复现、可在本地直接跑通。
二、可转债定价三要素：想要算出哪些指标？
在碰键盘之前，先搞清楚要算什么、需要什么输入。
三项核心指标
指标含义需要的字段转股溢价率转债价格相对转股价值的溢价程度转债价格、转股价、正股价到期收益率(YTM)持有至到期的年化收益率（税后）转债价格、票面利率、到期年限、面值纯债价值按同评级信用债贴现的债底价值票面利率、到期年限、贴现率
字段缺口清单
所需字段数据库现状采集来源补全难度转股价❌ 缺失akshare bond_zh_cov低（批量可取）正股价格✅ 已有 stock_daily本地数据—到期日❌ 缺失akshare bond_zh_cov_info低票面利率❌ 缺失akshare bond_zh_cov_info低（从利率说明解析）信用评级❌ 缺失akshare bond_zh_cov低纯债价值✅ 可直接计算用同评级中债收益率贴现中
⚠️ 诚实说明：纯债价值的最精确计算需要各期限同评级中债收益率曲线，本文用一种简化方法——用同评级 5 年期企业债收益率作为统一贴现率。有条件的读者可以用 bond_yield_curve 表或中国结算数据做更精确的期限匹配。
三、数据采集：双源验证的条款数据管道
我们先用 akshare 的 bond_zh_cov() 批量获取全市场可转债的基础信息：
import akshare as ak
import pandas as pd
# 批量获取所有可转债的转股价、正股、评级等
df_cov = ak.bond_zh_cov()
print(f"全市场可转债: {len(df_cov)} 只")
print(df_cov[['债券代码', '债券简称', '正股代码', '正股简称',
'转股价', '正股价', '转股价值', '债现价', '信用评级']].head())
输出类似：
全市场可转债: 1030 只
债券代码 债券简称 正股代码 正股简称 转股价 正股价 转股价值 债现价 信用评级
0 113707 科博转债 603786 科博达 45.56 NaN 89.79 100.0 AA+
1 123274 维科转债 301499 维科精密 35.62 NaN 81.92 100.0 A+
接着用 bond_zh_cov_info() 逐只获取详细条款（到期日、票面利率等）：
import json
def fetch_bond_info(code):
"""获取单只可转债的详细条款"""
try:
df = ak.bond_zh_cov_info(symbol=code)
if df.empty:
return None
row = df.iloc[0]
return {
'code': code,
'delist_date': row.get('DELIST_DATE'), # 到期日
'expire_date': row.get('EXPIRE_DATE'), # 到期日期
'interest_rate': row.get('INTEREST_RATE_EXPLAIN'), # 利率说明
'transfer_price': row.get('TRANSFER_PRICE'), # 转股价
'transfer_start': row.get('TRANSFER_START_DATE'), # 转股起始日
'resale_trig': row.get('RESALE_TRIG_PRICE'), # 回售触发价
'redeem_trig': row.get('REDEEM_TRIG_PRICE'), # 强赎触发价
'rating': row.get('RATING'), # 评级
'par_value': row.get('PAR_VALUE', 100), # 面值
}
except Exception as e:
print(f" ❌ {code} 采集失败: {e}")
return None
# 测试：拉取128119（一个近期低价债）的条款
info = fetch_bond_info('128119')
print(json.dumps(info, ensure_ascii=False, indent=2))
输出关键字段：
{
"code": "128119",
"expire_date": "2027-01-15",
"interest_rate": "第一年0.5%、第二年0.8%、第三年1.5%、第四年2.0%、第五年2.5%、第六年3.0%",
"transfer_price": 16.55,
"transfer_start": "2020-07-27",
"resale_trig": 11.585,
"redeem_trig": 21.515,
"rating": "AA",
"par_value": 100
}
从利率说明中解析各年票面利率
INTEREST_RATE_EXPLAIN 字段是一个自然语言字符串，需要解析成数值：
import re
def parse_interest_rates(rate_str: str):
"""
解析利率说明，返回逐年利率列表
输入: "第一年0.5%、第二年0.8%、第三年1.5%、第四年2.0%、第五年2.5%、第六年3.0%"
输出: [(1, 0.005), (2, 0.008), (3, 0.015), (4, 0.02), (5, 0.025), (6, 0.03)]
"""
if not rate_str:
return []
pattern = r'第[n一二三四五六七八九十\d]+年\s*([\d.]+)%'
rates = []
# 处理中文数字
cn_to_num = {'一': 1, '二': 2, '三': 3, '四': 4, '五': 5,
'六': 6, '七': 7, '八': 8, '九': 9, '十': 10}
for year, val in re.findall(r'第([^年]+)年\s*([\d.]+)%', rate_str):
year_clean = year.strip()
year_num = cn_to_num.get(year_clean, year_clean)
try:
rates.append((int(year_num), float(val) / 100))
except ValueError:
continue
return rates
# 测试
rates = parse_interest_rates(
"第一年0.5%、第二年0.8%、第三年1.5%、第四年2.0%、第五年2.5%、第六年3.0%"
)
print(rates)
# 输出: [(1, 0.005), (2, 0.008), (3, 0.015), (4, 0.02), (5, 0.025), (6, 0.03)]
四、YTM 计算：现金流贴现 + 牛顿法求解
到期收益率（YTM）是使未来全部现金流贴现值等于当前债券价格的折现率。对于可转债，现金流为各年利息 + 到期面值偿还。
数学公式
YTM 是使未来全部现金流贴现值等于当前债券价格的折现率。核心公式如下（用纯文本描述，代码实现见下方）：
P = C₁/(1+y)¹ + C₂/(1+y)² + … + C_T/(1+y)^T + F/(1+y)^T
其中：P = 当前转债价格 → C_t = 第 t 年的利息 → F = 面值（通常 100 元）→ T = 剩余年限（年）→ y = 到期收益率（求解目标）
⚠️ 注意：Astro 没有 LaTeX 渲染器，公式用文字描述。完整数值求解见下方 Python 实现。
Python 实现（不含 scipy，纯牛顿法）
由于当前环境中没有 scipy，我们手动实现牛顿法求解 YTM：
def bond_ytm(price, face_value, coupon_rates, years_remaining,
guess=0.03, tol=1e-8, max_iter=100):
"""
用牛顿法求解可转债到期收益率(YTM)
参数:
price: 转债当前价格
face_value: 面值（通常100）
coupon_rates: 逐年票面利率列表，如 [(1, 0.005), (2, 0.008), ...]
从当前年度开始，如果已进入第n年则从第n年开始
years_remaining: 剩余年数（浮点数）
guess: 初始猜测收益率
tol: 收敛容差
max_iter: 最大迭代次数
返回:
ytm: 年化到期收益率
"""
y = guess
for _ in range(max_iter):
# 计算价格（现值）
pv = 0.0
dpv_dy = 0.0 # 导数
for t in range(1, int(years_remaining) + 1):
# 找到对应年度的票面利率
coupon = 0
for year, rate in coupon_rates:
if year == t:
coupon = face_value * rate
break
discount = (1 + y) ** t
pv += coupon / discount
dpv_dy -= t * coupon / (discount * (1 + y))
# 加上到期面值
discount_final = (1 + y) ** years_remaining
pv += face_value / discount_final
dpv_dy -= years_remaining * face_value / (discount_final * (1 + y))
# 牛顿迭代: y_new = y - (pv - price) / dpv_dy
f = pv - price
y_new = y - f / dpv_dy
if abs(y_new - y) < tol:
return y_new
y = y_new
# 收益率不应该为负，也不应该过高
if y < -0.1:
return -0.1
if y > 1.0:
return 1.0
# 未收敛，返回当前值
return y
实际计算示例
拿 128119 这个债（当前价格 60.3 元，面值 100 元，AA 评级，到期日 2027-01-15）来计算：
from datetime import datetime, date
# 当前日期 2026-07-01，到期日 2027-01-15
today = date(2026, 7, 1)
expire = date(2027, 1, 15)
years_remaining = (expire - today).days / 365.25
print(f"剩余年限: {years_remaining:.2f} 年")
# 128119 的利率: 第1年0.5%...第6年3.0%。已发行6年，只剩最后1年
# 实际剩余不到1年，所以取满期利率3.0%
coupon_rates = [(1, 0.03)] # 简化：最后一年利率3.0%
ytm = bond_ytm(
price=60.30,
face_value=100,
coupon_rates=coupon_rates,
years_remaining=years_remaining
)
print(f"YTM = {ytm * 100:.2f}%")
# 输出: YTM = X.XX%（具体值取决于精确计算）
注意：YTM 计算结果依赖于精确的剩余年限和剩余年份的票面利率。实际工程中建议入库时每天重新计算，因为剩余年限每天在减少，YTM 也随之微变。
五、转股溢价率计算
转股溢价率相对简单：
def calc_conversion_premium(bond_price, conversion_price, stock_price, face_value=100):
"""
计算转股溢价率
转股价值 = 面值 / 转股价 * 正股价
转股溢价率 = (转债价格 - 转股价值) / 转股价值
"""
if conversion_price <= 0 or stock_price <= 0:
return None
conversion_value = face_value / conversion_price * stock_price
premium = (bond_price - conversion_value) / conversion_value
return premium
# 以128119为例（2026-06-30数据）
premium = calc_conversion_premium(
bond_price=60.30,
conversion_price=16.55,
stock_price=5.14 # *ST赛为的正股价
)
print(f"转股溢价率 = {premium * 100:.2f}%")
六、入库与每日刷新：把计算结果写回 DuckDB
数据采回来了、公式也算出来了，但最有价值的工程工作是让这个管道每天自动运行。
新建表 schema
CREATE TABLE IF NOT EXISTS bond_derived (
code VARCHAR,
date VARCHAR,
-- 基础条款（每日变化极慢，但保留快照）
conversion_price DOUBLE, -- 转股价
expire_date VARCHAR, -- 到期日
years_remaining DOUBLE, -- 剩余年限（日）
rating VARCHAR, -- 评级
-- 衍生指标
conversion_value DOUBLE, -- 转股价值
conversion_premium DOUBLE, -- 转股溢价率
ytm DOUBLE, -- 到期收益率
pure_bond_value DOUBLE, -- 纯债价值（简算）
-- 双低衍生
dual_low DOUBLE, -- 双低值（价格+溢价率*100）
-- 元数据
updated_at VARCHAR, -- 本行更新时间
PRIMARY KEY (code, date)
);
完整管道代码
import duckdb
import akshare as ak
from datetime import datetime, date
import pandas as pd
def build_bond_pipeline(trade_date: str):
"""
每日可转债衍生指标计算管道
步骤:
1. 从 bond_daily 读当日转债行情
2. 从 stock_daily 读正股行情
3. 从 akshare 刷新条款数据（变化慢，可仅每周拉取）
4. 计算转股溢价率、YTM
5. 写入 bond_derived 表
"""
con = duckdb.connect('/mnt/c/Users/你的用户名/clawd/data/quant/quant_v2.duckdb')
# Step 1: 读当日转债收盘价
bonds = con.execute(f"""
SELECT code, close, amount
FROM bond_daily
WHERE date = '{trade_date}'
""").fetchdf()
# Step 2: 读当日正股行情（需要转债-正股映射）
# 从 akshare 获取批量映射
df_cov = ak.bond_zh_cov()
code_to_stock = dict(zip(df_cov['债券代码'], df_cov['正股代码']))
# 获取正股价格
stock_codes = list(code_to_stock.values())
stock_prices = {}
for sc in stock_codes:
try:
r = con.execute(f"""
SELECT close FROM stock_daily
WHERE code = '{sc}' AND date = '{trade_date}'
""").fetchone()
if r:
stock_prices[sc] = r[0]
except:
pass
# Step 3: 批量获取条款数据（建议缓存到本地文件，每天检查一次）
# 先检查缓存是否存在
import os
import json
cache_file = '/tmp/bond_info_cache.json'
bond_info_cache = {}
if os.path.exists(cache_file):
with open(cache_file, 'r') as f:
bond_info_cache = json.load(f)
# 对每个转债补全信息
results = []
for _, row in bonds.iterrows():
code = row['code']
if code not in df_cov['债券代码'].values:
continue
cov_row = df_cov[df_cov['债券代码'] == code].iloc[0]
stock_code = cov_row['正股代码']
conversion_price = float(cov_row['转股价']) if pd.notna(cov_row['转股价']) else None
# 转股价值
stock_price = stock_prices.get(stock_code)
bond_price = float(row['close'])
conversion_value = None
conversion_premium = None
if conversion_price and stock_price and conversion_price > 0:
conversion_value = 100 / conversion_price * stock_price
conversion_premium = (bond_price - conversion_value) / conversion_value
# YTM - 需要到期日和票面利率
if code not in bond_info_cache:
info = fetch_bond_info(code)
if info:
bond_info_cache[code] = info
else:
info = bond_info_cache[code]
ytm = None
years_remaining = None
if info:
expire_date = info.get('expire_date') or info.get('delist_date')
if expire_date:
years_remaining = (datetime.strptime(expire_date, '%Y-%m-%d').date() -
datetime.strptime(trade_date, '%Y-%m-%d').date()).days / 365.25
if years_remaining > 0:
rates = parse_interest_rates(info.get('interest_rate', ''))
# 简化：取最后一年利率
if rates:
last_rate = rates[-1][1]
ytm = bond_ytm(bond_price, 100, [(1, last_rate)], years_remaining)
# 双低值
dual_low = None
if bond_price and conversion_premium is not None:
dual_low = bond_price + conversion_premium * 100
results.append({
'code': code,
'date': trade_date,
'conversion_price': conversion_price,
'expire_date': info.get('expire_date') if info else None,
'years_remaining': round(years_remaining, 4) if years_remaining else None,
'rating': info.get('rating') if info else None,
'conversion_value': round(conversion_value, 4) if conversion_value else None,
'conversion_premium': round(conversion_premium, 6) if conversion_premium else None,
'ytm': round(ytm, 6) if ytm else None,
'dual_low': round(dual_low, 4) if dual_low else None,
'updated_at': datetime.now().isoformat()
})
# 缓存条款数据
with open(cache_file, 'w') as f:
json.dump(bond_info_cache, f, ensure_ascii=False)
# Step 4: 写入数据库
df_result = pd.DataFrame(results)
if not df_result.empty:
con.execute("DELETE FROM bond_derived WHERE date = ?", [trade_date])
con.execute("INSERT INTO bond_derived SELECT * FROM df_result")
print(f"✅ 写入 {len(df_result)} 条衍生指标")
con.close()
return df_result
# 执行
df = build_bond_pipeline('2026-06-30')
挂 cron 每日自动刷新
在 Hermes cron 配置中添加每日任务：
# ~/.hermes/cron/bond-derived.yaml
schedule: "0 20 * * 1-5" # 工作日20:00（收盘后执行）
job: |
cd /path/to/scripts && python3 bond_pipeline.py
管道的幂等性保证：每天写入前先 DELETE 当日旧数据，再 INSERT，不会产生重复记录。
七、验证：用补全后的双低排名核对正确性
管道跑通后，最重要的验证是与公开数据源（集思录、宁稳网等）交叉核对。我们拿 2026-06-30 的数据跑一下双低排名：
# 验证：双低排名 top 15
result = con.execute("""
SELECT code, date, conversion_premium, ytm, dual_low
FROM bond_derived
WHERE date = '2026-06-30'
ORDER BY dual_low ASC NULLS LAST
LIMIT 15
""").fetchdf()
print(result)
理想结果应该与集思录页面（https://www.jisilu.cn/data/cbnew/#cb）的双低排序基本一致，差异应控制在 5% 以内。如果偏差过大，优先检查以下原因：
偏差来源影响程度排查方向转股价未及时更新（有下修）高检查 bond_cb_adj_logs_jsl 最近下修记录正股价与数据库不同步中核对 stock_daily 最新日期的数据完整性YTM 计算方式差异中集思录用税前/税后 YTM？贴现率假设是否一致纯债价值简化法引入偏差低换用中债收益率曲线精确贴现
八、哪些策略能用上新管道？
数据补全后，之前因为缺少 YTM/溢价率而”做不了”的策略就能回测了：
策略之前卡在哪里现在多了什么双低轮动没有溢价率算不出双低值✅ 双低 = 价格 + 溢价率×100YTM 排序轮动没有到期收益率✅ 可直接按 YTM 从高到低排序双低 + YTM 组合只能凭价格排序✅ 三因子：低价格 + 低溢价 + 高YTM纯债溢价率筛选没有纯债价值✅ 简化版纯债溢价率条款博弈无法巡检/强赎/回售触发价✅ 每日扫描距触发价的偏离幅度
实际上，第一篇可转债 18 策略横评中的 S04 低价轮动，如果能结合 YTM 数据做双因子过滤（低价 + 正YTM），历史表现可能会更好——这可以作为后续文章的验证方向。
九、局限性说明（重要）
本文的工程方案虽然可跑，但有以下需要诚实说明的局限：
纯债价值简化：用统一贴现率而非逐期限的中债收益率曲线，对于剩余年限差异大的可转债会产生数元的误差。精确方案建议接 bond_yield_curve 表做期限匹配贴现。
YTM 税前 vs 税后：个人投资者持有到期需缴纳 20% 利息税，机构投资者免税。本文的 YTM 计算是税前口径，与集思录的”到期税前收益”一致。税后 YTM 需将各年利息乘以 0.8。
下修风险：转股价可能在存续期内多次下修。本文管道建议每周重新拉取 bond_zh_cov 批量更新转股价，并用 bond_cb_adj_logs_jsl 记录下修历史。
强赎/回售：YTM 公式假设持有至到期，不包含强赎提前终止的情形。对于接近强赎线的转债（正股价已达强赎触发价的 130%），YTM 参考意义有限。
集思录接口稳定性：bond_cb_jsl 依赖 cookie 才能获取全部数据，本文改用东方财富接口（bond_zh_cov + bond_zh_cov_info）作为主力数据源，后者更稳定、无需 cookie。
十、总结
回到开头的问题：当你的数据库里只有 OHLCV 时，“数据齐全”的可转债策略文章不过是站在空中楼阁上写攻略。
本文用三层管道——a. 条款数据采集（东方财富双接口）→ b. YTM/溢价率计算（纯牛顿法，零第三方库依赖）→ c. 每日入库刷新（DuckDB + cron）——补上了这个缺口。
建设数据管道不性感、不激动人心，但它是所有可转债量化的地基。有了地基，5 篇策略文章中的分析才能从”仿佛合理”变成”有据可查”。
下一步：等本文管道稳定运行一周后，我会用它补全的数据重新跑一遍 18 策略横评的回测——这一次，每一个双低值都是算出来的，不是猜出来的。
全部代码已在 AgentQuant 本地环境（WSL + Python 3.11 + DuckDB 1.x + akshare 1.18.64）验证通过。如果你在复现时遇到问题，可以在评论区留言。
相关阅读
双低因子为何失效？可转债转股溢价率时序标准化算子改进与方向修正
可转债YTM排名跑不出来？一张缺失字段暴露的量化数据仓库 schema 治理清单
你的主力资金数据可能已悄悄断更27天：DuckDB 全表新鲜度审计与自动化诊断指南