指标编写文档

关于期魔方

期魔方量化投研平台是由四川赤壁量化科技有限公司自主研发，专为国内期货市场打造的全能量化交易与研究平台。它集成了市场行情分析、量化策略回测、数据分析、风险管理以及机器学习等多项强大功能，具备高度可扩展性，能够满足不同用户的多样化需求。

核心功能：

1. 专业的 K 线图表功能：期魔方支持自定义 Python 指标开发，具备强大的跨周期、跨品种、跨市场数据调用能力，并提供深度 DIV 可视化界面，帮助用户进行精准的市场分析。
2. 高效的 Python 量化与回测：期魔方采用自主研发的核心量化底层，性能相比市场上其他 Python 策略提升十倍以上。编写简单、快速入门，用户可在十分钟内轻松上手。
3. 全面的数据支持：期魔方支持超过 15 年的历史数据回测，且可以调用多维度数据，包括库存、仓单、现货等衍生数据，帮助用户进行深入分析与策略优化。
4. 盘手训练系统：期魔方提供多周期复盘与动态回放功能，用户可以通过模拟训练不断优化交易策略。同时，系统支持训练报告分析、查看与下载，便于用户进行复盘总结。
5. 机器学习集成：期魔方深度结合 sklearn，初学者也能快速上手模型训练与优化，助力用户实现智能化决策与策略提升。

一、指标编写

1. 新手入门 Demo

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

第一步是先定义外置参数 Params，第二步在 on_init 中定义输出图形对象，第三步编写 calculate_all 计算均线输出值的代码，第四步编写 calculate_last 计算均线输出值的代码。

# 双均线案例
# 外置参数
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):
    self.fast_line = self.Line('fast_line', 'rgba(255, 0, 0, 1)')
    self.slow_line = self.Line('slow_line', 'rgba(0, 255, 0, 1)')

# 第一次全量计算指标
def calculate_all(self, data):
    length, datetime_str, open, high, low, close, volume = data
    # 存储数据
    self.length = length
    self.datetime_str = datetime_str
    self.close = close
    for i in range(length):
        # 计算快线
        if i >= 4:
            self.fast_line.set_point(datetime_str[i], sum(close[i-4:i+1]) / 5)
        # 计算慢线
        if i >= 9:
            self.slow_line.set_point(datetime_str[i], sum(close[i-9:i+1]) / 10)

# 实时行情计算增量数据
def calculate_last(self, data):
    datetime_str, open, high, low, close, volume = data
    # 对比时间
    if datetime_str != self.datetime_str[-1]: # 新的k线来了
        self.length += 1
        self.datetime_str.append(datetime_str)
        self.close.append(close)
    else: # 更新数据
        self.close[-1] = close

    # 计算
    self.fast_line.set_point(datetime_str, sum(self.close[-5:]) / 5)
    self.slow_line.set_point(datetime_str, sum(self.close[-10:]) / 10)

2. 新建指标

2.1 创建指标文件夹

策略 -> 指标列表 -> [新增] -> 输入文件名和注释说明 -> 点击[确定]。

2.2 创建 Python 指标文件

a. 策略 -> 指标列表 -> 新增按钮 -> 新增指标弹窗 -> 填写指标名称，选择语言为 “Python” -> 点击确定；

b. 若是第一次使用编辑器的用户需先根据提示下载编辑器；

c. 等待编辑器自动解压完成后， vscode 编辑器会自动弹出系统自带的基础编程代码框架；

d. 用户根据自己的需求进行代码编写，编写完成后按 Ctrl + S 键保存，在空白处右键，点击[Python 编译]，提示[编译成功!]即可。

3. 指标结构

3.1 指标框架

按照指标框架，先导入外置参数，再通过 2 个主要的结构方法函数定义指标的输出对象并进行相应的计算：

# 外置参数
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

3.2 编写规范

‌a.缩进与空格‌

Python 使用缩进来表示代码块，通常使用 4 个空格进行缩进，而不是制表符（Tab）。避免混合使用空格和制表符进行缩进，以保持代码格式的一致性和可读性 ‌。

b.命名规范

（1）类名通常使用大驼峰命名法（UpperCamelCase），变量、函数和模块名应使用小写字母和下划线分隔（snake_case）。避免使用 Python 关键字作为变量名，以防止潜在的命名冲突。

3.3 内置参数

内置参数是指标对象本身（self）的属性，是期魔方在实例化指标对象时自动创建的，用户可以通过 self.xxx 的方式访问内置参数，内置参数包括：

self.name  # 指标名称
self.symbol  # 当前指标加载在行情页面的合约
self.period  # 当前指标加载在行情页面的周期
self.calculate_type  # 指标计算方式 'last' or 'all'
self.calculate_data_type  # 指标计算数据类型 'list' or 'dict' or 'pandas' or 'numpy'

3.4 外置参数

外置参数是指标可以在运行前，通过 UI 界面针对不同周期或合约配置的属性

若没有额外声明 is_main 外置参数，则该指标默认在副图加载：

# 导入参数映射模型
from pydantic import BaseModel

# 设置外置参数
class Params(BaseModel, validate_assignment=True):
    from pydantic import Field
    # 例：选择指标是否在主图显示 True 则为主图显示 False 则为副图显示
    is_main:bool = Field(default=True, title="是否是主图")

目前外置支持 4 种数据类型，分别为：

• 布尔型：bool
• 整型：int
• 浮点型：float
• 字符串：str

不同的类型在指标面板会自动生成对应的输入框，用户可直接在指标面板中修改参数值。

4. 结构性函数

描述

结构性函数为 python 指标的核心，是用户实现其代码的地方，需要用户在结构性函数中编写具体的指标配置以及计算逻辑，期魔方系统会按照指标加载的步骤调用用户编写的结构性函数。

4.1 on_init 初始化函数

描述

初始化函数在程序启动时调用一次： 1.可以初始化自定义属性 2.定义订阅数据的方式 3.定义接收数据的类型 4.初始化输出指标的图形对象

说明

接收数据的方式支持 2 种方式：

• 'last' - 最新数据订阅 (只推送 1 条增量数据，效率更高)【默认方式】
• 'all' - 全量数据订阅（每次推送 300 条 k 线数据）

接收数据的类型支持 4 种方式：

• 'list' - 列表数据接收（推荐）【默认方式】
• 'dict' - 字典数据接收
• 'pandas' - DataFrame 数据接收（推荐）
• 'numpy' - Numpy.ndarray 数据接收

输入参数

主输入参数对象

输出参数

无

接口案例

def on_init(self):
    # 自定义属性
    self.name1 = 'value1'
    self.name2 = 'value2'

    # 定义接收数据方式
    self.calculate_type = 'last'
    # 定义接收数据的类型
    self.calculate_data_type = 'list'

    # 定义输出对象
    # 画线
    """
    1. Line（线条）
    2. ArrowLine（箭头）
    3. ColorKline（彩色k线）
    4. MultiLine（线段）
    5. Text（文字）
    6. MultiText（多段文字）
    7. Point（圆点）
    8. MultiPoint（多段圆点）
    9. Bar（柱形）
    10. Polygon（多边形）
    """
    self.day5_line = self.Line('day5', 'rgba(255, 0, 0, 1)', line_width=2, line_dash=[5, 5])

    self.day10_line = self.Line('day10', 'rgba(0, 255, 0, 1)', line_width=4, line_dash=[10, 10])

    self.arrow1 = self.ArrowLine('arrow1', 'rgba(0, 255, 255, 1)', angle=30, length=10)

数据示例

无

4.2 calculate_all calculate_last 计算函数

描述

接收 k 线数据，计算画出图形的具体数值，并设置数据到图形对象中

说明

默认在指标加载时只推送一次全量数据，后续则只推送增量数据，用户除了实现calculate_all函数外，还需要实现calculate_last函数。

若在on_init(self)中定义了self.calculate_type = 'all'，则每次都推送全量数据，用户则无需实现calculate_last函数。

• calculate_all - 全量计算
• calculate_last - 增量计算

输入参数

主输入参数对象

输出参数

无

接口案例

# 第一次全量计算指标
def calculate_all(self, data):
    length, datetime_str, open, high, low, close, volume = data
    # 存储数据
    self.length = length
    self.datetime_str = datetime_str
    self.close = close
    # 获取外置参数
    fast = self.params.fast
    slow = self.params.slow
    # 列表起点为0，所以起始点为fast-1
    fast_start = fast - 1
    slow_start = slow - 1
    for i in range(length):
        # 计算快线
        if i >= fast_start:
            self.fast_line.set_point(datetime_str[i], sum(close[i-fast_start:i+1]) / fast)
        # 计算慢线
        if i >= slow_start:
            self.slow_line.set_point(datetime_str[i], sum(close[i-slow_start:i+1]) / slow)

# 实时行情计算增量数据
def calculate_last(self, data):
    datetime_str, open, high, low, close, volume = data
    # 对比时间
    if datetime_str != self.datetime_str[-1]: # 新的k线来了
        self.length += 1
        self.datetime_str.append(datetime_str)
        self.close.append(close)
    else: # 更新数据
        self.close[-1] = close
    # 取最新的数据
    last_fast = self.close[-self.params.fast:]
    last_slow = self.close[-self.params.slow:]
    # 计算
    self.fast_line.set_point(datetime_str, sum(last_fast) / self.params.fast)
    self.slow_line.set_point(datetime_str, sum(last_slow) / self.params.slow)

数据示例

无

4.3 on_auth 验证权限函数（非必须）

描述

用户自定义权限验证

说明

用户可以通过期魔方提供的功能性函数 self.get_userinfo() 获取当前运行指标的客户端的用户信息，以实现自定义权限的效果。

权限验证通过则返回值为 True ，权限未通过验证则返回字符串类型文字提示消息（该消息内容支持自定义）。

返回值

接口案例

def on_auth(self):
    # 通过期魔方提供的功能性函数获取用户信息
    userinfo = self.get_userinfo()
    if userinfo['nickname'] not in ['张三', '李四', '王五']:
        return '抱歉，您没有使用该指标的权限'

    return True

数据示例

无

4.4 on_subscribe 订阅多数据函数（非必须）

描述

订阅其他合约和周期的数据

说明

系统默认推送的为当前加载指标的合约以及当前选中的周期的数据，如果用户指标计算需要其他合约或周期的数据，则需要通过此函数订阅。

用户需要使用期魔方提供的功能性函数 self.subscribe() 去订阅多个合约或周期的数据。

周期参数：

• M1 - 1 分钟
• M3 - 3 分钟
• M5 - 5 分钟
• M10 - 10 分钟
• M15 - 15 分钟
• M30 - 30 分钟
• M45 - 45 分钟
• H1 - 1 小时
• H2 - 2 小时
• H4 - 4 小时
• D1 - 日线
• W1 - 周线
• m1 - 月线

返回值

无

接口案例

def on_subscribe(self):
    # 第一个参数为合约名称，后续参数为周期名称，可以订阅多个合约和周期
    self.subscribe(self.symbol, 'M15', 'M30')
    other_symbol = 'ag2504'
    self.subscribe(other_symbol, 'D1')

5. 功能性函数

描述

功能函数是期魔方封装好的函数，会实现某些功能或返回特定的结果，提供给用户使用。

当用户在编写代码时，需要用到这些功能时，可以直接调用。

5.1 subscribe 订阅数据

描述

订阅其他合约和周期的数据

说明

订阅其他品种和周期的数据，以用于计算多品种多周期的指标（非多品种多周期可以不写） 必须调用self.subscribe(symbol, period, period2, period3, ...)

输入参数

输出参数

无

接口案例

见 4.4 案例代码

数据示例

无

5.2 fields 订阅数据字段

描述

自定义订阅的数据字段

说明

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

输入参数

输出参数

无

接口案例

见 7.1 详细说明

数据示例

无

5.3 get_userinfo 获取当前登录用户信息

说明

a. 该函数返回当前登录期魔方用户的信息；

b. 该函数返回的以下 dict 字典中包含用户昵称，用户 ID，手机号，VIP 等级；

c. 在 on_auth() 函数中调用该函数，来判断该用户是否可以运行该指标文件。

注意，该函数返回的值都为字符串类型

输入参数

无

输出参数

DICT 对象

接口案例

见 4.3 案例代码

数据示例

无

5.4 send_get_request 发送 GET 请求

描述

发送 GET 请求

输入参数

输出参数

直接返回接口的返回值

接口案例

url = "xxx"
params = {
    "key1": 'value1',
    "key2": 'value2',
}
re = self.send_get_request(url, params=params)

数据示例

无

5.5 send_post_request 发送 GET 请求

描述

发送 POST 请求

输入参数

输出参数

直接返回接口的返回值

接口案例

url = "xxx"
data = {
    "key1": data1,
    "key2": data2,
}
re = self.send_post_request(url, data=data)

数据示例

无

5.6 Message 发送消息

描述

推送自定义指标预警消息

说明

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

注意，避免循环中使用，避免造成大量发消息卡死

输入参数

输出参数

无

接口案例

def on_init(self):
    # 自定义消息字典，记录发送消息状态
    self.msg_status = {}

def calculate_last(self, data):
    from datetime import datetime

    datetime_str, open, high, low, close, volume = data
    # 当一些条件达成时，这里比如价格大于100时，触发信号
    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)},触发XX信号'
        # 实例化一个消息实例，并设置内容为content变量中的值
        msg = self.Message(content, count=1) # count默认为1，可以不写
        msg.publish() # 发送消息
        self.msg_status[datetime_str] = 'published' # 更新字典，避免重复发送消息

数据示例

无

5.7 Buffer 数据序列

描述

一个数据序列

说明

当指标被当作一个函数使用时，需要定义一个数据序列(Buffer)，用于存储数据和当作返回值。

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

详细说明见 8.1 函数指标案例

输入参数

输出参数

无

接口案例

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]) / fast
            self.fast_buffer[i] = fast_value # 写入buffer

        # 计算慢线
        if i >= slow_start:
            slow_value = sum(self.close[i-slow_start:i+1]) / slow
            self.slow_buffer[i] = slow_value

数据示例

无

5.8 Indicator 函数指标

描述

一个函数指标

说明

当我们调用函数指标时，需要通过 self.Indicator() 方法来创建一个函数指标对象。

注意：函数指标的强大之处在于支持跨合约、跨周期！

详细说明见 8.3 函数指标案例

输入参数

输出参数

无

接口案例

def on_init(self):
    # 使用当前指标的symbol和period，以及默认外置参数
    self.ma_indicator = self.Indicator('ma')
    # 使用ag2506和M15，定义fast=5，slow=10
    self.ag_indicator = self.Indicator('ma', 'ma_ag', symbol='ag2506', period='M15', fast=5, slow=10)
    # 使用au2506和D1，定义fast=10，slow=20
    self.au_indicator = self.Indicator('ma', 'ma_au', symbol='au2506', period='D1', fast=10, slow=20)

数据示例

无

5.9 Logger 日志

描述

一个日志函数，会返回一个日志对象

说明

在开发指标的过程中，我们需要通过日志来记录一些信息，方便调试。

该日志对象提供几个函数用于记录不同等级的日志：

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

输入参数

注意：日志的路径必须为绝对路径，格式需要符合 Windows 的系统格式，如：C:\\Users\\Admin\\Desktop\\indicator.log

输出参数

无

接口案例

def on_init(self):
    # 在on_init中初始化日志对象
    self.logger = self.Logger('C:\\Users\\Admin\\Desktop\\indicator.log')

def calculate_all(self, data):
    # 在calculate_all中使用日志对象记录信息
    self.logger.info('这是一条普通信息')
    self.logger.error('这是一条错误信息')

6. 指标使用

6.1 行情模块应用

开发者在编辑好主图显示的指标文件代码后按 Ctrl + S 键保存并进行语言编译（Python 编译或麦语言编译），编译成功的指标，可在行情 -> 我的指标 -> 自编指标中查看并渲染到 K 线图上展示效果；

一个行情可以加载多个指标，您只需要继续点击加载别的指标即可，并在[我的加载]或行情左上方查看您添加的具体指标；

6.2 任务模块应用

区分好指标文件在主/副图显示后，在任务 -> 启动任务 -> 指标标识 -> 指标面板 -> 选择指标，应用指标在任务实时行情 K 线中。

7. 自定义数据字段

由于指标的计算并非需要使用所有的 K 线数据，为了提升指标的性能，期魔方量化交易平台支持自定义接收数据的方式，用户可以根据自己的需求选择订阅不同数据字段。

7.1 订阅数据字段

为了能够实现自定义数据字段，新增了一个 fields 函数，用于订阅数据的字段。

fields 函数有 3 种使用方法：

1. 不需要声明，则默认订阅 length datetime open high low close volume 这些数据字段，既为默认接收方式（主要是为了兼容旧版指标）。
2. 参数为 *，则订阅所有数据字段。包含length datetime open high low close volume total_turnover open_interest settlement 这些数据字段。
3. 参数为多个字段名，例如 fields('close', 'volume')，则只订阅 close 和 volume 这两个字段。

def on_init(self):
    # 1.不声明 fields 函数，订阅和接收默认数据字段

    # 2.订阅全部数据字段
    self.fields('*')

    # 3.订阅指定数据字段
    self.fields('close', 'volume')

7.2 接收数据字段

注意：不同的订阅数据字段方式，需要使用不同的方式接收数据

当使用第 1 种方式，既不声明fields函数，接收数据方式代码如下：

def calculate_all(self, data):
    # data 是一个元组，里面包含了默认数据字段
    # length 是一个值，表示数据长度
    # datetime/open/high/low/close/volume 是一个列表，里面包含了对应的数据
    length, datetime, open, high, low, close, volume = data

当使用第 2 种和第 3 种方式，接收数据方式代码如下：

def on_init(self):
    # 假设只订阅了 `close` 字段
    self.fields('close')

def calculate_all(self, data):
    # data 是一个元组，里面包含了 length 和 klines
    # length 是一个值，表示数据长度
    # klines 是一个字典，里面包含了对应的数据列表 {'datetime': list, 'close': list}
    # 无论是否订阅 datetime，都会默认包含
    length, klines = data

    self.length = length
    self.datetime = klines['datetime'] # datetime 无需订阅，默认包含
    self.close = klines['close'] # 获取 close 字段的数据

8. 函数指标（指标作为函数使用）

有时我们需要复用一套算法，我们可以将这套算法封装为一个指标函数，然后在其他指标中被调用。

函数指标是指可以不输出图形的指标，通过对数据的计算，返回一个数据序列或多个数据序列(Buffer)。

所以想让一个指标作为函数来使用，必须定义一个数据序列(Buffer)，用于存储数据和当作返回值。

8.1 定义数据序列

数据序列（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')
    # 定义慢线Buffer
    self.slow_buffer = self.Buffer('MA_SLOW')

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

8.2 计算函数的值

当指标被当作一个函数使用时，其计算方法和图形指标的函数一样，使用 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 = self.params.fast
    slow = self.params.slow

    # 列表起点为0，所以起始点为fast-1
    fast_start = fast - 1
    slow_start = slow - 1

    for i in range(length):
        # 计算快线
        if i >= fast_start:
            self.fast_buffer[i] = sum(self.close[i-fast_start:i+1]) / fast

        # 计算慢线
        if i >= slow_start:
            self.slow_buffer[i] = sum(self.close[i-slow_start:i+1]) / slow

def calculate_last(self, data):
    _, kline = data

    # 对比时间
    if kline['datetime'] != self.datetime[-1]: # 新的k线来了
        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 时，不需要像图形指标一样，调用 set_point 函数，而是直接使用 buffer[index] = value 即可。 图形指标的数据是一个字典，而 Buffer 的数据是一个列表。

8.3 调用函数指标

当我们编写好了一个函数指标后，我们可以在其他指标中调用它。

调用定义的函数指标时，我们需要使用 self.Indicator 方法来创建一个函数指标对象。

该方法有以下参数：

1. name：指标名称，必填，必须使用调用的函数指标文件名。
2. id：指标 ID，非必填，当使用多个同 name 的函数指标时需要填入 ID 来区分。
3. symbol：合约代码，非必填，不填时使用当前指标的合约。
4. period：周期，非必填，不填时使用当前指标的周期。
5. params：外置参数，非必填，不填时使用默认值。

注意：当不需要跨合约、跨周期时，不需要指定 symbol period

以下是一些调用案例：

1. 跟随当前指标，使用当前指标的数据

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

2. 同一个函数指标被使用多次时，需要指定 id 来区分

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)

8.4 获取函数指标的值

一共有 2 种方式来获取函数指标的值：

1. 第一种方式，每个函数指标对象有 2 个方法，buffer_all() 和 buffer_last()，分别用于获取所有数据和最新数据。当使用一个函数指标时，推荐使用这种方式获取值
2. 第二种方式，通过 self.gather_all() 和 self.gather_last() 方法来获取所有函数指标的值。当使用了多个函数指标时，推荐使用这种方式获取值

以我们刚才的双均线为例，使用第一种方式获取值的代码如下：

def calculate_all(self, data):
    length, klines = data
    datetime = klines['datetime']
    # 调用函数指标的 `buffer_all` 方法，获取所有数据
    # 假设在8.1和8.2中，我们已经定义了 `fast_buffer` 和 `slow_buffer` 两个Buffer对象，并且命名为 `MA_FAST` 和 `MA_SLOW`

    # ma_buffer 的数据结构为 {'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
    # ma_buffer 的数据结构为 {'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_FAST'])

对于跨品种、跨周期的情况，不用对每一个函数指标使用 buffer_all 方法，而可以使用 gather_all 一次性获取所有值，代码如下：

# 假设：我们有2个跨品种、跨周期的函数指标，分别为 ag_indicator 和 au_indicator
def calculate_all(self, data):
    length, klines = data
    datetime = klines['datetime']
    # 调用 `self.gather_all` 方法，获取所有函数指标的值
    # gather_all 的数据结构为 {'ma_ag': {'MA_FAST': list, 'MA_SLOW': list}, 'ma_au': {'MA_FAST': list, 'MA_SLOW': list}}
    gather_buffer = self.gather_all()
    # 获取ag2506的MA_FAST和MA_SLOW
    ag_fast = gather_buffer['ma_ag']['MA_FAST']
    ag_slow = gather_buffer['ma_ag']['MA_SLOW']
    # 然后做一些事情。。。

def calculate_last(self, data):
    _, kline = data
    # gather_buffer 的数据结构为 {'ma_ag': {'MA_FAST': value, 'MA_SLOW': value},'ma_au': {'MA_FAST': value, 'MA_SLOW': value}}
    gather_buffer = self.gather_last()

二、指标案例

带外置参数的双均线指标案例

在默认指标代码框架中，输出两条线，一条为 5 日均线，另一条为 10 日均线：

# 外置参数
from pydantic import BaseModel
class Params(BaseModel, validate_assignment=True):
    from pydantic import Field
    is_main:bool = Field(default=True, title="是否是主图")
    fast:int = Field(default=5, title="快线")
    slow:int = Field(default=10, title="慢线")

# 初始化图形对象
def on_init(self):
    self.fast_line = self.Line('fast_line', 'rgba(255, 0, 0, 1)')
    self.slow_line = self.Line('slow_line', 'rgba(0, 255, 0, 1)')

# 第一次全量计算指标
def calculate_all(self, data):
    length, datetime_str, open, high, low, close, volume = data
    # 存储数据
    self.length = length
    self.datetime_str = datetime_str
    self.close = close
    # 获取外置参数
    fast = self.params.fast
    slow = self.params.slow
    # 列表起点为0，所以起始点为fast-1
    fast_start = fast - 1
    slow_start = slow - 1
    for i in range(length):
        # 计算快线
        if i >= fast_start:
            self.fast_line.set_point(datetime_str[i], sum(close[i-fast_start:i+1]) / fast)
        # 计算慢线
        if i >= slow_start:
            self.slow_line.set_point(datetime_str[i], sum(close[i-slow_start:i+1]) / slow)

# 实时行情计算增量数据
def calculate_last(self, data):
    datetime_str, open, high, low, close, volume = data
    # 对比时间
    if datetime_str != self.datetime_str[-1]: # 新的k线来了
        self.length += 1
        self.datetime_str.append(datetime_str)
        self.close.append(close)
    else: # 更新数据
        self.close[-1] = close
    # 取最新的数据
    last_fast = self.close[-self.params.fast:]
    last_slow = self.close[-self.params.slow:]
    # 计算
    self.fast_line.set_point(datetime_str, sum(last_fast) / self.params.fast)
    self.slow_line.set_point(datetime_str, sum(last_slow) / self.params.slow)

三、指标图形对象

指标图形对象是指在量化交易中，用于绘制图形的 python 实例。这些对象通常包括各种线条、柱状图、K 线等元素，用以展示价格走势、成交量等信息。

期魔方量化交易平台提供了丰富的指标图形对象，以帮助用户进行技术分析、策略开发等。

图形对象：

1. 线条（Line）：根据 k 线时间点绘制线条，如均线。
2. k 线文字（Text）：根据 k 线时间点绘制文字，如移动止盈信号 ATP。
3. 圆点（Point）：根据 k 线时间点绘制圆点，如多空转折点。
4. 柱状图（Bar）：根据 k 线时间点绘制柱状图，如成交量。
5. 多边形（Polygon）：根据 k 线时间点绘制多边形，如区间突破。
6. 箭头（ArrowLine）：根据 k 线时间点绘制箭头，如突破信号。
7. 彩色 K 线（ColorKline）：根据 k 线时间点绘制彩色 K 线，如涨跌停板。
8. 多段线条（MultiLine）：根据 k 线时间点绘制线段，如区间压力线。
9. 多段文字（MultiText）：根据 k 线时间点绘制多段文字，如触发信号的文字说明。
10. 多段圆点（MultiPoint）：根据 k 线时间点绘制多段圆点，如多空转折点。

说明：

每一个图形对象的使用方式为：

1. 第一步先实例化图形对象，并配置其属性。
2. 第二步使用该图形对象提供的绘制点方法（一般为set_point(key, value, **styles)），写入对应点的值。

图形对象调用示例

1. 线条（Line）

当需绘制一条线条时，可以使用线条对象。

线条对象绘制的是完全连续的线条，可以配置颜色、宽度、虚线等属性。

如果需要绘制不连续的线段，可以使用多段线条（MultiLine）对象。

实例化参数

代码案例

def on_init(self):
    self.line1 = self.Line('line1', 'rgba(255, 0, 0, 1)')
    self.line2 = self.Line('line2', 'rgba(255, 0, 0, 1)', line_width=2, line_dash=[5, 5])

绘制参数

代码案例

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

    for i in range(length):
        self.line1.set_point(datetime_str[i], close[i])

2. k 线文字（Text）

当需要在 k 线上绘制文字时，可以使用 k 线文字对象。

文字会显示在设定的 k 线顶部，或者底部，我们可以配置文字的位置、内容、颜色、字体、居中方式、连线等。

实例化参数

代码案例

def on_init(self):
    self.text1 = self.Text('text1')

绘制参数

代码案例

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

    if close > 100:
        self.text1.set_point(datetime_str, close, 'B信号', color='rgb(255,0,0)', font='18px 微软雅黑', base_line=1, y_move=10)

3. 圆点（Point）

当需要在 k 线上绘制圆点时，可以使用圆点对象。

圆点会显示在设定的 k 线顶部，或者底部，我们可以配置圆点的颜色、弧度等属性。

实例化参数

代码案例

def on_init(self):
    self.point1 = self.Point('point1' , 'rgba(255, 0, 0, 1)', bg_color='rgba(255, 0, 0, 1)', point_radius=8)

绘制参数

代码案例

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

    if close > 100:
        self.point1.set_point(datetime_str, close)

4. 柱状图（Bar）

当需要在副图上绘制柱状图时，可以使用柱状图对象。

比如绘制成交量、MACD 这样的指标，可以配置柱子的颜色、宽度等属性。

实例化参数

代码案例

def on_init(self):
    self.bar1 = self.Bar('bar1', 'rgba(255, 0, 0, 1)', width=8, type=1)

绘制参数

代码案例

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

    for i in range(length):
        self.bar1.set_point(datetime_str[i], 0, volume[i])

5. 多边形（Polygon）

多边形由多边形的角构造而成，需要用到角对象CornerSegment。

多边形绘制需要先实例化角对象，然后使用set_segment()方法将角对象添加到多边形中，而不是使用set_point()方法。

角对象仍然以 k 线时间为键，以价格为值。

实例化参数

代码案例

def on_init(self):
    self.triangle = self.Polygon('triangle', '#FFCC00')

绘制参数

代码案例

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

    corner1 = self.CornerSegment('corner1', datetime_str[-30], high[-1] * 1.02, color='#99CCFF', radius=10, line_width=5, type=0)
    corner2 = self.CornerSegment('corner2', datetime_str[-10], high[-1] * 1.02, color='#99CCFF', radius=10, line_width=5, type=0)
    corner3 = self.CornerSegment('corner3', datetime_str[-20], high[-1] * 0.98, color='#99CCFF', radius=10, line_width=5, type=0)

    self.triangle.set_segment(corner1)
    self.triangle.set_segment(corner2)
    self.triangle.set_segment(corner3)

6. 箭头（ArrowLine）

箭头用于绘制箭头线，可以配置箭头的颜色、宽度、长度等属性。

绘制箭头请使用set_start_point()和set_end_point()方法，而不是使用set_point()方法。

实例化参数

代码案例

def on_init(self):
    self.arrow1 = self.ArrowLine('a1', '#FFCC00', start=True, end=True, angle=30, length=20, line_width=4)

    self.arrow2 = self.ArrowLine('a2', '#33CC66', start=False, end=True, angle=30, length=20, line_width=4)

绘制参数

代码案例

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

    self.arrow1.set_start_point(datetime_str[-50], close[-1])
    self.arrow1.set_end_point(datetime_str[-40], close[-1] * 1.02)

    self.arrow2.set_start_point(datetime_str[-30], close[-1] * 1.01)
    self.arrow2.set_end_point(datetime_str[-20], close[-1] * 0.98)

7. 彩色 K 线（ColorKline）

彩色 k 线可以绘制出不同颜色的 k 线，用于突出显示某些特定的 k 线，或者可以覆盖现有 k 线。

实例化参数

代码案例

def on_init(self):
    self.color_kline = self.ColorKline('main_kline')

绘制参数

代码案例

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

    for i in range(len(datetime_str)):
        self.color_kline.set_point(self.datetime_list[i], color='rgba(0, 255, 255)')

8. 多段线条（MultiLine）

当我们需要绘制不连续的线条时，可以使用多段线条。

多段线条依赖于子对象LineSegment，我们可以通过set_segment()方法添加子对象。

实例化参数

代码案例

def on_init(self):
    self.multi_line = self.MultiLine('multi_line', line_dash=[3, 3])

绘制参数

代码案例

if k == 1:
    self.line_segment = self.LineSegment(f'green_segment', 'rgb(0, 255, 127)')
else:
    self.line_segment = self.LineSegment('red_segment', 'rgb(244,55,50)')

self.line_segment.set_point(self.datetime_list[i], bottom)
self.multi_line.set_segment(self.line_segment)

9. 多段文字（MultiText）

当我们需要绘制不连续的文字时，可以使用多段文字。

多段文字依赖于子对象TextSegment，我们可以通过set_segment()方法添加子对象。

实例化参数

代码案例

def on_init(self):
    self.atp_multi_text = self.MultiText('atp_multi_text')
    self.ttp_multi_text = self.MultiText('ttp_multi_text')

绘制参数

代码案例

self.atp_segment = self.TextSegment('atp_segment')
self.ttp_segment = self.TextSegment('ttp_segment')

if self.low_list[i] <= atp_price:
    self.atp_segment.set_point(self.datetime_list[i], atp_price, atp_str, base_line=2, font=self.my_font)

if self.close_list[i] >= ttp_price:
    self.ttp_segment.set_point(self.datetime_list[i], ttp_price, ttp_str, base_line=2, font=self.my_font)

self.atp_multi_text.set_segment(self.atp_segment)
self.ttp_multi_text.set_segment(self.ttp_segment)

10. 多段圆点（MultiPoint）

当我们需要绘制不连续的圆点时，可以使用多段圆点。

多段圆点依赖于子对象PointSegment，我们可以通过set_segment()方法添加子对象。

实例化参数

代码案例

def on_init(self):
    self.multi_point = self.MultiPoint('multi_point')

绘制参数

代码案例

self.purple_point_segment = self.PointSegment('purple_segment', 'rgb(255,0,255)', point_radius=6)
self.green_point_segment = self.PointSegment('green_segment', 'rgb(0,255,0)', point_radius=6)

for i in range(self.length):
    status = self.generate_status(close[i])
    if status == 1:
        self.purple_point_segment.set_point(self.datetime_list[i], self.low_list[i] * 0.9995)
    else:
        self.green_point_segment.set_point(self.datetime_list[i], self.high_list[i] * 1.0005)

self.multi_point.set_segment(self.purple_point_segment)
self.multi_point.set_segment(self.green_point_segment)

11. 表格（Table）

表格可以用来展示更丰富的综合信息。

定义一个表格，需要定义表格的大小，既行数和列数。

然后需要定义表格的位置和表格的基础样式。

最后，填充表格的标题和内容。

我们可以通过set_table_data()方法一次性添加全部的表格内容，也可以通过set_table_cell()方法给指定的格子更新内容。

实例化参数

代码案例

def on_init(self):
    style = {
        "top": "10px",
        "left": "10px",
        "font_size": 18,
        "font_color": "#fffb8f",
        "border_size": 5,
        "border_color": "#fadb14",
        "background_color": "rgba(255, 255, 255, 0.1)",
        "text_align": "right",
    }
    # 定义一个2 x 2的表格
    self.table = self.Table("my_table", 2, 2, **style)
    title_style = {
        "font_size": 24,
        "font_color": "#f5222d",
        "text_align": "center",
    }
    # 设置表格标题
    self.table.set_table_title("Demo Table", **title_style)

绘制参数

代码案例

def calculate_all(self, data):
    _, klines = data

    # 给表格设定跟随k线时间点
    self.table.set_point(klines["datetime"][-1], klines["close"][-1])
    # 给表格设定表格内容
    table_data = [
        [
            {"content": f"Open: {klines['open'][-1]}", "font_size": 8},
            {"content": f"High: {klines['high'][-1]}", "font_size": 16},
        ],
        [
            {"content": f"Low: {klines['low'][-1]}", "font_size": 24},
            {"content": f"Close: {klines['close'][-1]}", "font_size": 32},
        ],
    ]
    self.table.set_table_data(table_data)

def calculate_last(self, data):
    _, kline = data
    # 给表格设定跟随k线时间点
    self.table.set_point(kline["datetime"], kline["close"])
    # 给表格设定表格内容
    self.table.set_table_cell(0, 0, f"Open: {kline['open']}")
    self.table.set_table_cell(0, 1, f"High: {kline['high']}")
    self.table.set_table_cell(1, 0, f"Low: {kline['low']}")
    self.table.set_table_cell(1, 1, f"Close: {kline['close']}")

四、开发指标

在这一章，我们使用一个完整的开发指标流程来讲解如何开发一个指标。

本教程通过编写我们所熟知的 RSI 指标来熟悉指标开发的步骤。

通过这个教程，您将知道开发指标的每一个步骤，以及遇到问题该如何处理。

1. 新建指标

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

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

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

详细操作截图请参考：https://www.qmfquant.com/app/index.html

2. 进入指标文件并调整模版

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

我们可以看到，新的 python 指标文件并不是空白的，而是包含了一些基础模板

模版代码如下：

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

"""
描述：指标外置参数
说明：
    用于定义指标的可配置参数，如周期、数值等
重要：
    请使用is_main参数区分主图指标和副图指标，
    主图指标的is_main为True，副图指标的is_main为False
"""
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

"""
描述：初始化方法，自定义属性及图形等
是否必须编写：必选
编写具体规范见官网XXXX帮助文档的指标编写的“初始化指标”部分
"""
def on_init(self): 
    pass

"""
描述：计算指标输出对象的值（全量计算）
是否必须编写：必选
编写具体规范见官网XXXX帮助文档的指标编写的“计算指标”部分
"""
def calculate_all(self, data):
    length, datetime_str, open, high, low, close, volume = data
    pass

"""
描述：计算指标输出对象的值（增量计算）
是否必须编写：可选
编写具体规范见官网XXXX帮助文档的指标编写的“计算指标”部分
"""
def calculate_last(self, data):
    length, datetime_str, open, high, low, close, volume = data
    pass

此时，我们可以看到，编辑器会在每次保存时自动触发编译，会输出编译结果（空白模版编译不通过）

第一步，我们先对模版文件进行调整，去掉我们不需要的代码

由于我们不需要验证权限，所以去除验证权限的代码（验证权限内容在1.4.3）

我们也不需要订阅跨品种跨周期数据，所以去除订阅数据的代码（订阅数据内容在1.4.4）

同时去除注释，然后我们得到了如下的代码

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

3. 定义外置参数 params

由于 RSI 指标是副图指标，所以我们需要在外置参数中指定其为副图

RSI 有一个参数是 天数（周期），所以我们在外置参数中声明一个属性来表明这个参数

from pydantic import BaseModel

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，中文名 "周期"
    

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

4. 编写初始化方法 on_init

RSI 不需要每次都计算其历史值，所以我们确定其计算方式为增量计算（定义计算方法内容在1.4.1）

RSI 指标是绘制一条线条，所以我们在初始化时创建一个线条对象（线条对象内容在3.1）

RSI 指标仅需要收盘价就可以完成计算，所以我们声明只需要收盘价数据（声明数据字段内容在1.7.1）

from pydantic import BaseModel

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

def on_init(self): 
    self.calculate_type = 'last' # 声明计算方式 'all' - 全量计算，'last' - 增量计算
    self.calculate_data_type = 'list' # 声明订阅数据类型，'list' - 列表
    self.fields('close') # 声明订阅数据字段，此出我们只订阅 'close' - 收盘价
    self.rsi_line = self.Line('rsi', '#f0f5ff', line_width=3) # 初始化一个线条对象
    #                          名称      颜色        宽度


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

5. 编写全量计算方法 calculate_all

虽然我们的指标是以增量计算来运行的，但是我们需要为其初始化图案时进行一次全量计算

其真正意思为：第一次会接收到600根K线数据，进入calculate_all方法，后续每次接收到1根K线，并进入calculate_last方法

RSI 的相关算法这里不做说明，网上有相关资料，请自行了解

我们需要在calculate_all方法中，处理获取的收盘价数据，并计算出我们需要的值，然后设置给图形对象

def calculate_all(self, data):
    length, klines = data # 1.接收平台推送的K线数据，length - 长度，klines - 一个包含了订阅数据的字典
    
    self.length = length
    self.datetime = klines['datetime'] # 无论是否订阅的 datetime，都会推送k线的时间
    self.close = klines['close'] # 在 on_init 中订阅了 close，所以这里会得到一个包含了收盘价的列表

    # 计算价差
    deltas = []
    for i in range(1, len(self.close)):
        deltas.append(self.close[i] - self.close[i - 1])

    # 若数据量少于周期，返回全为None的列表（防止数据不够）
    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 将其放入指标对象
    # 只要指标在运行，使用 self 放入指标对象的数据都会持续存在
    self.avg_up = up_sum / self.params.period
    self.avg_down = down_sum / self.params.period

    # 初始化RSI值列表，前period + 1个元素为None
    for i in range(0, self.params.period): # 使用 self.params.period 获取我们配置的外置参数，之前配置的是 14
        self.rsi_line.set_point(self.datetime[i], None) # 由于未满 14 天时，我们没有计算出 RSI 的值，所以前 14 天，我们需要给 None 值

    # 计算后续的RSI值
    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值
        rsi = 100 - (100 / (1 + rs))
        rsi = round(rsi, 2)
        self.rsi_line.set_point(self.datetime[i+1], rsi) # 调用 self.rsi_line 的 set_point 来设置我们计算好的值
        # 更新avg（为了增量计算服务）
        if i+1 != len(deltas):
            self.avg_up = avg_up
            self.avg_down = avg_down

虽然代码很多，但是我们把其归纳为3个步骤 第一步：获取数据，处理数据 第二步：使用自己的算法计算出值，既指标的信号 第三步：将数据放入图形对象

6. 编写增量计算方法 calculate_last

在增量计算中，我们需要用到 datetime 来比对当前接收到的K线数据是否为新的一根K线

如果是新的一根K线，那么更新 datetime 并且增加我们的数据长度，这里是 self.datetime 和 self.close

如果是当前K线，那么我们只需要更新当前的价格

对于 set_point 方法来说，第一个参数是时间，第二个参数是值，当检查到时间相同时，则不会新增，只会更新

def calculate_last(self, data):
    _, kline = data
    
    # 对比时间
    if kline['datetime'] != self.datetime[-1]: # 新的k线来了
        self.length += 1
        self.datetime.append(kline['datetime'])
        self.close.append(kline['close'])
        # 更新avg
        self.avg_up = self.last_avg_up
        self.avg_down = self.last_avg_down
    else: # 更新数据
        self.close[-1] = kline['close']
    
    # 计算RSI
    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.rsi_line 的 set_point 来设置我们计算好的值

    # 更新avg
    self.last_avg_up = avg_up
    self.last_avg_down= avg_down

7. 保存并编译

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

编译结果会在控制台输入，如果代码有误，或者无法通过测试数据，控制台会提示响应的错误

比如我们把 set_point 方法写错成了 set_points

self.rsi_line.set_points(self.datetime[i+1], rsi)

我们会得到如下错误：

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

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

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

8. 使用日志来调试

在开发指标的过程中，如果想要知道计算的值是多少，或者排查一些错误，我们需要添加日志

日志的详细内容参考1.5.9

我们在 on_init 中初始化一个日志对象，然后指定一个日志的地址

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 方法中使用日志记录下 rsi 的值

# 计算后续的RSI值
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值
    rsi = 100 - (100 / (1 + rs))
    rsi = round(rsi, 2)
    self.logger.info(f"rsi = {rsi}") # <----- 日志在这里
    self.rsi_line.set_point(self.datetime[i+1], rsi)
    # 更新avg
    if i+1 != len(deltas):
        self.avg_up = avg_up
        self.avg_down = avg_down

然后我们打开 D:\logs\my_rsi.log 查看日志，发现已经输出了对应的值

2025-06-13 14:25:43.697 | INFO     | rsi = 49.91
2025-06-13 14:25:43.697 | INFO     | rsi = 48.98
2025-06-13 14:25:43.697 | INFO     | rsi = 49.98
2025-06-13 14:25:43.697 | INFO     | rsi = 52.97
2025-06-13 14:25:43.697 | INFO     | rsi = 42.49
2025-06-13 14:25:43.698 | INFO     | rsi = 31.65
2025-06-13 14:25:43.698 | INFO     | rsi = 30.96
2025-06-13 14:25:43.698 | INFO     | rsi = 32.02
2025-06-13 14:25:43.698 | INFO     | rsi = 28.9
2025-06-13 14:25:43.698 | INFO     | rsi = 34.21
...

9. 在行情页面加载指标

当我们编写并调试完成后，我们就可以在行情页面查看刚才编写的指标的效果

因为 RSI 是副图指标，所以进入行情页面，选择一个合约，然后在K线图页面点击右键，指标窗口个数选择2个

然后在点击副图的指标图标，找到 我的指标 -> Py_my_rsi（注：所有的python指标会有Py_前缀）

然后填写外置参数，不填写会使用 14 这个默认值，再点击加载指标

最后，我们会在副图中看到一条白色的 RSI 线条，表明我们的指标运行正常

五、常见问题

1. 编写规范和编码语言选择问题

❓:为什么代码编写要规范

A：在代码编写时，一定要注意编写格式的规范性问题，变量、函数和类名在使用情景下，最好使用一眼看上去就能通俗易懂的名称。

例：在编写指标源码时，变量名最好不要使用单个字母命名，如：a、b 等，最好使用有意义的名称，例如：收盘价 close；

2. 代码错误问题

❓:为什么代码编写会出错

A：在编写技术指标源码时，可能会遇到语法错误、逻辑错误等问题。

例：变量未定义、循环未正确结束等，这些问题会导致程序无法正常运行或结果不准确，可在终端 Output 输出中根据报错类型进行相应优化；

3. 指标文件编译失败问题

❓:为什么指标文件会编译失败

A：出现编译失败时，先排查是否已保存好编辑后的代码，再查看编辑器右下角是否已选择好相应的语言编译方式

例：Python 指标代码选择成了麦语言编译，排除以上情况后，如果依旧编译不通过，可在终端 Output 输出中查看报错类型，并进行相应的错误修正；

4. 指标加载失败问题

❓:为什么编译通过的指标在加载时会出错

A: 指标在编译时，会带入测试数据对用户编写的指标代码进行计算并尝试检查输出结果，虽然这样能够保证指标的基本正确性，但是由于在实际加载中，真实的行情数据（不同的合约，不同的周期）更加复杂，所以编译通过的指标代码可能会在真实的行情数据中出现问题，导致指标加载失败。这时，我们可以通过右键期魔方的期货日志查看按钮，然后进入logs\strategy_indicator_server文件夹中，查看indicator.log文件，查看具体的报错信息，从而进行相应的错误修正。

例：测试数据为 5 分钟周期，包含有time字段，而日线周期的指标代码中没有time字段，导致指标通过编译但是无法在日线周期加载，通过查看indicator.log文件，我们可以看到报错信息，然后修正对应的代码。

5. 指标有时正常，有时失败的问题

❓:为什么指标在某个品种可以运行，另一个品种运行失败

A: 当我们编译时，只会加载测试数据，而测试数据并不全面，当我们运行时，也不能对所有品种进行测试，所以编写的代码是不能保证在每种情况下都可以正常运行的。因为不同的品种，数据，开盘时间不同，代码编写时，如果写的兼容性越好则越能适配更多的情况。或者当自己的算法在某些情况下计算会出错，也会导致指标运行失败。

❓:为什么指标刚开始运行正常，过了一会运行失败

A: 这跟指标代码的复杂性相关，也跟代码的编写水平相关，如果代码中算法很多，并且有不同的流程分支，那么某些情况下进入某些分支可能会导致运行失败，既代码考虑的还不够充分。不过我们可以使用 try except来捕获异常，这样指标会跳过计算错误的那一条数据，而不会导致运行失败。

6. 指标图形卡顿的问题

❓:为什么指标画出来后感觉卡顿

A: 这种情况下需要查看代码，是否创建了太多图形对象。我们应该在 on_init 中创建图形对象，如果在 calculate_last 中编写了创建图形对象的代码，则会每次更新价格时都创建一个图形对象，这样会导致图形对象在无限增加，导致画面卡顿。

❓:为什么运行了一段时间后变卡顿

A: 这种情况一般是没有使用增量计算导致，如果一直使用全量运算，并在 self（既指标对象本身）中保存了大量数据，那么会随着指标的运行，导致数据量越来越大。所以请尽量使用增量计算的方式来编写，每次只处理1条增量数据，效率得到极大提高。

7. 指标主/副图显示问题

❓:为什么指标在主图不显示

A：编译通过后的指标，如果在行情主图指标面板中没有找到该指标，可尝试在副图中指标面板里查看，这可能与外置参数中的 is_main 函数返回的布尔值(0 是主图，1 是副图)设置有关

例：当没有声明 is_main 函数则默认显示在副图指标面板中，可以详见 3.3 外置参数说明。

8. 指标文件的导入导出

❓：为什么导入导出文件时候提示失败？

A:（1）本地电脑上保存策略文件的目标路径问题：

例：在本地电脑上保存策略文件的目标路径：导入导出文件时首先要确定好开发者在本地电脑上保存策略文件的目标路径

（2）目标文件类型错误：

例：在策略页面“指标列表”中进行导入时，错误选择成了”策略文件“

（3）指标文件中的结构性函数缺失：

例：缺失关键结构性函数如缺失 on_init() 函数会存在导入失败的提示，此时您应补充完整缺失的结构性函数

（4）导入了外部库：

例：目前平台不支持导入外部库，目前平台集成了：numpy、pandas，可以使用 self.np 和 self.pd 直接进行调用。Python 语言自带的库请在函数内部导入，而不是在头部导入

ps：导出可以选择两种文件：源码文件（.py）和加密文件（.pqmf），导出文件为源码文件是能够被用户看到源码的，而加密文件则不行。因此需要用户根据自己的需求去选择导出两者中的哪一种格式文件

9. 引用外部库问题

❓:为什么使用外部 python 库会出错

A：使用了期魔方尚未支持的库

B：没有在文件内部导入库文件

正确的使用方法：

# 外置参数
from pydantic import BaseModel
class Params(BaseModel, validate_assignment=True):
    from pydantic import Field
    is_main:bool = Field(default=True, title="是否是主图")
    fast:int = Field(default=5, title="快线")
    slow:int = Field(default=10, title="慢线")

目前在期魔方编写指标支持的库文件及版本如下：

Package                   Version
------------------------- -----------
aiohappyeyeballs          2.4.8
aiohttp                   3.10.10
aiosignal                 1.3.2
altgraph                  0.17.4
annotated-types           0.7.0
anyio                     4.6.0
APScheduler               3.10.4
attrs                     25.1.0
certifi                   2025.1.31
cffi                      1.17.1
charset-normalizer        3.4.1
click                     8.1.8
colorama                  0.4.6
cryptography              43.0.0
Cython                    3.0.11
DBUtils                   3.1.0
dnspython                 2.7.0
email_validator           2.2.0
fastapi                   0.115.0
frozenlist                1.5.0
greenlet                  3.1.1
h11                       0.14.0
idna                      3.10
multidict                 6.1.0
mysql-connector-python    9.0.0
numpy                     2.1.2
packaging                 24.2
paho-mqtt                 1.6.1
pandas                    2.2.2
pefile                    2024.8.26
pip                       24.0
propcache                 0.3.0
pycparser                 2.22
pycryptodome              3.20.0
pydantic                  2.9.2
pydantic_core             2.23.4
pyinstaller               6.10.0
pyinstaller-hooks-contrib 2025.1
PyMySQL                   1.1.1
Pyro4                     4.82
python-dateutil           2.9.0.post0
pytz                      2025.1
pywin32-ctypes            0.2.3
requests                  2.32.3
serpent                   1.41
setuptools                75.8.2
six                       1.17.0
sniffio                   1.3.1
SQLAlchemy                2.0.35
starlette                 0.38.6
typing_extensions         4.12.2
tzdata                    2025.1
tzlocal                   5.3
urllib3                   2.3.0
uvicorn                   0.32.0
websockets                12.0
win11toast                0.35
winsdk                    1.0.0b10
yarl                      1.18.3