期魔方期魔方
注册
KB / ARTICLE

指标开发快速入门

指标文档 · 期货版修订 R2发布于 2026-10-14

本文通过两个示例带你快速上手期魔方 Python 指标开发：一个极简双均线 Demo + 一个完整的 RSI 八步实战（从新建到加载）。

新手入门 Demo：双均线

该 demo 以双均线为例，步骤如下：

  1. 第一步是先定义指标参数 Params
  2. 第二步在 on_init 中定义输出图形对象
  3. 第三步编写 calculate_all 计算均线输出值的代码
  4. 第四步编写 calculate_last 计算均线输出值的代码
# 双均线案例
# 外置参数
from pydantic import BaseModel
class Params(BaseModel, validate_assignment=True):
    """开发者无需修改Params类，只需关注Params类中的字段定义即可"""
    from pydantic import Field
    is_main:bool = Field(default=True, title="是否是主图")
    fast:int = Field(default=10, title="快线")
    slow:int = Field(default=20, title="慢线")

# 指标初始化
def on_init(self):
    self.calculate_type = 'last'
    self.calculate_data_type = 'list'
    self.fast_line = self.Line('fast_line', 'rgba(255, 0, 0, 1)')
    self.slow_line = self.Line('slow_line', 'rgba(0, 255, 0, 1)')
    self.fields('close')

# 全量计算方法
def calculate_all(self, data):
    length, klines = data
    self.length = length
    self.datetime = klines['datetime']
    self.close = klines['close']
    fast = self.params.fast
    slow = self.params.slow
    fast_start = fast - 1
    slow_start = slow - 1
    for i in range(length):
        if i >= fast_start:
            fast_value = sum(self.close[i-fast_start:i+1]) / fast
            self.fast_line.set_point(self.datetime[i], fast_value)
        if i >= slow_start:
            slow_value = sum(self.close[i-slow_start:i+1]) / slow
            self.slow_line.set_point(self.datetime[i], slow_value)

# 实时行情计算增量数据
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']
    last_fast = self.close[-self.params.fast:]
    last_slow = self.close[-self.params.slow:]
    fast_value = sum(last_fast) / self.params.fast
    slow_value = sum(last_slow) / self.params.slow
    self.fast_line.set_point(kline['datetime'], fast_value)
    self.slow_line.set_point(kline['datetime'], slow_value)

开发实战：RSI 指标八步

在这一章，我们通过编写 RSI 指标来熟悉指标开发的完整流程。

第一步：新建指标文件

登录期魔方客户端 → 点击 策略 → 点击 指标列表 → 在 我的指标文件夹的 操作中点击 新增指标

然后会弹出提示框，我们输入指标名称：my_rsi，选择语言：python，点击确定

如果您之前没有下载过编辑器，此时会提示您下载一个编辑器用于编写指标（整个过程会自动完成，请耐心等待一会）

image-20241107185212067

第二步：编写指标模板代码

点击刚才创建好的 my_rsi 指标旁边的 编辑指标 按钮，此时会自动唤起编辑器，并打开一个基础 python 指标模版文件。

新的 python 指标文件并不是空白的，而是包含了一些基础模板：

#------------------------------------------# 
#文件类型:技术指标
#帮助文档:https://qmfquant.com/static/doc/code/indicatorEdit.html
#期魔方，为您提供专业的量化服务
#------------------------------------------#

from pydantic import BaseModel

class Params(BaseModel, validate_assignment=True):
    from pydantic import Field
    is_main:bool = Field(default=True, title="是否为主图")

# def on_auth(self):
#     pass

# def on_subscribe(self):
#     pass

def on_init(self): 
    pass

def calculate_all(self, data):
    length, datetime_str, open, high, low, close, volume = data
    pass

def calculate_last(self, data):
    length, datetime_str, open, high, low, close, volume = data
    pass

第一步，对模版文件进行调整，去掉不需要的代码（权限验证、订阅数据），得到精简版本：

from pydantic import BaseModel

class Params(BaseModel, validate_assignment=True):
    from pydantic import Field
    is_main:bool = Field(default=True, title="是否为主图")

def on_init(self): 
    pass

def calculate_all(self, data):
    length, datetime_str, open, high, low, close, volume = data
    pass

def calculate_last(self, data):
    length, datetime_str, open, high, low, close, volume = data
    pass

第三步：定义外置参数（params）

由于 RSI 指标是副图指标，所以我们需要在外置参数中指定其为副图。RSI 有一个参数是 天数（周期），所以我们在外置参数中声明一个属性来表明这个参数：

class Params(BaseModel, validate_assignment=True):
    from pydantic import Field
    is_main:bool = Field(default=False, title="是否为主图")
    #                            ^^^^^ 使用 False 来表明副图
    period:int = Field(default=14, title="周期") 
    # 添加这行，声明一个类型为 int ，名称为 period 的外置参数，给一个默认值 14

第四步：实现 on_init 初始化

RSI 不需要每次都计算其历史值，所以确定其计算方式为增量计算。RSI 指标是绘制一条线条，所以我们在初始化时创建一个线条对象。RSI 仅需要收盘价就可以完成计算，所以声明只需要收盘价数据。

def on_init(self): 
    self.calculate_type = 'last' # 'all' - 全量计算，'last' - 增量计算
    self.calculate_data_type = 'list'
    self.fields('close') # 订阅收盘价
    self.rsi_line = self.Line('rsi', '#f0f5ff', line_width=3) # 线条对象

第五步：实现 calculate_all 全量计算

虽然指标是以增量计算来运行的，但是需要为其初始化图案时进行一次全量计算（第一次接收到600根K线数据）：

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

    deltas = []
    for i in range(1, len(self.close)):
        deltas.append(self.close[i] - self.close[i - 1])

    if len(deltas) < self.params.period: 
        for i in range(length):
            self.rsi_line.set_point(self.datetime[i], None) 

    up_sum = 0
    down_sum = 0
    for delta in deltas[:self.params.period]:
        if delta > 0:
            up_sum += delta
        elif delta < 0:
            down_sum += abs(delta)

    self.avg_up = up_sum / self.params.period
    self.avg_down = down_sum / self.params.period

    for i in range(0, self.params.period):
        self.rsi_line.set_point(self.datetime[i], None)

    for i in range(self.params.period, len(deltas)):
        delta = deltas[i]
        up_val = max(delta, 0)
        down_val = abs(min(delta, 0))
        avg_up = (self.avg_up * (self.params.period - 1) + up_val) / self.params.period
        avg_down = (self.avg_down * (self.params.period - 1) + down_val) / self.params.period
        rs = avg_up / avg_down if avg_down != 0 else float('inf')
        rsi = 100 - (100 / (1 + rs))
        rsi = round(rsi, 2)
        self.rsi_line.set_point(self.datetime[i+1], rsi)
        if i+1 != len(deltas):
            self.avg_up = avg_up
            self.avg_down = avg_down

归纳为 3 步：获取数据 → 算法计算 → 写入图形对象。

第六步：实现 calculate_last 增量计算

增量计算中，用 datetime 比对当前接收到的 K 线是否为新的一根：

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'])
        self.avg_up = self.last_avg_up
        self.avg_down = self.last_avg_down
    else:
        self.close[-1] = kline['close']
  
    delta = self.close[-1] - self.close[-2]
    up_val = max(delta, 0)
    down_val = abs(min(delta, 0))
    avg_up = (self.avg_up * (self.params.period - 1) + up_val) / self.params.period
    avg_down = (self.avg_down * (self.params.period - 1) + down_val) / self.params.period
    rs = avg_up / avg_down if avg_down != 0 else float('inf')
    rsi = 100 - (100 / (1 + rs))
    rsi = round(rsi, 2)
    self.rsi_line.set_point(kline['datetime'], rsi)

    self.last_avg_up = avg_up
    self.last_avg_down = avg_down

第七步：保存编译与调试

当开发完成，保存指标（会自动触发编译），或者在编辑器中点击右键 → 选择 python指标/策略编译。

编译结果会在控制台输出。如果代码有误，控制台会提示相应的错误。比如我们把 set_point 写成了 set_points：

编译失败，失败原因: 
<未通过测试数据: line 62, in calculate_all()>
错误描述：
AttributeError: 'Line' object has no attribute 'set_points'. Did you mean: 'set_point'?

编译成功则会显示 "编译成功！"

编译成功后会生成一个加密文件，该文件才是真实运行的指标。每次改动代码后，必须 保存 → 编译，才会让改动生效。

使用日志调试

def on_init(self): 
    self.calculate_type = 'last'
    self.calculate_data_type = 'list'
    self.fields('close')
    self.rsi_line = self.Line('rsi', '#f0f5ff', line_width=3)
    self.logger = self.Logger(r'D:\logs\my_rsi.log')

然后在 calculate_all 中记录日志：

self.logger.info(f"rsi = {rsi}")

第八步：绘制图形

编写并调试完成后，进入行情页面，选择合约，K 线图页面点击右键，指标窗口个数选择 2 个，然后在副图加载 Py_my_rsi（所有 python 指标会有 Py_ 前缀）。填写外置参数后加载指标，即可看到效果。

image-20241107185212067

相关阅读

已复制到剪贴板
指标文档 · 期货版修订 R2

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

评论