港股通专用接口

注意事项

1. 09:10分(市场初始化默认时间，自建行情需要询问券商)之前调用get_hks_list可能会取到历史数据。
2. 如果策略中频繁使用到get_hks_price_gap、 get_hks_unit_amount数据，建议在策略中缓存到全局变量中。
3. 调用港股通专用接口时参数校验不通过/API内部执行报错将会raise并抛出报错信息( qtcommon.exception.ParamsError/RuntimeError)，需要在策略中做保护。

港股通查询类接口

get_hks_list

中文名

获取港股通代码表

接口说明

用于获取行情返回的港股通代码列表。

接口定义

get_hks_list(market)

注意事项

1. market仅支持XHKG-SS和XHKG-SZ两个市场，不支持获取其他市场代码。不入参时，默认返回沪港通和深港通的代码列表。
2. 为减小对行情服务压力，该函数在同一分钟内多次调用返回当前分钟首次查询的缓存数据。

使用场景

❌研究 ❌回测 ✅交易

参数

market

• 类型：str

市场代码，仅支持XHKG-SS、XHKG-SZ、XHKG-SS;XHKG-SZ和XHKG-SZ;XHKG-SS，选填字段。

返回值

list:

• 查询成功返回对应市场代码的港股通代码列表。
• 查询失败，返回空列表。

异常

• qtcommon.exception.ParamsError：参数校验失败。
• RuntimeError：API执行错误。

示例

def initialize(context):
    g.security = "02899.XHKG-SS"
    set_universe(g.security)


def before_trading_start(context, data):
    g.flag = False


def handle_data(context, data):
    if not g.flag:
        try:
            g.hks_list = get_hks_list("XHKG-SS;XHKG-SZ")
            log.info("港股通代码表为：{}".format(g.hks_list))
        except BaseException as e:
            g.hks_list = []
            log.error("获取港股通代码表失败，错误信息为：{}".format(e))
        g.flag = True

get_hks_price_gap

中文名

获取港股通价差信息

接口说明

港股不同价格代码的最小价格变动幅度是不同的，该接口用于获取港股通代码的价差信息。

接口定义

get_hks_price_gap(market)

注意事项

1. market仅支持XHKG-SS和XHKG-SZ两个市场，不支持获取其他市场代码。
2. 为减小对柜台压力，该函数在同一分钟内多次调用且入参相同时返回当前分钟首次查询的缓存数据。

使用场景

❌研究 ❌回测 ✅交易

参数

market

• 类型：str

市场代码，仅支持XHKG-SS和XHKG-SZ，必填字段。

返回值

list:

• 查询成功返回对应市场代码的港股通价差信息列表，列表中每个元素包含的字段如下：

  字段名	类型	说明
  begin_price	float	分段起始价格
  end_price	float	分段结束价格
  step_price	float	最小价差

• 查询失败，返回空列表。

异常

• qtcommon.exception.ParamsError：参数校验失败。
• RuntimeError：API执行错误。

示例

def initialize(context):
    g.security = "02899.XHKG-SS"
    set_universe(g.security)


def before_trading_start(context, data):
    g.flag = False


def handle_data(context, data):
    if not g.flag:
        try:
            g.hks_price_gap = {"XHKG-SS": get_hks_price_gap("XHKG-SS"), "XHKG-SZ": get_hks_price_gap("XHKG-SZ")}
            log.info("港股通价差信息为：{}".format(g.hks_price_gap))
        except BaseException as e:
            g.hks_price_gap = {"XHKG-SS": [], "XHKG-SZ": []}
            log.error("获取港股通价差信息失败，错误信息为：{}".format(e))
        g.flag = True

get_hks_unit_amount

中文名

获取港股通委托单位数量

接口说明

港股通每个标的的最小交易单位不同，交易前需要调用该接口查询获取。

接口定义

get_hks_unit_amount(security, entrust_prop)

使用场景

❌研究 ❌回测 ✅交易

参数

security

• 类型：str

港股通标的代码，必填字段。

entrust_prop

• 类型：str

委托属性，仅支持HKN和HKO，必填字段。

返回值

dict:

• 查询成功返回对应港股通标的代码的交易单位信息，字典中包含的字段如下：

  字段名	类型	说明
  buy_unit	int	买入单位数量
  sell_unit	int	卖出单位数量

• 查询失败，返回空字典。

异常

• qtcommon.exception.ParamsError：参数校验失败。
• RuntimeError：API执行错误。

示例

def initialize(context):
    g.security = "02899.XHKG-SS"
    set_universe(g.security)


def before_trading_start(context, data):
    g.flag = False
    g.hks_unit_amount = {}


def handle_data(context, data):
    if not g.flag:
        try:
            for security in get_hks_list("XHKG-SS;XHKG-SZ"):
                g.hks_unit_amount[security] = get_hks_unit_amount(security, "HKN")
            log.info("港股通委托单位数量信息为：{}".format(g.hks_unit_amount))
        except BaseException as e:
            log.error("获取港股通委托单位数量信息失败，错误信息为：{}".format(e))
        g.flag = True

get_hks_enable_amount

中文名

获取港股通大约可买数量

接口说明

调用该接口获取当前账户港股通标的代码大约可买数量。

接口定义

get_hks_enable_amount(security, entrust_price, entrust_prop)

使用场景

❌研究 ❌回测 ✅交易

参数

security

• 类型：str

港股通标的代码，必填字段。

entrust_price

• 类型：float

委托价格，必须大于0，必填字段。

entrust_prop

• 类型：str

委托属性，仅支持HKN和HKO，必填字段。

返回值

int:

• 查询成功返回对应港股通标的代码的可买数量。
• 查询失败，返回0。

异常

• qtcommon.exception.ParamsError：参数校验失败。
• RuntimeError：API执行错误。

示例

def initialize(context):
    g.security = "02899.XHKG-SS"
    set_universe(g.security)


def before_trading_start(context, data):
    g.flag = False


def handle_data(context, data):
    if not g.flag:
        try:
            last_price = get_snapshot(g.security).get(g.security, {}).get("last_px", 0)
            if last_price > 0:
                hks_enable_amount = get_hks_enable_amount(g.security, last_price, "HKN")
                log.info("港股通代码 {} 可买数量为：{}".format(g.security, hks_enable_amount))
        except BaseException as e:
            log.error("获取港股通代码 {} 可买数量失败，错误信息为：{}".format(g.security, e))
        g.flag = True

get_kline_by_range

中文名

按时间范围获取港股通K线

接口说明

该接口用于获取指定日期范围内的K线数据，仅支持港股通代码

接口定义

get_kline_by_range(security_list, frequency, start_time, end_time=None, field=None)

注意事项

• 当查询日线数据时，start_time和end_time请传入年月日（长度为8）的字符串，当查询分钟线数据时，start_time和end_time请传入年月日小时分钟（长度为12）的字符串；

• 针对停牌场景，我们没有跳过停牌的日期，无论对单只股票还是多只股票进行调用，停牌时使用停牌前的价格填充，成交量、成交额为0，日K线可使用成交量为0的逻辑进行停牌日过滤。

• 该接口只能获取2025年后开始的数据。（需券商本地存在该段时间的数据）

• 该接口与get_kline_by_offset接口不支持多线程同时调用，即在run_daily或run_interval等函数中不要与handle_data等框架模块同一时刻调用get_kline_by_range或get_kline_by_offset接口，否则会偶现获异常raise的现象。

• 当日实时数据只有在9:30分之后才能获取到（不包含9:30分），请注意取数时间。

使用场景

❌研究 ❌回测 ✅交易

参数

security_list

• 类型： list[str]

港股通代码列表,必须传入list，支持单只或多只代码，必填字段

frequency

• 类型： str

单位时间长度，仅支持1分钟线(1m)、5分钟线(5m)、15分钟线(15m)、30分钟线(30m)、60分钟线(60m)、120分钟线(120m)、日线(1d)，必填字段

start_time

• 类型： str

开始时间，输入请小于当前日期，且均小于等于end_date的时间。查询日线数据仅支持（YYYYmmdd），如'20250901'; 查询分钟线数据仅支持（YYYYmmddHHMM）'202506011000'，必填字段

end_time

• 类型： str|None
• 默认值： str

结束时间，输入请小于等于当前日期。查询日线数据仅支持（YYYYmmdd），如'20250901'; 查询分钟线数据仅支持（YYYYmmddHHMM）'202506011000'，不传时默认取引擎的时间（例如框架函数handle_data在9:31分触发，而写在handle_data中的该函数直到9:32分才调用，此时引擎时间还是9:31）;选填字段

field

• 类型： list[str]|None
• 默认值： None

数据结果集中所支持输出字段，选填字段,当frequency为'1d'时，默认为['open','high','low','close','volume','money','price','is_open','preclose','high_limit','low_limit'],当frequency不为'1d'时，默认为['open','high','low','close','volume','money','price']

• 输出字段包括:
  • open -- 开盘价(numpy.float64)；
  • high -- 最高价(numpy.float64)；
  • low --最低价(numpy.float64)；
  • close -- 收盘价(numpy.float64)；
  • volume -- 交易量(numpy.float64)；
  • money -- 交易金额(numpy.float64)；
  • price -- 最新价(numpy.float64)；
  • is_open -- 是否开盘(numpy.int64)(仅日线返回)；
  • preclose -- 昨收盘价(numpy.float64)(仅日线返回)；
  • high_limit -- 涨停价(numpy.float64)(仅日线返回)；
  • low_limit -- 跌停价(numpy.float64)(仅日线返回)；

返回值

dict:

1. 正常获取到数据后,会返回一个dict，其中key是代码，value是获取到代码的K线数据，如下：

dict{股票代码(str): array([日期时间(numpy.int64), 开盘价(numpy.float64), 最高价(numpy.float64), 最低价(numpy.float64), 收盘价(numpy.float64), 成交量(numpy.float64), 成交额(numpy.float64), 最新价(numpy.float64)])}

dict{'00001.XHKG-SZ': array([(202509171610, 52.15, 52.4 , 52.15, 52.25, 1824129., 9.54154380e+07, 52.25),...])}

2. 当获取不到某只代码的情况下，代码的值返回的是空的数据(请对空数据做好保护)，如下：

dict{股票代码(str): array([])}

dict{'00001.XHKG-SZ': array([])}

3. 当数据异常，或函数处理异常的时候，该函数会raise异常，请调用该函数的时候做好保护。

示例

def initialize(context):
    g.security = ["00001.XHKG-SZ", "00001.XHKG-SS"]
    set_universe(g.security)

def handle_data(context, data):
    # 获得"00001.XHKG-SZ", "00001.XHKG-SS"的数据
    kline_data = get_kline_by_range(g.stock_list, "1m", "202509110930")
    log.info(kline_data)
    log.info("获取到00001.XHKG-SZ数据一共 {} 条".format(len(kline_data[g.stock_list[0]])))

get_kline_by_offset

中文名

按根数偏移获取港股通K线

接口说明

该接口用于获取根数偏移的K线数据，仅支持港股通代码

接口定义

get_kline_by_offset(security_list, frequency, count, field=None, include=False, query_time=None)

注意事项

• 当入参query_time时，query_time请传入年月日小时分钟（长度为12）的字符串；

• 针对停牌场景，我们没有跳过停牌的日期，无论对单只股票还是多只股票进行调用，停牌时使用停牌前的价格填充，成交量、成交额为0，日K线可使用成交量为0的逻辑进行停牌日过滤。

• 该接口只能获取2025年后开始的数据。（需券商本地存在该段时间的数据）

• 该接口与get_kline_by_range接口不支持多线程同时调用，即在run_daily或run_interval等函数中不要与handle_data等框架模块同一时刻调用get_kline_by_range或get_kline_by_offset接口，否则会偶现获异常raise的现象。

• 当日实时数据只有在9:30分之后才能获取到（不包含9:30分），请注意取数时间。

使用场景

❌研究 ❌回测 ✅交易

参数

security_list

• 类型： list[str]

港股通代码列表,必须传入list，支持单只或多只代码，必填字段

frequency

• 类型： str

单位时间长度，仅支持1分钟线(1m)、5分钟线(5m)、15分钟线(15m)、30分钟线(30m)、60分钟线(60m)、120分钟线(120m)、日线(1d)，周线(1w)、月线(mo)、季度线(1q)和年线(1y),必填字段

count

• 类型： int

K线数量，大于0，返回指定数量的K线行情，必填字段

field

• 类型： list[str]|None
• 默认值： None

数据结果集中所支持输出字段，选填字段,当frequency为'1d'时，默认为['open','high','low','close','volume','money','price','is_open','preclose','high_limit','low_limit'],当frequency不为'1d'时，默认为['open','high','low','close','volume','money','price']

• 输出字段包括:
  • open -- 开盘价(numpy.float64)；
  • high -- 最高价(numpy.float64)；
  • low --最低价(numpy.float64)；
  • close -- 收盘价(numpy.float64)；
  • volume -- 交易量(numpy.float64)；
  • money -- 交易金额(numpy.float64)；
  • price -- 最新价(numpy.float64)；
  • is_open -- 是否开盘(numpy.int64)(仅日线返回)；
  • preclose -- 昨收盘价(numpy.float64)(仅日线返回)；
  • high_limit -- 涨停价(numpy.float64)(仅日线返回)；
  • low_limit -- 跌停价(numpy.float64)(仅日线返回)；

include

• 类型： bool
• 默认值： False

是否包含查询周期的数据，True -包含，False-不包含，选填字段

query_time

• 类型： str|None
• 默认值： str

结束时间，输入请小于等于当前日期。仅支持（YYYYmmddHHMM）'202506011000'，不传时默认取引擎的时间（例如框架函数handle_data在9:31分触发，而写在handle_data中的该函数直到9:32分才调用，此时引擎时间还是9:31）;选填字段

返回值

dict:

1. 正常获取到数据后,会返回一个dict，其中key是代码，value是获取到代码的K线数据，如下：

dict{股票代码(str): array([日期时间(numpy.int64), 开盘价(numpy.float64), 最高价(numpy.float64), 最低价(numpy.float64), 收盘价(numpy.float64), 成交量(numpy.float64), 成交额(numpy.float64), 最新价(numpy.float64)])}

dict{'00001.XHKG-SZ': array([(202509171610, 52.15, 52.4 , 52.15, 52.25, 1824129., 9.54154380e+07, 52.25),...])}

2. 当获取不到某只代码的情况下，代码的值返回的是空的数据(请对空数据做好保护)，如下：

dict{股票代码(str): array([])}

dict{'00001.XHKG-SZ': array([])}

3. 当数据异常，或函数处理异常的时候，该函数会raise异常，请调用该函数的时候做好保护。

示例

def initialize(context):
    g.security = ["00001.XHKG-SZ", "00001.XHKG-SS"]
    set_universe(g.security)

def handle_data(context, data):
    # 获得"00001.XHKG-SZ"和"00001.XHKG-SS"的天数据
    kline_data = get_kline_by_range(g.stock_list, "1m", "202509110930")
    log.info(kline_data)
    log.info("获取到00001.XHKG-SZ数据一共 %s 条" % len(kline_data[g.stock_list[0]]))

港股通交易类接口

hks_order

中文名

港股通委托

接口说明

港股通标的买卖接口。支持港股通委托和港股通零股委托。

接口定义

hks_order(security, amount, entrust_prop, entrust_price, quote_type)

使用场景

❌研究 ❌回测 ✅交易

参数

security

• 类型：str

港股通标的代码，必填字段。

amount

• 类型：int

委托数量，买为正数、卖为负数，必填字段。

entrust_prop

• 类型：str

委托属性，仅支持HKN和HKO，必填字段。

entrust_price

• 类型：float

委托价格，必须大于0，必填字段。

quote_type

• 类型：str

报价方式，0为增强限价盘，1为竞价限价盘，必填字段。港股通零股委托时该参数不生效强制为增强限价盘。

返回值

order_id|None:

• 创建订单成功时返回Order对象的id。
• 创建订单失败时，返回None。

异常

• qtcommon.exception.ParamsError：参数校验失败。
• RuntimeError：API执行错误。

示例

def initialize(context):
    g.security = "02899.XHKG-SS"
    set_universe(g.security)


def before_trading_start(context, data):
    g.flag = False


def handle_data(context, data):
    if not g.flag:
        try:
            last_price = get_snapshot(g.security).get(g.security, {}).get("last_px", 0)
            if last_price > 0:
                hks_enable_amount = get_hks_enable_amount(g.security, last_price, "HKN")
                if hks_enable_amount > 0:
                    order_id = hks_order(g.security, hks_enable_amount, "HKN", last_price, "0")
                    log.info("港股通代码 {} 委托成功，order_id：{}".format(g.security, order_id))
        except BaseException as e:
            log.error("港股通代码 {} 委托失败，错误信息为：{}".format(g.security, e))
        g.flag = True

hks_cancel_order

中文名

港股通撤单

接口说明

该接口用于撤销港股通委托订单，根据order_id撤销订单。

接口定义

hks_cancel_order(order_id)

使用场景

❌研究 ❌回测 ✅交易

参数

order_id

• 类型：str

港股通委托返回的订单编号，必填字段。

返回值

None:

• 返回None。

异常

• qtcommon.exception.ParamsError：参数校验失败。
• RuntimeError：API执行错误。

示例

import time


def initialize(context):
    g.security = "02899.XHKG-SS"
    set_universe(g.security)


def before_trading_start(context, data):
    g.flag = False


def handle_data(context, data):
    if not g.flag:
        try:
            last_price = get_snapshot(g.security).get(g.security, {}).get("last_px", 0)
            if last_price > 0:
                hks_enable_amount = get_hks_enable_amount(g.security, last_price, "HKN")
                if hks_enable_amount > 0:
                    order_id = hks_order(g.security, hks_enable_amount, "HKN", last_price, "0")
                    log.info("港股通代码 {} 委托成功，order_id：{}".format(g.security, order_id))
                    time.sleep(1)
                    try:
                        hks_cancel_order(order_id)
                    except BaseException as e:
                        log.error("港股通代码 {} 撤单失败，错误信息为：{}".format(g.security, e))
        except BaseException as e:
            log.error("港股通代码 {} 委托失败，错误信息为：{}".format(g.security, e))
        g.flag = True

---

说明

接口支持的业务范围以及支持在引擎的哪些流程函数中调用，详见 接口列表