脉动数据行情官网地址：http://39.107.99.235:1008/market
联系QQ： 3132381695
Telegram(电报1)：@mdapi888 (脉动数据客服)
Telegram(电报2)：@dd111333 (火柴)
🔧 第一步：环境准备
在开始之前，请务必完成以下步骤：
- 安装Python库：
pip install requests pandas
📈 第二步：实战接口一 – 获取实时行情（HTTP）
此接口用于获取期货合约最新的报价、买卖盘和成交数据。
核心要点：
- 地址:
http://39.107.99.235:1008/getQuote.php - 方法: GET
- 性能优化: 务必在请求头中加入
"Accept-Encoding": "gzip"以启用压缩，加快响应速度并节省流量。
Python实现代码：
import requests
import time
def get_realtime_quote(product_code):
"""
获取合约实时行情数据
Args:
product_code (str): 产品代码，如 'HX_HST', 'btcusdt'
Returns:
dict: 解析后的行情数据字典，失败返回None
"""
url = "http://39.107.99.235:1008/getQuote.php"
params = {'code': product_code}
headers = {'Accept-Encoding': 'gzip'} # 关键：启用gzip压缩
try:
response = requests.get(url, params=params, headers=headers, timeout=5)
response.raise_for_status()
json_data = response.json()
if json_data.get('code') == 200:
# 核心行情数据在 data -> body 结构中
return json_data['data']['body']
else:
print(f"接口返回错误: {json_data.get('msg')}")
return None
except requests.exceptions.RequestException as e:
print(f"网络请求失败: {e}")
return None
except ValueError as e:
print(f"JSON解析失败: {e}")
return None
# 使用示例：获取恒生指数期货行情
if __name__ == '__main__':
# 请先通过接口五查询到具体代码
quote_data = get_realtime_quote('HX_HST')
if quote_data:
print(f"产品: {quote_data['StockCode']}")
print(f"时间: {quote_data['Time']}")
print(f"最新价: {quote_data['Price']} | 涨跌: {quote_data['Diff']} ({quote_data['DiffRate']}%)")
print(f"买卖盘: 买一 {quote_data['BP1']}({quote_data['BV1']}) | 卖一 {quote_data['SP1']}({quote_data['SV1']})")
print(f"当日成交量: {quote_data['TotalVol']}")
# 严格遵守频率限制：对同一产品，请求间隔应大于 0.33秒 (1/3)
time.sleep(0.34)📊 第三步：实战接口二 – 获取K线数据
此接口用于获取历史K线（蜡烛图）数据，用于分析、回测和指标计算。
核心要点：
- 地址:
http://39.107.99.235:1008/redis.php - 方法: GET
- 参数说明:
code: 产品代码time: K线周期。可选：1m,5m,15m,30m,1h,1d,1Mrows: 获取的K线条数
- 返回值: 一个二维数组，每条K线数据格式为：
[时间戳, 开盘价, 最高价, 最低价, 收盘价, 时间字符串, 成交量]
Python实现代码：
import requests
import pandas as pd
from datetime import datetime
def get_kline_data(product_code, period='1m', rows=100):
"""
获取合约K线历史数据
Args:
product_code (str): 产品代码
period (str): K线周期，如 '1m', '1h', '1d'
rows (int): 获取的条数
Returns:
pandas.DataFrame: 包含K线数据的DataFrame，失败返回空DataFrame
"""
url = "http://39.107.99.235:1008/redis.php"
params = {
'code': product_code,
'time': period,
'rows': rows
}
headers = {'Accept-Encoding': 'gzip'} # 关键：启用gzip压缩
try:
response = requests.get(url, params=params, headers=headers, timeout=10)
response.raise_for_status()
kline_array = response.json() # 注意：此接口直接返回数组，没有外层code/msg
# 将数据转换为规整的DataFrame
if kline_array:
df = pd.DataFrame(
kline_array,
columns=['timestamp_ms', 'open', 'high', 'low', 'close', 'time_str', 'volume']
)
# 转换数据类型
df['timestamp'] = pd.to_datetime(df['timestamp_ms'], unit='ms')
df.set_index('timestamp', inplace=True)
numeric_cols = ['open', 'high', 'low', 'close', 'volume']
df[numeric_cols] = df[numeric_cols].apply(pd.to_numeric, errors='coerce')
print(f"成功获取 {len(df)} 条 {period} K线数据")
return df[['open', 'high', 'low', 'close', 'volume']] # 返回常用列
return pd.DataFrame()
except requests.exceptions.RequestException as e:
print(f"K线数据请求失败: {e}")
return pd.DataFrame()
except ValueError as e:
print(f"K线数据解析失败: {e}")
return pd.DataFrame()
# 使用示例：获取比特币美元期货的1小时K线
if __name__ == '__main__':
# 获取最近100根1小时K线
kline_df = get_kline_data('btcusdt', period='1h', rows=100)
if not kline_df.empty:
print(kline_df.tail()) # 查看最新几条数据
# 进行简单的移动平均计算
kline_df['ma5'] = kline_df['close'].rolling(window=5).mean()
kline_df['ma20'] = kline_df['close'].rolling(window=20).mean()
print("\n计算移动平均后:")
print(kline_df[['close', 'ma5', 'ma20']].tail())
# 严格遵守频率限制：对同一产品，K线请求间隔应大于 0.2秒 (1/5)
time.sleep(0.25)⚠️ 第四步：重要注意事项与最佳实践
- 启用Gzip压缩：所有请求务必在Header中加上
"Accept-Encoding": "gzip"，这是提升性能的关键。 - 错误处理：网络环境不稳定，务必添加完善的超时和异常处理逻辑。
- 产品代码是核心：所有操作都依赖于正确的产品代码，请首先通过接口五（查询产品订阅代码） 确认。
- 数据用途：实时行情接口数据最全，包含盘口深度(
Depth)和成交明细(BS)；K线接口用于历史分析。
按照以上指南，你可以稳定、高效地对接脉动行情的核心数据接口。如果在对接过程中遇到具体问题，可以依据文档联系官方技术支持。