1 入门

1.1 快速入门

1.1.1 策略的新建与编译

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

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

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

from pydantic import BaseModel, Field
import numpy as np

# 判断当前是否持有指定方向的仓位            
def is_holding(context,direction):
    try:        
        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
    except Exception as e:
        put_log(f"[isHolding] => error ==> {e}", 'ERROR')
    return False

# 获取K线数据
def get_k_line_data(context):
    len1 = context.len1    
    len2 = context.len2
    context.klines = get_kline(context.base_instrument_id, context.base_period, max(len1, len2))
    context.dt_array = context.klines.get('datetime')    
    context.close_array   = context.klines.get("close")

# 计算均线指标数据
def calc_sma(context):
    import talib as ta
    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)    
    context.ma1_array = np.roll(context.ma1_array, -1)
    context.ma2_array = np.roll(context.ma2_array, -1)
    context.ma1_array[-1] = ma1_array[-1]
    context.ma2_array[-1] = ma2_array[-1]    

# 根据SMA指标产生开仓信号
def get_open_signal_by_ma(context):    
    if len(context.ma2_array) < 3 or len(context.ma1_array) < 3:
        return 0 # 数值不足 
    sig_open = 0
    if context.ma1_array[-3] <= context.ma2_array[-3] \
        and context.ma1_array[-2] > context.ma2_array[-2]: 
        sig_open = 1
    elif context.ma1_array[-3] >= context.ma2_array[-3] \
        and context.ma1_array[-2] < context.ma2_array[-2]:
        sig_open = -1
    return sig_open

# 根据指标计算平仓信号
def get_close_signal_by_ma(context):
    if len(context.ma2_array) < 3 or len(context.ma1_array) < 3:
        return 0 # 数值不足 
    sig_close = 0
    if context.ma1_array[-3] <= context.ma2_array[-3] \
        and context.ma1_array[-2] > context.ma2_array[-2]: 
        sig_close = 1
    elif context.ma1_array[-3] >= context.ma2_array[-3] \
        and context.ma1_array[-2] < context.ma2_array[-2]:
        sig_close = -1
    return sig_close

# 根据信号开仓
def open_position_by_signal(context, sig_open):
    if sig_open == 1:  # 开多信号        
        if not is_holding(context,"0"): # 没有持多仓
            buy_open(context.volume, context.base_instrument_id) 
    elif sig_open == -1:  # 开空信号
        if not is_holding(context,"1"): # 没有持空仓
            sell_open(context.volume, context.base_instrument_id) 

# 根据信号平仓
def close_position_by_signal(context, sig_close):
    if sig_close == 1:  # 平空信号
        if is_holding(context,"1"):  # 持有空仓
            buy_close(context.volume, context.base_instrument_id)
    elif sig_close == -1:  # 平多信号
        if is_holding(context,"0"):  # 持有多仓
            sell_close(context.volume, context.base_instrument_id)

# k线收盘运行逻辑
def on_bar(context):
    get_k_line_data(context)
    calc_sma(context)
    sig_open = get_open_signal_by_ma(context)
    sig_close = get_close_signal_by_ma(context)
    close_position_by_signal(context, sig_close)
    open_position_by_signal(context, sig_open)

########################  main programm  ##############################

class Params(BaseModel, validate_assignment=True):
    """参数映射模型"""
    len1:int               = Field(default=5,title="均线1周期")
    len2:int               = Field(default=20,title="均线2周期")
    volume:int            = Field(default=1,title="手数")

# 初始化一些参数
def on_init(context):
    import talib
    context.base_instrument_id = BASE_SETTING.get("Instrument")    
    context.base_period = BASE_SETTING.get("bPeriod")    
    context.positions = {}
    context.close_array = np.zeros(context.len2)
    context.ma1_array = np.zeros(context.len1)
    context.ma2_array = np.zeros(context.len2)

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

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

# 返回成交单信息 注意此处的报单信息是已成交状态 
def on_trade(context,trade):
    InstrumentID = trade.get("InstrumentID")
    OffsetFlag = trade.get("OffsetFlag")
    if OffsetFlag == "0":
        context.positions[InstrumentID] = trade
    else:
        context.positions.pop(InstrumentID)

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

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

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

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

1.1.2 回测运行与结果展示

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

回测完成后点击详情查看报告，综合报告，收益回撤，图表分析，明细详情，参数信息等，回测日志区内容在软件退出后会保存到日志目录，如用户目录的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_error，on_trade

# 外置参数
# 参数映射模型
from pydantic import BaseModel
class Params(BaseModel,validate_assignment=True):
    pass

# 行情报价每进来一次就会 onTick 刷新一次 
def on_tick(context):
    
    # 调用 获取K线 功能性函数举例：
    klines = get_kline("ru2501","M1",3)
    # 举例省略展示打印结果
    print(f"{klines}")

# 初始化默认参数
def on_init(context):
    
    # 调用 获取账户信息 功能性函数举例：
    account_info = get_account()
    # 举例省略展示打印结果
    print(f"{account_info}")

# 在程序停止时调用
def on_stop(context):
    pass
  
# 订单挂单后（订单未成交状态的报单信息） 返回报单信息
def on_order(context,order):
    pass

# 用于返回报错类型的方法，可在此对报错的类型进行调试与分析
def on_error(context,order):
    pass
    
# 订单实际成交后的成交单信息
def on_trade(context,trade):
    pass

1.2.1 on_tick 运算函数

说明

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

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

输入参数

输出参数

无

接口案例

def on_tick(context):
    pass

数据示例

无

1.2.2 on_init 初始化函数

说明

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

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

输入参数

输出参数

无

接口案例

def on_init(context):
    pass

数据示例

无

1.2.3 on_stop 策略停止函数

说明

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

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

输入参数

输出参数

无

接口案例

def on_stop(context):
    pass

数据示例

无

1.2.4 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()

数据示例

无

1.2.5 on_error 接收CTP错误

说明

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

输入参数

输出参数

无

接口案例

def on_error(context, data):
    pass

数据示例

无

1.2.6 on_trade 接收成交单信息

说明

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

输入参数

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

输出参数

无

接口案例

def on_trade(context, order):
    pass

数据示例

无

1.3 功能函数

1.3.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 衍生数据函数接口

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

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

1.3.2.1 fetch_futures_inventory_99获取大宗商品库存数据

1、说明

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

2、输入参数

库存品种字典

3、输出参数

库存数据JSON对象的data对象

4、接口调用案例

调用接口代码

from futures_data_client import fetch_futures_inventory_99

result1 = fetch_futures_inventory_99('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.2 fetch_futures_inventory_em获取仓单日报的库存数据

1、说明

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

2、输入参数

仓单日报库存品种字典

3、输出参数

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

4、接口调用案例

调用接口代码

from futures_data_client import fetch_futures_inventory_em

result1 = fetch_futures_inventory_em('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.3 fetch_futures_foreign_hist获取外盘期货历史行情数据

1、说明

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

2、输入参数

外盘品种数据字典

3、输出参数

查询接口返回对象

外盘数据JSON对象的data对象

4、接口调用案例

调用接口代码

from futures_data_client import fetch_futures_foreign_hist

result1 = fetch_futures_foreign_hist('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.4 fetch_futures_comex_inventory获取黄金和白银的库存数据

1、说明

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

2、输入参数

3、输出参数**

库存数据JSON对象data对象

4、接口调用案例

调用接口代码

from futures_data_client import fetch_futures_comex_inventory

result1 = fetch_futures_comex_inventory('白银', '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 fetch_futures_spot_price获取品种的现货和基差数据

1、说明

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

2、输入参数

现货和基差品种字典

3、输出参数

现货和基差数据JSON对象

4、接口调用案例

调用接口代码

from futures_data_client import fetch_futures_spot_price

result1 = fetch_futures_spot_price('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.3 get_account 获取CTP帐户信息

说明

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

输入参数

无

返回值

ACCOUNT_DATA 账户对象

接口案例

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.4 get_symbolinfo 获取合约信息

说明

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

输入参数

输出参数

SYMBOLINFO_DATA 产品对象明细字段信息

接口案例

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.5 get_position 获取持仓信息

说明

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

输入参数

无

输出参数

POSITION_DATA 持仓对象明细字段信息

接口案例

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.6 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.7 buy_open开多仓

说明

建立一个多头仓位，默认采用市价单发送

输入参数

返回值

接口案例

def on_tick(context):
    buy_open(1) #开单1手多单

数据示例

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

1.3.8 sell_open 开空仓

说明

建立一个空头仓位，默认采用市价单发送

输入参数**

输出参数

接口案例

def on_tick(context):
    sell_open(1) #开单1手空单

数据示例

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

1.3.9 buy_close 平空仓

说明

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

输入参数

输出参数

接口案例

def on_tick(context):
    buy_close(1) #平仓1手空单

数据示例

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

1.3.10 sell_close 平多仓

说明

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

输入参数

输出参数

接口案例

def on_tick(context):
    sell_close(1) #平仓1手多单

数据示例

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

1.3.11 send_order 自定义发单

说明

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

输入参数

输出参数

ORDER_INFO_DATA 报单对象明细字段信息

接口案例

def on_tick(context):
    # 假设这是一个单子完整的信息
    dict={
        "Direction":"0",
        "OrderPriceType":"2",
        "Comboffsetflag":"0",
        "Volumn":"1",
        "Symbol":"ag2412",
        "ExchangeID":"SHFE",
        "LimitPrice":66666,
    }
    # 执行报单
	result = send_order(dict)
    # 输出报单请求类型
    print(result)

数据示例

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

1.3.12 action_order 策略撤单

说明

涉及撤单时调用。

输入参数

输出参数

ACTION_ORDER_INFO 撤单对象明细字段信息

接口案例

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

数据示例

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

1.3.13 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.3.14 remove 策略停止任务

说明

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

输入参数

无

输出参数

无

接口案例

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

数据示例

无

1.3.15 put_log 推送日志

说明

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

输入参数

输出参数

无

接口案例

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

数据示例

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

1.3.16 get_userinfo 获取期魔方用户信息

说明

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

输入参数

无

输出参数

DICT 对象

接口案例

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

数据示例

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

1.3.17 time_current 获取当前时间

说明

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

输入参数

无

输出参数

接口案例

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

数据示例

无

1.3.18 get_current_main_symbol 获取当前主力合约

说明

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

输入参数

返回值

接口案例

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

数据示例

无

1.3.19 on_allorder 收到全局order消息

说明

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.3.20 load_model 加载机器学习模型

说明

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

输入参数

模型地址 model_file_path

输出参数

接口案例

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

数据示例

无

2 进阶

2.1 策略调试与日志打印

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

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

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

2.2 批量回测与综合报表

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

2.3 外部参数与全局变量

2.3.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 = "布尔")
    # 完成外置参数配置后的具体效果，可在期魔方添加任务和回测时的弹窗查看

2.3.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 = ""

接口案例

# 导入参数映射模型
from pydantic import BaseModel
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

2.4 自定义报单方式

参见 1.3.10 send_order 策略报单函数，用户可以自定义订单字段，如平今/平昨，交易所ID，报价方式等

2.5 多品种多周期

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

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

申明方式

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

接口案例

from pydantic import BaseModel

# 外置参数
# 参数映射模型
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):
	...

2.6 套利策略

套利策略涉及多品种或周期的合约数据订阅，订阅方法参见2.5中描述，通过传入的合约列表解析出具体合约，即可进行后续开仓信号的计算，一个基于j2505和jm2505合约的跨品种套利的案例参见策略案例3.3。

2.7 移仓换月

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

2.8 订单备份与恢复

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

2.9 大单拆分

大交易量用户在实际使用中，为了降低冲击成本，最大程度减小滑点损失，可以使用twap或vwap方式拆分订单，下面是一个twap拆单算法的例子：

    for _ in range(num_orders - context.twap_order_count_bull):
        if context.twap_executed_volume_bull >= context.volume:
            print("TWAP 多头交易完成，总下单量已达到目标")
            del context.twap_start_time_bull
            del context.twap_order_count_bull
            del context.twap_executed_volume_bull
            return
        # 记录交易时间
        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}")
            buy_open(actual_order_size, context.base_instrument_id, price=current_price)
            # 累加已下单量
            context.twap_order_count_bull += 1
            context.twap_executed_volume_bull += actual_order_size  

3 策略案例

3.1 网格策略

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

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

# 整体止损清仓 
def clear_position(context):
    for order in context.order_list:
        sell_close(order['volume'])
    context.quit_flag = True

# 整体止损处理
def over_all_stop_loss(context):
    total_vol = 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:
        clear_position(context)
        return True
    else:
        return False

# 判断网格是否发生改变，并进行处理
def wait_grid_change(context):
    if context.last_price >= context.min_open_price + context.plus_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):
    index = 0
    total_position = 0
    context.min_open_price = -1
    context.min_open_price_index = -1
    for order in context.order_list:
        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
        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")
    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)
    context.weit_trade = False  # 首笔订单已经成交

####################################  main program  ##################################

from pydantic import BaseModel, Field

class Params(BaseModel, validate_assignment=True):
    """参数映射模型"""
    """
    参数1：每笔开仓手数：1手
    参数2：加仓距离：10个点
    参数3：最大持仓手数：3手
    参数4：分笔止盈：10个点
    参数5：止损风控：30个点（相对于均价）
    """
    volume:int             = Field(default=1,title = "每笔开仓手数")
    plus_tick:int          = Field(default=10,title = "加仓距离(跳)")    
    max_volume:int         = Field(default=3,title = "最大持仓手数")
    take_tick:int          = Field(default=10,title = "分笔止盈(跳)")
    overall_stop_loss_point:int  = Field(default=30,title = "整体止损(跳)")
    
def on_init(context):
    context.order_list = [] # 保存开仓订单
    context.base_instrument_id = BASE_SETTING.get("Instrument")
    context.last_price = -1        # 保存最新价格
    context.price_tick = get_symbolinfo(context.base_instrument_id)['price_tick']
    context.quit_flag = False
    context.min_open_price = -1
    context.min_open_price_index = 0
    context.total_position = 0
    context.weit_trade = False

def on_tick(context):
    if context.quit_flag == True:   # 已经清仓止损过了
        return
    if context.weit_trade == True:  # 首笔订单已经发送了，还未成交
        return
    context.last_price = float(BASE_SYMBOL_TICK.get("LastPrice"))
    if len(context.order_list) == 0:
        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

3.2 日内清仓策略

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

详细代码如下：

from pydantic import BaseModel, Field
class Params(BaseModel, validate_assignment=True):
    """参数映射模型"""
    day_clear_mode:bool    = Field(default=False,title="日内清盘")
    day_clear_time:str     = Field(default="14:58",title="日内清盘开始时间")
    day_over_time:str      = Field(default="22:00",title="日内清盘结束时间")

def on_init(context):
    print("demo start")
    
def on_tick(context):
    """获取账户信息"""
    day_over_close(context)
        
def isSHFEoINE(context,exchangeid):
    return exchangeid in "SHFE INE"

def close_order(context,symbol,exchangid,direction,volume,comboffsetflag):
    try:
        tick = get_tick(symbol)
        if not tick.get("ActionDay"):
            # tick获取失败 返回等待下一次运行
            return
        upper_price = float(tick.get("UpperLimitPrice"))
        lower_price = float(tick.get("LowerLimitPrice"))
        price  =upper_price if direction=="0" else lower_price
        order  = {
            "symbol":symbol,
            "exchangeid":exchangid,
            "limitprice":price,
            "direction":direction,
            "orderpricetype":"2",
            "comboffsetflag":comboffsetflag,
            "volumn":volume
            }
        send_order(order)
    except Exception as e:
        print(f"开单出现错误=>请查看GridTrade.close_order[{e}]")
        print(e)
        return 0
    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

        yestoday_volume = item.get("YdPosition") - YdStrikeFrozen

        exchangeid = item.get("ExchangeID")

        InstrumentID = item.get("InstrumentID")

        today_position = 0

        if isSHFEoINE(context,exchangeid):

            today_position    = today_volume

            yestoday_position = yestoday_volume

        else:

            yestoday_position = today_volume + yestoday_volume

        """暂定一组订单同时报两次"""
        """不做回复确认 => 重获取持仓时再做重新分配 直到分配结束"""

        if today_position > 0:

            close_order(context,InstrumentID,exchangeid,direction,today_position,"3")
        
        if yestoday_position > 0:

            close_order(context,InstrumentID,exchangeid,direction,yestoday_position,"1")
    
def day_over_close(context):
    """到达时间执行平仓"""
    from datetime import datetime, time

    if not context.day_clear_mode: return
    now = datetime.now().time()
    
    # 定义时间段
    start_time = datetime.strptime(context.day_clear_time, "%H:%M").time()  # 开始时间为14:58

    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停止")

3.3 套利策略

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

详细代码如下：

                       
from pydantic import BaseModel, Field
import pandas as pd
import numpy as np

# 判断当前是否持有指定方向的仓位            
def isHolding(positions, instrument_id, direction):
    try:        
        if instrument_id in positions:
            # 把成交单传到全局变量 position里面
            position     = positions.get(instrument_id)
            position_type = position.get("Direction")
            if str(position_type) == str(direction):   
                return True
    except Exception as e:
        put_log(f"[isHolding] => error ==> {e}", 'ERROR')
    return False

# 获取K线数据
def get_k_line_data(context):
    context.klines = get_kline(context.base_instrument_id, context.base_period, context.calc_len)  
    context.close_array   = context.klines.get("close")
    context.klines2 = get_kline(context.instrument_id2, context.base_period, context.calc_len)
    context.close_array2   = context.klines2.get("close")

# 计算比例数据
def calc_indicator(context):
    import talib as ta
    if (not context.close_array) or (not context.close_array2):
        return -1
    if len(context.close_array) < context.calc_len or len(context.close_array2) < context.calc_len:
        return
    ratio_array = np.asarray(context.close_array) / np.asarray(context.close_array2)
    series = pd.Series(ratio_array)
    rolling_mean = series.rolling(window=context.calc_len).mean().values
    rolling_std = series.rolling(window=context.calc_len).std().values
    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):    
    if len(context.ratio_array) < 3 or len(context.mean_array) < 3 or len(context.std_array) < 3:
        return 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]: 
        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]: 
        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):
    if len(context.ratio_array) < 3 or len(context.mean_array) < 3 or len(context.std_array) < 3:
        return 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]: 
        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]: 
        sig_close = -1
    return sig_close

def isHoldingPosition(context):
    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(context)  
    calc_indicator(context)        
    sig_open = get_open_signal(context)
    sig_close = get_close_signal(context)
    close_position_by_signal(context, sig_close)
    open_position_by_signal(context, sig_open)

########################  main programm  ##############################

class Params(BaseModel, validate_assignment=True):
    calc_len:int           = Field(default=100,title="统计周期")
    deviation:float        = Field(default=1.0,title="触发交易的偏离度(标准差倍数)")
    volume:int             = Field(default=1,title="手数")
    QMF_SUBSCRIBE_SYMBOLS:str = Field(default='j2505:M5&jm2505:M5',title="订阅合约和周期")

# 初始化一些参数
def on_init(context):
    import talib
    symbols = QMF_SUBSCRIBE_SYMBOLS.split("&")
    context.base_instrument_id = symbols[0].split(":")[0]
    context.instrument_id2 = symbols[1].split(":")[0]
    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):
    on_bar_run(on_bar, context)

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

# 返回成交单信息 注意此处的报单信息是已成交状态 
def on_trade(context,trade):
    InstrumentID = trade.get("InstrumentID")
    OffsetFlag = trade.get("OffsetFlag")
    if OffsetFlag == "0":
        context.positions[InstrumentID] = trade
    else:
        context.positions.pop(InstrumentID)

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

3.4 自动移仓换月

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

详细代码如下：

from pydantic import BaseModel, Field

# K线交易逻辑
def custom_trade_on_k_line(context):    
    pass 

# 是否在指定合约上持有仓位
def is_holding_on_symbol(context,direction, symbol):
    try:        
        if symbol in context.positions:
            # 把成交单传到全局变量 position里面
            position     = context.positions.get(symbol)
            position_type = position.get("Direction")
            if str(position_type) == str(direction):   
                return True
    except Exception as e:
        put_log(f"[isHolding] => error ==> {e}", 'ERROR')
    return False

# 是否需要移仓换月
def need_deal_hot_change(context):
    if ('888' in context.hot_symbol) and context.auto_hot_change == True:
        cur_hot = get_current_main_symbol(context.base_instrument_id)
        if (context.hot_symbol != cur_hot): # 基于主连合约交易，并且发生了主力换月            
            return True
    else:
        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:
        custom_trade_on_k_line(context) 

########################  main programm  ##############################

class Params(BaseModel, validate_assignment=True):
    volume:int             = Field(default=1,title="手数")
    auto_hot_change:bool   = Field(default=False, title='是否自动移仓换月')

# 初始化一些参数
def on_init(context):
    import talib
    context.base_instrument_id = BASE_SETTING.get("Instrument")    
    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)

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

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

# 返回成交单信息 更新仓位
def on_trade(context,trade):
    InstrumentID = trade.get("InstrumentID")
    OffsetFlag = trade.get("OffsetFlag")
    if OffsetFlag == "0":
        context.positions[InstrumentID] = trade
    else:
        context.positions.pop(InstrumentID)

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

3.5 订单备份与恢复

把订单保存到本地一个指定目录，当策略启动时，从本地磁盘读取历史订单信息，加载到内存中，方便用户基于历史订单信息执行不同的交易逻辑。

详细代码如下：

from pydantic import BaseModel, Field
import pandas as pd
import numpy as np

# 判断当前是否持有指定方向的仓位            
def isHolding(context,direction):
    try:        
        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
    except Exception as e:
        put_log(f"[isHolding] => error ==> {e}", 'ERROR')
    return False

# 获取K线数据
def get_k_line_data(context):
    len1 = context.len1    
    len2 = context.len2
    context.klines = get_kline(context.base_instrument_id, context.base_period, max(len1, len2))
    context.dt_array = context.klines.get('datetime')    
    context.close_array   = context.klines.get("close")

# 计算均线指标数据
def calc_sma(context):
    import talib as ta
    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)    
    context.ma1_array = np.roll(context.ma1_array, -1)
    context.ma2_array = np.roll(context.ma2_array, -1)
    context.ma1_array[-1] = ma1_array[-1]
    context.ma2_array[-1] = ma2_array[-1]    

# 根据SMA指标产生开仓信号
def get_open_signal_by_ma(context):    
    if len(context.ma2_array) < 3 or len(context.ma1_array) < 3:
        return 0 # 数值不足 
    sig_open = 0
    if context.ma1_array[-3] <= context.ma2_array[-3] \
        and context.ma1_array[-2] > context.ma2_array[-2]: 
        sig_open = 1
    elif context.ma1_array[-3] >= context.ma2_array[-3] \
        and context.ma1_array[-2] < context.ma2_array[-2]:
        sig_open = -1
    return sig_open

# 根据指标计算平仓信号
def get_close_signal_by_ma(context):
    if len(context.ma2_array) < 3 or len(context.ma1_array) < 3:
        return 0 # 数值不足 
    sig_close = 0
    if context.ma1_array[-3] <= context.ma2_array[-3] \
        and context.ma1_array[-2] > context.ma2_array[-2]: 
        sig_close = 1
    elif context.ma1_array[-3] >= context.ma2_array[-3] \
        and context.ma1_array[-2] < context.ma2_array[-2]:
        sig_close = -1
    return sig_close

# 根据信号开仓
def open_position_by_signal(context, sig_open):
    if sig_open == 1:  # 开多信号        
        if not isHolding(context,"0"): # 没有持多仓
            ret = buy_open(context.volume, context.base_instrument_id) 
    elif sig_open == -1:  # 开空信号
        if not isHolding(context,"1"): # 没有持空仓
            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(context.volume, context.base_instrument_id)
    elif sig_close == -1:  # 平多信号
        if isHolding(context,"0"):  # 持有多仓
            sell_close(context.volume, context.base_instrument_id)

# k线收盘运行逻辑
def on_bar(context):
    get_k_line_data(context)
    calc_sma(context)
    sig_open = get_open_signal_by_ma(context)
    sig_close = get_close_signal_by_ma(context)
    close_position_by_signal(context, sig_close)
    open_position_by_signal(context, sig_open)

# 记录订单到磁盘文件
def record_order(context, order):
    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.dump(context.order_list, f)

# 从磁盘文件读取订单到内存
def restore_orders(context):
    import json
    import os
    if os.path.exists(context.order_dir):
        with open(context.order_dir, "r") as f:            
            if os.path.getsize(context.order_dir) != 0:
                order_dicts = json.load(f)
                context.order_list = order_dicts
                # put_log(f'restored history orders : \n{context.order_list}', 'INFO')

###############################  main programm  ###############################

class Params(BaseModel, validate_assignment=True):
    """参数映射模型"""
    len1:int               = Field(default=5,title="均线1周期")
    len2:int               = Field(default=20,title="均线2周期")
    volume:int             = Field(default=1,title="手数")
    order_dir:str          = Field(default=r'E:\order.json', title='订单保存地址') # 这里要改成本机地址
    max_order_num:int      = Field(default=10, title='最大保存订单个数')

# 初始化一些参数
def on_init(context):
    import talib
    context.base_instrument_id = BASE_SETTING.get("Instrument")    
    context.base_period = BASE_SETTING.get("bPeriod")    
    context.positions = {}
    context.close_array = np.zeros(context.len2)
    context.ma1_array = np.zeros(context.len1)
    context.ma2_array = np.zeros(context.len2)
    context.order_list = []
    restore_orders(context)

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

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

# 返回成交单信息 记录仓位变化
def on_trade(context,trade):
    InstrumentID = trade.get("InstrumentID")
    OffsetFlag = trade.get("OffsetFlag")
    if OffsetFlag == "0":
        context.positions[InstrumentID] = trade
    else:
        context.positions.pop(InstrumentID)
    record_order(context, trade)
    
# 返回报错        
def on_error(context,order):
    pass
        
# 策略回测中途发生错误 or 回撤完毕时调用
def on_stop(context):
    pass

3.6 与外部数据联动

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

具体对应数据参考akshare官网：https://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')

'''
股票000001历史行情:
日期    股票代码     开盘     收盘     最高     最低      成交量           成交额    振幅   涨跌幅   涨跌额   换手率
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

获取科创板实时行情：

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
..   ...     ...     ...    ...    ...    ...       ...           ...    ...     ...    ...     ...     ...   ...    ...     ...   ...           ...           ...   ...    ...     ...      ...

获取美股历史行情

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

获取美股实时行情

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
...      ...                              ...   ...    ...     ...    ...    ...    ...     ...          ...    ...          ...          ...     ...      ...         ...

4 常见问题

4.1 策略文件编译失败

可能的原因：

a) 编辑代码后没有保存

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

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

解决方法：

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

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

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

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

可能的原因：

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

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

4.4 回测无数据

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

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

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

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

4.5 策略执行效率问题

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

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

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

4.6 策略报单失败

可能的原因：

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

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

4.7 收盘后发单失败

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

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

4.8 同一根K线内多次开仓

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

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

4.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

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

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

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

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

4.11 引用外部库报错

可能原因：

a) 使用了期魔方尚未支持的库

b) 没有在函数内部导入库文件

正确的使用方法如下：

# 计算均线指标数据
def calc_sma(context):
    
    import talib as ta

    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