可转债回测卡在"没有YTM"？从零搭建转股溢价率与到期收益率计算管道：条款数据采集+清洗+每日刷新全流程
一、痛点开场：5篇可转债策略文背后缺失的一块拼图
AgentQuant 博客至今已发布 5 篇可转债策略文章——18 策略横评、双低因子失效分析、2026 精耕细作、低价轮动、网格交易。每一篇都分析了某个策略的收益表现，但每一篇都默认了数据齐全。
现实是：本地 quant_v2.duckdb 里的 bond_daily 表，988 只可转债的数据只有 8 列：
| 字段 | 类型 | 说明 |
|---|---|---|
code | VARCHAR | 转债代码 |
date | VARCHAR | 日期 |
open / high / low / close | DOUBLE | 价量 |
vol / amount | DOUBLE | 成交量/额 |
没有转股价、没有到期日、没有票面利率、没有到期收益率(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/Administrator/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/溢价率而”做不了”的策略就能回测了：
| 策略 | 之前卡在哪里 | 现在多了什么 |
|---|---|---|
| 双低轮动 | 没有溢价率算不出双低值 | ✅ 双低 = 价格 + 溢价率×100 |
| YTM 排序轮动 | 没有到期收益率 | ✅ 可直接按 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）验证通过。如果你在复现时遇到问题，可以在评论区留言。