FQuant 后端开发指南

1. 项目架构概述

FQuant 采用计算与展示分离的架构，由两个核心部分组成：

• FQuant.Server: 后端计算引擎。基于 Python，使用 Celery 执行定时的、耗时的数据获取、清洗、计算和分析任务。
• FQuantWeb2: 前端展示应用。基于 Nuxt.js，负责数据的可视化和用户交互。

两者之间完全解耦，通过静态 JSON 文件进行通信。

数据流

整个系统的数据流如下：

1. 计算: FQuant.Server 中的 Celery 定时任务被触发，执行各种数据分析和计算。
2. 生成: 任务执行完毕后，将结果生成为静态的 .json 文件。这些文件主要有两种格式：
   • 图表配置: 直接由 pyecharts 库生成的、ECharts 可直接渲染的图表配置对象。
   • 原始数据: 结构化的原始数据，供前端进行表格渲染或其他自定义展示。
3. 发布: 一个独立的 Nginx 服务器将这些生成的 JSON 文件作为静态资源发布。
4. 访问: FQuantWeb2 前端应用通过标准的 HTTP 请求访问 Nginx 服务器上的 JSON 文件，获取数据并渲染页面。

2. 开发工作流

我们采用本地开发，服务器调试的工作模式。

1. 本地编码: 在本地环境中完成新功能的代码编写。
2. 同步代码: 将本地编写好的代码同步到测试服务器。
3. Jupyter Notebook 调试:
   • 这是最核心的调试步骤。在测试服务器上，打开 Jupyter Notebook。
   • 在 Notebook 中，导入你新编写的函数或模块。
   • 直接调用你的函数，传入测试参数，执行它。
   • 检查函数的返回值或其生成的临时 JSON 文件，验证其输出是否完全符合预期。
   • 反复修改代码并在此调试，直到功能正确无误。
4. 集成到 Celery (可选):
   • 只有当一个函数的功能在 Jupyter Notebook 中被完全验证通过后，才能考虑将其集成到自动化流程中。
   • 如果该功能需要定时执行，再将其封装成一个 Celery 任务，添加到 TaskData.py 或 TaskMessage.py 中，并配置好相应的执行时间。

核心理念: 始终将核心业务逻辑与任务调度（Celery）分离。先确保函数本身是可独立运行且正确的，再考虑如何去调用它。

3. 后端代码规范与实践

数据源

• 项目的主要数据来源是通达信 (TongDaXin)，通过 pytdx 库进行数据获取。
• 相关的数据获取逻辑主要封装在 FQData 模块中。在进行常规功能开发时，您无需关心 FQData 的内部实现细节，可以将其视为一个可靠的数据提供方。

数据访问

• 标准方式: 在开发新功能时，获取项目内各种预计算数据（如日线、扩展数据、因子数据等）的标准方式是调用 FQuant.Server/FQMarket/FQMarket/FQUtil/ToolsGetData.py 文件中提供的函数。
• 实现细节: ToolsGetData.py 中的函数封装了对 MongoDB 和 Redis 的查询逻辑，将原始的数据库查询转换为了易于调用的 Python 函数，返回 Pandas DataFrame 格式的数据。

图表生成

• 当新功能需要输出图表时，应使用 pyecharts 库来生成 ECharts 的 JSON 配置。
• 最终输出的应是一个完整的、前端可直接使用的 JSON 文件。

临时文件

• 在开发和调试过程中，您可以将生成的 JSON 文件输出到任意临时目录，以便在 Jupyter Notebook 中进行检查和验证。

已排除的组件

• 位于 FQuant.Server/FQServer/Flask/server.py 的 Flask 应用是一个临时的测试服务，不是项目的主服务，在开发中应忽略它。

核心配置文件

• 项目的大部分硬编码配置（如文件路径、金融参数、黑白名单、消息推送密钥等）都集中在 FQuant.Server/FQMarket/FQMarket/FQUtil/Parameter.py 文件中。在进行新功能开发前，应首先查阅此文件以了解相关的全局变量和配置。

关键参数说明

• GLOBALMAP.TODAY(): 这是项目中处理交易日切换的唯一标准函数。它的逻辑是：每日早上 8:00 会将返回的日期切换到新的交易日。所有需要根据交易日进行计算的功能都必须调用此函数来获取日期，以确保逻辑的统一性。TODAYEND() 和 DAY() 是已废弃或未使用的函数，应避免使用。

• defaultGrade 字典: 此字典中的值（如 '连板': 56）不包含复杂的金融学意义。它们的主要用途是作为前端 ECharts 雷达图视觉展示的**“刻度”或“最大值”**，目的是让图表在视觉上更匀称美观。开发时无需深究其数值的业务逻辑。

• n1name_list 列表: 这是一个用于简化输出的常量列表，主要在生成多个相关的 JSON 文件时，作为文件名或字段名使用，以避免在代码中重复定义字符串。

• 微信推送函数: 文件中多个 send*2Wechat 函数使用了不同的密钥，是为了实现信息分发，将不同主题的消息推送到企业微信中不同的应用或群组，属于业务层面的信息隔离。

4. 策略开发范式

项目中添加新策略遵循一套标准的编写模式，主要涉及 FQuant.Server/FQMarket/FQMarket/StrategyPools/ 目录。

开发一个新策略（例如 mynewstrategy）的步骤如下：

1. 创建策略文件: 在 StrategyPools 目录下创建新文件 sp_mynewstrategy.py。

2. 编写策略逻辑函数: 在文件中实现策略的核心筛选逻辑。

   • 函数定义: 创建一个函数（如 def sp_mynewstrategy(date, lists=None):），它接收日期和可选的股票列表作为输入。
   • 数据获取: 函数内部通过 DATABASE.collection.find() 从 stock_data_factor 或 stock_data_extent 等集合获取数据。
   • 筛选逻辑: 使用 Pandas 对获取的 DataFrame 进行条件过滤。
   • 函数返回: 函数最终返回一个符合策略条件的股票代码列表 (code.tolist())。

3. 编写数据保存函数: 在同一文件中，实现将策略结果每日保存到数据库的调度函数。

   • 函数定义: 创建一个名为 saveStrategyPools_mynewstrategy(end_date=None, renew=0) 的函数。
   • 核心功能: 此函数负责调度上面的策略逻辑函数，并管理数据的增量更新。
   • 实现模式:
     • 定义一个唯一的策略名称字符串，如 code = 'mynewstrategy'。
     • 查询 strategy_pools 数据库，找到该策略最后一次保存的日期。
     • 从该日期的下一天开始，循环调用 sp_mynewstrategy(date) 函数，直到指定的 end_date。
     • 将每日返回的结果（股票列表、策略名、日期、数量）作为一个新文档插入到 strategy_pools 集合中。

通过遵循这种“策略逻辑”与“保存调度”分离的“两函数”模式，可以确保新策略与现有框架的兼容性，并能方便地通过 Jupyter Notebook 调用 saveStrategyPools_* 函数进行回测和数据生成。

5. 核心函数解析

关于项目中一些核心函数的详细解析，请参考独立的文档页面：

• 核心函数解析：get_open_select_stock_list