如何在 JoinQuant 回测中读取研究环境的 CSV 文件？(read_file)

在聚宽（JoinQuant）的回测或模拟交易环境中，读取外部文件（如股票列表）主要通过 read_file() 函数来实现。

以下是具体的实现步骤和代码示例：

前置步骤：上传文件

在编写代码之前，您必须先将文件上传到聚宽的 “研究” (Research) 环境中。

1. 进入聚宽官网的 “我的策略” -> “研究”。
2. 点击右上角的 “上传” 按钮，将您的股票列表文件（如 .csv, .txt, .json 等）上传到根目录或指定文件夹中。

代码实现

read_file(path) 函数会读取文件并返回二进制内容（bytes），因此通常需要结合 pandas 或 json 库进行解析。

以下是针对常见文件格式（CSV、TXT、JSON）的完整策略代码示例：

# -*- coding: utf-8 -*-
from jqdata import *
import pandas as pd
import json
from io import BytesIO  # 用于将二进制流转换为文件对象

def initialize(context):
    set_benchmark('000300.XSHG')
    set_option('use_real_price', True)
    
    # 定义全局变量存储股票池
    g.my_stocks = []
    
    # --- 场景 1：读取 CSV 文件 ---
    # 假设文件名为 'my_stocks.csv'，位于研究根目录，有一列名为 'code'
    # 文件内容示例：
    # code
    # 000001.XSHE
    # 600519.XSHG
    try:
        # 1. 读取文件原始内容 (bytes)
        content = read_file('my_stocks.csv')
        
        # 2. 使用 pandas 读取二进制流
        df = pd.read_csv(BytesIO(content))
        
        # 3. 提取股票代码列并转为 list
        g.my_stocks = df['code'].tolist()
        log.info("成功从CSV读取 %d 只股票" % len(g.my_stocks))
        
    except Exception as e:
        log.error("读取CSV文件失败: " + str(e))

    # --- 场景 2：读取 TXT 文件 ---
    # 假设文件名为 'my_stocks.txt'，每行一个股票代码
    # try:
    #     content = read_file('my_stocks.txt')
    #     # 解码并按行分割，同时去除空白字符
    #     stock_list = content.decode('utf-8').strip().split('\n')
    #     # 去除每行可能的空格
    #     g.my_stocks = [s.strip() for s in stock_list if s.strip()]
    #     log.info("成功从TXT读取 %d 只股票" % len(g.my_stocks))
    # except Exception as e:
    #     log.error("读取TXT文件失败: " + str(e))

    # --- 场景 3：读取 JSON 文件 ---
    # 假设文件名为 'my_stocks.json'，内容为 ["000001.XSHE", "600519.XSHG"]
    # try:
    #     content = read_file('my_stocks.json')
    #     g.my_stocks = json.loads(content)
    #     log.info("成功从JSON读取 %d 只股票" % len(g.my_stocks))
    # except Exception as e:
    #     log.error("读取JSON文件失败: " + str(e))

    # 过滤掉不存在或已退市的股票，并设置股票池
    if g.my_stocks:
        # 规范化代码（防止代码格式不标准）
        g.my_stocks = normalize_code(g.my_stocks)
        # 过滤掉停牌、退市等不可交易的股票（可选）
        # current_data = get_current_data()
        # g.my_stocks = [stock for stock in g.my_stocks if not current_data[stock].paused]
        
        # 设置股票池（仅用于history等API的默认列表，非必须）
        set_universe(g.my_stocks)
    else:
        log.warn("股票列表为空，请检查文件内容")

def handle_data(context, data):
    # 打印读取到的第一只股票，验证是否成功
    if g.my_stocks:
        log.info("当前股票池第一只: " + g.my_stocks[0])

关键点说明

1. 路径问题：read_file 的参数是相对于“研究”根目录的相对路径。如果文件在子文件夹中，请加上文件夹名，例如 'data/stock_list.csv'。
2. 编码问题：read_file 返回的是 bytes 类型。
   • 如果是文本文件（txt/json），通常需要 .decode('utf-8') 转换为字符串。
   • 如果是 CSV/Excel，建议使用 io.BytesIO 包装后传给 pandas 读取。
3. 代码规范化：外部文件的股票代码格式可能不统一（例如缺少后缀 .XSHE 或 .XSHG），建议读取后调用 normalize_code() 进行格式化。
4. 本地文件：回测是在聚宽的服务器上运行的，无法直接读取您本地电脑（C盘/D盘）的文件，必须先上传到“研究”环境。