1 快速入门

1.1 软件操作

1.1.1 策略的新建与编译

策略 -> 策略列表 -> 我的策略 -> 新增策略 -> 填写策略名称，选择语言为 Python（已固定），添加注释说明 -> 确定

如果系统提示下载编写工具，就点击下载安装包，编辑器解压进度条 100% 后，会自动跳转到 VSCode 编辑器页面

以下是双均线策略代码举例： 当 5 日均线上穿 20 日均线时，平空买多； 当 5 日均线下穿 20 日均线时，平多卖空；

"""
文件类型: 量化策略
帮助文档: https://qmfquant.com/static/doc/code/strategyEdit.html
期魔方，为您提供专业的量化服务
"""

"""
描述：导入需要使用的库
是否必须编写：可选
"""
import numpy as np  # 导入 numpy 库便于进行数值计算
import talib as ta  # 导入 talib 库用于计算均线指标

"""
描述：策略外置参数配置的部分
是否必须编写：可选
编写具体规范见官网“策略编写”文档的策略编写的“外置参数编写”部分
"""
from pydantic import BaseModel, Field

class Params(BaseModel, validate_assignment=True):
    """参数映射模型"""
    """设置外置参数:手数"""
    # 举例：交易手数的文字文本框的配置代码：volume:int = Field(default=1, title="交易手数")
    pass

"""
描述：该函数在策略被启动前运行一次；用于配置策略中需要的初始化逻辑；
必要的初始化部分请查询帮助文档
是否必须编写：必选
"""
def on_init(context):
    context.base_instrument_id = BASE_SETTING.get("Instrument")  # 获取运行合约的名称
    context.base_period = BASE_SETTING.get("bPeriod")  # 获取运行合约的周期
    context.ma5_array = np.zeros(5)  # 初始化 5 周期均线数组，长度为 5，初始值为 0
    context.ma20_array = np.zeros(20)  # 初始化 20 周期均线数组，长度为 20，初始值为 0
    context.position = 0  # 理论持仓量（0 表示不持仓，正数表示多头持仓，负数表示空头持仓）

"""
描述：主要的运行逻辑需要编写在 on_tick 函数中；策略成功启动后，每一次报价都会调用一次该函数；
是否必须编写：必选
"""
def on_tick(context):
    # on_bar_run 让 on_bar 函数只在每一个 K 线新产生时运行一次，而不是每一个 tick 都运行一次，这样可以使策略收线运行
    on_bar_run(on_bar, context)

"""
描述：每一次新的 K 线数据产生时，调用该函数；
是否必须编写：可选
"""
def on_bar(context):
    # 获取当前合约当前周期的 K 线数据(计算 5 周期和 20 周期均线分别需要 5 根和 20 根 K 线数据，所以这里直接取 300 根 K 线数据进行计算)
    context.klines = get_kline(context.base_instrument_id, context.base_period, 300)
    # 获取 K 线数据中的收盘价数据
    context.close_array = context.klines.get("close")
    # 计算 5 周期和 20 周期均线
    ma5_array = ta.SMA(np.asarray(context.close_array), 5) # np.asarray() 将列表转化为数组，方便 talib 库计算
    ma20_array = ta.SMA(np.asarray(context.close_array), 20) # np.asarray() 将列表转化为数组，方便 talib 库计算

    # 移动是因为计算出来的数据包括了当前未闭合的 K 线数据,所以需要将当前 K 线数据去掉
    context.ma5_array = np.roll(context.ma5_array, -1) # np.roll() 将数组中的元素向前移动一个位置
    context.ma20_array = np.roll(context.ma20_array, -1) # np.roll() 将数组中的元素向前移动一个位置
    
    context.ma5_array[-1] = ma5_array[-1] # 将计算得到的 5 周期均线赋值给 context.ma5_array 数组的最后一个元素
    context.ma20_array[-1] = ma20_array[-1] # 将计算得到的 20 周期均线赋值给 context.ma20_array 数组的最后一个元素

    volume = 1 # 这里我们每次交易 1 手，也可以在外置参数中设置并在 on_init 函数中获取外置参数中设置的手数

    # 计算出来的 5 周期均线和20 周期均线数量不足以判断时，不进行交易
    if len(context.ma5_array) < 3 or len(context.ma20_array) < 3:
        return

    # 发生了死叉情况（即排除最新一根 K 线后，往前数第二根 5 周期均线小于 20 周期均线且往前数第一根 5 周期均线大于20 周期均线），如果当前持有多仓则平多仓
    if context.ma5_array[-3] >= context.ma20_array[-3] and context.ma5_array[-2] < context.ma20_array[-2]:
        if context.position > 0: # 如果持有多仓
            sell_close(volume, context.base_instrument_id) # 平多仓
            context.position = 0 # 更新持仓量
    # 发生了金叉情况（即排除最新一根 K 线后，往前数第二根 5 周期均线大于20 周期均线且往前数第一根 5 周期均线小于20 周期均线），如果当前持有空仓则平空仓
    elif context.ma5_array[-3] <= context.ma20_array[-3] and context.ma5_array[-2] > context.ma20_array[-2]:
        if context.position < 0: # 如果持有空仓
            buy_close(volume, context.base_instrument_id) # 平空仓
            context.position = 0 # 更新持仓量

    # 发生了金叉情况，如果当前没有持仓则开多仓
    if context.ma5_array[-3] <= context.ma20_array[-3] and context.ma5_array[-2] > context.ma20_array[-2]: 
        if context.position == 0:  # 如果没有持仓
            buy_open(volume, context.base_instrument_id)  # 开多仓
            context.position = volume # 更新持仓量
    # 发生了死叉情况，如果当前没有持仓则开空仓
    elif context.ma5_array[-3] >= context.ma20_array[-3] and context.ma5_array[-2] < context.ma20_array[-2]:
        if context.position == 0: # 如果没有持仓
            sell_open(volume, context.base_instrument_id) # 开空仓
            context.position = -volume # 更新持仓量

"""
描述：当策略触发停止或报错时，调用该函数；可在函数中做一些结算的逻辑；
是否必须编写：可选
"""
def on_stop(context):
    pass

"""
描述：报单回复接口，每次接收到报单时，该函数会被调用，报单信息在 order 对象中；
详细请查询帮助文档；
是否必须编写：可选
"""
def on_order(context,order):
    pass

"""
描述：报单回复接口，每次接收到成交单时，该函数会被调用，成交信息在 trade 对象中；
详细请查询帮助文档；
是否必须编写：可选
"""
def on_trade(context,trade):
    pass

"""
描述：当挂载账号出现错误操作时，该函数会被调用，错误信息在 err 对象中；
是否必须编写：可选
"""
def on_error(context,err):
    pass

编辑好策略文件后，保存文件，在文件中单击右键，选择 Python 编译，编译输出中显示编译成功即可。

【注意】在编辑器中不要运行代码，否则会报错。

1.1.2 回测运行与结果展示

回测栏 -> 添加回测单元 -> 在参数设置页面填写基础参数和策略参数后保存 -> 点击回测

1.1.2.1 批量回测与综合报表

在回测功能视图右上角有综合报表按钮，综合报表结构与单策略类似

回测完成后点击详情查看报告，综合报告，收益回撤，图表分析，明细详情，参数信息等，回测日志区内容在软件退出后会保存到日志目录，如用户目录的 AppData\Local\qmfquant\logs\backtest_server 下面

1.1.3 任务启动

任务栏 -> 双击策略 -> 填写任务参数 -> 提交任务 -> 启动

可以在任务界面查看实时 K 线和日志内容，日志内容在客户端退出后会保存到日志目录，如用户目录的 AppData\Local\qmfquant\logs\task_server 下面

1.2 策略结构

一个完整的策略，一般包含 6 个结构性函数：on_tick，on_init，on_stop，on_order，on_trade，on_error，如果需要收线执行，还需添加 on_bar 函数

如果需要输入外置参数，还需添加外置参数配置类：class Params

"""
文件类型: 量化策略
帮助文档: https://qmfquant.com/static/doc/code/strategyEdit.html
期魔方，为您提供专业的量化服务
"""

"""
描述：策略外置参数配置的部分
是否必须编写：可选
编写具体规范见官网“策略编写”文档的策略编写的“外置参数编写”部分
"""
from pydantic import BaseModel, Field
class Params(BaseModel, validate_assignment=True):
    """参数映射模型"""
    """设置外置参数:手数"""
    #举例：交易手数的文字文本框的配置代码：volume:int = Field(default=1.0,title = "交易手数")
    pass

"""
描述：该函数在策略被启动前运行一次；用于配置策略中需要的初始化逻辑；
必要的初始化部分请查询帮助文档
是否必须编写：必选
"""
def on_init(context):
    pass

"""
描述：主要的运行逻辑需要编写在 on_tick 函数中；策略成功启动后，每一次报价都会调用一次该函数；
是否必须编写：必选
"""
def on_tick(context):
    on_bar_run(on_bar, context) # 如不需要收线执行，则可以不写这行代码

"""
描述：每一次新的 K 线数据产生时，调用该函数；
是否必须编写：可选
"""
def on_bar(context):
    pass

"""
描述：当策略触发停止或报错时，调用该函数；可在函数中做一些结算的逻辑；
是否必须编写：可选
"""
def on_stop(context):
    pass

"""
描述：报单回复接口，每次接收到报单时，该函数会被调用，报单信息在 order 对象中；
详细请查询帮助文档；
是否必须编写：可选
"""
def on_order(context,order):
    pass
    
"""
描述：报单回复接口，每次接收到成交单时，该函数会被调用，成交信息在 trade 对象中；
详细请查询帮助文档；
是否必须编写：可选
"""
def on_trade(context,trade):
    pass

"""
描述：当挂载账号出现错误操作时，该函数会被调用，错误信息在 err 对象中；
是否必须编写：可选
"""
def on_error(context,err):
    pass

1.2.1 on_init 初始化函数

说明

a. 该方法每次任务启动只执行一次；

b.用于设置策略交易逻辑中的初始化部分。

输入参数

返回值

无

接口案例

def on_init(context):
    pass

数据示例

无

1.2.2 on_tick 运算函数

说明

a. 策略的品种合约的 TICK 变化时调用的方法；

b. 行情报价每进来一次就会调用函数 on_tick 刷新一次。

输入参数

返回值

无

接口案例

def on_tick(context):
    pass

数据示例

无

1.2.1.1 on_bar_run 运行 K 线处理逻辑

说明

用来注册 K 线处理回调函数，此回调函数只在每根 K 线收线时运行一次

输入参数

返回值

无

接口案例

def on_tick(context):
    ...
    # tick 数据处理逻辑
    tick_data_process(context)
    ...
    # on_bar 函数中处理 K 线数据
    on_bar_run(on_bar, context, other_params)


def on_bar(context, other_params):
    """需要被逐根 K 线运行的策略代码"""
    ...

执行流程图解：

行情 Tick 到达
  ↓
执行 on_tick()
  ├─ 实时平仓逻辑（每个 Tick 都执行）
  └─ on_bar_run()  # 检查 K 线闭合
      ↓
      if 新 K 线生成:
          执行 on_bar() 中的策略逻辑
      else:
          跳过

通过这种设计，既能保证实时性的平仓操作，又能确保开仓信号在 K 线闭合时准确触发，避免在未成形 K 线上做出错误决策。

数据示例

无

1.2.3 on_order 接收委托单信息

说明

a. 返回委托单信息的方法，注意此处的 order 报单信息是未成交状态（包含字段详见以下对象 ORDER_DATA，与 on_trade 中 order 对象的值会有一定差异）；

b. 用于存储和访问交易过程中的各种状态和信息；

c. 可在此处对订单状态进行分析和调控。

输入参数

ORDER_DATA 委托单对象明细字段信息

返回值

无

接口案例

def on_order(context, order):
    # 获取订单状态 
    order_status = order.get("OrderStatus")
    CombOffsetFlag = order.get("CombOffsetFlag")
    StatusMsg = order.get("StatusMsg")
    if order_status == "5":
        if CombOffsetFlag == "0":
            # 清除开启任务等待序列
            context.open_task.clear()
        else:
            # 清除关闭任务等待序列
            context.close_task.clear()

数据示例

无

注意事项

每次 OrderStatus 发生改变，该函数会返回一次，以 OrderStatus 为特征，目前该函数的返回情况只有以下几种：

a -> 0

a -> 3 -> 0

a -> 3 -> 4 -> 0

a -> 5

a -> 3 -> 5

1.2.3.1 on_allorder 接收订单发送请求的回复

说明

on_order 只能收到本任务发送请求的回复，on_allorder 会接收任务挂在账户的所有的订单回复，仅在任务中使用，回测中不可用

输入参数

返回值

无

接口案例

def on_init(context):
    pass

def on_tick(context):
    pass

def on_allorder(context,order):
    put_log("收到order信息","INFO")
    put_log(f"{order}","INFO")

数据示例

无

1.2.4 on_trade 接收成交单信息

说明

接收成交单信息的方法，注意此处的 order 报单信息是已成交状态，包含字段详见对象 ORDER_DATA（同 on_order 中的 order 对象，但值会有一定差异）。

输入参数

ORDER_DATA 成交单对象明细字段信息

返回值

无

接口案例

def on_trade(context, order):
    pass

数据示例

无

1.2.5 on_error 接收 CTP 错误

说明

接收 CTP 端口发送过来的错误信息数据 data（暂只供接收信息，无法输出该信息数据内容）。

输入参数

返回值

无

接口案例

def on_error(context, data):
    pass

数据示例

无

1.2.6 on_stop 策略停止函数

说明

a. 策略结束时调用，可以执行一些结尾统计操作 ；

b. 保存策略的回测数据或策略回测中途发生错误或者停止都会调用 on_stop，并返回信息。

输入参数

返回值

无

接口案例

def on_stop(context):
    pass

数据示例

无

1.2.7 外部参数与全局变量

1.2.7.1 外置参数

‌描述

外置参数是策略文件中自定义的变量，用于配置和控制策略的行为

作用‌

a. 允许用户根据需求动态调整策略；

b. 提供灵活性，使策略能适应不同环境和需求；

c. 通过修改参数，可以控制策略在特定条件下的表现。

配置步骤

a. 导入参数映射模型；

b. 设置外置参数。

以下是策略外置参数目前可支持 5 种类型：

接口案例

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

class Params(BaseModel, validate_assignment=True):
    # b.设置外置参数
    # 配置外置参数可支持类型：
    INT_TEST: int = Field(default=1, title="整数")
    FLOAT_TEST: float = Field(default=1.0, title="浮点数")
    STR_TEST: str = Field(default="test", title ="字符串")
    DICT_TEST: dict = Field(default={"options":["选项1", "选项2", "选项3"], "value":"选项1"}, title ="下拉条")
    BOOL_TEST:bool = Field(default=True, title = "布尔")
    # 完成外置参数配置后的具体效果，可在期魔方添加任务和回测时的弹窗查看

1.2.7.2 BASE 全局变量

描述

在策略文件（尤其是涉及编程或脚本的策略文件，如某些自动化策略、交易策略等）中，BASE 全局变量通常指的是一个基础或核心的变量，它在整个策略文件或程序中都是可访问和可修改的。BASE 变量的具体含义和用途可能因策略的不同而有所差异。

作用

全局变量在程序中具有广泛的作用，主要体现在以下几个方面：

a. 数据共享‌：全局变量能够在不同的函数或代码块中共享数据，实现多个函数对相同数据的访问和修改，从而加强函数之间的联系‌；

b. 配置信息存储‌：全局变量可用于存储程序的配置信息，如数据库连接信息、API 密钥等，便于在整个程序中轻松访问这些信息‌；

c. 扩大作用域‌：全局变量的作用域是整个程序，可以在程序的任何地方访问，这增加了变量的可用性和灵活性‌。

然而，全局变量的使用也需谨慎，因其作用范围广，容易被不同函数修改，可能导致代码的可读性和可维护性下降，还可能引发命名冲突和命名空间污染等问题‌。

用户可以在编写策略文件中根据需求自行调用以下的 BASE 全局变量：

"""执行模式"""
BASE_MODE = "REAL"

"""期货交易账户id"""
BASE_USERID = "1111111"

"""每个回测任务后端唯一的标识id，不在前端展示"""
BASE_ID = "135DAWQE654DAQWE321D5XAS654"

"""基础产品交易所"""
BASE_EXCHANGEID = "simnow"

"""基础产品"""
BASE_SYMBOLS = ['c2503', 'c2505']

"""基础设置"""
BASE_SETTING = {}

"""日志"""
BASE_LOG_DATA = []

"""写日志对象"""
BASE_LOG = None

"""产品细则对象"""
BASE_SYMBOLS_INFO = {}

"""获取产品tick"""
SYMBOLS_TICKS = {}

"""基础产品运行tick 该对象可获取当前基础产品的最新tick对象"""
BASE_SYMBOL_TICK = {}

"""多产品调用申明"""
"""格式: 合约id1:周期1&合约id2:周期2
   例如: ag2512:D1&rb2510:D1&ag2512:M15
   例子定义了3个子产品数据;
   这里的定义不会影响BASE产品的数据获取；
"""
QMF_SUBSCRIBE_SYMBOLS = ""

常用对象

BASE_SETTING：

BASE_SYMBOL_TICK：

接口案例

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


class Params(BaseModel, validate_assignment=True):
    # 设置外置参数，此处仅以 int 举例...
    INT_TEST: int = Field(default=1, title="整数")
    ...
# 调用全局变量以获取 BASE_USERID 账户id
BASE_USERID = "12121212"
# 打印 BASE_USERID 账户id
print(f"BASE_USERID:{BASE_USERID}")

#  BASE_USERID 账户id 的打印结果:
BASE_USERID:12121212

1.2.7.3 多品种多周期

运行多产品策略时，运算过程中会存在各标的合约运算频率不一致的情况，为避免此等情况，多产品标的合约均统一以“运行品种”为主频率进行运算，但会存在所选择“运行品种”的频率运算会比多产品策略中的部分品种低，因此在回测页面或任务页面运行多产品策略时，建议开发者在选择“运行品种”时设置交易量活跃的高频率主力品种。

而在多产品策略下，要想实现一个策略同时订阅多个标的合约 tick 时，需在策略外置参数类中申明 QMF_SUBSCRIBE_SYMBOLS 全局变量。

申明方式

QMF_SUBSCRIBE_SYMBOLS = "First_symbol:First_period&second_symbol:second_period&....."

接口案例

from pydantic import BaseModel, Field

# 外置参数
# 参数映射模型
class Params(BaseModel, validate_assignment=True):
    volume: int = Field(default=1, title="手数")
    QMF_SUBSCRIBE_SYMBOLS: str = Field(default='j2505:M5&jm2505:M5', title="订阅合约和周期")
    # ...
    
def on_tick(context):
    ...

1.3 策略编写

1.3.1 引用外部库

分为两种情况，一种是期魔方软件已经安装的库，一种是软件没有安装的库。

1.3.1.1 软件中已安装的库

比如 talib 库，已经安装在期魔方软件内部，直接在函数中导入即可使用：

import numpy as np
import talib as ta

# 计算均线指标数据
def calc_sma(context):
    len1 = context.len1
    len2 = context.len2   
    context.close_array = context.klines.get("close")
    ma1_array = ta.SMA(np.asarray(context.close_array),len1)
    ma2_array = ta.SMA(np.asarray(context.close_array),len2)

目前在期魔方平台编写策略支持的库文件及版本如下：

aiohttp                   3.9.5
aiosignal                 1.3.1
akshare                   1.16.9
altgraph                  0.17.4
annotated-types           0.6.0
anyio                     4.3.0
APScheduler               3.10.4
arch                      7.2.0
astropy                   7.0.1
astropy-iers-data         0.2025.2.17.0.34.13
asttokens                 2.4.1
attrs                     23.2.0
beautifulsoup4            4.12.3
certifi                   2024.7.4
cffi                      1.17.1
charset-normalizer        3.3.2
clarabel                  0.10.0
click                     8.1.7
colorama                  0.4.6
contourpy                 1.3.0
cryptography              43.0.1
cvxpy                     1.6.1
cycler                    0.12.1
DBUtils                   3.1.0
decorator                 5.1.1
dnspython                 2.6.1
email_validator           2.1.1
et_xmlfile                2.0.0
executing                 2.1.0
fastapi                   0.110.1
fonttools                 4.54.1
frozendict                2.4.6
frozenlist                1.4.1
h11                       0.14.0
html5lib                  1.1
idna                      3.7
ipython                   8.29.0
jedi                      0.19.1
joblib                    1.4.2
jsonpath                  0.82.2
kiwisolver                1.4.7
loguru                    0.7.2
lxml                      5.3.0
matplotlib                3.9.2
matplotlib-inline         0.1.7
mini-racer                0.12.4
multidict                 6.0.5
multitasking              0.0.11
mysql-connector-python    8.3.0
networkx                  3.4.2
ntplib                    0.4.0
numpy                     1.26.4
objprint                  0.2.3
openpyxl                  3.1.5
osqp                      0.6.7.post3
packaging                 24.1
paho-mqtt                 1.6.1
pandas                    2.2.2
parso                     0.8.4
patsy                     1.0.1
peewee                    3.17.7
pefile                    2023.2.7
pillow                    11.0.0
pip                       25.0.1
platformdirs              4.3.6
prompt_toolkit            3.0.48
pure_eval                 0.2.3
py-mini-racer             0.6.0
pybind11                  2.13.6
pycparser                 2.22
pydantic                  2.6.4
pydantic_core             2.16.3
pyerfa                    2.0.1.5
Pygments                  2.18.0
pyinstaller               6.10.0
pyinstaller-hooks-contrib 2024.9
PyMySQL                   1.1.0
pyparsing                 3.2.0
python-dateutil           2.9.0.post0
pytz                      2024.1
pywin32-ctypes            0.2.2
PyYAML                    6.0.2
qdldl                     0.1.7.post5
QuantStats                0.0.64
requests                  2.31.0
Riskfolio-Lib             7.0.0
scikit-learn              1.6.1
scipy                     1.14.1
scs                       3.2.7.post2
seaborn                   0.13.2
setuptools                72.1.0
six                       1.16.0
sniffio                   1.3.1
soupsieve                 2.6
stack-data                0.6.3
starlette                 0.37.2
statsmodels               0.14.4
TA-Lib                    0.4.28
tabulate                  0.9.0
threadpoolctl             3.5.0
tqdm                      4.67.1
traitlets                 5.14.3
typing_extensions         4.12.2
tzdata                    2024.1
uvicorn                   0.29.0
wcwidth                   0.2.13
webencodings              0.5.1
websockets                12.0
win32-setctime            1.1.0
xlrd                      2.0.1
XlsxWriter                3.2.2
yarl                      1.9.4
yfinance                  0.2.48

1.3.1.2 软件中没有的库

比如 statsmodels 库，期魔方软件中没有安装，用户可以首先把库安装到本地电脑的 Python 环境中，然后在策略中把本地 Python 库的 lib 路径添加到环境变量中。

def register_python_env(env_path):
    import sys, os
    if not os.path.isabs(env_path):
        put_log(f'必须使用绝对路径', 'ERROR')
        return -1
    # 避免重复添加
    normalized_path = os.path.normcase(os.path.abspath(env_path))
    if normalized_path not in [os.path.normcase(p) for p in sys.path]:
        sys.path.append(normalized_path)
    return 0

def on_init(context):
    register_python_env(r'python_dir\Lib\site-packages')

def on_tick(context):
    on_bar_run(on_bar, context)

def on_bar(context):
    import statsmodels.api as sm
    # 创建数据
    X = [1, 2, 3, 4, 5]
    y = [2, 3, 5, 7, 11]
    # 添加常数项
    X = sm.add_constant(X)
    # 拟合线性回归模型
    model = sm.OLS(y, X).fit()
    result = model.summary()

其中，python_dir 是用户电脑的 Python 路径。如果出现兼容性问题，可以查看下是否因为期魔方 Python 版本不支持。策略程序中使用 sys.version 可以查看当前调用 Python 的版本。

1.3.2 数据获取

1.3.2.1 实时数据

1.3.2.1.1 get_tick 获取合约报价

说明

获取指定合约 id 的当前最新报价

输入参数

返回值

返回一个字典

接口案例

def on_tick(context):
    # 获取合约 ag2412 当前的最新报价
    ag2412_tick = get_tick("ag2412")

数据示例

{
    'ActionDay':  "", 
                    'AskPrice1': '0', 
                    'AskPrice2': 0, 
                    'AskPrice3': 0, 
                    'AskPrice4': 0, 
                    'AskPrice5': 0, 
                    'AskVolume1': 0
    # 以下内容已省略
    # ...
}

1.3.2.1.2 get_symbolinfo 获取合约信息

说明

返回一个 dict，从 CTP 获取指定品种代码或合约代码产品细则数据

输入参数

返回值

返回一个字典

接口案例

def on_tick(context):
    # 获取沪银产品信息
    ag_symbolinfo = get_symbolinfo("ag")
    print(f"{ag_symbolinfo}")

数据示例

{
    'id': 1,
    'exchange_id': 'SHFE',
    'product_class': 'ag',
    'product_name': '沪银',
    'volume_multiple': 15,
    # 以下内容已省略
    # ...
}

1.3.2.1.3 get_current_main_symbol 获取当前主力合约

说明

获取当前主连合约映射的具体合约

输入参数

返回值

接口案例

def on_tick(context):
    ...
    # 获取玉米当前的主力合约
    hot_symbol = get_current_main_symbol('c888')

数据示例

无

1.3.2.2 历史数据

1.3.2.2.1 get_kline 获取 K 线数据

说明

获取 K 线的开高低收、时间周期、成交量的序列。

输入参数

返回值

返回一个字典

接口案例

klines = get_kline("ru2501","M1",3)
print(f"{klines}")

数据示例

{
    'close': [7687.0, 7686.0, 7647.0], 
    'open': [7687.0, 7687.0, 7686.0],
    'high': [7687.0, 7688.0, 7686.0],
    'low': [7687.0, 7686.0, 7647.0],
    'datetime': [Timestamp('2024-11-15 14:57:00'), 
                 Timestamp('2024-11-15 14:58:00'), 
                 Timestamp('2024-11-15 14:59:00')],
    'volume': [0.0, 3.0, 3966.0]
}

1.3.2.2.2 time_current 获取当前时间

说明

获取当前时间，在回测中返回当前行情报价时间，在实际任务中返回当前的系统时间

输入参数

无

返回值

接口案例

def on_tick(context):
    ...
    # 获取当前时间
    current_dt = time_current()

数据示例

无

1.3.2.3 账户数据

1.3.2.3.1 get_account 获取 CTP 帐户信息

说明

获取当下已登录期货公司账户信息

输入参数

无

返回值

返回一个字典

接口案例

def on_init(context):
    account_info = get_account()
    print(f"{account_info}")

数据示例

{
    'AccountID': '182958',
    'Available': 20833162.898450002,
    'Balance': 20940449.198450003,
    'BizType': '',
    'BrokerID': '9999',
    'CashIn': 0.0, 
    # 以下展示内容已省略
    # ...
}

1.3.2.3.2 get_position 获取持仓信息

说明

返回一个 list，用于获取用户目前期货交易账户中所有持仓明细详情的函数。

输入参数

无

返回值

返回一个字典

接口案例

def on_init(context):
    # 获取当前持仓的明细
    position_info = get_position()
    print(f"{position_info}")

数据示例

[
    {
        'AbandonFrozen': 0,
        'BrokerID': '9999',
        'CashIn': 0.0,
        'CloseAmount': 0,
        'CloseProfit': 0
    	# 以下展示内容已省略
    	# ...
    }
]

1.3.2.5 衍生数据

a) 衍生数据接口有使用频率限制，在策略回测模式中，建议在 on_init 函数中一次性把需要的数据都取出来，不要在 on_tick 函数中频繁调用接口；

b) 所有衍生数据都是通过市场公开数据收集而来，数据准确性期魔方不做保证，仅做参考

1.3.2.5.1 get_futures_inventory_all 获取大宗商品库存数据

说明

返回指定 symbol 的具体品种大宗商品库存数据

输入参数

库存品种字典

返回值

库存数据 JSON 对象的 data 对象

接口调用案例

调用接口代码

from futures_data_client import fetch_futures_inventory_99

result1 = get_futures_inventory_all('ag', '2025-02-1', '2025-02-27')
print(result1)

数据返回案例

{
	'code': 0,
	'data': {
		'columns': ['日期', '收盘价', '库存'],
		'data': [
			['2025-02-07', 8055.0, 120081],
			['2025-02-21', 8102.0, 122764]
		]
	},
	'msg': '获取数据成功'
}

1.3.2.5.2 get_futures_inventory_60days 获取仓单日报的库存数据

说明

获取仓单日报的库存数据，目前是近 60 个交易日的数据

输入参数

仓单日报库存品种字典

返回值

仓单日报库存数据JSON对象data对象

接口调用案例

调用接口代码

from futures_data_client import fetch_futures_inventory_em

result1 = get_futures_inventory_60days('ag', '2025-02-27', '2025-02-28')
print(result1)

数据返回案例

{
	'code': 0,
	'data': {
		'columns': ['日期', '库存', '增减'],
		'data': [
			['2025-02-27', 1293568, -7845.0],
			['2025-02-28', 1277102, -16466.0]
		]
	},
	'msg': '获取数据成功'
}

1.3.2.5.3 get_futures_foreign_data 获取外盘期货历史行情数据

说明

获取期货外盘历史行情数据

输入参数

外盘品种数据字典

返回值

查询接口返回对象

外盘数据 JSON 对象的 data 对象

接口调用案例

调用接口代码

from futures_data_client import fetch_futures_foreign_hist

result1 = get_futures_foreign_data('XAU', '2025-03-06', '2025-03-07')
print(result1)

数据返回案例

{
	'code': 0,
	'data': {
		'columns': ['日期', 'open', 'high', 'low', 'close', 'volume', 'position', 's'],
		'data': [
			['2025-03-06', 2918.54, 2926.44, 2891.19, 2910.76, 0, 0, 0],
			['2025-03-07', 2911.04, 2930.12, 2896.56, 2912.0, 0, 0, 0]
		]
	},
	'msg': '获取数据成功'
}

1.3.2.5.4 get_futures_comex_inventory_data 获取黄金和白银的库存数据

说明

查询黄金/白银的所有库存数据

输入参数

返回值

库存数据 JSON 对象 data 对象

接口调用案例

调用接口代码

from futures_data_client import fetch_futures_comex_inventory

result1 = get_futures_comex_inventory_data('白银', '2025-3-6', '2025-3-7')
print(result1)

数据返回案例

{
	'code': 0,
	'data': {
		'columns': ['序号', '日期', 'COMEX白银库存量-吨', 'COMEX白银库存量-盎司'],
		'data': [
			[1242, '2025-03-06', 13072.104555172, 420277607.188]
		]
	},
	'msg': '获取数据成功'
}

1.3.2.5.5 get_futures_spot_price_data 获取品种的现货和基差数据

说明

查询品种在指定日期现货和基差数据

输入参数

现货和基差品种字典

返回值

现货和基差数据 JSON 对象

接口调用案例

调用接口代码

from futures_data_client import fetch_futures_spot_price

result1 = get_futures_spot_price_data('ag', '2025-3-6', '2025-3-7')
print(result1)

数据返回案例

{
	'code': 0,
	'data': {
		'columns': ['日期', 'symbol', 'spot_price', 'near_contract', 'near_contract_price', 'dominant_contract', 'dominant_contract_price', 'near_month', 'dominant_month', 'near_basis', 'dom_basis', 'near_basis_rate', 'dom_basis_rate'],
		'data': [
			['2025-03-06', 'A', 4180.0, 'a2503', 4100.0, 'a2505', 4175.0, '2503', '2505', -80.0, -5.0, -0.019138756, -0.0011961722]
		]
	},
	'msg': '获取数据成功'
}

1.3.2.5 期魔方账户数据

1.3.2.5.1 get_userinfo 获取期魔方用户信息

说明

返回当前登录期魔方用户的信息

输入参数

无

返回值

返回一个字典

接口案例

# 与 on_auth() 函数一起搭配使用
def on_auth(self):
    # 先通过系统接口获取用户信息
    userinfo = self.get_userinfo()
    # 再通过三个返回值字段来判断会员等级
    # 通过会员等级判断 0：普通会员，1：黄金会员，2：超级会员
    return vip_level

数据示例

# 会员等级判断 0：普通会员，1：黄金会员，2：超级会员
1

1.3.3 报单方式

1.3.3.1 快捷报单

1.3.3.1.1 buy_open 开多仓

说明

建立一个多头仓位，默认采用市价单发送，会返回该订单唯一值

输入参数

返回值

接口案例

def on_tick(context):
    res = buy_open(1) # 开单 1 手多单
    req_id = res["data"]["result"]["data"]["req_id"] # 通过这个方式可以拿到每次报单的唯一id

数据示例

# 以下是请求成功返回的结果。若请求失败：code 的值为非零 且 msg 会标记请求失败的原因)
{'code': 0, 'data': {...}, 'msg': 'ok'}

1.3.3.1.2 sell_open 开空仓

说明

建立一个空头仓位，默认采用市价单发送，会返回该订单唯一值

输入参数

返回值

接口案例

def on_tick(context):
    res = sell_open(1) # 开单 1 手空单
    req_id = res["data"]["result"]["data"]["req_id"] # 通过这个方式可以拿到每次报单的唯一id

数据示例

# 以下是请求成功返回的结果。若请求失败：code 的值为非零 且 msg 会标记请求失败的原因)
{'code': 0, 'data': {...}, 'msg': 'ok'}

1.3.3.1.3 buy_close 平空仓

说明

结束一个空头仓位，默认采用市价单发送，如果存在今昨仓的持仓，平仓时会先平今，会返回该订单唯一值

输入参数

返回值

接口案例

def on_tick(context):
    res = buy_close(1) # 平仓 1 手空单
    req_id = res["data"]["result"]["data"]["req_id"] # 通过这个方式可以拿到每次报单的唯一id

数据示例

# 以下是请求成功返回的结果。若请求失败：code 的值为非零 且 msg 会标记请求失败的原因)
{'code': 0, 'data': {...}, 'msg': 'ok'}

1.3.3.1.4 sell_close 平多仓

说明

结束一个多头仓位，默认采用市价单发送，如果存在今昨仓的持仓，平仓时会先平今，会返回该订单唯一值

输入参数

返回值

接口案例

def on_tick(context):
    res = sell_close(1) # 平仓 1 手多单
    req_id = res["data"]["result"]["data"]["req_id"] # 通过这个方式可以拿到每次报单的唯一id

数据示例

# 以下是请求成功返回的结果。若请求失败：code 的值为非零且 msg 会标记请求失败的原因)
{'code': 0, 'data': {...}, 'msg': 'ok'}

1.3.3.2 send_order 自定义报单

说明

做多、做空、平仓等动作都可以通过该方法函数进行报单，用户自己封装订单，会返回该订单唯一值

输入参数

注意:

需要使用 send_order 跨产品且未为该产品进行有效订阅操作然后去报单时，需要填写 draw_signal 值为 false，否则会因为该产品在策略中获取不到价格而导致无法报单。

返回值

ORDER_INFO_DATA 报单对象明细字段信息

接口案例

def on_tick(context):
    # 假设这是一个单子完整的信息
    dict_data = {
        "Direction":"0",
        "OrderPriceType":"2",
        "Comboffsetflag":"0",
        "Volumn":"1",
        "Symbol":"ag2412",
        "ExchangeID":"SHFE",
        "LimitPrice":66666,
    }
    # 执行报单
    res = send_order(dict_data)
    req_id = res["data"]["result"]["data"]["req_id"] # 通过这个方式可以拿到每次报单的唯一id
    # 输出报单请求类型
    print(res)

数据示例

# 以下是请求成功返回的结果。若请求失败：code 的值为非零 且 msg 会标记请求失败的原因)
{'code': 0, 'data': {...}, 'msg': 'ok'}

1.3.3.3 action_order 撤单

说明

涉及撤单时调用

输入参数

返回值

ACTION_ORDER_INFO 撤单对象明细字段信息

接口案例

def on_tick(context):
    # 单子具体信息
    dict_data = {"Direction": "0",
          "OrderPriceType": "2",
          # ...
          # 单子中 key 必须包含 Symbol 、ExchangeID 、OrderSysID ，否则会存在撤单失败的情况，同 send_order 方法函数的用法举例，订单号 OrderSysID 为共12位右置的字符串，该字符串存在于 on_order 接口返回的订单信息
          "OrderSysID": "1863",
          "ExchangeID": "SHFE",
          "Symbol": "rb2501",
         }
    # 执行撤单
    res = action_order(dict)
    print(result)

数据示例

# 以下是请求成功返回的结果。若请求失败：code 的值为非零 且 msg 会标记请求失败的原因)
{'code': 0, 'data': {...}, 'msg': 'ok'}

1.3.4 其他功能函数

1.3.4.1 remove 策略停止任务

说明

调用该函数会暂停当前的策略运行

输入参数

无

返回值

无

接口案例

def on_tick(context):
    # 此处运行交易逻辑代码
    ...
    # 需要退出任务运行时调用
    remove()

数据示例

无

1.3.4.2 put_log 推送日志

说明

推送日志到前端展示的方法

用户可以使用 put_log 函数打印日志，如果没有显示出来，可以将level参数提高级别到'ERROR'级别；

基于回测的日志信息在软件退出后会写入到用户目录的 AppData\Local\qmfquant\logs\backtest_server 下面

基于任务的日志信息在软件退出后会写入到用户目录的 AppData\Local\qmfquant\logs\task_server 下面

输入参数

返回值

无

接口案例

# 以 INFO 级别示例：
def on_tick(context):
    text_data="这是一条日志消息内容"
    # 推送日志
    put_log(text_data,level="INFO")
    print(text_data)

数据示例

"""这是一条日志消息内容"""

1.3.4.3 send_message_ex 推送自定义消息

说明

调用该函数可以通过期魔方软件推送指定消息。

输入参数

接口案例

def on_tick(context):
    # 此处运行交易逻辑代码
    ...
    # 需要退出任务运行时调用
    send_message_ex('策略发出了信号')

1.3.4.4 load_model 加载机器学习模型

说明

调用该函数可以加载指定机器学习模型，并返回模型

输入参数

返回值

接口案例

def on_tick(context):
    # 此处运行交易逻辑代码
    ...
    # 需要退出任务运行时调用
    model = load_model(model_file_path)

数据示例

无

1.3.4.5 is_trading_time 获取当前是否处于交易时间范围

说明

通过该函数判断当前时间是否可处于交易时间范围中，一般用于报单前，只有处于报单时间范围时，才触发报单操作

输入参数

返回值

接口案例

def on_tick(context):
    # 此处运行交易逻辑代码
    ...
    # 需要退出任务运行时调用
    model = is_trading_time('ag')

数据示例

无

2 策略案例

2.1 网格策略

这是一个网格策略，其原理是确定网格的大小和价格区间。网格的大小是指相邻两个价格之间的距离，而价格区间则是指每个网格的买入和卖出价格。这些参数的设定往往需要一定的经验和技巧，通常根据市场的波动性和投资者的风险承受能力来确定网格大小，而价格区间则可以根据自己的预期收益率来设定。

下面是一个只做多的网格策略：

"""
文件类型: 量化策略
帮助文档: https://qmfquant.com/static/doc/code/strategyEdit.html
期魔方，为您提供专业的量化服务
"""


# 整体止损清仓函数，当触发整体止损条件时，将所有持仓进行平仓操作，并标记退出标志为 True
def clear_position(context):
    # 遍历订单列表
    for order in context.order_list:
        # 对每个订单的持仓数量进行平仓操作
        sell_close(order["volume"])
    # 将退出标志设置为 True，表示已经进行了清仓止损操作
    context.quit_flag = True


# 整体止损处理函数，计算持仓的平均开仓价格，根据当前价格判断是否触发整体止损条件
def over_all_stop_loss(context):
    # 初始化总持仓数量为 0
    total_vol = 0
    # 初始化加权开仓价格为 0，用于计算平均开仓价格
    weighted_open_price = 0
    # 遍历订单列表
    for order in context.order_list:
        # 累加每个订单的持仓数量到总持仓数量
        total_vol += order["volume"]
        # 计算每个订单的开仓价格乘以持仓数量，并累加到加权开仓价格
        weighted_open_price += order["open_price"] * order["volume"]
    # 计算平均开仓价格，即加权开仓价格除以总持仓数量
    avg_open_price = weighted_open_price / total_vol
    # 判断当前价格是否小于等于平均开仓价格减去整体止损点数乘以价格跳动单位
    if context.last_price <= avg_open_price - context.overall_stop_loss_point * context.price_tick:
        # 如果触发整体止损条件，调用清仓函数进行平仓操作
        clear_position(context)
        # 返回 True 表示已经触发整体止损
        return True
    else:
        # 未触发整体止损条件，返回 False
        return False


# 判断网格是否发生改变，并进行相应的处理，包括分笔止盈和加仓操作
def wait_grid_change(context):
    # 判断当前价格是否大于等于最小开仓价格加上分笔止盈点数乘以价格跳动单位
    if context.last_price >= context.min_open_price + context.take_tick * context.price_tick:
        # 若向上突破网格，进行分笔止盈操作，平仓指定数量的持仓
        sell_close(context.volume)  # 向上突破网格，分笔止盈
    # 判断当前价格是否小于等于最小开仓价格减去加仓距离点数乘以价格跳动单位
    elif context.last_price <= context.min_open_price - context.plus_tick * context.price_tick:
        # 若向下突破网格，且当前总持仓数量小于最大持仓手数
        if context.total_position < context.max_volume:
            # 进行加仓操作，开仓指定数量的持仓
            buy_open(context.volume)  # 向下突破网格，加仓


# 更新最小开仓价格等变量，遍历订单列表，找出最小开仓价格及其索引，并更新总持仓数量
def update_variables(context):
    # 初始化订单索引为 0
    index = 0
    # 初始化总持仓数量为 0
    total_position = 0
    # 初始化最小开仓价格为 -1，表示尚未找到有效价格
    context.min_open_price = -1
    # 初始化最小开仓价格对应的订单索引为 -1，表示尚未找到有效索引
    context.min_open_price_index = -1
    # 遍历订单列表
    for order in context.order_list:
        # 如果最小开仓价格仍为 -1，说明是第一次遍历，直接将当前订单的开仓价格作为最小开仓价格
        if context.min_open_price == -1:
            context.min_open_price = order["open_price"]
            context.min_open_price_index = index
        # 如果当前订单的开仓价格小于最小开仓价格，更新最小开仓价格及其索引
        elif order["open_price"] < context.min_open_price:
            context.min_open_price = order["open_price"]
            context.min_open_price_index = index
        # 订单索引加 1
        index += 1
        # 累加每个订单的持仓数量到总持仓数量
        total_position += order["volume"]
    # 更新总持仓数量
    context.total_position = total_position


# 订单成交处理回调函数，处理订单成交后的相关操作，包括更新订单列表和变量
def on_trade_callback(context, trade):
    # 从成交信息中获取开平标志
    offset = trade.get("OffsetFlag")
    # 从成交信息中获取成交价格
    price = trade.get("Price")
    # 从成交信息中获取成交数量
    volume = trade.get("Volume")
    # 判断开平标志是否为开仓（这里 "0" 表示开仓）
    if offset == "0":  # 开仓就增加订单列表
        # 创建一个新的订单字典，包含开仓价格和成交数量
        order = {"open_price": price, "volume": volume}
        # 将新订单添加到订单列表
        context.order_list.append(order)
    else:  # 平仓单就删除订单列表
        # 从订单列表中移除最小开仓价格对应的订单
        context.order_list.pop(context.min_open_price_index)
    # 调用更新变量函数，更新最小开仓价格等相关变量
    update_variables(context)
    # 将首笔订单是否成交的标志设置为 False，表示首笔订单已经成交
    context.weit_trade = False  # 首笔订单已经成交


################################## 主程序部分  ##################################


# 从 pydantic 库中导入 BaseModel 和 Field 类，用于定义参数映射模型
from pydantic import BaseModel, Field


# 定义一个参数映射模型类
class Params(BaseModel, validate_assignment=True):
    """参数映射模型"""

    """
    参数1：每笔开仓手数：1手
    参数2：加仓距离：10个点
    参数3：最大持仓手数：3手
    参数4：分笔止盈：10个点
    参数5：止损风控：30个点（相对于均价）
    """
    # 定义每笔开仓手数，默认值为 1
    volume: int = Field(default=1, title="每笔开仓手数")
    # 定义加仓距离，单位为跳，默认值为 10
    plus_tick: int = Field(default=10, title="加仓距离(跳)")
    # 定义最大持仓手数，默认值为 3
    max_volume: int = Field(default=3, title="最大持仓手数")
    # 定义分笔止盈点数，单位为跳，默认值为 10
    take_tick: int = Field(default=10, title="分笔止盈(跳)")
    # 定义整体止损点数，单位为跳，默认值为 30
    overall_stop_loss_point: int = Field(default=30, title="整体止损(跳)")


# 初始化函数，在策略启动时进行一些初始化操作
def on_init(context):
    # 初始化订单列表，用于保存开仓订单
    context.order_list = []  # 保存开仓订单
    # 从基础设置中获取合约 ID
    context.base_instrument_id = BASE_SETTING.get("Instrument")
    # 初始化最新价格为 -1，表示尚未获取到有效价格
    context.last_price = -1  # 保存最新价格
    # 获取合约的价格跳动单位
    context.price_tick = get_symbolinfo(context.base_instrument_id)["price_tick"]
    # 初始化退出标志为 False，表示尚未进行清仓止损操作
    context.quit_flag = False
    # 初始化最小开仓价格为 -1，表示尚未找到有效价格
    context.min_open_price = -1
    # 初始化最小开仓价格对应的订单索引为 0
    context.min_open_price_index = 0
    # 初始化总持仓数量为 0
    context.total_position = 0
    # 初始化首笔订单是否成交的标志为 False，表示首笔订单尚未发送或成交
    context.weit_trade = False


# 处理 tick 数据的函数，根据当前状态和价格信息进行相应的操作
def on_tick(context):
    # 判断是否已经进行了清仓止损操作，如果是则直接返回，不再进行后续操作
    if context.quit_flag:  # 已经清仓止损过了
        return
    # 判断首笔订单是否已经发送但还未成交，如果是则直接返回，不再进行后续操作
    if context.weit_trade:  # 首笔订单已经发送了，还未成交
        return
    # 从基础 tick 数据中获取最新价格，并转换为浮点数保存
    context.last_price = float(BASE_SYMBOL_TICK.get("LastPrice"))
    # 判断订单列表是否为空，如果为空说明还没有开仓
    if len(context.order_list) == 0:
        # 进行开仓操作，开仓指定数量的持仓，并将首笔订单是否成交的标志设置为 True
        buy_open(context.volume, context.base_instrument_id)
        context.weit_trade = True
    else:
        # 调用整体止损处理函数，判断是否触发整体止损条件
        ret = over_all_stop_loss(context)
        # 如果没有触发整体止损条件
        if ret == False:  # 没有整体止损，就监控网格变化
            # 调用判断网格是否改变的函数，进行相应的处理
            wait_grid_change(context)


# 处理订单状态变化的函数，这里暂时不做任何处理
def on_order(context, order):
    pass


# 处理订单成交的函数，调用订单成交处理回调函数进行具体操作
def on_trade(context, trade):
    on_trade_callback(context, trade)


# 处理错误信息的函数，这里暂时不做任何处理
def on_error(context, data):
    pass


# 处理策略停止的函数，这里暂时不做任何处理
def on_stop(context):
    pass

2.2 日内清仓策略

这是一个日内清仓策略，其核心是在预设的时间段内对所有持仓进行平仓操作，以避免持仓过夜可能带来的风险或根据特定的交易策略需求结束当日的交易活动。

详细代码如下：

"""
文件类型: 量化策略
帮助文档: https://qmfquant.com/static/doc/code/strategyEdit.html
期魔方，为您提供专业的量化服务
"""


# 从 pydantic 库中导入 BaseModel 和 Field 类，用于定义参数映射模型
from pydantic import BaseModel, Field


# 定义一个参数映射模型类
class Params(BaseModel, validate_assignment=True):
    """参数映射模型"""

    # 定义日内清盘启动选项，默认值为 False
    day_clear_mode: bool = Field(default=False, title="日内清盘")
    # 定义日内清盘开始时间，默认值为 “14:58”
    day_clear_time: str = Field(default="14:58", title="日内清盘开始时间")
    # 定义日内清盘结束时间，默认值为 “22:00”
    day_over_time: str = Field(default="22:00", title="日内清盘结束时间")


# 定义初始化函数，当程序初始化时会调用该函数
def on_init(context):
    # 打印初始化完成策略开始的提示信息
    print("demo start")


# 定义 tick 事件处理函数，当收到 tick 数据时会调用该函数
def on_tick(context):
    """获取账户信息"""
    # 调用日内清盘函数
    day_over_close(context)


# 定义一个函数，用于判断交易所 ID 是否为上海期货交易所（SHFE）或上海国际能源交易中心（INE）
def isSHFEoINE(context, exchangeid):
    # 判断 exchangeid 是否在字符串"SHFE INE"中
    return exchangeid in "SHFE INE"


# 定义一个函数，用于提交平仓订单，symbol 为合约代码，exchangid 为交易所 ID，direction 为买卖方向，volume 为下单数量，comboffsetflag 为开平标志
def close_order(context, symbol, exchangid, direction, volumn, comboffsetflag):
    try:
        # 获取指定合约的 tick 数据
        tick = get_tick(symbol)
        # 判断 tick 数据中是否包含 ActionDay 字段，如果不包含则表示 tick 获取失败
        if not tick.get("ActionDay"):
            # tick 获取失败，返回等待下一次运行
            return
        # 获取 tick 数据中的涨停价，并转换为浮点数类型
        upper_price = float(tick.get("UpperLimitPrice"))
        # 获取 tick 数据中的跌停价，并转换为浮点数类型
        lower_price = float(tick.get("LowerLimitPrice"))
        # 根据买卖方向选择下单价格，如果方向为 “0” 则使用涨停价，否则使用跌停价
        price = upper_price if direction == "0" else lower_price
        # 构建订单字典
        order = {
            "symbol": symbol,  # 合约代码
            "exchangeid": exchangid,  # 交易所ID
            "limitprice": price,  # 下单价格
            "direction": direction,  # 买卖方向
            "orderpricetype": "2",  # 订单价格类型
            "comboffsetflag": comboffsetflag,  # 开平标志
            "volumn": volumn,  # 下单数量
        }
        # 发送订单
        send_order(order)
    except Exception as e:
        # 打印开单出现错误的提示信息，包含错误信息
        print(f"开单出现错误=>请查看GridTrade.close_order[{e}]")
        # 打印具体的错误信息
        print(e)
        # 开单失败，返回 0
        return 0
    # 开单成功，返回 1
    return 1


# 定义一个函数，用于清空持仓
def clear_position(context):
    # 获取当前持仓信息
    position = get_position()

    # 定义平仓方向映射字典，将持仓方向映射为平仓方向
    CLOSE_DIRECTION_MAP = {"2": "1", "3": "0"}

    # 遍历持仓信息列表
    for item in position:
        # 获取持仓方向
        PosiDirection = item.get("PosiDirection")
        # 根据持仓方向获取平仓方向
        direction = CLOSE_DIRECTION_MAP[PosiDirection]
        # 获取昨仓冻结数量
        YdStrikeFrozen = item.get("YdStrikeFrozen")
        # 获取多仓冻结数量
        LongFrozen = item.get("LongFrozen")
        # 获取空仓冻结数量
        ShortFrozen = item.get("ShortFrozen")
        # 根据平仓方向选择冻结数量
        Frozen = LongFrozen if direction == "0" else ShortFrozen
        # 计算今仓冻结数量
        todayFrozen = Frozen - YdStrikeFrozen
        # 计算今仓可平数量
        today_volume = item.get("TodayPosition") - todayFrozen
        # 计算昨仓可平数量
        yesterday_volume = item.get("YdPosition") - YdStrikeFrozen
        # 获取交易所ID
        exchangeid = item.get("ExchangeID")
        # 获取合约代码
        InstrumentID = item.get("InstrumentID")
        # 初始化今仓平仓数量为 0
        today_position = 0
        # 判断交易所是否为上海期货交易所或上海国际能源交易中心
        if isSHFEoINE(context, exchangeid):
            # 如果是，则今仓平仓数量为今仓可平数量，昨仓平仓数量为昨仓可平数量
            today_position = today_volume
            yesterday_position = yesterday_volume
        else:
            # 如果不是，则昨仓平仓数量为今仓可平数量和昨仓可平数量之和
            yesterday_position = today_volume + yesterday_volume
        """暂定一组订单同时报两次"""
        """不做回复确认 => 重获取持仓时再做重新分配 直到分配结束"""
        # 如果今仓平仓数量大于 0，则提交今仓平仓订单
        if today_position > 0:
            close_order(context, InstrumentID, exchangeid, direction, today_position, "3")
        # 如果昨仓平仓数量大于 0，则提交昨仓平仓订单
        if yesterday_position > 0:
            close_order(context, InstrumentID, exchangeid, direction, yesterday_position, "1")


# 定义一个函数，用于在到达指定时间时执行平仓操作
def day_over_close(context):
    """到达时间执行平仓"""
    # 从datetime模块中导入 datetime 和 time 类，用于处理日期和时间
    from datetime import datetime, time

    # 判断日内清盘模式是否开启，如果未开启则直接返回
    if not context.day_clear_mode:
        return
    # 获取当前时间
    now = datetime.now().time()

    # 定义时间段
    # 将日内清盘开始时间字符串转换为 time 对象
    start_time = datetime.strptime(context.day_clear_time, "%H:%M").time()  # 开始时间为 14:58
    # 将日内清盘结束时间字符串转换为 time 对象
    end_time = datetime.strptime(context.day_over_time, "%H:%M").time()  # 结束时间为 15:00
    # 判断当前时间是否在指定的时间段内
    if start_time <= now < end_time:
        # 如果在时间段内，则调用清空持仓函数
        clear_position(context)


# 定义停止函数，当程序停止时会调用该函数
def on_stop(context):
    # 打印程序停止的提示信息
    print("demo停止")

2.3 套利策略

基于 j2505 和 jm2505 的跨品种套利，监控两种合约比值的变化，动态计算其均值和标准差，当比值偏离均值超过一倍标准差时，就进行反向开仓，预测比值回归到一倍标准差之内。

套利策略涉及多品种或周期的合约数据订阅，订阅方法参见 1.2.7.3 多品种多周期 中描述，通过传入的合约列表解析出具体合约，即可进行后续开仓信号的计算。

详细代码如下：

"""
文件类型: 量化策略
帮助文档: https://qmfquant.com/static/doc/code/strategyEdit.html
期魔方，为您提供专业的量化服务
"""


# 导入NumPy库，用于高效的数值计算
import numpy as np

# 导入Pandas库，用于数据处理和分析
import pandas as pd

# 从 pydantic 库中导入 BaseModel 和 Field 类，用于定义参数映射模型
from pydantic import BaseModel, Field


# 判断当前是否持有指定方向的仓位
def isHolding(positions, instrument_id, direction):
    # 检查指定的合约 ID 是否在持仓信息字典中
    if instrument_id in positions:
        # 从持仓信息字典中获取指定合约的持仓信息
        position = positions.get(instrument_id)
        # 从持仓信息中获取持仓方向
        position_type = position.get("Direction")
        # 判断持仓方向是否与指定方向一致
        if str(position_type) == str(direction):
            return True
    return False


# 获取 K 线数据
def get_k_line_data(context):
    # 调用g et_kline 函数获取基础合约的 K 线数据，并将结果存储在klines中
    context.klines = get_kline(context.base_instrument_id, context.base_period, context.calc_len)
    # 从 K 线数据中提取收盘价数据，并存储在 close_array 中
    context.close_array = context.klines.get("close")
    # 调用 get_kline 函数获取第二个合约的 K 线数据，并将结果存储在 klines2 中
    context.klines2 = get_kline(context.instrument_id2, context.base_period, context.calc_len)
    # 从第二个合约的 K 线数据中提取收盘价数据，并存储在 close_array2 中
    context.close_array2 = context.klines2.get("close")


# 计算比例数据
def calc_indicator(context):
    # 检查基础合约或第二个合约的收盘价数据是否为空
    if (not context.close_array) or (not context.close_array2):
        # 若为空，返回 -1 表示数据不足
        return -1
    # 检查基础合约或第二个合约的收盘价数据长度是否小于统计周期
    if len(context.close_array) < context.calc_len or len(context.close_array2) < context.calc_len:
        # 若长度不足，直接返回
        return
    # 将基础合约和第二个合约的收盘价数据转换为 NumPy 数组，并计算比例数组
    ratio_array = np.asarray(context.close_array) / np.asarray(context.close_array2)
    # 将比例数组转换为 Pandas 的 Series 对象，方便进行滚动计算
    series = pd.Series(ratio_array)
    # 计算比例数组的滚动均值，并将结果转换为NumPy数组
    rolling_mean = series.rolling(window=context.calc_len).mean().values
    # 计算比例数组的滚动标准差，并将结果转换为NumPy数组
    rolling_std = series.rolling(window=context.calc_len).std().values
    # 将 context 中的比例数组、均值数组和标准差数组分别向后滚动一位
    context.ratio_array = np.roll(context.ratio_array, -1)
    context.mean_array = np.roll(context.mean_array, -1)
    context.std_array = np.roll(context.std_array, -1)
    # 将最新的比例值、均值和标准差分别赋值给数组的最后一个元素
    context.ratio_array[-1] = ratio_array[-1]
    context.mean_array[-1] = rolling_mean[-1]
    context.std_array[-1] = rolling_std[-1]


# 根据偏离度产生开仓信号
def get_open_signal(context):
    # 检查比例数组、均值数组或标准差数组的长度是否小于 3
    if len(context.ratio_array) < 3 or len(context.mean_array) < 3 or len(context.std_array) < 3:
        # 若长度不足，返回 0 表示数值不足，无法产生开仓信号
        return 0
    # 初始化开仓信号为 0，表示无开仓信号
    sig_open = 0
    # 判断是否满足开空仓的条件
    if (
        context.ratio_array[-3] <= context.mean_array[-3] + context.deviation * context.std_array[-3]
        and context.ratio_array[-2] > context.mean_array[-2] + context.deviation * context.std_array[-2]
    ):
        # 若满足条件，将开仓信号设置为 -1，表示开空仓
        sig_open = -1
    # 判断是否满足开多仓的条件
    elif (
        context.ratio_array[-3] >= context.mean_array[-3] - context.deviation * context.std_array[-3]
        and context.ratio_array[-2] < context.mean_array[-2] - context.deviation * context.std_array[-2]
    ):
        # 若满足条件，将开仓信号设置为 1，表示开多仓
        sig_open = 1
    # 计算上轨线的值
    upper = context.mean_array[-2] + context.deviation * context.std_array[-2]
    # 计算下轨线的值
    lower = context.mean_array[-2] - context.deviation * context.std_array[-2]
    return sig_open


# 根据指标计算平仓信号
def get_close_signal(context):
    # 检查比例数组、均值数组或标准差数组的长度是否小于 3
    if len(context.ratio_array) < 3 or len(context.mean_array) < 3 or len(context.std_array) < 3:
        # 若长度不足，返回 0 表示数值不足，无法产生平仓信号
        return 0
    # 初始化平仓信号为 0，表示无平仓信号
    sig_close = 0
    # 判断是否满足平空仓的条件
    if (
        context.ratio_array[-3] >= context.mean_array[-3] + context.deviation * context.std_array[-3]
        and context.ratio_array[-2] < context.mean_array[-2] + context.deviation * context.std_array[-2]
    ):
        # 若满足条件，将平仓信号设置为 1，表示平空仓
        sig_close = 1
    # 判断是否满足平多仓的条件
    elif (
        context.ratio_array[-3] <= context.mean_array[-3] - context.deviation * context.std_array[-3]
        and context.ratio_array[-2] > context.mean_array[-2] - context.deviation * context.std_array[-2]
    ):
        # 若满足条件，将平仓信号设置为 -1，表示平多仓
        sig_close = -1
    return sig_close


# 判断当前是否持有仓位
def isHoldingPosition(context):
    # 检查基础合约和第二个合约的多仓和空仓情况，只要有一个合约持有仓位，则返回 True
    if (
        isHolding(context.positions, context.base_instrument_id, "0")
        or isHolding(context.positions, context.base_instrument_id, "1")
        or isHolding(context.positions, context.instrument_id2, "0")
        or isHolding(context.positions, context.instrument_id2, "1")
    ):
        return True
    else:
        return False


# 开一对正向套利仓位
def open_long_position(context):
    # 以指定手数买入基础合约
    buy_open(context.volume, context.base_instrument_id)
    # 以指定手数卖出第二个合约
    sell_open(context.volume, context.instrument_id2)


# 开一对反向套利仓位
def open_short_position(context):
    # 以指定手数卖出基础合约
    sell_open(context.volume, context.base_instrument_id)
    # 以指定手数买入第二个合约
    buy_open(context.volume, context.instrument_id2)


# 根据信号开仓
def open_position_by_signal(context, sig_open):
    if sig_open == 1:  # 开多信号
        # 检查当前是否没有持仓
        if not isHoldingPosition(context):
            # 若没有持仓，则开一对正向套利仓位
            open_long_position(context)
    elif sig_open == -1:  # 开空信号
        # 检查当前是否没有持仓
        if not isHoldingPosition(context):
            # 若没有持仓，则开一对反向套利仓位
            open_short_position(context)


# 结束一对正向套利仓位
def close_long_position(context):
    # 检查是否持有基础合约的多仓
    if isHolding(context.positions, context.base_instrument_id, "0"):
        # 若持有多仓，则以指定手数卖出基础合约进行平仓
        sell_close(context.volume, context.base_instrument_id)
    # 检查是否持有第二个合约的空仓
    if isHolding(context.positions, context.instrument_id2, "1"):
        # 若持有空仓，则以指定手数买入第二个合约进行平仓
        buy_close(context.volume, context.instrument_id2)


# 结束一对反向套利仓位
def close_short_position(context):
    # 检查是否持有基础合约的空仓
    if isHolding(context.positions, context.base_instrument_id, "1"):
        # 若持有空仓，则以指定手数买入基础合约进行平仓
        buy_close(context.volume, context.base_instrument_id)
    # 检查是否持有第二个合约的多仓
    if isHolding(context.positions, context.instrument_id2, "0"):
        # 若持有多仓，则以指定手数卖出第二个合约进行平仓
        sell_close(context.volume, context.instrument_id2)


# 根据信号平仓
def close_position_by_signal(context, sig_close):
    if sig_close == 1:  # 平空信号
        # 若收到平空信号，则结束一对反向套利仓位
        close_short_position(context)
    elif sig_close == -1:  # 平多信号
        # 若收到平多信号，则结束一对正向套利仓位
        close_long_position(context)


#  K 线收盘运行逻辑
def on_bar(context):
    # 调用 get_k_line_data 函数获取 K 线数据
    get_k_line_data(context)
    # 调用 calc_indicator 函数计算比例数据
    calc_indicator(context)
    # 调用 get_open_signal 函数获取开仓信号
    sig_open = get_open_signal(context)
    # 调用 get_close_signal 函数获取平仓信号
    sig_close = get_close_signal(context)
    # 调用 close_position_by_signal 函数根据平仓信号进行平仓操作
    close_position_by_signal(context, sig_close)
    # 调用 open_position_by_signal 函数根据开仓信号进行开仓操作
    open_position_by_signal(context, sig_open)


############################## 主程序部分  ##############################


# 定义一个参数映射模型类
class Params(BaseModel, validate_assignment=True):
    # 统计周期，默认为 100
    calc_len: int = Field(default=100, title="统计周期")
    # 触发交易的偏离度(标准差倍数)，默认为 1.0
    deviation: float = Field(default=1.0, title="触发交易的偏离度(标准差倍数)")
    # 手数，默认为1
    volume: int = Field(default=1, title="手数")
    # 订阅合约和周期，默认为 "j2505:M5&jm2505:M5"
    QMF_SUBSCRIBE_SYMBOLS: str = Field(default="j2505:M5&jm2505:M5", title="订阅合约和周期")


# 初始化一些参数
def on_init(context):
    # 将订阅合约和周期字符串按 "&" 分割成列表
    symbols = QMF_SUBSCRIBE_SYMBOLS.split("&")
    # 从分割后的列表中提取基础合约的 ID，并存储在 context.base_instrument_id 中
    context.base_instrument_id = symbols[0].split(":")[0]
    # 从分割后的列表中提取第二个合约的 ID，并存储在 context.instrument_id2 中
    context.instrument_id2 = symbols[1].split(":")[0]
    # 从 BASE_SETTING 中获取基础周期，并存储在 context.base_period 中
    context.base_period = BASE_SETTING.get("bPeriod")
    # 初始化持仓信息字典为空字典
    context.positions = {}
    # 初始化基础合约的收盘价数组为长度为统计周期的全零数组
    context.close_array = np.zeros(context.calc_len)
    # 初始化比例数组为长度为统计周期的全零数组
    context.ratio_array = np.zeros(context.calc_len)
    # 初始化均值数组为长度为统计周期的全零数组
    context.mean_array = np.zeros(context.calc_len)
    # 初始化标准差数组为长度为统计周期的全零数组
    context.std_array = np.zeros(context.calc_len)
    # 初始化第二个合约的收盘价数组为长度为统计周期的全一数组
    context.close_array2 = np.ones(context.calc_len)
    # 初始化第二个合约的比例数组为长度为统计周期的全零数组
    context.ratio_array2 = np.zeros(context.calc_len)
    # 初始化第二个合约的均值数组为长度为统计周期的全零数组
    context.mean_array2 = np.zeros(context.calc_len)
    # 初始化第二个合约的标准差数组为长度为统计周期的全零数组
    context.std_array2 = np.zeros(context.calc_len)


# 每次有行情进来，便会调用 on_tick 函数刷新一次
def on_tick(context):
    # 每产生新的 K 线时，调用 on_bar_run 函数执行 K 线收盘运行逻辑
    on_bar_run(on_bar, context)


# 返回委托单信息 注意此处的报单信息是未成交状态
def on_order(context, order):
    pass


# 返回成交单信息 注意此处的报单信息是已成交状态
def on_trade(context, trade):
    # 从成交单信息对象中获取合约ID
    InstrumentID = trade.get("InstrumentID")
    # 从成交单信息对象中获取开平标志
    OffsetFlag = trade.get("OffsetFlag")
    # 判断开平标志是否为开仓标志"0"
    if OffsetFlag == "0":
        # 若为开仓标志，则将成交单信息存储在持仓信息字典中
        context.positions[InstrumentID] = trade
    else:
        # 若为平仓标志，则从持仓信息字典中移除该合约的持仓信息
        context.positions.pop(InstrumentID)


# 返回报错
def on_error(context, order):
    pass


# 策略回测中途发生错误 or 回撤完毕时调用
def on_stop(context):
    pass

2.4 自动移仓换月

对于订阅主连合约进行交易的策略，当主力合约映射发生改变后，可以通过 1.3.2.1.3 get_current_main_symbol 获取当前主力合约函数检测到这种改变，从而进行必要的处理。需要注意在回测模式，主力连续合约作为一个独立合约存在，不存在移仓换月，不需处理，只在任务模式才需要这个处理。

当有基于主连合约（后缀888）进行交易并发生主力映射改变，如下案例是自动关闭原主力合约，在新主力合约开同样数量的仓位。

详细代码如下：

"""
文件类型: 量化策略
帮助文档: https://qmfquant.com/static/doc/code/strategyEdit.html
期魔方，为您提供专业的量化服务
"""


# 从 pydantic 库中导入 BaseModel 和 Field 类，用于定义参数映射模型
from pydantic import BaseModel, Field


# 定义 K 线交易逻辑的函数，目前函数体为空，可根据实际需求实现具体逻辑
def custom_trade_on_k_line(context):
    # 此处可添加基于 K 线的交易逻辑代码
    pass


# 判断是否在指定合约上持有仓位的函数
def is_holding_on_symbol(context, direction, symbol):
    # 检查指定合约是否在持仓字典中
    if symbol in context.positions:
        # 从持仓字典中获取指定合约的持仓信息
        position = context.positions.get(symbol)
        # 从持仓信息中获取持仓方向
        position_type = position.get("Direction")
        # 比较持仓方向是否与传入的方向一致
        if str(position_type) == str(direction):
            # 如果一致，说明持有该方向的仓位，返回 True
            return True
    # 否则，返回 False，表示未持有该方向的仓位
    return False


# 判断是否需要进行移仓换月操作的函数
def need_deal_hot_change(context):
    # 检查合约 ID 中是否包含 "888" 且自动移仓换月开关是否打开
    if ("888" in context.base_instrument_id) and context.auto_hot_change == True:
        # 获取当前的主力合约代码
        cur_hot = get_current_main_symbol(context.base_instrument_id)
        # 比较当前记录的主力合约代码与最新的主力合约代码是否不同
        if context.hot_symbol != cur_hot:  # 基于主连合约交易，并且发生了主力换月
            # 如果不同，说明需要进行移仓换月操作，返回 True
            return True
    else:
        # 否则，返回 False，表示不需要进行移仓换月操作
        return False


# 关闭原主力合约仓位的函数
def close_position_on_prev_hot(context):
    # 检查原主力合约是否持有多头仓位
    if is_holding_on_symbol(context, "0", context.hot_symbol):
        # 如果持有多头仓位，执行卖出平仓操作
        sell_close(context.volume, context.hot_symbol)
    # 检查原主力合约是否持有空头仓位
    if is_holding_on_symbol(context, "1", context.hot_symbol):
        # 如果持有空头仓位，执行买入平仓操作
        buy_close(context.volume, context.hot_symbol)


# 在新主力合约建立仓位的函数
def open_position_on_new_hot(context):
    # 获取当前的主力合约代码
    cur_hot = get_current_main_symbol(context.base_instrument_id)
    # 检查新主力合约是否未持有多头仓位
    if not is_holding_on_symbol(context, "0", cur_hot):
        # 如果未持有多头仓位，执行买入开仓操作
        buy_open(context.volume, cur_hot)
    # 检查新主力合约是否未持有空头仓位
    if not is_holding_on_symbol(context, "1", cur_hot):
        # 如果未持有空头仓位，执行卖出开仓操作
        sell_open(context.volume, cur_hot)


# 如果原主力合约有仓位，将其移到新主力合约的函数
def move_position(context):
    # 检查原主力合约是否持有空头仓位
    holding_short = is_holding_on_symbol(context, "1", context.hot_symbol)
    # 检查原主力合约是否持有多头仓位
    holding_long = is_holding_on_symbol(context, "0", context.hot_symbol)
    # 如果原主力合约持有空头或多头仓位
    if holding_short or holding_long:
        # 关闭原主力合约的仓位
        close_position_on_prev_hot(context)
        # 在新主力合约建立仓位
        open_position_on_new_hot(context)


# K 线收盘时的运行逻辑函数
def on_bar(context):
    # 判断是否需要进行移仓换月操作
    if need_deal_hot_change(context):
        # 如果需要，执行移仓操作
        move_position(context)
        # 更新记录最新的主力合约映射
        context.hot_symbol = get_current_main_symbol(context.base_instrument_id)
    else:
        # 如果不需要，执行基于 K 线的交易逻辑
        custom_trade_on_k_line(context)


############################## 主程序部分 ##############################


# 定义一个参数映射模型类
class Params(BaseModel, validate_assignment=True):
    # 定义手数字段，默认值为 1，字段标题为 "手数"
    volume: int = Field(default=1, title="手数")
    # 定义是否自动移仓换月字段，默认值为 False，字段标题为 "是否自动移仓换月"
    auto_hot_change: bool = Field(default=False, title="是否自动移仓换月")


# 初始化一些参数的函数，在策略启动时调用
def on_init(context):
    # 从 BASE_SETTING 中获取基础合约 ID
    context.base_instrument_id = BASE_SETTING.get("Instrument")
    # 从 BASE_SETTING 中获取基础周期
    context.base_period = BASE_SETTING.get("bPeriod")
    # 初始化持仓字典为空
    context.positions = {}
    # 初始化获取当前的主力合约
    context.hot_symbol = get_current_main_symbol(context.base_instrument_id)
    # 初始化获取当前的主力合约
    context.cur_symbol = get_current_main_symbol(context.base_instrument_id)


# 每次有行情进来时调用的函数，用于刷新策略状态
def on_tick(context):
    # 注册 K 线处理回调函数，当 K 线收盘时调用 on_bar 函数
    on_bar_run(on_bar, context)


# 返回委托单信息的函数，注意此处的报单信息是未成交状态
def on_order(context, order):
    # 可根据实际需求处理委托单信息
    pass


# 返回成交单信息并更新仓位的函数
def on_trade(context, trade):
    # 从成交单中获取合约 ID
    InstrumentID = trade.get("InstrumentID")
    # 从成交单中获取开平标志
    OffsetFlag = trade.get("OffsetFlag")
    # 如果开平标志为 "0"，表示开仓操作
    if OffsetFlag == "0":
        # 将成交单信息存入持仓字典中
        context.positions[InstrumentID] = trade
    else:
        # 否则，表示平仓操作，从持仓字典中移除该合约的持仓信息
        context.positions.pop(InstrumentID)


# 返回报错信息的函数
def on_error(context, order):
    # 可根据实际需求处理报错信息
    pass


# 策略回测中途发生错误或回撤完毕时调用的函数
def on_stop(context):
    # 可根据实际需求处理策略停止时的操作
    pass

2.5 订单备份与恢复

在跑自动交易途中，如果突然不小心关了电脑或客户端，重启后不做处理的情况下，之前的订单和任务就丢了。希望接之前跑的任务和持仓继续跑，其中一种解决方法是，把订单保存到本地一个指定目录，当策略启动时，从本地磁盘读取历史订单信息，加载到内存中，方便用户基于历史订单信息继续执行之前的交易逻辑或者执行不同的交易逻辑。

详细代码如下：

"""
文件类型: 量化策略
帮助文档: https://qmfquant.com/static/doc/code/strategyEdit.html
期魔方，为您提供专业的量化服务
"""


# 导入 NumPy 库，用于数组操作等
import numpy as np

# 导入 TA-Lib 库，用于技术分析指标的计算
import talib as ta

# 从 pydantic 库中导入 BaseModel 和 Field 类，用于定义参数映射模型
from pydantic import BaseModel, Field


# 判断当前是否持有指定方向的仓位
def isHolding(context, direction):
    # 检查当前交易品种是否存在于持仓信息中
    if context.base_instrument_id in context.positions:
        # 从持仓信息中获取当前交易品种的成交单信息，并赋值给变量 position
        position = context.positions.get(context.base_instrument_id)
        # 从成交单信息中获取仓位方向
        position_type = position.get("Direction")
        # 判断获取到的仓位方向是否与指定的方向一致
        if str(position_type) == str(direction):
            return True
    return False


# 获取 K 线数据
def get_k_line_data(context):
    # 从外置参数中获取均线 1 的周期
    len1 = context.len1
    # 从外置参数中获取均线 2 的周期
    len2 = context.len2
    # 调用 get_kline 函数获取指定交易品种、指定周期的 K 线数据，数据长度为均线周期的最大值
    context.klines = get_kline(context.base_instrument_id, context.base_period, max(len1, len2))
    # 从 K 线数据中获取日期时间数组
    context.dt_array = context.klines.get("datetime")
    # 从 K 线数据中获取收盘价数组
    context.close_array = context.klines.get("close")


# 计算均线指标数据
def calc_sma(context):
    # 从外置参数中获取均线 1 的周期
    len1 = context.len1
    # 从外置参数中获取均线 2 的周期
    len2 = context.len2
    # 从 K 线数据中获取收盘价数组
    context.close_array = context.klines.get("close")
    # 使用 TA-Lib 库的 SMA 函数计算收盘价数组的均线 1 数据
    ma1_array = ta.SMA(np.asarray(context.close_array), len1)
    # 使用 TA-Lib 库的 SMA 函数计算收盘价数组的均线 2 数据
    ma2_array = ta.SMA(np.asarray(context.close_array), len2)
    # 将均线 1 数组元素循环左移一位
    context.ma1_array = np.roll(context.ma1_array, -1)
    # 将均线 2 数组元素循环左移一位
    context.ma2_array = np.roll(context.ma2_array, -1)
    # 将最新计算的均线1值赋值给均线 1 数组的最后一个元素
    context.ma1_array[-1] = ma1_array[-1]
    # 将最新计算的均线2值赋值给均线 2 数组的最后一个元素
    context.ma2_array[-1] = ma2_array[-1]


# 根据 SMA 指标产生开仓信号
def get_open_signal_by_ma(context):
    # 检查均线 1 数组和均线 2 数组的长度是否小于 3，如果小于 3 则说明数据不足
    if len(context.ma2_array) < 3 or len(context.ma1_array) < 3:
        return 0  # 数值不足，返回信号值 0 表示无开仓信号
    # 初始化开仓信号值为 0，表示无开仓信号
    sig_open = 0
    # 判断均线 1 是否在倒数第三个数据点小于等于均线 2，且在倒数第二个数据点大于均线 2
    if context.ma1_array[-3] <= context.ma2_array[-3] and context.ma1_array[-2] > context.ma2_array[-2]:
        sig_open = 1  # 满足条件则表示产生开多信号，将信号值设为 1
    # 判断均线1是否在倒数第三个数据点大于等于均线 2，且在倒数第二个数据点小于均线 2
    elif context.ma1_array[-3] >= context.ma2_array[-3] and context.ma1_array[-2] < context.ma2_array[-2]:
        sig_open = -1  # 满足条件则表示产生开空信号，将信号值设为 -1
    return sig_open


# 根据指标计算平仓信号
def get_close_signal_by_ma(context):
    # 检查均线 1 数组和均线 2 数组的长度是否小于 3，如果小于 3 则说明数据不足
    if len(context.ma2_array) < 3 or len(context.ma1_array) < 3:
        return 0  # 数值不足，返回信号值 0 表示无平仓信号
    # 初始化平仓信号值为 0，表示无平仓信号
    sig_close = 0
    # 判断均线 1 是否在倒数第三个数据点小于等于均线 2，且在倒数第二个数据点大于均线 2
    if context.ma1_array[-3] <= context.ma2_array[-3] and context.ma1_array[-2] > context.ma2_array[-2]:
        sig_close = 1  # 满足条件则表示产生平空信号，将信号值设为 1
    # 判断均线 1 是否在倒数第三个数据点大于等于均线 2，且在倒数第二个数据点小于均线 2
    elif context.ma1_array[-3] >= context.ma2_array[-3] and context.ma1_array[-2] < context.ma2_array[-2]:
        sig_close = -1  # 满足条件则表示产生平多信号，将信号值设为 -1
    return sig_close


# 根据信号开仓
def open_position_by_signal(context, sig_open):
    if sig_open == 1:  # 开多信号
        # 检查是否没有持有多仓
        if not isHolding(context, "0"):
            # 调用 buy_open 函数进行开多操作
            ret = buy_open(context.volume, context.base_instrument_id)
    elif sig_open == -1:  # 开空信号
        # 检查是否没有持有空仓
        if not isHolding(context, "1"):
            # 调用 sell_open 函数进行开空操作
            sell_open(context.volume, context.base_instrument_id)


# 根据信号平仓
def close_position_by_signal(context, sig_close):
    if sig_close == 1:  # 平空信号
        # 检查是否持有空仓
        if isHolding(context, "1"):
            # 调用 buy_close 函数进行平空操作
            buy_close(context.volume, context.base_instrument_id)
    elif sig_close == -1:  # 平多信号
        # 检查是否持有多仓
        if isHolding(context, "0"):
            # 调用 sell_close 函数进行平多操作
            sell_close(context.volume, context.base_instrument_id)


#  K 线收盘运行逻辑
def on_bar(context):
    # 调用 get_k_line_data 函数获取 K 线数据
    get_k_line_data(context)
    # 调用 calc_sma 函数计算均线指标数据
    calc_sma(context)
    # 调用 get_open_signal_by_ma 函数根据 SMA 指标产生开仓信号
    sig_open = get_open_signal_by_ma(context)
    # 调用 get_close_signal_by_ma 函数根据 SMA 指标产生平仓信号
    sig_close = get_close_signal_by_ma(context)
    # 调用 close_position_by_signal 函数根据平仓信号进行平仓操作
    close_position_by_signal(context, sig_close)
    # 调用 open_position_by_signal 函数根据开仓信号进行开仓操作
    open_position_by_signal(context, sig_open)


# 记录订单到磁盘文件
def record_order(context, order):
    # 导入 json 库，用于处理 JSON 数据
    import json

    # 检查订单列表的长度是否达到最大保存订单个数
    if len(context.order_list) >= context.max_order_num:
        # 如果达到最大个数，则截取订单列表，只保留最近的最大个数的订单
        context.order_list = context.order_list[1 - context.max_order_num :]
    # 将新订单添加到订单列表中
    context.order_list.append(order)
    # 以写入模式打开订单保存文件
    with open(context.order_dir, "w") as f:
        # 将订单列表以 JSON 格式写入文件
        json.dump(context.order_list, f)


# 从磁盘文件读取订单到内存
def restore_orders(context):
    # 导入 json 库，用于处理 JSON 数据
    import json

    # 导入 os 库，用于操作系统相关功能，如文件路径操作等
    import os

    # 检查订单保存文件是否存在
    if os.path.exists(context.order_dir):
        # 以只读模式打开订单保存文件
        with open(context.order_dir) as f:
            # 检查文件大小是否不为 0，即文件是否有内容
            if os.path.getsize(context.order_dir) != 0:
                # 从文件中加载 JSON 数据，并将其转换为 Python 字典列表
                order_dicts = json.load(f)
                # 将加载的订单字典列表赋值给 order_list
                context.order_list = order_dicts
                # put_log(f'restored history orders : \n{context.order_list}', 'INFO')


############################## 主程序部分 ##############################


# 定义一个参数映射模型类
class Params(BaseModel, validate_assignment=True):
    """参数映射模型"""

    # 定义均线 1 的周期，默认值为 5
    len1: int = Field(default=5, title="均线 1 周期")
    # 定义均线 2 的周期，默认值为 20
    len2: int = Field(default=20, title="均线 2 周期")
    # 定义交易手数，默认值为1
    volume: int = Field(default=1, title="手数")
    # 定义订单保存地址，默认值为 r"E:\order.json"，注意这里要改成本机地址
    order_dir: str = Field(default=r"E:\order.json", title="订单保存地址")
    # 定义最大保存订单个数，默认值为 10
    max_order_num: int = Field(default=10, title="最大保存订单个数")


# 初始化一些参数
def on_init(context):
    # 从 BASE_SETTING 中获取交易品种 ID
    context.base_instrument_id = BASE_SETTING.get("Instrument")
    # 从 BASE_SETTING 中获取 K 线周期
    context.base_period = BASE_SETTING.get("bPeriod")
    # 初始化持仓信息为空字典
    context.positions = {}
    # 初始化收盘价数组，长度为均线2的周期，元素初始值都为 0
    context.close_array = np.zeros(context.len2)
    # 初始化均线1数组，长度为均线1的周期，元素初始值都为 0
    context.ma1_array = np.zeros(context.len1)
    # 初始化均线2数组，长度为均线2的周期，元素初始值都为 0
    context.ma2_array = np.zeros(context.len2)
    # 初始化订单列表为空列表
    context.order_list = []
    # 调用 restore_orders 函数从磁盘文件读取订单到内存
    restore_orders(context)


# 每次有行情进来，便会调用 on_tick 函数刷新一次
def on_tick(context):
    # 注册 K 线处理回调，当 K 线收盘时调用 on_bar 函数处理 K 线数据
    on_bar_run(on_bar, context)


# 返回委托单信息 注意此处的报单信息是未成交状态
def on_order(context, order):
    pass


# 返回成交单信息 记录仓位变化
def on_trade(context, trade):
    # 从成交单信息中获取交易品种ID
    InstrumentID = trade.get("InstrumentID")
    # 从成交单信息中获取开平标志
    OffsetFlag = trade.get("OffsetFlag")
    # 判断开平标志是否为开仓标志 "0"
    if OffsetFlag == "0":
        # 如果是开仓，则将成交单信息添加到持仓信息中
        context.positions[InstrumentID] = trade
    else:
        # 如果是平仓，则从持仓信息中移除该交易品种的持仓信息
        context.positions.pop(InstrumentID)
    # 调用 record_order 函数将成交单信息记录到磁盘文件
    record_order(context, trade)


# 返回报错
def on_error(context, order):
    pass


# 策略回测中途发生错误 or 回撤完毕时调用
def on_stop(context):
    pass

2.6 大单拆分

大交易量用户在实际使用中，为了降低冲击成本，最大程度减小滑点损失，可以使用twap或vwap方式拆分订单，twap通过在固定时间间隔内平均开仓，达到在指定时间段内均匀执行订单的目的，下面是一个twap拆单算法的例子：

# 循环执行，循环次数为总订单数减去当前已执行的 TWAP 多头订单数
for _ in range(num_orders - context.twap_order_count_bull):
    # 检查当前已执行的 TWAP 多头交易量是否达到了目标交易量
    if context.twap_executed_volume_bull >= context.volume:
        # 若达到目标交易量，打印提示信息，表示 TWAP 多头交易已完成
        print("TWAP 多头交易完成，总下单量已达到目标")
        # 删除 TWAP 多头交易的开始时间记录
        del context.twap_start_time_bull
        # 删除 TWAP 多头交易的订单计数记录
        del context.twap_order_count_bull
        # 删除 TWAP 多头交易的已执行交易量记录
        del context.twap_executed_volume_bull
        # 函数返回，结束当前操作
        return
    # 计算下一次交易的时间，即 TWAP 多头交易开始时间加上当前已执行的订单数对应的分钟数
    next_trade_time = context.twap_start_time_bull + timedelta(minutes=context.twap_order_count_bull)
    # 检查当前时间是否已经达到或超过下一次交易的时间
    if current_time >= next_trade_time:
        # 获取当前的收盘价作为下单价格
        current_price = context.close_array[-1]
        # 计算实际的下单量，取预设订单量和剩余目标交易量中的较小值
        actual_order_size = min(order_size, context.volume - context.twap_executed_volume_bull)
        # 打印当前下单信息，包括订单序号、下单价格和下单手数
        print(f"TWAP 多单 {context.twap_order_count_bull+1}/{num_orders}: 价格 {current_price}, 手数 {actual_order_size}")
        # 执行买入开仓操作，传入实际下单量、基础合约 ID 和下单价格
        buy_open(actual_order_size, context.base_instrument_id, price=current_price)
        # 累加已执行的 TWAP 多头订单数
        context.twap_order_count_bull += 1
        # 累加已执行的 TWAP 多头交易量
        context.twap_executed_volume_bull += actual_order_size

vwap 是在 twap 的基础上，加入成交量权重，动态调整每次开仓手数，根据成交量变化优化开仓效率：

# 异步 VWAP 拆单算法：根据成交量调整开仓，该算法会在指定的总时间内，按照一定的时间间隔进行拆单操作，并根据实时成交量动态调整每次下单的手数
async def async_vwap_order(context, direction, volume, total_time, interval):
    # 计算总拆单次数，通过总时间除以时间间隔得到，并将结果转换为整数
    times = int(total_time / interval)
    # 计算每次下单的基础手数，通过总开仓手数除以拆单次数得到，并将结果转换为整数
    volume_per_order = int(volume / times)
    # 初始化一个变量，用于记录已经开仓的手数，初始值为 0
    total_opened = 0
    # 初始化成交量数组，调用 get_kline 函数获取指定交易标的、周期和时间段的 K 线数据
    klines = get_kline(context.base_instrument_id, context.base_period, context.vwap_period)
    # 从 K 线数据中提取成交量数据，并将其转换为numpy数组
    volume_data = np.array(klines.get("volume"))
    # 开始循环，循环次数为拆单次数
    for i in range(times):
        # 获取最新的 K 线数据，只获取最近的一根 K 线
        latest_klines = get_kline(context.base_instrument_id, context.base_period, 1)
        # 从最新的 K 线数据中提取最新的成交量
        current_volume = latest_klines.get("volume")[-1]
        # 更新成交量数组，将数组元素向后滚动一位，去掉最早的成交量数据
        volume_data = np.roll(volume_data, -1)
        # 将最新的成交量数据添加到数组的最后一位
        volume_data[-1] = current_volume
        # 实时计算成交量数组的平均值，作为当前的均量
        avg_volume = np.mean(volume_data)
        # 计算剩余需要开仓的手数，通过总开仓手数减去已经开仓的手数得到
        remaining_volume = volume - total_opened
        # 判断剩余手数是否小于等于 0，如果是，则说明所有仓位已开完，停止开仓
        if remaining_volume <= 0:
            # 记录日志，提示所有仓位已开完，停止开仓
            put_log("所有仓位已开完，停止开仓", "INFO")
            # 跳出循环
            break
        # 动态调整下单量的系数，将其限制在 0.5 到 1 倍之间
        volume_factor = max(0.5, min(1, current_volume / avg_volume))
        # 根据调整系数计算调整后的下单手数
        adjusted_volume = int(volume_per_order * volume_factor)
        # 确保调整后的下单手数不超过剩余需要开仓的手数
        adjusted_volume = min(adjusted_volume, remaining_volume)
        # 确保调整后的下单手数最小为 1 手
        adjusted_volume = max(1, adjusted_volume)
        # 防止超开总手数，如果调整后的下单手数加上已经开仓的手数超过总开仓手数，则进行调整
        if total_opened + adjusted_volume > volume:
            adjusted_volume = volume - total_opened
        # 根据交易方向进行开仓操作，如果方向为 1，则进行买入开仓
        if direction == 1:
            buy_open(adjusted_volume, context.base_instrument_id)
        # 如果方向为 -1，则进行卖出开仓
        elif direction == -1:
            sell_open(adjusted_volume, context.base_instrument_id)
            # 更新已经开仓的手数，将本次开仓的手数累加到总开仓手数中
        total_opened += adjusted_volume

2.7 与外部数据联动

平台默认支持 AKShare 库调用，如果需要引用 AKShare 库所支持数据，比如A股或美股数据做参考的用户，可以调用其接口来获取数据。

具体对应数据参考 AKShare 官网：https://akshare.akfamily.xyz

获取个股历史行情：

import akshare as ak

stock_zh_a_hist_df = ak.stock_zh_a_hist(symbol="000001", period="daily", start_date="2025-03-15", end_date="2025-03-20", adjust="")
put_log(f"\n{stock_zh_a_hist_df}", "INFO")

日期    股票代码     开盘     收盘     最高     最低      成交量           成交额    振幅   涨跌幅   涨跌额   换手率
0  2025-03-17  000001  11.63  11.50  11.67  11.46  4603612  5.315523e+09  1.75 -3.93 -0.47  2.37
1  2025-03-18  000001  11.52  11.49  11.54  11.48  1605290  1.846500e+09  0.52 -0.09 -0.01  0.83
2  2025-03-19  000001  11.48  11.52  11.53  11.46  1362455  1.566452e+09  0.61  0.26  0.03  0.70
3  2025-03-20  000001  11.51  11.49  11.61  11.49  1101157  1.268849e+09  1.04 -0.26 -0.03  0.57

获取科创板实时行情：

import akshare as ak

stock_kc_a_spot_em_df = ak.stock_kc_a_spot_em()
put_log(f"科创板实时行情:\n{stock_kc_a_spot_em_df}", "INFO")

科创板实时行情:
      序号      代码      名称    最新价    涨跌幅    涨跌额       成交量           成交额     振幅      最高     最低      今开      昨收    量比    换手率  市盈率-动态   市净率           总市值          流通市值    涨速  5分钟涨跌  60日涨跌幅  年初至 今涨跌幅
0      1  688793     倍轻松  35.90   6.65   2.24   68981.0  2.509844e+08  13.93   38.57  33.88   33.88   33.66  1.89   8.03  327.04  8.29  3.085441e+09  3.085441e+09 -0.39  -0.58   21.20    20.47
1      2  688118    普元信息  26.60   5.14   1.30   76647.0  2.010637e+08  10.28   27.20  24.60   24.99   25.30  2.74   8.35  382.49  2.85  2.442349e+09  2.442349e+09  0.26  -0.11   23.38    26.49
2      3  688776    国光电气  72.80   5.13   3.55   41949.0  3.035973e+08   8.58   74.18  68.24   68.29   69.25  1.47   3.87  163.91  4.21  7.890313e+09  7.890313e+09  0.17   0.12   33.07    54.40
..   ...     ...     ...    ...    ...    ...       ...           ...    ...     ...    ...     ...     ...   ...    ...     ...   ...           ...           ...   ...    ...     ...      ...

获取美股历史行情

import akshare as ak

stock_us = ak.stock_us_hist(symbol="106.TTE", period="daily", start_date="20250301", end_date="20250310", adjust="qfq")
put_log(f"{stock_us}", "INFO")

日期     开盘     收盘      最高      最低      成交量          成交额    振幅   涨跌幅   涨跌额   换手率
0  2025-03-03  61.57  59.83  61.670  59.360  1607167   96877978.0  3.84 -0.66 -0.40  0.07
1  2025-03-04  59.22  59.87  60.550  58.966  2320214  138720830.0  2.65  0.07  0.04  0.10
2  2025-03-05  60.97  60.94  61.230  60.100  2092218  127068015.0  1.89  1.79  1.07  0.09
3  2025-03-06  60.64  60.78  61.508  60.540  1556678   94709372.0  1.59 -0.26 -0.16  0.07
4  2025-03-07  61.39  61.39  62.050  60.570  2835416  174262938.0  2.44  1.00  0.61  0.13
5  2025-03-10  61.67  61.07  61.950  60.585  1961626  120370695.0  2.22 -0.52 -0.32  0.09

获取美股实时行情

import akshare as ak

stock_us_spot_em_df = ak.stock_us_spot_em()
put_log(f"{stock_us_spot_em_df}", "INFO")

          序号                               名称   最新价    涨跌额     涨跌幅    开盘价    最高价    最低价     昨收价          总市值    市盈率          成交量          成交额      振幅      换手率          代码
0          1            Impact BioMedical Inc  2.08  1.550  292.45  0.530  2.480  0.530   0.530   23928226.0  20.54   30679574.0   54278090.0  367.92   266.69     107.IBO
1          2      Innovative Eyewear Inc Wt-A  0.14  0.098  229.41  0.042  0.191  0.042   0.043          NaN    NaN      41062.0       4693.0  350.12      NaN   105.LUCYW
2          3            Plus Therapeutics Inc  1.38  0.869  170.06  0.570  2.080  0.550   0.511    8136940.0  -0.63  357747872.0  454892368.0  299.37  6067.29    105.PSTV
...      ...                              ...   ...    ...     ...    ...    ...    ...     ...          ...    ...          ...          ...     ...      ...         ...

3 常见问题

3.1 策略文件编译失败

可能的原因：

a) 编辑代码后没有保存

b) 不符合 Python 语法，例：将 def on_stop 写成了 deff on-stop

c) 期魔方平台函数接口使用错误，例：get_kline("ru2501","M1")，少了一个参数

解决方法：

借助终端的 output 选择 “Python 编译输出”，在输出结果查看原因，分析处理

3.2 策略修改后无法重新回测

修改策略后，在策略回测页面点击编辑参数，点击保存，即可重新回测

3.3 找不到历史导入的策略文件

可能的原因：

a) 策略文件的命名格式问题，比如在同一文件夹路径下按前后顺序添加了两个命名都为 MACD 策略文件导致后面添加的策略文件会把前面已添加的文件覆盖掉

b) 系统本地保存的策略目录缓存问题，比如当把本地缓存的路径 C:\期魔方\strategy 下策略目录文件夹清空后，重启软件会把原先在软件上导入后的策略文件给清空而无法找回

3.4 回测无数据

如果在回测进度 100% 后，点开详情没有数据，可能的原因：

a) 错误地调用了功能函数以获取目标合约信息，比如没有正确使用功能函数 get_symbolinfo，自然无法获取到数据去进行计算或查询工作

b) 策略文件只有策略原生的结构性框架，比如一个策略文件的代码中仅仅保留使用策略原生的结构性函数框架，此时保存策略文件代码后右键点击 Python 编译依旧能够显示编译通过的提示， 而后在回测过程中使用该策略文件依旧可以支持回测，但回测进度完成至 100% 后，会存在未统计出相应的交易数据的情况，而导致点击详情时无法正常打开详情页并强制清除该回测进度

c) 使用的策略不支持回测，比如“风控模型”策略的逻辑目前暂时只能在任务模块中运行，因此在回测模块中应用该策略会导致回测过程中无法产生交易数据

3.5 策略执行效率问题

如果回测时候策略的进度太慢，可能是因为：

a) 策略文件中涉及使用多种复杂的计算或大量的数据查询的模块

b) 运行周期选择很小，比如当策略在回测过程中选择 “1分钟” 的运行时间周期上时，回测效率会很慢。当选择分钟运行周期时，信号点增加了，因此产生的交易数据量暴增便会导致策略执行效率慢的问题，建议选择 “1天” 为运行周期

3.6 策略报单失败

可能的原因：

a) 错误使用功能方法函数，比如在使用方法函数 send_order 时缺乏必需的 order_info 参数，在传递过程中这张单子可能就无法正确报单成功

b) 系统在处理报单任务时可能存在无法即时成交而出现单子堆积的问题，系统增加了报单超过 3 秒不成交会自动撤单的机制，比如当你的系统的时间为 “11:00:50” 本地时间为 “11:00:55” 时，误差超过了 3 秒，单子就被自动撤单。此时校准本地时间即可

3.7 收盘后发单失败

可能原因：对于收盘后收到的溢出数据，没有判断时间，仍执行发单逻辑，导致失败

解决办法：判断当前时间，如果不在交易时间内，则忽略处理，不执行发单逻辑

3.8 同一根 K 线内多次开仓

可能原因：基于 K 线触发的开仓逻辑，写在了 on_tick 函数中，没有判断是否生成新 K 线，导致开仓逻辑多次执行

解决办法：K 线的处理逻辑调用 on_bar_run 处理，就能保证只在新 K 线产生时才执行开仓逻辑运算

3.9 开仓信号错误或提前了一根 K 线

可能原因：引用 K 线数据时，访问了当前 K 线的数据。例如：

if context.ma1_array[-2] <= context.ma2_array[-2] and context.ma1_array[-1] > context.ma2_array[-1]:
    sig_open = 1

解决办法：当前最新的 K 线是实时变动的，需要改为访问已经收线的 K 线数据

if context.ma1_array[-3] <= context.ma2_array[-3] and context.ma1_array[-2] > context.ma2_array[-2]: 
    sig_open = 1

3.10 策略平仓失败或重复开仓

a) 平仓失败的可能原因是没有持仓

b) 重复开仓的可能原因是策略出现了非预期情况，已经持有仓位

解决办法：开仓前判断是否已持有仓位，平仓前判断是否并没持有仓位

3.11 任务停止后仍在发单

目前策略中一般不响应策略停止信号，如果在策略中有循环，虽然页面看到任务退出日志，但策略会在循环结束后才真正退出任务，如果循环中有发单逻辑，就会看到任务已经停止，但策略仍在发单。

3.12 put_log/print 函数在策略代码写了输出代码而日志文件没有输出的问题

问题：为什么在策略里面，使用了 put_log 或 print 函数输出日志，而任务/回测日志文件没有相关日志记录

a) 首先，put_log 函数是否在完整策略方法体系下输出的，其次 put_log 参数是否录入正确

b) 确认下设置页面的“策略详情日志”是否勾选，如果没勾选，则需勾选保存后，把软件重启下，再去运行任何和回测，策略详情日志配置的方法：

（1）软件右上角有一个滚轮设置按钮，点击它

（2）弹出页面，选择选项卡“日志设置”

（3）勾选“策略详情日志”的复选框

（4）点击“保存”按钮

（5）重启软件

3.13 量化任务的功能操作没反应

问题：量化任务的功能操作没反应？

问题原因：可能因为运行量化任务的部分后端服务异常导致服务中止运行了

问题排查步骤：

确认下任务的服务是否正常启动，验证方式：在桌面的任务栏底部右侧的“系统托盘”状态栏区域找到“期魔方”图标，右键点击它，选择“服务运行状态”会弹出系统各个服务运行情况，我们这重点关注“任务服务”“任务消息服务”“持仓服务”是否是“正常”状态

a) 任务服务：主要负责量化任务的执行过程，负责任务开始到停止所有业务流程传统，如果遇到页面没反应，基本是否服务断开了

b) 持仓服务：它是通过期货账户登录期货公司后启动的服务进程，报单实际交易是通过这个服务发起的，当出现报单交易失败时，一般是这个服务导致的问题。

c) 任务消息服务：它是负责任务和跟持仓服务消息通信的消息通信服务，它如果断开了，在获取持仓，货期期货账户信息就会失败。

问题解决方式：

（1）确认当前电脑的操作系统是否满足软件安装要求，要求 windows10 以及以上

（2）打开任务管理器，看下 cpu 和内存使用情况，cpu 和内存使用超过80%上就要注意了，系统就在报警了，这种情况可能导致服务启动失败的情况，这个时候要关闭下不必要的服务或者升级下电脑的硬件资源

（3）如果确认以上都没问题的话，可通过加期魔方的 QQ 群 “744013072” 反馈问题，或者期魔方官网联系客服。

3.14 报单异常内容错误

问题：任务启动后，如遇到报单异常内容错误“已撤单报单被拒绝SHFE:当前状态禁止此项操作”错误时？

问题原因：在交易报单操作时，而报单的时间不在 CTP 报单交易时间范围内，就会出现该错误提示

问题排查步骤：检查下发生交易时间是否在该品种的交易时间范围内。

问题解决方式：我们系统提供了函数 is_trading_time 把品种作为参数插入就可以验证处于报单交易时间范围内，使用方式：

is_in_time_flag = is_trading_time('au')

if is_in_time_flag:
    print(“当前成就报单时间范围内，可以调用报单接口去报单交易”)

3.15 查看当前任务运行的所有日志

问题：任务运行后，任务页面的日志列表只能看到最新 50 条日志，如果看当前任务运行的所有日志，在什么地方看？

问题解决方式：任务运行后，会在系统缓存目录有记录当前任务所有日志内容，用于问题分析，完整日志目录如下

（1）打开日志开启的开关：软件右上角的顶部，点击“设置”按钮，进入设置页面，选择“日志设置”，勾选“策略详情日志”，保存，然后重启软件。

（2）查看日志：

日志文件路径：C:\Users\windows用户登录账号\AppData\Local\qmfquant\logs\task_server

日志文件：每个任务的文件夹命令：任务运行品种 + 任务运行周期 + 任务运行唯一值.log

3.16 查看 报单/撤单 是否交易成功

问题：报单/撤单后，如何知道是否交易成功，或者交易失败？

问题解决方式：我们的策略框架提供了相关结构函数，分别介绍如下：

通过on_order结构函数的数据得知报单成功，它是我们把报单交易操作报给期货公司后，期货公司会返回订单信息，我们会把订单数据推给on_order函数，用户端接收它的数据即可，具体请看下on_order函数。

通过on_error结构函数返回的数据获悉报单失败，它是我们把报单交易操作报给期货公司后，期货公司认为报单有异常时，系统后台会接收到异常信息，系统然后会把异常信息推送给策略的 on_error 函数，策略开发人员可以通过该函数接收和打印出异常信息，用于问题分析。

问题描述：任务发起品种报单到期货公司后，如何调试获取报单过程ctp推送订单日志和成交单日志

问题解决方式：通过持仓这边记录期货公司的日志数据可获取，日志文件如下：

日志路径：C:\Users\windows登录用户账号\AppData\Local\qmfquant\logs\ctp_server

日志文件：

3.17 查看每次报单的唯一键值

问题：任务发起每一次报单有没一个唯一键值，以便区分on_order和on_trade是那次的报单请求的数据？

问题解决方式：通过可以通过报单函数（比如 buy_open）的返回值获取 req_id 唯一值，它会在 on_order 和 on_trade 回调返回的 data 数据包含 req_id 值，这样就能区分出从报单请求，到报单成功/报单失败的唯一关联。

举例：买多报单函数获取 req_id 的代码样例：

buy_open_result = buy_open(7, context.base_instrument_id)

print(f"req_id: {buy_open_result['data']['result']['data']['req_id']}")

3.18 获取最新的 20 根 K 线用于策略判断

问题：如何获取最新的 20 根 K 线用于策略判断

问题解决方式：在 on_tick/on_bar 结构函数的实操代码：

# 获取最新 10 根 K 线数据
kline_data = get_kline(context.base_instrument_id, context.base_period, 20) 

# 如你还需获取最新一根 K 线数据的成交量，通过以下方式获取
current_volume = kline_data.get("volume")[-1]

3.19 on_bar 函数中代码未执行

问题：on_bar 函数中代码没执行？

问题解决方式：

（1）在 on_tick 方法中加了代码 on_bar_run(on_bar, context)

（2）在 on_bar 方法，通过代码 print 输出调试日志验证

3.20 代码中如何获取任务/回测页面添加任务时参数配置页面的参数数据

问题：策略代码如何获取任务/回测页面添加任务时参数配置页面的参数数据？

问题解决方式：通过 BASE_SETTING 字典去获取数据，具体见本文档的该字典的具体使用，比如我要获取当前任务选择的品种参数，获取代码案例: BASE_SETTING.get('Instrument')

3.21 on_tick 方法的代码未执行

问题：任务的策略运行后，on_tick 方法的代码没执行？

问题处理方式：首先盘口服务是否正常运行，它是负责给任务的 on_tick 方法提供 tick 服务，当我们的策略的on_tick 没有收到数据时，一般是这个服务出异常了，排查如下：

首先，在桌面的任务栏底部右侧的“系统托盘”状态栏区域找到“期魔方”图标，右键点击它，选择“服务运行状态”会弹出系统各个服务运行情况，我们这重点关注“任务服务“盘口服务”是否是“正常”状态，否则盘口服务被关闭了。

其次，如果盘口服务是“正常”状态，那就是我们订阅 CTP 行情服务失败了，我们可以通过日志文件分析原因，我们可以去日志文件去分析问题，日志路径 C:\Users\windows登录用户账号\AppData\Local\qmfquant\logs\ctp_tick_server，查找最新的日志文件，看下我们的品种是否订阅成功，非行情时间下，是订阅失败的。