函数指标：定义、计算与跨周期调用

本文讲解如何把一套算法封装成可复用的函数指标（不输出图形，返回 Buffer 数据），以及如何在其他指标中跨周期、跨合约调用它。

什么是函数指标

函数指标是指可以不输出图形的指标，通过对数据的计算，返回一个或多个数据序列（Buffer）。它的核心特征是：

当一个指标中定义了 Buffer，这个指标就能够被作为一个函数来使用。

典型用法：

• 把均线、RSI、ATR 等算法封装成函数指标
• 在多个主图/副图指标中调用同一个函数指标，避免重复计算
• 跨周期、跨合约调用不同品种的指标数据

定义数据序列（Buffer）

Buffer 需要在 on_init 中定义：self.xxx = self.Buffer('xxx')。

我们把之前的双均线指标改造为一个函数：

def on_init(self):
    self.calculate_type = 'last'
    self.calculate_data_type = 'list'
    self.fields('close')
    # 定义 Buffer
    self.fast_buffer = self.Buffer('MA_FAST')
    self.slow_buffer = self.Buffer('MA_SLOW')

计算函数的值

计算方法和图形指标一样使用 calculate_all 和 calculate_last，只是把值写入 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:
            self.fast_buffer[i] = sum(self.close[i-fast_start:i+1]) / self.params.fast
        if i >= slow_start:
            self.slow_buffer[i] = sum(self.close[i-slow_start:i+1]) / self.params.slow

def calculate_last(self, data):
    _, kline = data
    if kline['datetime'] != self.datetime[-1]:
        self.length += 1
        self.datetime.append(kline['datetime'])
        self.close.append(kline['close'])
    else:
        self.close[-1] = kline['close']

    self.fast_buffer[-1] = sum(self.close[-self.params.fast:]) / self.params.fast
    self.slow_buffer[-1] = sum(self.close[-self.params.slow:]) / self.params.slow

注意：写入 buffer 用 buffer[index] = value，图形指标用 set_point。图形指标的数据是一个字典，而 Buffer 的数据是一个列表。

调用函数指标

在其他指标中调用已定义的函数指标，使用 self.Indicator() 方法创建对象：

参数：

参数	必填	说明
name	是	函数指标文件名
id	否	多个同 name 指标时用 id 区分
symbol	否	合约代码（不填用当前指标合约）
period	否	周期（不填用当前指标周期）
params	否	外置参数覆盖（不填用默认值）

调用案例

1. 跟随当前指标

def on_init(self):
    self.ma_indicator = self.Indicator('ma')

2. 同一指标多次使用（不同参数）

def on_init(self):
    self.ma_indicator1 = self.Indicator('ma', 'ma_1', fast=5, slow=10)
    self.ma_indicator2 = self.Indicator('ma', 'ma_2', fast=20, slow=30)

3. 跨周期、跨合约调用

def on_init(self):
    self.ag_indicator = self.Indicator('ma', 'ma_ag', symbol='ag2506', period='M15', fast=5, slow=10)
    self.au_indicator = self.Indicator('ma', 'ma_au', symbol='au2506', period='D1', fast=10, slow=20)

不需要跨合约、跨周期时，不用指定 symbol 和 period。

获取函数指标的值

有两种方式：

方式一：单指标对象 .buffer_all() / .buffer_last()

当使用一个函数指标时，推荐使用这种方式获取值。

def calculate_all(self, data):
    length, klines = data
    datetime = klines['datetime']

    # buffer_all() 返回 {'MA_FAST': list, 'MA_SLOW': list}
    ma_buffer = self.ma_indicator.buffer_all()
    for i in range(length):
        self.fast_line.set_point(datetime[i], ma_buffer['MA_FAST'][i])
        self.slow_line.set_point(datetime[i], ma_buffer['MA_SLOW'][i])

def calculate_last(self, data):
    _, kline = data
    # buffer_last() 返回 {'MA_FAST': value, 'MA_SLOW': value}
    ma_buffer = self.ma_indicator.buffer_last()
    self.fast_line.set_point(kline['datetime'], ma_buffer['MA_FAST'])
    self.slow_line.set_point(kline['datetime'], ma_buffer['MA_SLOW'])

方式二：多指标 .gather_all() / .gather_last()

当使用了多个函数指标时（尤其是跨品种、跨周期的情况），用 gather_all() 一次性获取所有值：

def calculate_all(self, data):
    length, klines = data
    datetime = klines['datetime']
    # gather_all() 返回 {'ma_ag': {'MA_FAST': list, 'MA_SLOW': list}, 'ma_au': {'MA_FAST': list, 'MA_SLOW': list}}
    gather_buffer = self.gather_all()
    ag_fast = gather_buffer['ma_ag']['MA_FAST']
    ag_slow = gather_buffer['ma_ag']['MA_SLOW']
    au_fast = gather_buffer['ma_au']['MA_FAST']
    # 根据需要进行后续计算...

def calculate_last(self, data):
    _, kline = data
    # gather_last() 返回 {'ma_ag': {'MA_FAST': value, 'MA_SLOW': value}, 'ma_au': {...}}
    gather_buffer = self.gather_last()
    ag_fast = gather_buffer['ma_ag']['MA_FAST']

---

相关阅读

• 数据订阅、Buffer 与辅助功能
• 图形绘制通用函数