🚀 AI 一键生成 joinquant 策略代码
立即体验
👉 AI 生成代码 👈

JoinQuant聚宽平台策略间函数调用与数据共享指南

joinquant | 发布时间: 2026/5/29
以下内容由 EasyQuant 生成。

问题描述

请问下,两个策略之间的函数如何能够相互调用?

如题,我在一个策略中写了一个函数,不想在另一个策略中再次重新写一遍;
有没有啥方法,可以在另一个策略中调用这个策略中的函数?
或者说这个策略中的输出的数据,可以直接被另外一个策略引用;

解决方案

在 JoinQuant(聚宽)平台上,直接在一个策略中“调用”另一个正在运行的策略的函数是不支持的,因为每个策略运行在独立的进程/沙箱中。但是,你可以通过以下两种标准方法完美解决你的需求:函数复用(自定义Python库)数据共享(文件读写)


方法一:函数复用(通过自定义 Python 库)

如果你不想在多个策略中重复编写相同的函数,你可以将这些公共函数提取出来,写成一个独立的 .py 文件,放在聚宽的**“研究”模块根目录**下。然后在你的各个策略中通过 import 导入使用。

步骤 1:在“研究”根目录创建公共函数库

在聚宽的“投资研究”模块根目录下,新建一个 Python 文件,例如命名为 mylib.py

# mylib.py
#-*- coding: utf-8 -*-
# 如果你的文件包含中文, 请在文件的第一行使用上面的语句指定你的文件编码

# 如果需要用到聚宽的API,请加入下面的语句
from kuanke.user_space_api import *

def my_custom_function(security):
    """
    这是一个自定义的公共函数示例
    """
    # 获取股票的上市时间
    info = get_security_info(security)
    return info.start_date

步骤 2:在策略中调用该函数

在你的回测或模拟交易策略代码中,直接 import 这个库即可调用。

# 策略代码
# 导入自己创建的库
from mylib import my_custom_function

def initialize(context):
    g.security = '000001.XSHE'
    
    # 调用自定义库中的函数
    start_date = my_custom_function(g.security)
    log.info(f"{g.security} 的上市时间是: {start_date}")

注意:目前只能 import 研究根目录下的 .py 文件,暂不支持 import 子目录下的文件。


方法二:数据共享(通过文件读写 API)

如果你希望策略 A 运行过程中产生的数据,能够被策略 B 直接引用,你可以使用聚宽提供的 write_fileread_file 函数。策略 A 将数据写入到“研究”目录的文件中,策略 B 从该文件中读取。

策略 A:输出数据并写入文件

使用 write_file 将数据(如 JSON 或 CSV 格式)保存到研究目录。

# 策略 A 代码
import json

def handle_data(context, data):
    # 假设这是策略A计算出的重要信号数据
    signal_data = {
        '000001.XSHE': 'buy',
        '000002.XSHE': 'sell',
        'update_time': str(context.current_dt)
    }
    
    # 将字典转换为 JSON 字符串并写入文件
    # 文件会保存在投资研究的根目录
    write_file('strategy_a_signals.json', json.dumps(signal_data), append=False)
    log.info("信号数据已写入文件")

策略 B:读取并引用数据

使用 read_file 读取策略 A 生成的文件。

# 策略 B 代码
import json

def handle_data(context, data):
    try:
        # 读取策略A生成的文件
        content = read_file('strategy_a_signals.json')
        signal_data = json.loads(content)
        
        log.info(f"读取到策略A的信号更新时间: {signal_data.get('update_time')}")
        
        # 根据策略A的数据执行操作
        if signal_data.get('000001.XSHE') == 'buy':
            order_target_value('000001.XSHE', 10000)
            
    except Exception as e:
        log.error(f"读取数据失败: {e}")

总结与建议

  1. 代码逻辑复用:强烈建议使用自定义 Python 库 (mylib.py) 的方式,这符合编程中的模块化思想,便于维护。
  2. 动态信号传递:如果必须是两个同时运行的模拟盘之间传递动态信号,使用 write_fileread_file 是唯一可行的方法。但请注意文件读写的频率,避免过于频繁的 IO 操作影响策略性能。