3. 流动性挖矿工具：Python量化回测框架搭建、Wind/聚宽数据接口调用、实时行情监控
做信用债做市，说白了就是跟市场赛跑。你比别人快一秒看到机会，就能多吃一口肉。我刚开始做的时候，全靠手工盯盘，眼睛都快盯瞎了。后来我意识到，必须得用工具，用代码把重复劳动自动化掉。
这一章，我就跟你聊聊我的工具箱。不是什么高大上的东西，都是实战里摸爬滚打出来的。
3.1 量化回测框架搭建
回测是什么？就是拿历史数据模拟你的策略。我见过太多人，策略回测时赚翻了，实盘一跑就亏成狗。为什么？因为回测框架没搭对。
我个人习惯用 Python 搭一个轻量级的回测框架。别整那些复杂的，够用就行。
核心思路：事件驱动 + 数据流处理。说白了，就是模拟真实交易环境，一笔一笔地处理订单。
我分享一个我常用的骨架代码：
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class CreditBondBacktest:
"""信用债做市回测框架"""
def __init__(self, initial_capital=10000000):
self.capital = initial_capital # 初始资金
self.position = {} # 持仓
self.trades = [] # 交易记录
self.pnl = [] # 每日盈亏
def load_data(self, data_source='wind'):
"""加载历史数据"""
# 我一般从Wind拉取中债估值和成交数据
pass
def run_strategy(self, strategy_func):
"""运行策略"""
for date, market_data in self.data.iterrows():
signal = strategy_func(market_data, self.position)
if signal:
self.execute_trade(signal, market_data)
self.update_pnl(date)
def execute_trade(self, signal, market_data):
"""执行交易"""
# 这里要处理滑点、手续费等现实因素
pass
def update_pnl(self, date):
"""更新盈亏"""
pass
def get_performance(self):
"""输出绩效指标"""
pass我的经验：回测框架里一定要加「滑点模型」。信用债流动性差，你看到的成交价不一定能成交。我一开始没加，回测年化收益8%，实盘只有3%。加了滑点后，回测直接降到4%，这才对得上。
3.2 Wind/聚宽数据接口调用
数据是量化交易的命根子。信用债的数据尤其难搞，不像股票那么透明。
我主要用两个数据源：Wind 和 聚宽。
Wind 接口调用
Wind 在债券圈是标配。但它的 Python 接口有点坑，我踩过不少雷。
from WindPy import w
w.start()
# 获取信用债估值数据
def get_bond_valuation(bond_code, start_date, end_date):
"""拉取中债估值"""
data = w.wsd(bond_code,
"b_anal_maturity,b_ytm_b,carry_bp",
start_date, end_date,
"returnType=1")
return data.Data
# 获取成交数据
def get_bond_trade(bond_code, trade_date):
"""拉取银行间成交"""
data = w.wst(bond_code,
"rt_last,rt_volume",
trade_date, trade_date)
return data.Data避坑指南：我曾经因为 Wind 接口返回的数据格式问题，debug 了一整天。记住，Wind 返回的是元组，不是 DataFrame。建议你拿到数据后，第一时间转成 pandas 格式。
聚宽数据接口
聚宽的好处是免费，而且数据清洗得比较好。但信用债数据不如 Wind 全。
import jqdatasdk as jq
# 登录聚宽
jq.auth('your_account', 'your_password')
# 获取债券基础信息
def get_bond_info(bond_code):
"""获取债券基本信息"""
info = jq.get_security_info(bond_code)
return {
'code': bond_code,
'name': info.display_name,
'start_date': info.start_date,
'type': info.type
}
# 获取行情数据
def get_bond_price(bond_code, start_date, end_date):
"""获取日频行情"""
df = jq.get_price(bond_code,
start_date=start_date,
end_date=end_date,
frequency='daily',
fields=['close', 'volume', 'money'])
return df我的建议：两个数据源都接上。Wind 做主数据源，聚宽做备份和校验。我经常拿两个数据源交叉验证，发现过好几次数据异常。
3.3 实时行情监控
做市交易，实时行情就是你的眼睛。信用债的实时行情比股票难拿，因为银行间市场没有统一的交易所。
我搭建了一套实时监控系统，核心逻辑是这样的：
实时监控的代码，我习惯用 WebSocket 或者轮询的方式：
import time
import threading
from datetime import datetime
class RealTimeMonitor:
"""实时行情监控器"""
def __init__(self, bonds_list):
self.bonds = bonds_list # 监控的债券列表
self.alerts = [] # 报警记录
self.running = False
def start_monitor(self):
"""启动监控"""
self.running = True
thread = threading.Thread(target=self._monitor_loop)
thread.start()
print(f"[{datetime.now()}] 监控已启动，共监控 {len(self.bonds)} 只债券")
def _monitor_loop(self):
"""监控主循环"""
while self.running:
for bond in self.bonds:
# 获取实时行情
price = self._get_realtime_price(bond)
# 检查是否触发条件
if self._check_alert_condition(bond, price):
self._trigger_alert(bond, price)
time.sleep(1) # 每秒轮询一次
def _get_realtime_price(self, bond):
"""获取实时价格"""
# 这里调用 Wind 或聚宽的实时接口
pass
def _check_alert_condition(self, bond, price):
"""检查报警条件"""
# 比如：收益率偏离估值超过10bp
pass
def _trigger_alert(self, bond, price):
"""触发报警"""
alert_msg = f"债券 {bond} 触发报警，当前价格 {price}"
self.alerts.append(alert_msg)
print(f"[ALERT] {alert_msg}")
# 可以发邮件、钉钉通知等核心要点：实时监控不是看价格涨跌，而是看「偏离度」。信用债的合理价格区间很窄，一旦偏离超过你的阈值，就是机会或者风险。
我的经验：监控频率别设太高。信用债不像股票，1秒刷一次就够了。我曾经设了毫秒级刷新，结果把 Wind 接口给封了...嗯，后来学乖了。
好了，工具就这些。说白了，回测框架帮你验证想法，数据接口给你弹药，实时监控让你抓住机会。这三样东西搭好了，你的做市交易就有了坚实的基础。
下一章，我会聊聊具体的做市策略，怎么用这些工具赚钱。
公众号:蓝海资料掘金营，微信deep3321