数据订阅、Buffer 与辅助功能

本文讲解指标开发中的数据订阅（subscribe / fields）、数据序列 Buffer 的用法，以及辅助功能 API（用户信息 / HTTP / 消息 / 日志）。

数据订阅

`subscribe(symbol, period, ...)` — 周期合约数据订阅

描述：订阅其他合约和周期的数据，用于计算多品种多周期的指标。非多品种多周期可以不写。

函数签名：self.subscribe(symbol, period, period2, period3, ...)，支持一次订阅多个周期。

输入参数：

参数	中文描述	类型	必填
symbol	合约名称	str	是
period	周期	str	是

接口案例：

def on_subscribe(self):
    self.subscribe('ag2506', 'M15')  # 订阅白银 2506 合约 15 分钟周期
    self.subscribe('au2506', 'D1')   # 订阅黄金 2506 合约日线

`fields(*field)` — 自定义数据字段

说明：当计算指标不需要所有的 K 线数据时，使用该函数自定义订阅的数据字段，可提高性能。

接口案例：

def on_init(self):
    self.fields('close')            # 只需要收盘价
    self.fields('open', 'high', 'low', 'close', 'volume')  # 需要 OHLCV
    # 默认全部字段（open, high, low, close, volume, datetime）

数据序列 Buffer

描述

Buffer 是一个数据序列。当指标被当作函数使用时（不输出图形，而是返回数据给其他指标调用），需要定义 Buffer 来存储和返回数据。

如果定义并写入了 Buffer，在行情页面左上角，Buffer 数据会跟随行情序列（K 线）一一对应。

输入参数

参数	中文描述	类型
name	序列名称	str

接口案例

def on_init(self):
    # 定义 Buffer
    self.fast_buffer = self.Buffer('fast_buffer')
    self.slow_buffer = self.Buffer('slow_buffer')

def calculate_all(self, data):
    length, klines = data
    self.length = length
    self.datetime = klines['datetime']
    self.close = klines['close']
    fast_start = self.params.fast - 1
    slow_start = self.params.slow - 1

    for i in range(length):
        if i >= fast_start:
            fast_value = sum(self.close[i-fast_start:i+1]) / self.params.fast
            self.fast_buffer[i] = fast_value  # 写入 buffer

        if i >= slow_start:
            slow_value = sum(self.close[i-slow_start:i+1]) / self.params.slow
            self.slow_buffer[i] = slow_value

注意

Buffer 写入时直接用 buffer[index] = value，不需要 set_point
图形指标的数据是字典，Buffer 的数据是列表
定义了 Buffer 的指标可以被其他指标作为函数调用

辅助功能 API

`get_userinfo()` — 获取当前登录用户信息

在 on_auth() 中调用，用于判断用户是否可以运行指标（实现权限控制）。

返回字段	说明
nickname	用户昵称
user_id	用户 ID
phone_number	手机号
vip_level	VIP 等级（0 普通 / 1 黄金 / 2 超级）

注意：所有返回值均为字符串类型。

def on_auth(self):
    user = self.get_userinfo()
    vip = user.get('vip_level', '0')
    if vip == '0':
        # 普通用户可能无法运行某些高级指标
        pass

HTTP 请求

`send_get_request(url, params=None, **kwargs)` — GET 请求

url = "https://api.example.com/data"
params = {"key1": 'value1', "key2": 'value2'}
result = self.send_get_request(url, params=params)

`send_post_request(url, data=None, **kwargs)` — POST 请求

url = "https://api.example.com/submit"
data = {"name": "test", "value": 123}
result = self.send_post_request(url, data=data)

发送消息通知

当触发某些条件时，可以调用该函数生成消息实例，然后调用 publish() 发送。

输入参数：

参数	中文描述	类型
content	消息内容	str
count	消息次数（默认为 1）	int

避免在循环中使用，防止大量发消息卡死。

def on_init(self):
    self.msg_status = {}

def calculate_last(self, data):
    from datetime import datetime
    datetime_str, open_, high, low, close, volume = data
    if (close > 100) and (self.msg_status.get(datetime_str, None) is None):
        now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        content = f'XXX指标,{self.symbol}.{self.period}周期,于{now},价格在{round(close, 2)},触发信号'
        msg = self.Message(content, count=1)
        msg.publish()
        self.msg_status[datetime_str] = 'published'

日志记录

在开发调试中记录信息。日志对象提供多个等级函数：

debug() — 调试信息
info() — 普通信息
success() — 成功信息
warning() — 警告信息
error() — 错误信息
critical() — 严重错误信息

参数	中文描述	类型	必填
file_path	绝对路径	str	是

注意：日志路径必须为 Windows 绝对路径，需使用双反斜杠，如：C:\\Users\\Admin\\Desktop\\indicator.log

def on_init(self):
    # 路径必须使用双反斜杠 \\
    self.logger = self.Logger('C:\\Users\\Admin\\Desktop\\indicator.log')

def calculate_all(self, data):
    self.logger.info('这是一条普通信息')
    self.logger.error('这是一条错误信息')

相关阅读

已复制到剪贴板

指标文档 · 期货版修订 R2

返回指标文档 · 期货版板块