指标编写文档

关于期魔方

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

核心功能：

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

指标外置参数配置部分，若没有额外声明 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)

四、常见问题

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

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

6. 指标文件的导入导出

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

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

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

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

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

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

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

（4）导入了外部库：

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

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

7. 引用外部库问题

❓:为什么使用外部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