gj-kingstar-api 1.0.5

金仕达行情封装

gj_kingstar_api

金仕达行情API封装 - 支持多站点高可用、自动故障切换的Python SDK

功能特性

• 多站点高可用：支持配置多个站点，自动选择最优站点连接
• 自动故障切换：主站点故障时自动切换到备用站点
• 站点探活：定时探测所有站点状态，保证连接质量
• 重连补偿：断线重连后自动恢复订阅状态
• 心跳机制：维持长连接，检测连接状态
• 配置灵活：支持代码配置、配置文件（yaml/json）两种方式

快速开始

基本用法

from gj_kingstar_api import KingstarClient, MarketListener

class MyMarketListener(MarketListener):
    def on_tick(self, flow_no, ticks):
        for tick in ticks:
            print(f'{tick.unique_instrument_id} 最新价: {tick.last_price}')
    
    def on_site_switch(self, old_site, new_site, reason):
        print(f'【站点切换】{old_site} -> {new_site} (原因: {reason})')
    
    def on_connection_status(self, status, message):
        print(f'【连接状态】{status}: {message}')

# 初始化客户端
client = KingstarClient(url="ws://your-kingstar-server:port", user="username", pwd="password")

# 注册监听器
client.add_listener(MyMarketListener())

# 连接并订阅
client.connect()
client.subscribe_tick('SHFE', 'Future', ['ag2506', 'ag2508'])

多站点模式

sites = [
    {'url': 'ws://primary.kingstar.com:8080', 'weight': 10, 'name': '主站点'},
    {'url': 'ws://secondary.kingstar.com:8080', 'weight': 5, 'name': '备站点'},
    {'url': 'ws://backup.kingstar.com:8080', 'weight': 3, 'name': '灾备站'}
]

client = KingstarClient(
    sites=sites,
    user="username",
    pwd="password",
    auto_switch=True,
    probe_interval=30,
    max_reconnect=10,
    heartbeat_interval=30
)
client.add_listener(MyMarketListener())
client.connect()
client.subscribe_tick('SHFE', 'Future', ['ag2506'])

配置文件方式

创建配置文件 kingstar_config.yaml：

sites:
  - url: ws://primary.kingstar.com:8080
    weight: 10
    name: 主站点
  - url: ws://secondary.kingstar.com:8080
    weight: 5
    name: 备站点
connection:
  auto_switch: true
  probe_interval: 30
  max_reconnect: 10
  heartbeat_interval: 30

client = KingstarClient.from_config("kingstar_config.yaml", user="username", pwd="password")
client.add_listener(MyMarketListener())
client.connect()

API 文档

KingstarClient

初始化参数

参数	类型	说明	默认值
url	str	单站点URL（兼容旧版）	None
user	str	用户名	None
pwd	str	密码	None
sites	List[Dict]	多站点配置列表	None
auto_switch	bool	是否自动切换站点	True
probe_interval	int	站点探活间隔（秒）	30
max_reconnect	int	最大重连次数	10
heartbeat_interval	int	心跳间隔（秒）	30

方法

方法	说明
connect()	启动客户端连接
close()	关闭客户端
subscribe_tick(exchange_id, type, instrument_id)	订阅行情tick
unsubscribe_tick(exchange_id, type, instrument_id)	取消订阅
add_listener(listener)	注册监听器
get_current_site()	获取当前连接站点
get_site_status()	获取所有站点状态
add_site(url, name, weight)	动态添加站点
remove_site(url)	动态移除站点

类方法

方法	说明
from_config(config_path, user, pwd)	从配置文件创建客户端

MarketListener

自定义监听器需要继承 MarketListener 并实现以下方法：

方法	说明
on_subscribe_tick_data(flow_no)	订阅成功回调
on_unsubscribe_tick_data(flow_no)	取消订阅回调
on_tick(flow_no, ticks)	行情tick数据回调
on_site_switch(old_site, new_site, reason)	站点切换回调
on_connection_status(status, message)	连接状态变更回调

支持的合约类型

类型	说明
Future	期货
Option	期权
CS	股票

依赖

• Python >= 3.7
• protobuf >= 4.24.4
• websockets >= 11.0.3